You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "laskoviymishka (via GitHub)" <gi...@apache.org> on 2026/07/31 14:05:03 UTC
Re: [PR] Kafka Connect: Fix invalid decimal type inferred for some BigDecimal values [iceberg]
laskoviymishka commented on code in PR #16606:
URL: https://github.com/apache/iceberg/pull/16606#discussion_r3690736007
##########
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;
Review Comment:
One more edge on this subtraction: it can wrap for pathological scales.
`precision -= scale` with a very negative scale overflows int — `new BigDecimal(BigInteger.ONE, Integer.MIN_VALUE)` gives `precision - (-2147483648)`, which wraps to a large negative, and `Math.max(..., 0)` then hands `DecimalType.of` a garbage precision. A precision <= 38 guard at the end won't catch this, since the wrapped value can look small and valid. These BigDecimals are legal and could arrive from a decoder outside the normal Connect Decimal path. Rejecting `|scale| > 38` up front (or `Math.subtractExact`) closes it cleanly.
##########
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);
Review Comment:
I'd add an upper-bound guard here before merge — right now this turns a controlled failure into a task crash for large-magnitude values.
`DecimalType.of` enforces precision <= 38, so `new BigDecimal("1E+38")` (precision 1, scale -38) normalizes to precision 39 and throws `IllegalArgumentException`. Same on the other branch: `new BigDecimal("1E-39")` (scale 39) hits `Math.max(1, 39)` = 39 and throws. That exception propagates uncaught out of `inferIcebergType` and aborts the sink task. Before this change the old `DecimalType.of(1, -38)` was accepted (negative scale wasn't validated) and only failed later at write time — so it's a regression in failure mode, from "the write errors" to "the task dies."
I'd cap at 38 (or return null) before constructing the type. An early return in the negative-scale branch plus a bound on the final precision also makes each branch's invariant legible instead of leaning on the single `Math.max`:
```java
if (scale < 0) {
precision -= scale; // scale is negative, so this widens precision by |scale|
scale = 0;
}
int p = Math.max(precision, scale);
if (p > 38) {
return null; // or DecimalType.of(38, Math.min(scale, 38))
}
return DecimalType.of(p, scale);
```
Whichever way we land, could we add `1E+38` and `1E-39` to `testInferIcebergTypeSmallDecimal` so the boundary is actually covered? wdyt?
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSinkWriter.java:
##########
@@ -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<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)
+ Type added = catalog.loadTable(TABLE_IDENTIFIER).schema().findType("amount");
Review Comment:
Small thing — I'd use `findField("amount").type()` here to match the idiom in TestIcebergWriterFactory, and assert the field isn't null first.
`findType` returns null for a missing column, so if evolution failed to add it at all the `isEqualTo` fails with an opaque "expected ... but was null" instead of a clear "column wasn't added." An `assertThat(added).isNotNull()` before the type check makes that failure legible.
##########
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
Review Comment:
Not blocking, but worth a thought: widening precision to exactly the scale gives the column zero integer digits.
`0.001` infers as decimal(3,3), which holds -0.999..0.999 but not `1.001` — a later record >= 1.0 in the same column would fail to write. Spark's DDL inference pads one integer digit (`max(precision, scale) + 1`, so decimal(4,3)). Since this changes the inferred type it's a real design call, not just a nit — do we want to match that padding here, or is the tight type intentional? wdyt?
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSinkWriter.java:
##########
@@ -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<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)
+ Type added = catalog.loadTable(TABLE_IDENTIFIER).schema().findType("amount");
+ assertThat(added).isEqualTo(Types.DecimalType.of(3, 3));
Review Comment:
Could we add a companion e2e for the exponential case while we're here?
`1E+2` is only covered as a unit test; the negative-scale branch is never exercised end-to-end, so the write path (convertDecimal calling `setScale` on a negative-scale input) isn't confirmed. A `testEvolveAddsExponentialDecimalColumn` writing `1E+2` and asserting the catalog column is decimal(3,0) would round it out.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestSinkWriter.java:
##########
@@ -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.
Review Comment:
This comment describes the wrong failure point.
`DecimalType.of(1, 3)` throws immediately because Iceberg validates scale <= precision at construction, so the failure happens during type inference / schema evolution — the write below is never reached. I'd reword to something like "before the fix, `inferIcebergType` calls `DecimalType.of(1, 3)`, which throws because Iceberg requires scale <= precision; after the fix it infers decimal(3, 3) and the record is written."
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org