From 59c81c955ca595e8de741574a74a44ef4e4420c4 Mon Sep 17 00:00:00 2001 From: Vova Kolmakov Date: Fri, 29 May 2026 13:05:53 +0700 Subject: [PATCH 1/2] Kafka Connect: Fix invalid decimal type inferred for some BigDecimal values Co-Authored-By: Claude Opus 4.8 (1M context) --- .../iceberg/connect/data/SchemaUtils.java | 10 ++++++++- .../iceberg/connect/data/TestSchemaUtils.java | 15 +++++++++++++ .../iceberg/connect/data/TestSinkWriter.java | 22 +++++++++++++++++++ 3 files changed, 46 insertions(+), 1 deletion(-) diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SchemaUtils.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SchemaUtils.java index 89d7878172cb..1d022f45d69e 100644 --- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SchemaUtils.java +++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SchemaUtils.java @@ -306,7 +306,15 @@ Type inferIcebergType(Object value) { return BooleanType.get(); } else if (value instanceof BigDecimal) { BigDecimal bigDecimal = (BigDecimal) value; - return DecimalType.of(bigDecimal.precision(), bigDecimal.scale()); + int scale = bigDecimal.scale(); + int precision = bigDecimal.precision(); + // BigDecimal may use a negative scale (e.g. "1E+2" has scale -2) + if (scale < 0) { + precision -= scale; + scale = 0; + } + // a value < 1 may have precision < scale (e.g. "0.001"); widen precision to the scale + return DecimalType.of(Math.max(precision, scale), scale); } else if (value instanceof Integer || value instanceof Long) { return LongType.get(); } else if (value instanceof Float || value instanceof Double) { diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSchemaUtils.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSchemaUtils.java index 9443ed467696..85c0e4f92fcd 100644 --- a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSchemaUtils.java +++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSchemaUtils.java @@ -340,4 +340,19 @@ public void testToIcebergTypeUUIDLogicalTypeOnString() { Schema uuidSchema = SchemaBuilder.string().name("uuid").build(); assertThat(SchemaUtils.toIcebergType(uuidSchema, config)).isInstanceOf(UUIDType.class); } + + @Test + public void testInferIcebergTypeSmallDecimal() { + IcebergSinkConfig config = mock(IcebergSinkConfig.class); + + // BigDecimal("0.001") has precision 1, smaller than its scale 3; + // Iceberg requires scale <= precision, so precision is widened to the scale + assertThat(SchemaUtils.inferIcebergType(new BigDecimal("0.001"), config)) + .isEqualTo(DecimalType.of(3, 3)); + + // BigDecimal("1E+2") has a negative scale (-2); normalized to scale 0, + // the same decimal(3, 0) as new BigDecimal("100") + assertThat(SchemaUtils.inferIcebergType(new BigDecimal("1E+2"), config)) + .isEqualTo(DecimalType.of(3, 0)); + } } diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSinkWriter.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSinkWriter.java index 09f7a373d5f2..d9fdd5bd1b74 100644 --- a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSinkWriter.java +++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSinkWriter.java @@ -26,6 +26,7 @@ import static org.mockito.Mockito.when; import java.io.IOException; +import java.math.BigDecimal; import java.time.Instant; import java.time.ZoneOffset; import java.time.temporal.ChronoUnit; @@ -41,6 +42,7 @@ import org.apache.iceberg.inmemory.InMemoryCatalog; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.types.Type; import org.apache.iceberg.types.Types; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.record.TimestampType; @@ -235,6 +237,26 @@ public void testDynamicNoRoute() { assertThat(writerResults).hasSize(0); } + @Test + public void testEvolveAddsFractionalDecimalColumn() { + IcebergSinkConfig config = mock(IcebergSinkConfig.class); + when(config.tableConfig(any())).thenReturn(mock(TableSinkConfig.class)); + when(config.tables()).thenReturn(ImmutableList.of(TABLE_IDENTIFIER.toString())); + when(config.evolveSchemaEnabled()).thenReturn(true); + + // a new column whose value is a fractional decimal < 1: BigDecimal("0.001") has precision 1 + // and scale 3, so before the fix the column evolves to a malformed decimal(1, 3) and the + // write below fails; after the fix it evolves to decimal(3, 3) and the record is written. + Map value = ImmutableMap.of("amount", new BigDecimal("0.001")); + + List writerResults = sinkWriterTest(value, config); + assertThat(writerResults).isNotEmpty(); + + // the column evolved to a valid decimal that can hold 0.001 (scale <= precision) + Type added = catalog.loadTable(TABLE_IDENTIFIER).schema().findType("amount"); + assertThat(added).isEqualTo(Types.DecimalType.of(3, 3)); + } + private List sinkWriterTest( Map value, IcebergSinkConfig config) { IcebergWriterResult writeResult = From e081d1513f6bcbe2321c6b766d737c22af06d2ab Mon Sep 17 00:00:00 2001 From: Vova Kolmakov Date: Fri, 31 Jul 2026 21:51:22 +0700 Subject: [PATCH 2/2] Kafka Connect: Bound the precision of an inferred decimal type Co-Authored-By: Claude Opus 5 (1M context) Generated-by: Claude Code (claude-opus-5) --- .../iceberg/connect/data/SchemaUtils.java | 38 ++++++++++++++----- .../iceberg/connect/data/TestSchemaUtils.java | 22 +++++++++++ .../iceberg/connect/data/TestSinkWriter.java | 30 ++++++++++++--- 3 files changed, 75 insertions(+), 15 deletions(-) diff --git a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SchemaUtils.java b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SchemaUtils.java index 1d022f45d69e..5bcaed64acaf 100644 --- a/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SchemaUtils.java +++ b/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SchemaUtils.java @@ -72,6 +72,8 @@ class SchemaUtils { private static final Pattern TRANSFORM_REGEX = Pattern.compile("(\\w+)\\((.+)\\)"); + private static final int MAX_DECIMAL_PRECISION = 38; + static PrimitiveType needsDataTypeUpdate(Type currentIcebergType, Schema valueSchema) { if (currentIcebergType.typeId() == TypeID.FLOAT && valueSchema.type() == Schema.Type.FLOAT64) { return DoubleType.get(); @@ -305,16 +307,7 @@ Type inferIcebergType(Object value) { } else if (value instanceof Boolean) { return BooleanType.get(); } else if (value instanceof BigDecimal) { - BigDecimal bigDecimal = (BigDecimal) value; - int scale = bigDecimal.scale(); - int precision = bigDecimal.precision(); - // BigDecimal may use a negative scale (e.g. "1E+2" has scale -2) - if (scale < 0) { - precision -= scale; - scale = 0; - } - // a value < 1 may have precision < scale (e.g. "0.001"); widen precision to the scale - return DecimalType.of(Math.max(precision, scale), scale); + return inferDecimalType((BigDecimal) value); } else if (value instanceof Integer || value instanceof Long) { return LongType.get(); } else if (value instanceof Float || value instanceof Double) { @@ -357,6 +350,31 @@ Type inferIcebergType(Object value) { } } + /** + * BigDecimal does not satisfy Iceberg's 0 <= scale <= precision <= 38 invariant: a + * value < 1 has a precision smaller than its scale ("0.001" is precision 1, scale 3), and an + * exponential value has a negative scale ("1E+2" is scale -2). Both are normalized here, the + * same way Spark normalizes a BigDecimal in Decimal.set. A value that needs more than 38 digits + * cannot be represented, so its type is reported as unknown. + */ + private static Type inferDecimalType(BigDecimal value) { + // widened to long because the subtraction below overflows int for a pathological scale, + // e.g. new BigDecimal(BigInteger.ONE, Integer.MIN_VALUE) + long scale = value.scale(); + long precision = value.precision(); + if (scale < 0) { + precision -= scale; + scale = 0; + } + + precision = Math.max(precision, scale); + if (precision > MAX_DECIMAL_PRECISION) { + return null; + } + + return DecimalType.of((int) precision, (int) scale); + } + private int nextId() { return fieldId++; } diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSchemaUtils.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSchemaUtils.java index 85c0e4f92fcd..602228c481bf 100644 --- a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSchemaUtils.java +++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSchemaUtils.java @@ -30,6 +30,7 @@ import static org.mockito.Mockito.when; import java.math.BigDecimal; +import java.math.BigInteger; import java.time.LocalDate; import java.time.LocalDateTime; import java.time.LocalTime; @@ -355,4 +356,25 @@ public void testInferIcebergTypeSmallDecimal() { assertThat(SchemaUtils.inferIcebergType(new BigDecimal("1E+2"), config)) .isEqualTo(DecimalType.of(3, 0)); } + + @Test + public void testInferIcebergTypeDecimalOutOfRange() { + IcebergSinkConfig config = mock(IcebergSinkConfig.class); + + // the widest values that can still be represented + assertThat(SchemaUtils.inferIcebergType(new BigDecimal("1E+37"), config)) + .isEqualTo(DecimalType.of(38, 0)); + assertThat(SchemaUtils.inferIcebergType(new BigDecimal("1E-38"), config)) + .isEqualTo(DecimalType.of(38, 38)); + + // one digit past the limit on either branch: no type can be inferred + assertThat(SchemaUtils.inferIcebergType(new BigDecimal("1E+38"), config)).isNull(); + assertThat(SchemaUtils.inferIcebergType(new BigDecimal("1E-39"), config)).isNull(); + + // a scale of Integer.MIN_VALUE must not overflow the precision normalization. it is built + // here rather than parsed from "1E+2147483648", which only yields this scale on Java 21+ + // (Java 17 rejects that exponent with a NumberFormatException) + BigDecimal minScale = new BigDecimal(BigInteger.ONE, Integer.MIN_VALUE); + assertThat(SchemaUtils.inferIcebergType(minScale, config)).isNull(); + } } diff --git a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSinkWriter.java b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSinkWriter.java index d9fdd5bd1b74..97eb912e897f 100644 --- a/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSinkWriter.java +++ b/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSinkWriter.java @@ -42,7 +42,6 @@ import org.apache.iceberg.inmemory.InMemoryCatalog; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; -import org.apache.iceberg.types.Type; import org.apache.iceberg.types.Types; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.record.TimestampType; @@ -245,16 +244,37 @@ public void testEvolveAddsFractionalDecimalColumn() { when(config.evolveSchemaEnabled()).thenReturn(true); // a new column whose value is a fractional decimal < 1: BigDecimal("0.001") has precision 1 - // and scale 3, so before the fix the column evolves to a malformed decimal(1, 3) and the - // write below fails; after the fix it evolves to decimal(3, 3) and the record is written. + // and scale 3. DecimalType.of validates only precision <= 38, so before the fix the column + // evolved to a malformed decimal(1, 3) without error and the write below failed in the + // Parquet writer; after the fix it evolves to decimal(3, 3) and the record is written. Map value = ImmutableMap.of("amount", new BigDecimal("0.001")); List writerResults = sinkWriterTest(value, config); assertThat(writerResults).isNotEmpty(); // the column evolved to a valid decimal that can hold 0.001 (scale <= precision) - Type added = catalog.loadTable(TABLE_IDENTIFIER).schema().findType("amount"); - assertThat(added).isEqualTo(Types.DecimalType.of(3, 3)); + Types.NestedField added = catalog.loadTable(TABLE_IDENTIFIER).schema().findField("amount"); + assertThat(added).isNotNull(); + assertThat(added.type()).isEqualTo(Types.DecimalType.of(3, 3)); + } + + @Test + public void testEvolveAddsExponentialDecimalColumn() { + IcebergSinkConfig config = mock(IcebergSinkConfig.class); + when(config.tableConfig(any())).thenReturn(mock(TableSinkConfig.class)); + when(config.tables()).thenReturn(ImmutableList.of(TABLE_IDENTIFIER.toString())); + when(config.evolveSchemaEnabled()).thenReturn(true); + + // BigDecimal("1E+2") has a negative scale (-2), which is normalized to decimal(3, 0), the + // same type inferred for new BigDecimal("100"); the value is rescaled when it is written + Map value = ImmutableMap.of("amount", new BigDecimal("1E+2")); + + List writerResults = sinkWriterTest(value, config); + assertThat(writerResults).isNotEmpty(); + + Types.NestedField added = catalog.loadTable(TABLE_IDENTIFIER).schema().findField("amount"); + assertThat(added).isNotNull(); + assertThat(added.type()).isEqualTo(Types.DecimalType.of(3, 0)); } private List sinkWriterTest(