source-mysql: fix regression involving boolean values in CDC

airbytehq · Jan 13, 2025 · 0511eef · 0511eef
1 parent b23448f
commit 0511eef
Show file tree

Hide file tree

Showing 4 changed files with 29 additions and 40 deletions.
diff --git a/airbyte-integrations/connectors/source-mysql/metadata.yaml b/airbyte-integrations/connectors/source-mysql/metadata.yaml
@@ -9,7 +9,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad
-  dockerImageTag: 3.10.0
+  dockerImageTag: 3.10.1
   dockerRepository: airbyte/source-mysql
   documentationUrl: https://docs.airbyte.com/integrations/sources/mysql
   githubIssueLabel: source-mysql

diff --git a/...ql/src/main/kotlin/io/airbyte/integrations/source/mysql/MySqlSourceCdcBooleanConverter.kt b/...ql/src/main/kotlin/io/airbyte/integrations/source/mysql/MySqlSourceCdcBooleanConverter.kt
diff --git a/...sql/src/main/kotlin/io/airbyte/integrations/source/mysql/MySqlSourceDebeziumOperations.kt b/...sql/src/main/kotlin/io/airbyte/integrations/source/mysql/MySqlSourceDebeziumOperations.kt
@@ -32,6 +32,7 @@ import io.airbyte.cdk.read.cdc.DebeziumState
 import io.airbyte.cdk.read.cdc.DeserializedRecord
 import io.airbyte.cdk.ssh.TunnelSession
 import io.airbyte.cdk.util.Jsons
+import io.debezium.connector.binlog.converters.TinyIntOneToBooleanConverter
 import io.debezium.connector.mysql.MySqlConnector
 import io.debezium.connector.mysql.gtid.MySqlGtidSet
 import io.debezium.document.DocumentReader
@@ -422,10 +423,9 @@ class MySqlSourceDebeziumOperations(
                 .withDatabase("include.list", databaseName)
                 .withOffset()
                 .withSchemaHistory()
-                .withConverters(
-                    MySqlSourceCdcBooleanConverter::class,
-                    MySqlSourceCdcTemporalConverter::class
-                )
+                .with("converters", "tinyint-one,temporal")
+                .with("tinyint-one.type", TinyIntOneToBooleanConverter::class.qualifiedName!!)
+                .with("temporal.type", MySqlSourceCdcTemporalConverter::class.qualifiedName!!)
 
         val serverTimezone: String? =
             (configuration.incrementalConfiguration as CdcIncrementalConfiguration).serverTimezone

diff --git a/...rc/test/kotlin/io/airbyte/integrations/source/mysql/MySqlSourceDatatypeIntegrationTest.kt b/...rc/test/kotlin/io/airbyte/integrations/source/mysql/MySqlSourceDatatypeIntegrationTest.kt
@@ -202,6 +202,20 @@ object MySqlSourceDatatypeTestOperations :
             "NULL" to "null",
         )
 
+    val tinyInt1Values =
+        mapOf(
+            "1" to "true",
+            "0" to "false",
+            "NULL" to "null",
+        )
+
+    val tinyInt1DefaultTrueValues =
+        mapOf(
+            "1" to "true",
+            "0" to "false",
+            "NULL" to "true",
+        )
+
     val enumValues =
         mapOf(
             "'a'" to """"a"""",
@@ -294,6 +308,16 @@ object MySqlSourceDatatypeTestOperations :
                     tinyintValues,
                     LeafAirbyteSchemaType.INTEGER,
                 ),
+                MySqlSourceDatatypeTestCase(
+                    "TINYINT(1)",
+                    tinyInt1Values,
+                    LeafAirbyteSchemaType.BOOLEAN,
+                ),
+                MySqlSourceDatatypeTestCase(
+                    "TINYINT(1) NOT NULL DEFAULT 1",
+                    tinyInt1DefaultTrueValues,
+                    LeafAirbyteSchemaType.BOOLEAN,
+                ),
                 MySqlSourceDatatypeTestCase(
                     "SMALLINT",
                     tinyintValues,