Set default Decimal handling mode to double (#519)

* Set default Decimal handling mode to `double` * Set default Decimal handling mode to `double` * Set default Decimal handling mode to `double` * Set default Decimal handling mode to `double`
memiiso · Feb 23, 2025 · bb6a528 · bb6a528
1 parent ae76cfb
commit bb6a528
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 7 deletions.
diff --git a/debezium-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/DebeziumConfig.java b/debezium-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/DebeziumConfig.java
@@ -1,23 +1,24 @@
 package io.debezium.server.iceberg;
 
 import io.debezium.jdbc.TemporalPrecisionMode;
+import io.debezium.relational.RelationalDatabaseConnectorConfig;
 import io.quarkus.runtime.annotations.ConfigRoot;
 import io.smallrye.config.ConfigMapping;
 import io.smallrye.config.WithDefault;
 import io.smallrye.config.WithName;
-import io.smallrye.config.WithParentName;
 
 @ConfigRoot
 @ConfigMapping
 public interface DebeziumConfig {
 
-  @WithParentName
-  IcebergConfig sinkConfig();
-
   @WithName("debezium.source.time.precision.mode")
   @WithDefault("isostring")
   TemporalPrecisionMode temporalPrecisionMode();
 
+  @WithName("debezium.source.decimal.handling.mode")
+  @WithDefault("double")
+  RelationalDatabaseConnectorConfig.DecimalHandlingMode decimalHandlingMode();
+
   // Event format
   @WithName("debezium.format.value")
   @WithDefault("json")

diff --git a/.../src/test/java/io/debezium/server/iceberg/IcebergChangeConsumerTemporalIsoStringTest.java b/.../src/test/java/io/debezium/server/iceberg/IcebergChangeConsumerTemporalIsoStringTest.java
@@ -56,7 +56,7 @@ public void testConsumingVariousDataTypes() throws Exception {
         "VALUES \n" +
         "(1, null, null, null, null) \n" +
         ",(2, CURRENT_DATE , CURRENT_TIME, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP ) \n" +
-        ",(3, '2024-01-02'::DATE , CURRENT_TIME, '2023-10-11 10:30:00'::timestamp, '2023-11-12 10:30:00+02'::timestamptz ) ";
+        ",(3, '2024-01-02'::DATE , '18:04:00'::TIME, '2023-10-11 10:30:00'::timestamp, '2023-11-12 10:30:00+02'::timestamptz ) ";
 
     SourcePostgresqlDB.runSQL(sql);
     Awaitility.await().atMost(Duration.ofSeconds(320)).until(() -> {
@@ -76,6 +76,9 @@ public void testConsumingVariousDataTypes() throws Exception {
         Assertions.assertEquals(1, df.filter("c_id = 3 AND c_timestamp = to_timestamp('2023-10-11 10:30:00')").count());
         Assertions.assertEquals(DataTypes.TimestampType, getSchemaField(df, "c_timestamptz").dataType());
         Assertions.assertEquals(1, df.filter("c_id = 3 AND c_timestamptz = to_timestamp('2023-11-12 10:30:00+02')").count());
+        // time type is kept as string, because spark does not support time type
+        Assertions.assertEquals(DataTypes.StringType, getSchemaField(df, "c_time").dataType());
+        Assertions.assertEquals(1, df.filter("c_id = 3 AND c_time = '18:04:00Z'").count());
         return true;
       } catch (Exception e) {
 //        e.printStackTrace();

diff --git a/...rver-iceberg-sink/src/test/java/io/debezium/server/iceberg/IcebergChangeConsumerTest.java b/...rver-iceberg-sink/src/test/java/io/debezium/server/iceberg/IcebergChangeConsumerTest.java
@@ -9,6 +9,7 @@
 package io.debezium.server.iceberg;
 
 import com.google.common.collect.Lists;
+import io.debezium.relational.RelationalDatabaseConnectorConfig;
 import io.debezium.server.iceberg.testresources.BaseSparkTest;
 import io.debezium.server.iceberg.testresources.CatalogJdbc;
 import io.debezium.server.iceberg.testresources.S3Minio;
@@ -79,8 +80,8 @@ public void testConsumingVariousDataTypes() throws Exception {
         "            c_timestamp TIMESTAMP,\n" +
         "            c_timestamptz TIMESTAMPTZ,\n" +
         "            c_float FLOAT,\n" +
-        "            c_decimal DECIMAL(18,4),\n" +
-        "            c_numeric NUMERIC(18,4),\n" +
+        "            c_decimal DECIMAL(18,6),\n" +
+        "            c_numeric NUMERIC(18,6),\n" +
         "            c_interval INTERVAL,\n" +
         "            c_boolean BOOLEAN,\n" +
         "            c_uuid UUID,\n" +
@@ -128,8 +129,13 @@ public void testConsumingVariousDataTypes() throws Exception {
 
         Assertions.assertEquals(DataTypes.DateType, getSchemaField(df, "c_date").dataType());
         Assertions.assertEquals(1, df.filter("c_id = 2 AND c_date = to_date('2024-05-05', 'yyyy-MM-dd')").count());
+        Assertions.assertEquals(consumer.config.debezium().decimalHandlingMode(), RelationalDatabaseConnectorConfig.DecimalHandlingMode.DOUBLE);
+        Assertions.assertEquals(1, df.filter("c_id = 2 AND c_float = CAST('1.23' AS DOUBLE)").count(), "c_float not matching");
+        Assertions.assertEquals(1, df.filter("c_id = 2 AND c_decimal = CAST('1234566.34456' AS DOUBLE)").count(), "c_decimal not matching");
+        Assertions.assertEquals(1, df.filter("c_id = 2 AND c_numeric = CAST('345672123.452' AS DOUBLE)").count(), "c_numeric not matching");
         return true;
       } catch (Exception e) {
+        e.printStackTrace();
         return false;
       }
     });