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

[wombatu-kun]

Closes #16605

Problem

SchemaUtils.inferIcebergType inferred a decimal column type directly from a BigDecimal's precision and scale. Iceberg requires 0 <= scale <= precision <= 38, but a BigDecimal does not guarantee that: a value < 1 (e.g. 0.001) has precision 1 and scale 3, and an exponential-form value (e.g. 1E+2) has a negative scale. DecimalType.of only validates precision <= 38, so these malformed types were built silently and then failed downstream at write time.

Solution

Normalize precision and scale before building the type: rescale a negative scale to 0, and widen precision to at least the scale. Valid inputs are unchanged (e.g. 12.345 stays decimal(5, 3)); 0.001 becomes decimal(3, 3) and 1E+2 becomes decimal(3, 0). A value needing more than 38 digits cannot be represented at all, so no type is inferred for it and the existing "cannot infer" path applies, rather than an IllegalArgumentException escaping inference. The normalization runs in long so a pathological scale cannot overflow it.

Tests

• TestSchemaUtils.testInferIcebergTypeSmallDecimal (unit): asserts the inferred type is valid for both the small-fraction and the negative-scale cases.
• TestSchemaUtils.testInferIcebergTypeDecimalOutOfRange (unit): 1E+37 and 1E-38 still infer as decimal(38, 0) and decimal(38, 38); 1E+38, 1E-39 and a scale of Integer.MIN_VALUE infer no type.
• TestSinkWriter.testEvolveAddsFractionalDecimalColumn (end-to-end): with schema evolution enabled, writes a schemaless record with a fractional decimal for a new column. Before the change the write fails with DataException caused by Invalid DECIMAL scale: 3 cannot be greater than precision: 1; after the change the record is written and the column is decimal(3, 3).
• TestSinkWriter.testEvolveAddsExponentialDecimalColumn (end-to-end): the same for 1E+2, which fails before the change and evolves to decimal(3, 0) after it.

---

AI Disclosure

• Model: Claude Opus 4.8, Claude Opus 5
• Platform/Tool: Claude Code
• Human Oversight: fully reviewed
• Prompt Summary: Fix the invalid decimal type inferred for BigDecimal values whose scale exceeds their precision, then bound the inferred precision and address review feedback.

[AnatolyPopov]

@wombatu-kun Thanks for the PR. Looks good to me.

[github-actions]

This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.

[wombatu-kun]

no stale

[laskoviymishka]

Thanks for the fix — normalizing the negative-scale and precision-less-than-scale cases into a single DecimalType.of call is the right approach, and the unit coverage for 0.001 and 1E+2 is a good start. Almost there; one blocking thing before merge.

The normalized precision has no upper bound, so this trades a controlled failure for a task crash on large-magnitude values. new BigDecimal("1E+38") normalizes to precision 39 and new BigDecimal("1E-39") lands on scale 39 — both make DecimalType.of throw IllegalArgumentException (it enforces precision <= 38), and that exception propagates uncaught out of inferIcebergType and aborts the sink task. Before this change those values built an invalid-but-non-throwing decimal that only failed later at write time, so it's a regression in failure mode. I'd cap the precision at 38 (or return null) before constructing the type, and add 1E+38 / 1E-39 to the unit test so the boundary is covered.

A few smaller things inline: an integer-overflow edge in the same block for pathological scales, the test comment that points at the wrong failure step, a findType null-guard, and an optional design question about whether decimal(3,3) for 0.001 is too tight. None of those block.

Fix the precision cap and this is good to land — happy to take another pass once it's in.

[laskoviymishka]

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:

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?

[wombatu-kun]

Done e081d15. Out-of-range returns null rather than a capped type: null is this method's existing "cannot infer" signal, already returned for empty lists, empty maps and unrecognized classes, so RecordConverter skips the column and autoCreateTable raises a DataException instead of an IllegalArgumentException escaping inference. Clamping to decimal(38, scale) would silently drop significant digits. Boundaries are covered in testInferIcebergTypeDecimalOutOfRange.

[laskoviymishka]

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.

[wombatu-kun]

Done e081d15. Normalization runs in long, so the subtraction cannot overflow and no separate |scale| bound is needed - the final precision > 38 check catches every out-of-range case.

Note that Math.abs(scale) > 38 would not have worked here: Math.abs(Integer.MIN_VALUE) is negative, so it passes the very input it should reject.

[laskoviymishka]

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?

[wombatu-kun]

Spark does not pad. Decimal.set(scala.math.BigDecimal) sets _precision = scale when precision < scale, so 0.001 is decimal(3, 3) there too, and for scale < 0 it sets _precision = precision - scale, _scale = 0 - the same normalization on both branches as this PR. JsonInferSchema also uses max(precision, scale), falling back to DoubleType past MAX_PRECISION rather than padding. Keeping the tight type.

[laskoviymishka]

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."

[wombatu-kun]

DecimalType.of(1, 3) does not throw. The constructor carries a single precondition, precision <= 38 (api/src/main/java/org/apache/iceberg/types/Types.java:526-529) - no scale <= precision check and no negative-scale check, which is why this reaches the writer at all. Invalid DECIMAL scale: 3 cannot be greater than precision: 1 is Parquet's message: reverting the fix locally fails at parquet.schema.Types$BasePrimitiveBuilder.decimalMetadata, via iceberg.parquet.TypeToMessageType.primitive, so the evolution does commit and the write is reached. Reworded in e081d15 to say why the failure is deferred, since that is easy to misread.

[laskoviymishka]

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.

[wombatu-kun]

Done e081d15

[laskoviymishka]

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.

[wombatu-kun]

Done e081d15

[laskoviymishka]

I think this is good to merge now!
Thanks for this @wombatu-kun

🌬️ ⛵