fix(iceberg): map PK NumberType to StringType instead of DecimalType (#74328)

benmoriceau · devin-ai-integration[bot] · web-flow · commit 419b0a7efd2b · 2026-03-05T15:45:21.000-08:00
Co-authored-by: Devin AI &lt;158243242+devin-ai-integration[bot]@users.noreply.github.com&gt;
diff --git a/airbyte-cdk/bulk/core/load/changelog.md b/airbyte-cdk/bulk/core/load/changelog.md
@@ -9,6 +9,7 @@ The Load CDK provides functionality for destination connectors including stream-
 
 | Version | Date       | Pull Request | Subject                                                                                         |
 |---------|------------|--------------|-------------------------------------------------------------------------------------------------|
+| 1.0.4   | 2026-03-05 | [#74328](https://github.com/airbytehq/airbyte/pull/74328) | Fix iceberg dedup: map PK NumberType to StringType instead of DecimalType for identifier field compatibility. |
 | 1.0.3   | 2026-03-05 | [#74272](https://github.com/airbytehq/airbyte/pull/74272) | Fix iceberg dedup.                                                                              |
 | 1.0.2   | 2026-02-24 | | Bump bulk-cdk-core-base to 1.0.1 to pick up CVE fixes (CVE-2021-47621, CVE-2022-36944).         |
 | 1.0.1   | 2026-02-09 | [#72959](https://github.com/airbytehq/airbyte/pull/72959) | Fix: CVE-2026-25526 (Jinjava dependency bump).                                                  |
diff --git a/airbyte-cdk/bulk/core/load/version.properties b/airbyte-cdk/bulk/core/load/version.properties
@@ -1 +1 @@
-version=1.0.3
+version=1.0.4
diff --git a/airbyte-cdk/bulk/toolkits/load-iceberg-parquet/src/main/kotlin/io/airbyte/cdk/load/data/iceberg/parquet/AirbyteTypeToIcebergSchema.kt b/airbyte-cdk/bulk/toolkits/load-iceberg-parquet/src/main/kotlin/io/airbyte/cdk/load/data/iceberg/parquet/AirbyteTypeToIcebergSchema.kt
@@ -67,7 +67,7 @@ class AirbyteTypeToIcebergSchema {
             is BooleanType -> Types.BooleanType.get()
             is DateType -> Types.DateType.get()
             is IntegerType -> Types.LongType.get()
-            is NumberType -> Types.DecimalType.of(38, 18)
+            is NumberType -> Types.DoubleType.get()
             // Schemaless types are converted to string
             is ArrayTypeWithoutSchema,
             is ObjectTypeWithEmptySchema,
@@ -108,7 +108,13 @@ fun ObjectType.toIcebergSchema(primaryKeys: List<List<String>>): Schema {
         // But we should leave the _airbyte_meta field as an actual object.
         val stringifyObjects = name != Meta.COLUMN_NAME_AB_META
         val icebergType =
-            icebergTypeConverter.convert(field.type, stringifyObjects = stringifyObjects)
+            if (isPrimaryKey && field.type is NumberType) {
+                // Override PK NumberType fields to StringType so they can be used as
+                // Iceberg identifier fields (float/double are disallowed as identifiers).
+                Types.StringType.get()
+            } else {
+                icebergTypeConverter.convert(field.type, stringifyObjects = stringifyObjects)
+            }
         fields.add(
             NestedField.of(
                 id,
diff --git a/airbyte-cdk/bulk/toolkits/load-iceberg-parquet/src/main/kotlin/io/airbyte/cdk/load/data/iceberg/parquet/AirbyteValueToIcebergRecord.kt b/airbyte-cdk/bulk/toolkits/load-iceberg-parquet/src/main/kotlin/io/airbyte/cdk/load/data/iceberg/parquet/AirbyteValueToIcebergRecord.kt
@@ -65,8 +65,8 @@ class AirbyteValueToIcebergRecord {
             is NullValue -> return null
             is NumberValue ->
                 return when (type.typeId()) {
-                    // NumberType is mapped to DecimalType in Iceberg.
-                    Type.TypeID.DECIMAL -> airbyteValue.value
+                    // PK NumberType fields are mapped to StringType in Iceberg.
+                    Type.TypeID.STRING -> airbyteValue.value.toPlainString()
                     else -> airbyteValue.value.toDouble()
                 }
             is StringValue -> return airbyteValue.value
diff --git a/airbyte-cdk/bulk/toolkits/load-iceberg-parquet/src/main/kotlin/io/airbyte/cdk/load/toolkits/iceberg/parquet/IcebergSuperTypeFinder.kt b/airbyte-cdk/bulk/toolkits/load-iceberg-parquet/src/main/kotlin/io/airbyte/cdk/load/toolkits/iceberg/parquet/IcebergSuperTypeFinder.kt
@@ -23,7 +23,7 @@ import org.apache.iceberg.types.Types.*
  */
 @Singleton
 class IcebergSuperTypeFinder(private val icebergTypesComparator: IcebergTypesComparator) {
-    private val unsupportedTypeIds = setOf(BINARY, FIXED, UUID, MAP, TIMESTAMP_NANO)
+    private val unsupportedTypeIds = setOf(BINARY, FIXED, UUID, MAP, DECIMAL, TIMESTAMP_NANO)
 
     /**
      * Returns a supertype for [existingType] and [incomingType] if one exists.
@@ -106,16 +106,6 @@ class IcebergSuperTypeFinder(private val icebergTypesComparator: IcebergTypesCom
         // If both are the same type ID, we just use the existing type
         if (existingTypeId == incomingTypeId) {
             // For timestamps, you'd want to reconcile UTC. This is simplified here.
-            // For decimals, return the wider precision (Iceberg allows widening precision).
-            if (existingTypeId == DECIMAL) {
-                val existingDecimal = existingType as DecimalType
-                val incomingDecimal = incomingType as DecimalType
-                return if (incomingDecimal.precision() > existingDecimal.precision()) {
-                    incomingType
-                } else {
-                    existingType
-                }
-            }
             return existingType
         }
 
diff --git a/airbyte-cdk/bulk/toolkits/load-iceberg-parquet/src/main/kotlin/io/airbyte/cdk/load/toolkits/iceberg/parquet/IcebergTypesComparator.kt b/airbyte-cdk/bulk/toolkits/load-iceberg-parquet/src/main/kotlin/io/airbyte/cdk/load/toolkits/iceberg/parquet/IcebergTypesComparator.kt
@@ -222,13 +222,7 @@ class IcebergTypesComparator {
                 // but for this function's purpose, we only check the existing fields.
                 true
             }
-            Type.TypeID.DECIMAL -> {
-                require(existingType is Types.DecimalType && incomingType is Types.DecimalType) {
-                    "Expected DECIMAL types, got $existingType and $incomingType."
-                }
-                existingType.precision() == incomingType.precision() &&
-                    existingType.scale() == incomingType.scale()
-            }
+            Type.TypeID.DECIMAL,
             Type.TypeID.BINARY,
             Type.TypeID.FIXED,
             Type.TypeID.UUID,
diff --git a/airbyte-cdk/bulk/toolkits/load-iceberg-parquet/src/test/kotlin/io/airbyte/cdk/load/toolkits/iceberg/parquet/IcebergSuperTypeFinderTest.kt b/airbyte-cdk/bulk/toolkits/load-iceberg-parquet/src/test/kotlin/io/airbyte/cdk/load/toolkits/iceberg/parquet/IcebergSuperTypeFinderTest.kt
@@ -126,30 +126,11 @@ class IcebergSuperTypeFinderTest {
     }
 
     @Test
-    fun testIdenticalDecimalTypes() {
-        val decimalType = Types.DecimalType.of(38, 18)
-        val result = superTypeFinder.findSuperType(decimalType, decimalType, "column_name")
-
-        // Identical decimals => return existing
-        assertThat(result).isSameAs(decimalType)
-    }
-
-    @Test
-    fun testDecimalPrecisionWidening() {
-        val narrowDecimal = Types.DecimalType.of(10, 2)
-        val wideDecimal = Types.DecimalType.of(20, 2)
-
-        val result = superTypeFinder.findSuperType(narrowDecimal, wideDecimal, "column_name")
-        // Wider precision should be returned
-        assertThat(result).isSameAs(wideDecimal)
-    }
-
-    @Test
-    fun testDecimalToIntIsNotAllowed() {
+    fun testDecimalIsUnsupported() {
         val decimalType = Types.DecimalType.of(10, 2)
         val intType = Types.IntegerType.get()
 
-        // Decimal to Int promotion is not allowed by Iceberg
+        // Fails in validateTypeIds => DECIMAL is not supported
         assertThatThrownBy { superTypeFinder.findSuperType(decimalType, intType, "column_name") }
             .isInstanceOf(ConfigErrorException::class.java)
             .hasMessageContaining(
diff --git a/airbyte-cdk/bulk/toolkits/load-iceberg-parquet/src/test/kotlin/io/airbyte/cdk/load/toolkits/iceberg/parquet/IcebergTypesComparatorTest.kt b/airbyte-cdk/bulk/toolkits/load-iceberg-parquet/src/test/kotlin/io/airbyte/cdk/load/toolkits/iceberg/parquet/IcebergTypesComparatorTest.kt
@@ -391,7 +391,7 @@ class IcebergTypesComparatorTest {
     }
 
     @Test
-    fun testDecimalTypesEqual() {
+    fun testDecimalTypeIsUnsupported() {
         val existingSchema =
             buildSchema(
                 field("decimal_col", Types.DecimalType.of(10, 2), false),
@@ -401,52 +401,10 @@ class IcebergTypesComparatorTest {
                 field("decimal_col", Types.DecimalType.of(10, 2), false),
             )
 
-        val diff = comparator.compareSchemas(incomingSchema, existingSchema)
-
-        assertThat(diff.newColumns).isEmpty()
-        assertThat(diff.updatedDataTypes).isEmpty()
-        assertThat(diff.removedColumns).isEmpty()
-        assertThat(diff.newlyOptionalColumns).isEmpty()
-    }
-
-    @Test
-    fun testDecimalTypesDifferentPrecision() {
-        val existingSchema =
-            buildSchema(
-                field("decimal_col", Types.DecimalType.of(10, 2), false),
-            )
-        val incomingSchema =
-            buildSchema(
-                field("decimal_col", Types.DecimalType.of(20, 2), false),
-            )
-
-        val diff = comparator.compareSchemas(incomingSchema, existingSchema)
-
-        // Different precision means the type has been updated
-        assertThat(diff.updatedDataTypes).containsExactly("decimal_col")
-        assertThat(diff.newColumns).isEmpty()
-        assertThat(diff.removedColumns).isEmpty()
-        assertThat(diff.newlyOptionalColumns).isEmpty()
-    }
-
-    @Test
-    fun testDecimalTypesDifferentScale() {
-        val existingSchema =
-            buildSchema(
-                field("decimal_col", Types.DecimalType.of(10, 2), false),
-            )
-        val incomingSchema =
-            buildSchema(
-                field("decimal_col", Types.DecimalType.of(10, 5), false),
-            )
-
-        val diff = comparator.compareSchemas(incomingSchema, existingSchema)
-
-        // Different scale means the type has been updated
-        assertThat(diff.updatedDataTypes).containsExactly("decimal_col")
-        assertThat(diff.newColumns).isEmpty()
-        assertThat(diff.removedColumns).isEmpty()
-        assertThat(diff.newlyOptionalColumns).isEmpty()
+        // The code in typesAreEqual() throws for TypeID.DECIMAL
+        assertThatThrownBy { comparator.compareSchemas(incomingSchema, existingSchema) }
+            .isInstanceOf(IllegalArgumentException::class.java)
+            .hasMessageContaining("Unsupported or unmapped Iceberg type: DECIMAL")
     }
 
     @Test
diff --git a/airbyte-cdk/bulk/toolkits/load-iceberg-parquet/src/testFixtures/kotlin/io/airbyte/cdk/load/data/icerberg/parquet/AirbyteTypeToIcebergSchemaTest.kt b/airbyte-cdk/bulk/toolkits/load-iceberg-parquet/src/testFixtures/kotlin/io/airbyte/cdk/load/data/icerberg/parquet/AirbyteTypeToIcebergSchemaTest.kt
@@ -101,7 +101,7 @@ class AirbyteTypeToIcebergSchemaTest {
     @Test
     fun `convert handles NumberType`() {
         assertEquals(
-            Types.DecimalType.of(38, 18),
+            Types.DoubleType.get(),
             converter.convert(NumberType, stringifyObjects = false)
         )
     }
@@ -215,7 +215,7 @@ class AirbyteTypeToIcebergSchemaTest {
     }
 
     @Test
-    fun `toIcebergSchema maps PK NumberType to DecimalType for identifier compatibility`() {
+    fun `toIcebergSchema maps PK NumberType to StringType for identifier compatibility`() {
         val objectType =
             ObjectType(
                 linkedMapOf(
@@ -229,17 +229,17 @@ class AirbyteTypeToIcebergSchemaTest {
         val idColumn = schema.findField("id")
         val amountColumn = schema.findField("amount")
 
-        // PK NumberType field should be DecimalType (for Iceberg identifier compatibility)
+        // PK NumberType field should be StringType (for Iceberg identifier compatibility)
         assertNotNull(idColumn)
         assertFalse(idColumn!!.isOptional)
-        assertEquals(Types.DecimalType.of(38, 18), idColumn.type())
+        assertEquals(Types.StringType.get(), idColumn.type())
 
-        // Non-PK NumberType field should also be DecimalType
+        // Non-PK NumberType field should remain DoubleType
         assertNotNull(amountColumn)
         assertTrue(amountColumn!!.isOptional)
-        assertEquals(Types.DecimalType.of(38, 18), amountColumn.type())
+        assertEquals(Types.DoubleType.get(), amountColumn.type())
 
-        // PK field should be in identifier fields (DecimalType is allowed)
+        // PK field should be in identifier fields (StringType is allowed)
         val identifierFieldIds = schema.identifierFieldIds()
         assertEquals(1, identifierFieldIds.size)
         assertTrue(identifierFieldIds.contains(idColumn.fieldId()))
