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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
