chore: mappers also rename source default cursor and source defined pk. (#14002)

Author: gosusnp

airbyte-mappers/src/main/kotlin/io/airbyte/mappers/transformations/DestinationCatalogGenerator.kt

@@ -54,6 +54,8 @@ class DestinationCatalogGenerator(
     stream.stream.jsonSchema = Jsons.deserialize(jsonSchema)
     stream.cursorField = updateFields.cursor
     stream.primaryKey = updateFields.primaryKey
+    stream.stream.sourceDefinedPrimaryKey = updateFields.sourceDefinedPrimaryKey
+    stream.stream.defaultCursorField = updateFields.sourceDefaultCursor
 
     return errors
   }
@@ -79,7 +81,13 @@ class DestinationCatalogGenerator(
       }
         .fold(
           MapperToFieldAccumulator(
-            SlimStream(stream.fields ?: listOf(), stream.cursorField, stream.primaryKey),
+            SlimStream(
+              fields = stream.fields ?: listOf(),
+              cursor = stream.cursorField,
+              primaryKey = stream.primaryKey,
+              sourceDefinedPrimaryKey = stream.stream.sourceDefinedPrimaryKey,
+              sourceDefaultCursor = stream.stream.defaultCursorField,
+            ),
             listOf(),
             mapOf(),
           ),

airbyte-mappers/src/main/kotlin/io/airbyte/mappers/transformations/SlimStream.kt

@@ -6,10 +6,18 @@ import io.airbyte.config.FieldType
 /**
  * Subset of Stream configuration that is relevant to mappers.
  */
-class SlimStream(fields: List<Field>, cursor: List<String>? = null, primaryKey: List<List<String>>? = null) {
+class SlimStream(
+  fields: List<Field>,
+  cursor: List<String>? = null,
+  primaryKey: List<List<String>>? = null,
+  sourceDefaultCursor: List<String>? = null,
+  sourceDefinedPrimaryKey: List<List<String>>? = null,
+) {
   private val _fields: MutableList<Field> = fields.toMutableList()
   private val _cursor: MutableList<String>? = cursor?.toMutableList()
   private val _primaryKey: MutableList<MutableList<String>>? = primaryKey?.map { it.toMutableList() }?.toMutableList()
+  private val _sourceDefaultCursor: MutableList<String>? = sourceDefaultCursor?.toMutableList()
+  private val _sourceDefinedPrimaryKey: MutableList<MutableList<String>>? = sourceDefinedPrimaryKey?.map { it.toMutableList() }?.toMutableList()
 
   val fields: List<Field>
     get() = _fields.toList()
@@ -20,16 +28,26 @@ class SlimStream(fields: List<Field>, cursor: List<String>? = null, primaryKey:
   val primaryKey: List<List<String>>?
     get() = _primaryKey?.map { it.toList() }?.toList()
 
+  val sourceDefaultCursor: List<String>?
+    get() = _sourceDefaultCursor?.toList()
+
+  val sourceDefinedPrimaryKey: List<List<String>>?
+    get() = _sourceDefinedPrimaryKey?.map { it.toList() }?.toList()
+
   fun deepCopy(
     fields: List<Field>? = null,
     cursor: List<String>? = null,
     primaryKey: List<List<String>>? = null,
+    sourceDefaultCursor: List<String>? = null,
+    sourceDefinedPrimaryKey: List<List<String>>? = null,
   ): SlimStream =
     // This relies on the fact that the constructor is copying into a mutable list
     SlimStream(
       fields = fields ?: _fields.map { it.copy() },
       cursor = cursor ?: _cursor,
       primaryKey = primaryKey ?: _primaryKey,
+      sourceDefaultCursor = sourceDefaultCursor ?: _sourceDefaultCursor,
+      sourceDefinedPrimaryKey = sourceDefinedPrimaryKey ?: _sourceDefinedPrimaryKey,
     )
 
   /**
@@ -63,13 +81,30 @@ class SlimStream(fields: List<Field>, cursor: List<String>? = null, primaryKey:
       throw IllegalStateException("Field $oldName not found in stream fields")
     }
 
-    _cursor?.apply {
+    _cursor?.apply { renameInSimpleList(this, oldName, newName) }
+    _primaryKey?.apply { renameInNestedList(this, oldName, newName) }
+    _sourceDefaultCursor?.apply { renameInSimpleList(this, oldName, newName) }
+    _sourceDefinedPrimaryKey?.apply { renameInNestedList(this, oldName, newName) }
+  }
+
+  private fun renameInSimpleList(
+    list: MutableList<String>,
+    oldName: String,
+    newName: String,
+  ) {
+    list.apply {
       if (this == listOf(oldName)) {
         this[0] = newName
       }
     }
+  }
 
-    _primaryKey?.apply {
+  private fun renameInNestedList(
+    list: MutableList<MutableList<String>>,
+    oldName: String,
+    newName: String,
+  ) {
+    list.apply {
       this.replaceAll {
         if (it == listOf(oldName)) mutableListOf(newName) else it
       }

airbyte-mappers/src/test/kotlin/io/airbyte/mappers/transformations/DestinationCatalogGeneratorTest.kt

@@ -280,7 +280,7 @@ class DestinationCatalogGeneratorTest {
                 "additionalProperties": true
               },
               "default_cursor_field": [
-                "modified_at"
+                "modified_at_test"
               ],
               "supported_sync_modes": [
                 "full_refresh",
@@ -289,7 +289,7 @@ class DestinationCatalogGeneratorTest {
               "source_defined_cursor": true,
               "source_defined_primary_key": [
                 [
-                  "id"
+                  "id_test"
                 ]
               ]
             },

airbyte-mappers/src/test/kotlin/io/airbyte/mappers/transformations/SlimStreamTest.kt

@@ -23,9 +23,13 @@ class SlimStreamTest {
             Field(CURSOR_NAME, FieldType.INTEGER),
             Field(PRIMARY_KEY_NAME, FieldType.STRING),
             Field(PRIMARY_KEY_OTHER_ATTR_NAME, FieldType.STRING),
+            Field(SOURCE_CURSOR_NAME, FieldType.STRING),
+            Field(SOURCE_PRIMARY_KEY_NAME, FieldType.STRING),
           ),
         cursor = listOf(CURSOR_NAME),
         primaryKey = listOf(listOf(PRIMARY_KEY_NAME), listOf(PRIMARY_KEY_OTHER_ATTR_NAME)),
+        sourceDefaultCursor = listOf(SOURCE_CURSOR_NAME),
+        sourceDefinedPrimaryKey = listOf(listOf(SOURCE_PRIMARY_KEY_NAME), listOf(PRIMARY_KEY_OTHER_ATTR_NAME)),
       )
   }
 
@@ -83,11 +87,33 @@ class SlimStreamTest {
     assertEquals(listOf(listOf(renamedPrimaryKey), listOf(PRIMARY_KEY_OTHER_ATTR_NAME)), slimStream.primaryKey)
   }
 
+  @Test
+  fun `renaming a field also updates source default cursor if relevant`() {
+    val renamedSourceFieldKey = "newSourceField"
+    slimStream.redefineField(SOURCE_CURSOR_NAME, renamedSourceFieldKey)
+
+    assertTrue(slimStream.fields.contains(Field(renamedSourceFieldKey, FieldType.STRING)))
+    assertFalse(slimStream.fields.any { it.name == SOURCE_CURSOR_NAME })
+    assertEquals(listOf(renamedSourceFieldKey), slimStream.sourceDefaultCursor)
+  }
+
+  @Test
+  fun `renaming a field also updates source defined PK if relevant`() {
+    val renamedSourceFieldKey = "newSourceField"
+    slimStream.redefineField(SOURCE_PRIMARY_KEY_NAME, renamedSourceFieldKey)
+
+    assertTrue(slimStream.fields.contains(Field(renamedSourceFieldKey, FieldType.STRING)))
+    assertFalse(slimStream.fields.any { it.name == SOURCE_PRIMARY_KEY_NAME })
+    assertEquals(listOf(listOf(renamedSourceFieldKey), listOf(PRIMARY_KEY_OTHER_ATTR_NAME)), slimStream.sourceDefinedPrimaryKey)
+  }
+
   companion object {
     const val FIELD1_NAME = "field1"
     const val FIELD2_NAME = "field2"
     const val CURSOR_NAME = "cursor_field"
     const val PRIMARY_KEY_NAME = "primary_key"
     const val PRIMARY_KEY_OTHER_ATTR_NAME = "primary_key_other"
+    const val SOURCE_CURSOR_NAME = "source_cursor_field"
+    const val SOURCE_PRIMARY_KEY_NAME = "source_primary_key"
   }
 }