cdk changes for mysql dv2

Author: edgao

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/AbstractJdbcDestination.kt

@@ -196,6 +196,11 @@ abstract class AbstractJdbcDestination<DestinationState : MinimumDestinationStat
         rawTableSchema: String
     ): JdbcDestinationHandler<DestinationState>
 
+    protected open fun getV1V2Migrator(
+        database: JdbcDatabase,
+        databaseName: String
+    ): DestinationV1V2Migrator = JdbcV1V2Migrator(namingResolver, database, databaseName)
+
     /**
      * Provide any migrations that the destination needs to run. Most destinations will need to
      * provide an instande of
@@ -306,6 +311,7 @@ abstract class AbstractJdbcDestination<DestinationState : MinimumDestinationStat
             typerDeduper,
             getDataTransformer(parsedCatalog, defaultNamespace),
             optimalBatchSizeBytes,
+            parsedCatalog,
         )
     }
 
@@ -317,7 +323,7 @@ abstract class AbstractJdbcDestination<DestinationState : MinimumDestinationStat
         val sqlGenerator = getSqlGenerator(config)
         val databaseName = getDatabaseName(config)
         val v2TableMigrator = NoopV2TableMigrator()
-        val migrator = JdbcV1V2Migrator(namingResolver, database, databaseName)
+        val migrator = getV1V2Migrator(database, databaseName)
         val destinationHandler: DestinationHandler<DestinationState> =
             getDestinationHandler(
                 databaseName,
@@ -434,7 +440,7 @@ abstract class AbstractJdbcDestination<DestinationState : MinimumDestinationStat
                     if (attemptInsert) {
                         sqlOps.insertRecords(
                             database,
-                            java.util.List.of(dummyRecord),
+                            listOf(dummyRecord),
                             outputSchema,
                             outputTableName,
                         )

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/JdbcBufferedConsumerFactory.kt

@@ -24,6 +24,8 @@ import io.airbyte.cdk.integrations.destination.buffered_stream_consumer.OnCloseF
 import io.airbyte.cdk.integrations.destination.buffered_stream_consumer.OnStartFunction
 import io.airbyte.cdk.integrations.destination.buffered_stream_consumer.RecordWriter
 import io.airbyte.commons.json.Jsons
+import io.airbyte.integrations.base.destination.typing_deduping.ParsedCatalog
+import io.airbyte.integrations.base.destination.typing_deduping.StreamConfig
 import io.airbyte.integrations.base.destination.typing_deduping.StreamId.Companion.concatenateRawTableName
 import io.airbyte.integrations.base.destination.typing_deduping.TyperDeduper
 import io.airbyte.protocol.models.v0.*
@@ -54,6 +56,9 @@ object JdbcBufferedConsumerFactory {
 
     const val DEFAULT_OPTIMAL_BATCH_SIZE_FOR_FLUSH = 25 * 1024 * 1024L
 
+    /**
+     * @param parsedCatalog Nullable for v1 destinations. Required for v2 destinations.
+     */
     fun createAsync(
         outputRecordCollector: Consumer<AirbyteMessage>,
         database: JdbcDatabase,
@@ -65,9 +70,10 @@ object JdbcBufferedConsumerFactory {
         typerDeduper: TyperDeduper,
         dataTransformer: StreamAwareDataTransformer = IdentityDataTransformer(),
         optimalBatchSizeBytes: Long = DEFAULT_OPTIMAL_BATCH_SIZE_FOR_FLUSH,
+        parsedCatalog: ParsedCatalog? = null
     ): SerializedAirbyteMessageConsumer {
         val writeConfigs =
-            createWriteConfigs(namingResolver, config, catalog, sqlOperations.isSchemaRequired)
+            createWriteConfigs(namingResolver, config, catalog, sqlOperations.isSchemaRequired, parsedCatalog)
         return AsyncStreamConsumer(
             outputRecordCollector,
             onStartFunction(database, sqlOperations, writeConfigs, typerDeduper),
@@ -89,19 +95,28 @@ object JdbcBufferedConsumerFactory {
         namingResolver: NamingConventionTransformer,
         config: JsonNode,
         catalog: ConfiguredAirbyteCatalog?,
-        schemaRequired: Boolean
+        schemaRequired: Boolean,
+        parsedCatalog: ParsedCatalog?
     ): List<WriteConfig> {
         if (schemaRequired) {
             Preconditions.checkState(
                 config.has("schema"),
                 "jdbc destinations must specify a schema."
             )
         }
-        return catalog!!
-            .streams
-            .stream()
-            .map(toWriteConfig(namingResolver, config, schemaRequired))
-            .collect(Collectors.toList())
+        return if (parsedCatalog == null) {
+            catalog!!
+                .streams
+                .stream()
+                .map(toWriteConfig(namingResolver, config, schemaRequired))
+                .collect(Collectors.toList())
+        } else {
+            // we should switch this to kotlin-style list processing, but meh for now
+            parsedCatalog.streams
+                .stream()
+                .map(parsedStreamToWriteConfig(namingResolver))
+                .collect(Collectors.toList())
+        }
     }
 
     private fun toWriteConfig(
@@ -150,6 +165,26 @@ object JdbcBufferedConsumerFactory {
         }
     }
 
+    private fun parsedStreamToWriteConfig(
+        namingResolver: NamingConventionTransformer
+    ): Function<StreamConfig, WriteConfig> {
+        return Function { streamConfig: StreamConfig ->
+            // TODO We should probably replace WriteConfig with StreamConfig?
+            // The only thing I'm not sure about is the tmpTableName thing,
+            // but otherwise it's a strict improvement (avoids people accidentally
+            // recomputing the table names, instead of just treating the output of
+            // CatalogParser as canonical).
+            WriteConfig(
+                streamConfig.id.originalName,
+                streamConfig.id.originalNamespace,
+                streamConfig.id.rawNamespace,
+                namingResolver.getTmpTableName(streamConfig.id.rawNamespace),
+                streamConfig.id.rawName,
+                streamConfig.destinationSyncMode,
+            )
+        }
+    }
+
     /**
      * Defer to the [AirbyteStream]'s namespace. If this is not set, use the destination's default
      * schema. This namespace is source-provided, and can be potentially empty.
@@ -160,7 +195,7 @@ object JdbcBufferedConsumerFactory {
     private fun getOutputSchema(
         stream: AirbyteStream,
         defaultDestSchema: String,
-        namingResolver: NamingConventionTransformer
+        namingResolver: NamingConventionTransformer,
     ): String {
         return if (isDestinationV2) {
             namingResolver.getNamespace(
@@ -252,8 +287,10 @@ object JdbcBufferedConsumerFactory {
             records: List<PartialAirbyteMessage> ->
             require(pairToWriteConfig.containsKey(pair)) {
                 String.format(
-                    "Message contained record from a stream that was not in the catalog. \ncatalog: %s",
-                    Jsons.serialize(catalog)
+                    "Message contained record from a stream that was not in the catalog. \ncatalog: %s, \nstream identifier: %s\nkeys: %s",
+                    Jsons.serialize(catalog),
+                    pair,
+                    pairToWriteConfig.keys
                 )
             }
             val writeConfig = pairToWriteConfig.getValue(pair)

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/JdbcSqlOperations.kt

@@ -142,7 +142,11 @@ abstract class JdbcSqlOperations : SqlOperations {
                     val uuid = UUID.randomUUID().toString()
 
                     val jsonData = record.serialized
-                    val airbyteMeta = Jsons.serialize(record.record!!.meta)
+                    val airbyteMeta = if (record.record!!.meta == null) {
+                        "{\"changes\":[]}"
+                    } else {
+                        Jsons.serialize(record.record!!.meta)
+                    }
                     val extractedAt =
                         Timestamp.from(Instant.ofEpochMilli(record.record!!.emittedAt))
                     if (isDestinationV2) {

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/typing_deduping/JdbcDestinationHandler.kt

@@ -41,7 +41,11 @@ import org.slf4j.LoggerFactory
 
 @Slf4j
 abstract class JdbcDestinationHandler<DestinationState>(
-    protected val databaseName: String,
+    // This feels hacky. MySQL doesn't have a notion of "schema".
+    // In this case, the JDBC interface translates the schema parameter to mysql database.
+    // This allows us to pass e.g. stream namespace into metadata.get(schema=?) for mysql.
+    // So it's misleading, but mysql will use a null databaseName here to avoid problemss.
+    protected val databaseName: String?,
     protected val jdbcDatabase: JdbcDatabase,
     protected val rawTableSchemaName: String,
     private val dialect: SQLDialect
@@ -50,10 +54,13 @@ abstract class JdbcDestinationHandler<DestinationState>(
         get() = DSL.using(dialect)
 
     @Throws(Exception::class)
-    private fun findExistingTable(id: StreamId): Optional<TableDefinition> {
+    protected open fun findExistingTable(id: StreamId): Optional<TableDefinition> {
         return findExistingTable(jdbcDatabase, databaseName, id.finalNamespace, id.finalName)
     }
 
+    protected open fun getTableFromMetadata(dbmetadata: DatabaseMetaData, id: StreamId): ResultSet =
+        dbmetadata.getTables(databaseName, id.rawNamespace, id.rawName, null)
+
     @Throws(Exception::class)
     private fun isFinalTableEmpty(id: StreamId): Boolean {
         return !jdbcDatabase.queryBoolean(
@@ -80,9 +87,8 @@ abstract class JdbcDestinationHandler<DestinationState>(
                     id.rawName
                 )
                 try {
-                    dbmetadata!!.getTables(databaseName, id.rawNamespace, id.rawName, null).use {
-                        table ->
-                        return@executeMetadataQuery table.next()
+                    getTableFromMetadata(dbmetadata!!, id).use {
+                        table -> return@executeMetadataQuery table.next()
                     }
                 } catch (e: SQLException) {
                     LOGGER.error("Failed to retrieve table info from metadata", e)
@@ -340,22 +346,25 @@ abstract class JdbcDestinationHandler<DestinationState>(
         }
     }
 
-    private fun isAirbyteRawIdColumnMatch(existingTable: TableDefinition): Boolean {
-        return existingTable.columns.containsKey(JavaBaseConstants.COLUMN_NAME_AB_RAW_ID) &&
-            toJdbcTypeName(AirbyteProtocolType.STRING) ==
-                existingTable.columns[JavaBaseConstants.COLUMN_NAME_AB_RAW_ID]!!.type
+    protected open fun isAirbyteRawIdColumnMatch(existingTable: TableDefinition): Boolean {
+        return toJdbcTypeName(AirbyteProtocolType.STRING).equals(
+            existingTable.columns[JavaBaseConstants.COLUMN_NAME_AB_RAW_ID]!!.type,
+            ignoreCase = true,
+        )
     }
 
-    private fun isAirbyteExtractedAtColumnMatch(existingTable: TableDefinition): Boolean {
-        return existingTable.columns.containsKey(JavaBaseConstants.COLUMN_NAME_AB_EXTRACTED_AT) &&
-            toJdbcTypeName(AirbyteProtocolType.TIMESTAMP_WITH_TIMEZONE) ==
-                existingTable.columns[JavaBaseConstants.COLUMN_NAME_AB_EXTRACTED_AT]!!.type
+    protected open fun isAirbyteExtractedAtColumnMatch(existingTable: TableDefinition): Boolean {
+        return toJdbcTypeName(AirbyteProtocolType.TIMESTAMP_WITH_TIMEZONE).equals(
+            existingTable.columns[JavaBaseConstants.COLUMN_NAME_AB_EXTRACTED_AT]!!.type,
+            ignoreCase = true,
+        )
     }
 
-    private fun isAirbyteMetaColumnMatch(existingTable: TableDefinition): Boolean {
-        return existingTable.columns.containsKey(JavaBaseConstants.COLUMN_NAME_AB_META) &&
-            toJdbcTypeName(Struct(LinkedHashMap<String, AirbyteType>())) ==
-                existingTable.columns[JavaBaseConstants.COLUMN_NAME_AB_META]!!.type
+    protected open fun isAirbyteMetaColumnMatch(existingTable: TableDefinition): Boolean {
+        return toJdbcTypeName(Struct(LinkedHashMap<String, AirbyteType>())).equals(
+            existingTable.columns[JavaBaseConstants.COLUMN_NAME_AB_META]!!.type,
+            ignoreCase = true,
+        )
     }
 
     private fun existingSchemaMatchesStreamConfig(
@@ -364,9 +373,9 @@ abstract class JdbcDestinationHandler<DestinationState>(
     ): Boolean {
         // Check that the columns match, with special handling for the metadata columns.
         if (
-            !isAirbyteRawIdColumnMatch(existingTable) ||
-                !isAirbyteExtractedAtColumnMatch(existingTable) ||
-                !isAirbyteMetaColumnMatch(existingTable)
+            !(existingTable.columns.containsKey(JavaBaseConstants.COLUMN_NAME_AB_RAW_ID) && isAirbyteRawIdColumnMatch(existingTable)) ||
+            !(existingTable.columns.containsKey(JavaBaseConstants.COLUMN_NAME_AB_EXTRACTED_AT) && isAirbyteExtractedAtColumnMatch(existingTable)) ||
+            !(existingTable.columns.containsKey(JavaBaseConstants.COLUMN_NAME_AB_META) && isAirbyteMetaColumnMatch(existingTable))
         ) {
             // Missing AB meta columns from final table, we need them to do proper T+D so trigger
             // soft-reset
@@ -392,7 +401,7 @@ abstract class JdbcDestinationHandler<DestinationState>(
                     {
                         map: LinkedHashMap<String?, String>,
                         column: Map.Entry<String?, ColumnDefinition> ->
-                        map[column.key] = column.value.type
+                        map[column.key] = column.value.type.lowercase()
                     },
                     { obj: LinkedHashMap<String?, String>, m: LinkedHashMap<String?, String>? ->
                         obj.putAll(m!!)
@@ -475,9 +484,9 @@ abstract class JdbcDestinationHandler<DestinationState>(
      * @param airbyteType
      * @return
      */
-    protected abstract fun toJdbcTypeName(airbyteType: AirbyteType?): String
+    protected abstract fun toJdbcTypeName(airbyteType: AirbyteType): String
 
-    protected abstract fun toDestinationState(json: JsonNode?): DestinationState
+    protected abstract fun toDestinationState(json: JsonNode): DestinationState
 
     companion object {
         private val LOGGER: Logger = LoggerFactory.getLogger(JdbcDestinationHandler::class.java)