source-mysql: fix regression involving boolean values in CDC

airbytehq · Jan 30, 2025 · 103d967 · 103d967
1 parent 8c2ed63
commit 103d967
Show file tree

Hide file tree

Showing 5 changed files with 245 additions and 255 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.11.1
+  dockerImageTag: 3.11.2
   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
@@ -34,6 +34,7 @@ import io.airbyte.cdk.read.cdc.ResetDebeziumWarmStartState
 import io.airbyte.cdk.read.cdc.ValidDebeziumWarmStartState
 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
@@ -420,10 +421,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!!)
 
         cdcIncrementalConfiguration.serverTimezone
             ?.takeUnless { it.isBlank() }

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,