Kafka Connect: Fix invalid decimal type inferred for some BigDecimal values

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,8 +307,7 @@ Type inferIcebergType(Object value) {
       } else if (value instanceof Boolean) {
         return BooleanType.get();
       } else if (value instanceof BigDecimal) {
-        BigDecimal bigDecimal = (BigDecimal) value;
-        return DecimalType.of(bigDecimal.precision(), bigDecimal.scale());
+        return inferDecimalType((BigDecimal) value);
       } else if (value instanceof Integer || value instanceof Long) {
         return LongType.get();
       } else if (value instanceof Float || value instanceof Double) {
@@ -349,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++;
     }

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;
@@ -340,4 +341,40 @@ 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));
+  }
+
+  @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();
+  }
 }

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;
@@ -235,6 +236,47 @@ 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. 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<String, Object> value = ImmutableMap.of("amount", new BigDecimal("0.001"));
+
+    List<IcebergWriterResult> writerResults = sinkWriterTest(value, config);
+    assertThat(writerResults).isNotEmpty();
+
+    // the column evolved to a valid decimal that can hold 0.001 (scale <= precision)
+    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<String, Object> value = ImmutableMap.of("amount", new BigDecimal("1E+2"));
+
+    List<IcebergWriterResult> 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<IcebergWriterResult> sinkWriterTest(
       Map<String, Object> value, IcebergSinkConfig config) {
     IcebergWriterResult writeResult =