Add support for Iceberg decimals (#537)

Write decimals as actual Iceberg decimals (and not byte arrays) when
Kafka source connector is configured to use `precise` decimal handling
mode.

Author: zilder

debezium-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/RecordConverter.java

@@ -28,6 +28,8 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.math.BigDecimal;
+import java.math.BigInteger;
 import java.nio.ByteBuffer;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
@@ -205,6 +207,16 @@ private Object jsonValToIcebergVal(Types.NestedField field, JsonNode node) {
         return node.floatValue();
       case DOUBLE: // double is represented in 64 bits
         return node.asDouble();
+      case DECIMAL: {
+          BigDecimal decimalVal = null;
+          try {
+              int scale = ((Types.DecimalType) field.type()).scale();
+              decimalVal = new BigDecimal(new BigInteger(node.binaryValue()), scale);
+          } catch (IOException e) {
+              throw new RuntimeException("Failed to convert decimal value to iceberg value, field: " + field.name(), e);
+          }
+          return decimalVal;
+      }
       case BOOLEAN:
         return node.asBoolean();
       case UUID:

debezium-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/SchemaConverter.java

@@ -96,7 +96,7 @@ private IcebergSchemaInfo debeziumFieldToIcebergField(JsonNode fieldSchema, Stri
         return schemaData;
       default:
         // its primitive field
-        final Types.NestedField field = Types.NestedField.of(schemaData.nextFieldId().getAndIncrement(), !isPkField, fieldName, icebergPrimitiveField(fieldName, fieldType, fieldTypeName));
+        final Types.NestedField field = Types.NestedField.of(schemaData.nextFieldId().getAndIncrement(), !isPkField, fieldName, icebergPrimitiveField(fieldName, fieldType, fieldTypeName, fieldSchema));
         schemaData.fields().add(field);
         if (isPkField) schemaData.identifierFieldIds().add(field.fieldId());
         return schemaData;
@@ -182,7 +182,7 @@ public Schema icebergSchema() {
 
   }
 
-  private Type.PrimitiveType icebergPrimitiveField(String fieldName, String fieldType, String fieldTypeName) {
+  private Type.PrimitiveType icebergPrimitiveField(String fieldName, String fieldType, String fieldTypeName, JsonNode fieldSchema) {
     // Debezium Temporal types: https://debezium.io/documentation//reference/connectors/postgresql.html#postgresql-temporal-types
     switch (fieldType) {
       case "int8":
@@ -247,6 +247,16 @@ private Type.PrimitiveType icebergPrimitiveField(String fieldName, String fieldT
       case "uuid":
         return Types.UUIDType.get();
       case "bytes":
+        // With `decimal.handling.mode` set to `precise` debezium relational source connector would encode decimals
+        // as byte arrays. We want to write them out as Iceberg decimals.
+        if (fieldTypeName.equals("org.apache.kafka.connect.data.Decimal")) {
+          JsonNode params = fieldSchema.get("parameters");
+          if (!params.isNull() && !params.isEmpty()) {
+            int precision = params.get("connect.decimal.precision").asInt();
+            int scale = params.get("scale").asInt();
+            return Types.DecimalType.of(precision, scale);
+          }
+        }
         return Types.BinaryType.get();
       default:
         // default to String type

debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/IcebergChangeConsumerDecimalTest.java

@@ -0,0 +1,90 @@
+/*
+ *
+ *  * Copyright memiiso Authors.
+ *  *
+ *  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ */
+
+package io.debezium.server.iceberg;
+
+import com.google.common.collect.Lists;
+import io.debezium.relational.RelationalDatabaseConnectorConfig;
+import io.debezium.server.iceberg.testresources.CatalogJdbc;
+import io.debezium.server.iceberg.testresources.S3Minio;
+import io.debezium.server.iceberg.testresources.SourcePostgresqlDB;
+import io.quarkus.test.common.QuarkusTestResource;
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.QuarkusTestProfile;
+import io.quarkus.test.junit.TestProfile;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.types.DataTypes;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import static io.debezium.server.iceberg.TestConfigSource.ICEBERG_CATALOG_TABLE_NAMESPACE;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.when;
+
+/**
+ * Integration test that verifies basic reading from PostgreSQL database and writing to iceberg destination.
+ *
+ * @author Ismail Simsek
+ */
+@QuarkusTest
+@QuarkusTestResource(value = S3Minio.class, restrictToAnnotatedClass = true)
+@QuarkusTestResource(value = SourcePostgresqlDB.class, restrictToAnnotatedClass = true)
+@QuarkusTestResource(value = CatalogJdbc.class, restrictToAnnotatedClass = true)
+@TestProfile(IcebergChangeConsumerDecimalTest.TestProfile.class)
+public class IcebergChangeConsumerDecimalTest extends BaseSparkTest {
+
+  @Test
+  public void testConsumingNumerics() throws Exception {
+    assertEquals(sinkType, "iceberg");
+    String sql = "\n" +
+        "        DROP TABLE IF EXISTS inventory.data_types;\n" +
+        "        CREATE TABLE IF NOT EXISTS inventory.data_types (\n" +
+        "            c_id INTEGER ,\n" +
+        "            c_decimal DECIMAL(18,6)\n" +
+        "          );";
+    SourcePostgresqlDB.runSQL(sql);
+    sql = "INSERT INTO inventory.data_types (c_id, c_decimal) " +
+        "VALUES (1, '1234566.34456'::decimal)";
+    SourcePostgresqlDB.runSQL(sql);
+    Awaitility.await().atMost(Duration.ofSeconds(320)).until(() -> {
+      try {
+        Dataset<Row> df = getTableData("testc.inventory.data_types");
+        df.show(false);
+
+        Assertions.assertEquals(1, df.count());
+        Assertions.assertEquals(1, df.filter("c_id = 1 AND c_decimal = CAST('1234566.344560' AS DECIMAL(18,6))").count(), "c_decimal not matching");
+        return true;
+      } catch (Exception e) {
+        e.printStackTrace();
+        return false;
+      }
+    });
+  }
+
+  public static class TestProfile implements QuarkusTestProfile {
+    @Override
+    public Map<String, String> getConfigOverrides() {
+      Map<String, String> config = new HashMap<>();
+      config.put("debezium.sink.iceberg.destination-regexp", "\\d");
+      config.put("debezium.source.decimal.handling.mode", "precise");
+      return config;
+    }
+  }
+
+}