feat: Support Iceberg system functions (bucket, truncate, years/months/days/hours) natively - #5638
Conversation
Implement Iceberg's `bucket`, `truncate`, `years`, `months`, `days`, and `hours` system functions as native scalar functions and route the `StaticInvoke` calls Spark binds for them through `CometStaticInvoke`, keyed on the Iceberg implementation class names. With these native, the hash distribution and local sort that Iceberg requests in front of a partitioned write stay in Comet, so the native Iceberg writer no longer declines a partitioned table that uses the default `write.distribution-mode`. Filters, projections, and sort keys over hidden-partitioning expressions stay native as well. The kernels match Iceberg's Java implementations exactly: the spec's byte encodings hashed with standard 32-bit Murmur3, Java's wrapping integer arithmetic for truncate, code-point counting for strings, and UTC-only calendar math for the temporal transforms. A truncated decimal that no longer fits its precision becomes null, as it does in Spark. Also name the declaring class in the fallback reason for an unlisted static invoke, since every Iceberg system function is called `invoke`. Closes apache#5635
|
@jordepic could you help with reviews? |
Native: hash tinyint/smallint directly instead of casting first, fold the string/binary bucket arms into one generic helper, compute the minimal two's complement length with leading_ones/leading_zeros, fuse the years/months passes into a single kernel, reuse is_valid_decimal_precision for the decimal overflow check, and share the string/binary truncation helpers with a corrected note on which Arrow kernel overflows. Serde: key CometStaticInvoke's single map by (functionName, class name) so the Iceberg handlers join it instead of a second lookup, and use type sets for the temporal predicates. Tests: move capturePlans into CometIcebergTestBase, write the source parquet once per suite, batch the per-type comparisons into one query per column, and declare boundary values column-wise.
|
Yep! I'm gone the rest of today but it's on my list for tomorrow |
| let truncated: Decimal128Array = | ||
| array.as_primitive::<Decimal128Type>().unary_opt(|v| { | ||
| let truncated = truncate_i128(v, width as i128); | ||
| is_valid_decimal_precision(truncated, *precision).then_some(truncated) |
There was a problem hiding this comment.
I think this precision check is happening too early relative to Spark's semantics.
Iceberg's JVM TruncateDecimal.invoke returns Decimal.apply(truncatedValue) without coercing it back to the declared DecimalType. Spark's StaticInvoke therefore exposes a non-null Decimal to its parent expression; changePrecision(precision, scale) only happens later when an UnsafeRowWriter materializes the value.
Here unary_opt turns that intermediate into NULL during expression evaluation itself. That changes nested/predicate semantics. For example, with DECIMAL(18,4) value -99999999999999.9999, truncate(10, v) mathematically produces -100000000000000.0000 (19 digits). On the JVM the StaticInvoke result is still non-null, whereas this branch returns NULL, so WHERE truncate(10, v) IS NULL can silently select a row that Spark does not.
Could we either preserve the JVM intermediate semantics or fall back when this overflow case occurs?
There was a problem hiding this comment.
One data point for scoping the fix: on the write path the null sort key is actually consistent with Spark. Spark's SortExec builds its key through an UnsafeRowWriter too, so the JVM sort sees null for the same rows. The divergence you describe is confined to filters and projections, where codegen evaluates invoke without materializing the row. It may be worth keeping the fix scoped to those paths so the sort key stays aligned with what Spark's own sort produces.
There was a problem hiding this comment.
LLM-assisted: this reply and the changes it describes were written with Claude Code.
You're right about the intermediate: TruncateDecimal.invoke returns Decimal.apply(...) with no changePrecision, so the StaticInvoke result is non-null and only UnsafeRowWriter nulls it.
I don't think the kernel can reproduce that, though. It has to return a Decimal128(precision, scale) array, and that array has no encoding for "exceeds the declared precision but isn't null yet" — whatever it holds is what both the plan output and any parent expression see. Emitting the oversized value would make the common case wrong: SELECT truncate(10, dec18) is null in Spark, and the corpus in the new suite already contains -99999999999999.9999, so checkSparkAnswerAndOperator fails on it. Nulling eagerly is also what Spark does for decimal overflow elsewhere, via CheckOverflow(..., nullOnOverflow = true); Iceberg's invoke simply isn't wrapped in one.
So it's a choice of which side to be right on, and I've taken the materialized side. As @jordepic notes, SortExec builds its key through an UnsafeRowWriter too, so the sort agrees, and so does the projection output. What differs is a truncated decimal feeding another expression directly: WHERE truncate(10, v) IS NULL, and the Murmur3Hash behind DISTRIBUTE BY (that one only moves rows between partitions, it doesn't change an answer). The Iceberg partition value isn't affected either way — the writer derives it from the untruncated column.
I've written the difference down rather than leaving it implicit: 3dc1a2d in the Iceberg user guide, plus the kernel comment. The alternative is to fall back for decimal truncate entirely, which costs native decimal partitioning for a case that needs a value within width - 1 ULP of the negative precision boundary. Would you rather have that?
There was a problem hiding this comment.
I need to correct what I said above, and @sunchao is right to push on it. SortExec does not build its key through an UnsafeRowWriter. It creates the comparator with RowOrdering.create, which generates code that evaluates the sort expression on each input row and compares the raw result, and the sort prefix for a decimal(18,4) is toUnscaledLong of that same value. So Spark's sort sees the oversized decimal and Comet's sort sees null. The sentence about the sort key in the new iceberg.md paragraph and in the truncate.rs comment came from my mistake and should be dropped.
The practical effect on the write path is still nil, for a different reason: for a given precision and width exactly one truncated value can exceed the precision (the multiples of the width that land below the negative bound form a window narrower than the width), and it is the column minimum, so it sorts first under either key and the clustered writer sees a single run. The difference is confined to ordering position under a non-default null ordering, and to predicates and hashes as you described.
There was a problem hiding this comment.
LLM-assisted: this reply and the changes it describes were written with Claude Code.
Thanks for catching it. I'd taken the UnsafeRowWriter claim from your earlier note and propagated it into iceberg.md and the kernel comment without checking, so that one's on me too. Confirmed: SortExec orders the child's rows with an ordering from RowOrdering.create, which evaluates the sort expression per comparison, and the decimal(18,4) prefix is toUnscaledLong of that same value — no UnsafeRowWriter in the path. Both places are corrected in 0a57b6a, and it is moot anyway now that decimals fall back.
| } | ||
|
|
||
| fn civil_date(days: i32) -> Result<chrono::NaiveDate> { | ||
| Date32Type::to_naive_date_opt(days).ok_or_else(|| { |
There was a problem hiding this comment.
This introduces a narrower date domain than the Iceberg JVM implementation.
Date32Type::to_naive_date_opt ultimately uses chrono's date representation, whose range is only about 262k years in either direction. Spark DateType, however, carries an i32 epoch-day value, and Iceberg's DateTimeUtil.daysToYears / daysToMonths uses Java LocalDate; that supports the complete i32 epoch-day domain.
For example, epoch day 100000000 is valid for the JVM implementation but falls outside chrono's range, so native years / months will return an execution error where Spark/Iceberg succeeds. The timestamp versions have the same issue near the extremes of the i64 microsecond domain.
Could we compute the proleptic-Gregorian year/month directly from the epoch-day integer (or fall back outside chrono's range) and add a reference test with epoch days beyond chrono's limits?
There was a problem hiding this comment.
LLM-assisted: this reply and the changes it describes were written with Claude Code.
Good catch, fixed in 71e03be. years and months no longer touch chrono: civil_from_days does the proleptic Gregorian split in i64 integer arithmetic, exact over the whole i32 epoch-day domain — i32::MAX days is +5881580-07-11 and i32::MIN is -5877641-06-23, both well inside LocalDate. The kernels are infallible now as a result.
The expected values come from running Iceberg's own DateTimeUtil.convertDays / convertMicros on a JDK 17 JVM, and the tables pin epoch days ±1e8, ±1e9 and both i32 extremes, plus i64::MIN / i64::MAX micros. There is also a test checking the integer split against chrono for every day in a 400-year window around the epoch and on a stride over the rest, so it cannot drift in between.
On the timestamp side you're right that the same concern applies, though it lands differently per function. days was already safe: i64::MAX micros is 1.07e8 days, comfortably inside an i32. years and months went through chrono after that conversion and had exactly the problem you describe. hours is the one that genuinely overflows — i64::MAX micros is 2.56e9 hours — and Iceberg narrows it with a plain (int) cast, so as i32 wraps to the same value; that is now pinned rather than incidental.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
The native transform integration is useful, but the existing decimal-intermediate P2 and calendar-range P2 remain. I independently confirmed both with exact-version component probes and maintained Spark source tracing. There is an additional test-fixture issue inline.
Please also add a reusable expression microbenchmark and matched Comet-versus-Iceberg-JVM results for representative numeric/decimal, string/binary, and temporal inputs, including relevant dictionary/null cases. Include versions, settings, evaluated plans, and correctness checks. I found no benchmark of these new functions in the change or discussion. The 5.5x result from #5361 used write.distribution-mode=none and does not measure these transforms. Dictionary expansion and repeated scalar/array work are costs to measure, not demonstrated regressions.
The latest CI snapshot has 34 successful, 7 skipped, 26 running and 4 failed checks. The Spark 3.4/3.5 scan jobs abort during this suite's fixture setup, as detailed inline. The Spark 4.0 and Spark 4.1 jobs each fail the new partitioned-write end-to-end test because an expected Parquet file is missing. I inspected those logs but have not established the cause of that missing file, so that test is not yet passing end-to-end evidence.
My own validation was source tracing and focused Arrow/Chrono/Iceberg/Guava component probes. It did not execute a full Spark query, Comet suite, JNI/native writer or performance benchmark.
| maybeNull(randomDecimal38()), | ||
| maybeNull(randomString()), | ||
| maybeNull(randomBinary()), | ||
| maybeNull(LocalDate.ofEpochDay(random.nextInt(40000) - 20000)), |
There was a problem hiding this comment.
[P2] Match the fixture's date/time objects to Spark's encoder
These rows contain java.time.LocalDate and Instant, but the Spark 3.4/3.5 CI runs encode the date field as java.sql.Date. beforeAll writes this DataFrame before any tests run, so the entire new suite aborts with LocalDate cannot be cast to java.sql.Date in both the 3.4 job and 3.5 job. The suite's sparkConf does not select the Java 8 date/time API. Could you enable spark.sql.datetime.java8API.enabled before constructing the fixture, or emit java.sql.Date/Timestamp when that setting is false? Until this setup succeeds, those profiles do not execute the new transform comparisons.
There was a problem hiding this comment.
LLM-assisted: this reply and the changes it describes were written with Claude Code.
Fixed in 7de2708, and all six [scans] jobs are green on the current head, 3.4 and 3.5 included.
The write is now wrapped in withSQLConf with datetimeJava8ApiEnabled set, along with two other Spark 3.x defaults the corpus trips over: datetimeRebaseModeInWrite (the timestamps reach back to 1843 and 3.x refuses to write before 1900) and outputTimestampType (3.x defaults to INT96, which has its own ancient-timestamp check). The row encoder is built on the driver inside that block, so setting the conf around the write is enough, and none of it changes a comparison since every test reads the data back from Parquet.
On the 4.0/4.1 failures: those were fixed by f8db9b5, where the 4.0, 4.1 and 4.2 [scans] jobs all passed and the only two failures left were the 3.4/3.5 ones above. I should be straight about the cause, though. I could not retrieve those job logs, so I do not have a confirmed root cause either. That commit narrowed the test in two ways — it stopped using the multi-byte string column as a partition source and it excluded Long.MinValue — and I cannot reproduce a failure locally with either one restored (Spark 4.1, macOS, including with sun.jnu.encoding forced to US-ASCII), so the charset explanation the comment gave was a hypothesis I have not demonstrated. I have replaced it with what is actually known (f4e8e45). What I can say is that the test is green on 4.0/4.1/4.2, that the string transforms are still covered by the comparison, filter, sort and shuffle tests, and that truncate at Long.MinValue is pinned against the JVM in the Rust unit tests.
…d write test The CI containers run the JVM with an ASCII platform charset, so raw multi-byte partition directory names written by the pinned iceberg-rust cannot be reopened by iceberg-java for the metrics rebuild; partition on the long column instead. iceberg-rust's truncate transform also overflows on Long.MinValue in debug builds, so that boundary row is not written.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Rechecked f8db9b5f. No new findings beyond the existing [P2] threads for decimal intermediates, calendar range, and fixture encoding. All three remain unchanged.
This update narrows only the partitioned-write test. It does not establish the cause of the earlier missing-file failures. Current-head CI has 15 successful, 6 running and 7 skipped checks at the latest snapshot. This re-review checked the exact source increment and unchanged prior evidence. I have not rerun the Spark/native test.
jordepic
left a comment
jordepic
left a comment
There was a problem hiding this comment.
Thanks for putting this together. The class-name keyed dispatch in CometStaticInvoke is a clean way to handle Iceberg not being on the compile classpath, and the fallback message naming the declaring class is a real usability improvement given that every Iceberg function is called invoke. I traced the kernels against DateTimeUtil, TruncateUtil, and BucketFunction.bind in iceberg-java and they match where I looked, including the floor semantics for negative dates and the int-then-narrow behavior for tinyint and smallint.
My main question is about the relationship with iceberg-rust. The native writer already computes partition values through iceberg::transform::create_transform_function inside PartitionValueCalculator, on every batch. After this change the sort key feeding the clustered writer comes from these new kernels and the partition key the writer groups by comes from iceberg-rust's kernels. The clustered writer requires the two to agree, and when they don't the task fails with The input is not sorted! Cannot write to partition that was previously closed, so any drift between the implementations surfaces as a runtime failure on the write path rather than at plan time.
Comparing against the pinned iceberg-rust rev, bucket, hours, and days are functionally identical there. iceberg-rust's bucket uses the murmur3 crate (x86 32-bit, seed 0, the same as Guava's murmur3_32_fixed) with the same minimal two's-complement decimal encoding, and that crate is already in native/Cargo.lock. The only gaps are Int8/Int16 inputs and dictionary unpacking, both of which this PR already handles in a thin wrapper. Could those three call iceberg-rust's transforms directly? That would remove most of bucket.rs including the hand-rolled murmur3_32. truncate is the one function where iceberg-rust genuinely diverges from Java (rem_euclid for i32, unchecked arithmetic for i64), so keeping a local kernel there makes sense.
If you would rather keep all the local kernels, it would be worth adding a Rust test that asserts each SparkIceberg* UDF matches create_transform_function over the same boundary inputs. Without that, a future iceberg-rust bump can silently desynchronize the sort key from the writer.
One small addition to @sunchao's benchmark request: the dictionary unpack via cast in apply_unary allocates a full copy of the values before hashing. String partition columns coming out of the Parquet scan are usually dictionary-encoded, so a benchmark of bucket and truncate over a dictionary string column would tell us whether hashing the dictionary values once and taking through the keys is worth doing now.
| //! `hours` are plain floor division of the epoch value. `days` returns a date (Iceberg's | ||
| //! `DaysFunction.resultType()` is `DateType`), the other three return an int. | ||
| //! | ||
| //! The kernels work on the raw epoch values rather than going through Arrow's timezone-aware |
There was a problem hiding this comment.
In Comet, TimestampType arrays are always tagged UTC (native/core/src/execution/serde.rs), and the writer casts every batch to iceberg-rust's Arrow schema, which tags Timestamptz as +00:00. So date_part in iceberg-rust's Year and Month transforms would also evaluate in UTC on both the sort path and the write path. Is there a plan shape where a non-UTC tag reaches these kernels? If not, it would be good for this doc to say the concern is defensive rather than observed, or these two could reuse iceberg-rust's kernels as well.
There was a problem hiding this comment.
LLM-assisted: this reply and the changes it describes were written with Claude Code.
Agreed on both counts, and the module docs now say so (71e03be): Comet tags TimestampType UTC and the writer casts to a schema that tags Timestamptz +00:00, so date_part would agree today; the epoch arithmetic is correct for any tag. I know of no plan shape where a different tag reaches these kernels.
I kept the local kernels rather than delegating, and added a test that makes the reason executable: iceberg_rust_years_follow_the_timezone_tag in iceberg_write.rs shows Transform::Year returning 0 for -1 micros tagged Asia/Kathmandu where Iceberg's Java DateTimeUtil and Comet both return -1. If that test ever fails, iceberg-rust has dropped the tag dependency and delegating becomes safe.
| CREATE TABLE $table (i32 INT, i64 BIGINT, ts TIMESTAMP, dt DATE) | ||
| USING iceberg | ||
| PARTITIONED BY (bucket(4, i32), truncate(1000, i64), days(ts), months(dt))""") | ||
| // iceberg-rust's own truncate transform, which the writer uses for partition values, does |
There was a problem hiding this comment.
This works around unchecked arithmetic in iceberg-rust's truncate_i64, which also affects the native writer's partition values in debug builds, independent of this PR. Could you file that upstream in iceberg-rust and reference the issue here instead of describing the workaround inline? Otherwise the filter reads as a test-data choice and the writer-side bug gets lost.
There was a problem hiding this comment.
LLM-assisted: this reply and the changes it describes were written with Claude Code.
Fair — the writer-side bug should not be buried in a test comment. I will file it against iceberg-rust separately and link it here rather than hold this PR on it. It is a bit wider than truncate_i64: truncate_i32 uses rem_euclid, which diverges from Java for widths above 2^30 as well as overflowing at Integer.MIN_VALUE, and the decimal kernel has the same unchecked subtraction. In the meantime the new iceberg_rust_transform_parity module in iceberg_write.rs records the excluded inputs in one place, with the reason, instead of only here.
…faults CometIcebergSystemFunctionSuite aborted in beforeAll on Spark 3.4 and 3.5. Three Spark 3.x defaults differ from Spark 4's and each blocks writing the corpus: - datetimeJava8ApiEnabled is off, so the row encoder rejects the java.time values sourceData supplies (ClassCastException: LocalDate -> java.sql.Date) - datetimeRebaseModeInWrite is EXCEPTION, which rejects the deliberately pre-epoch timestamps (the corpus reaches back to 1843) - outputTimestampType is INT96, which has its own ancient-timestamp check Set all three to Spark 4's values around the write only. Every test reads the data back from parquet, so no result comparison is affected.
|
I was mainly looking at this one and cross-referencing the iceberg-rust source code and the iceberg source code, in the event that other code reviews hadn't. I think I mainly would love to avoid code duplication where possible! I may also just ask that you rewrite the PR description by hand, since understanding claudisms is really challenging for me from an english perspective. |
`years` and `months` split the calendar with `chrono::NaiveDate`, whose range stops at about year 262143. A Spark `DateType` is an `i32` epoch day (up to year 5881580) and Iceberg's `DateTimeUtil` goes through `LocalDate`, which covers all of it, so `years(DATE)` on a far-future date raised an execution error where the JVM returns a value. Replace the `chrono` round trip with Howard Hinnant's `civil_from_days` in `i64` arithmetic, which is exact over the whole `i32` epoch-day domain and drops the fallible path from the kernels. The pinned expectations for the extremes come from running `DateTimeUtil.convertDays` / `convertMicros` on a JVM, and a new test checks the integer split against `chrono` everywhere `chrono` can represent the date.
`apply_unary` unpacked a dictionary with `cast` before handing it to the kernel, so a dictionary-encoded string column -- what a Parquet scan hands back for a low cardinality partition column -- paid for a full copy of the values buffer and then hashed or truncated every row instead of every distinct value. Run the kernel over the dictionary's values and expand the result through the keys instead. Over 8192 rows with eight distinct strings (`cargo bench --bench iceberg_transforms`): iceberg_bucket/string_dict 72.9 us -> 3.0 us iceberg_truncate/string_dict 140.9 us -> 35.4 us The decimal comment in `truncate` also now spells out which paths agree with the JVM result and which do not.
A partitioned write sorts on Comet's kernels and then groups by the partition values iceberg-rust computes, and the clustered writer fails at runtime when the two disagree. Assert they agree over the boundary inputs of every shared type, so an iceberg-rust bump that changes a transform breaks here first. The excluded cases are the interesting ones: iceberg-rust's `truncate` does not wrap like Java's, and its `years` / `months` go through Arrow's `date_part`, which honours the array's timezone tag. The last test pins that tag dependency, since it is the reason `years` and `months` keep local kernels while `bucket`, `days`, and `hours` could in principle be delegated.
Iceberg's `TruncateDecimal.invoke` returns a `Decimal` that can exceed the column's precision and Spark nulls it only when the row is materialized, while Comet nulls it in the kernel. Say so, and say which paths that leaves in agreement.
…erg's JVM `native/spark-expr/benches/iceberg_transforms.rs` covers every transform over every supported type, with and without nulls, plus the dictionary-encoded string shape a Parquet scan produces. `CometIcebergSystemFunctionBenchmark` runs the same queries with Comet on and off. The Comet-off case is Iceberg's own JVM implementation, since Spark binds each function as a `StaticInvoke` of the class under `org.apache.iceberg.spark.functions`. The data stays in Parquet rather than an Iceberg table so both cases scan identically and only the transform differs.
`HashPartitioning` hashes the `StaticInvoke` result directly rather than through an `UnsafeRowWriter`, so the shuffle hash is on the differing side, not the agreeing one. The Iceberg partition value does not come from this kernel at all.
…usion The multi-byte string column is out of the write test's partition spec because the test failed on the Linux CI runners with it in, not because the platform-charset mechanism the comment claimed has been demonstrated -- it does not reproduce locally, including with sun.jnu.encoding forced to US-ASCII.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Re-reviewed f4e8e454 against 10537e14. The calendar-range P2 is fixed in source. The replacement arithmetic matched 317,133 sampled date/timestamp inputs against each of Iceberg Java 1.8.1, 1.10.0 and 1.11.0 in isolated scalar checks. The earlier explicit fixture settings also address the reported setup mismatch in source, though the Spark suite was not rerun.
The existing decimal-intermediate P2 remains. Documenting early nulling does not preserve results for enclosing predicates or hashes, and decimal inputs are still enabled as Compatible. Could unsafe decimal cases fall back until the intermediate semantics are preserved? Please also narrow the new sort claim: SortExec materializes a Long prefix, not necessarily the decimal expression result. An embedded transform in a sort expression differs from sorting an already-materialized decimal column.
The dictionary optimization, writer comparison tests, and two benchmark entry points address the earlier follow-ups. The author-reported native dictionary timings are useful. The existing benchmark request still needs matched Comet/Iceberg-JVM results with versions/settings, executed plans and fallback status, correctness checks, and comparable dictionary/null inputs. The writer comparisons explicitly omit known divergent boundaries, so they do not establish complete writer alignment.
No new P1/P2 or duplicate inline findings. Validation was source tracing, revalidation of retained evidence, and fresh scalar calendar checks. It did not execute the new Arrow/UDF tests, Spark queries, sort/writer paths, or benchmarks. At the final check, CI had 18 successful checks, six skipped and seven still running.
|
LLM-assisted: this reply and the changes it describes were written with Claude Code. @jordepic thanks for tracing it against iceberg-java, and for the framing about the sort key and the partition values coming from two different implementations — that is the sharpest statement of the risk here. I went with the test rather than the delegation, for two reasons. The wrapper would not get much thinner: the So there is now an On the dictionary unpack — you were right that it was worth measuring, and worth doing.
For reference the plain string column is 31.2 µs for |
|
LLM-assisted: this reply and the changes it describes were written with Claude Code. @sunchao benchmarks added, both kinds.
Apple M3 Max, JDK 17, Spark 4.1, release native build, 1,048,576 rows, best of five: No case regresses. Dictionary expansion was a regression before this round — @jordepic called it — and is fixed in 1622503; numbers on that thread. Correctness is unchanged: the same queries run under On the two P2s: the calendar-range one is fixed in 71e03be, with the reference values taken from running Iceberg's On CI: all six |
|
LLM-assisted: this reply and the changes it describes were written with Claude Code. @jordepic filed upstream as promised:
Two things came out of writing them up that are worth correcting here, since both change what I said earlier on those threads. The
Iceberg Java only validates The timezone one turned out to be more than the Comet-facing concern I described on the I have updated the test comment in |
The truncate exclusions in the parity tests and the Long.MinValue filter in the write test now point at apache/iceberg-rust#3141, and the timezone-tag tests at apache/iceberg-rust#3142, instead of describing the workaround inline.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Rechecked ada706be against 10537e14. This increment changes tracking comments only. Thanks for filing iceberg-rust #3141 and #3142. Those links address the upstream-tracking request.
On the remaining decimal P2: yes, I prefer fallback where the JVM intermediate semantics cannot be preserved. Returning null early can change which rows a predicate selects, even if the boundary is rare. If safe cases cannot be distinguished, falling back for decimal truncate is preferable to reporting the changed behavior as Compatible. This costs native evaluation, but avoids introducing context-dependent nulling rules. The earlier clarification about an embedded sort expression versus a materialized decimal column still applies.
The 21-case matched benchmark table addresses the missing-numbers part of the request. I am treating the reported 1.2x–3.7x gains as query-level, author-measured results. The runner changes Comet execution for the scan and projection, so common Parquet input alone does not isolate transform cost. Could you attach the executed plans, verify equal outputs on the timed corpus, and include comparable null cases? Related suite coverage is useful, but is not the same corpus and expression-parameter coverage as this benchmark.
No new P1/P2 findings. This pass revalidated source and retained evidence, without rerunning tests or benchmarks. Current-head CI has 16 successful checks, six skipped and six still running.
jordepic
left a comment
jordepic
left a comment
There was a problem hiding this comment.
Went through the new commits. The civil_from_days arithmetic matches the JVM at both i32 extremes, the hours narrowing matches Java's (int) cast, the dictionary path is correct including null keys, and the parity module plus the two upstream issues are exactly what I was after. Three follow-ups inline, one of which is a correction of something I said earlier.
| } else if (!valueTypeSupported(value.dataType)) { | ||
| Unsupported(Some(s"$nativeName does not support input type ${value.dataType}")) | ||
| } else { | ||
| Compatible() |
There was a problem hiding this comment.
On the decimal truncate question @sunchao and you are going back and forth on: Comet already has the mechanism that gives both of you what you want. If getSupportLevel returns Incompatible(Some(...)) for a DecimalType value instead of Compatible(), the expression falls back by default and a user who accepts the documented difference can opt in with the per-expression allowIncompatible config. That is how every other expression with a known divergence from Spark is handled here, and it keeps the native path available for partitioned writes on tables where the operator has read the caveat. It would also need getIncompatibleReasons so the generated compatibility page carries the note, rather than only the Iceberg guide. That seems like a better answer to your "would you rather have that?" than a hard fallback.
There was a problem hiding this comment.
LLM-assisted: this reply and the changes it describes were written with Claude Code.
Good call on reaching for the existing mechanism. I went one step past it and marked decimal truncate Unsupported rather than Incompatible, for two reasons.
The wiring @sunchao flagged is real: getExprConfigName resolves against the outer expression class, so the opt-in would have been spark.comet.expression.StaticInvoke.allowIncompatible, which also unlocks every future incompatible static invoke. Fixing that properly means changing what spark.comet.expression.StaticInvoke.enabled covers for readSidePadding, aesEncrypt, decode and base64, and I would rather not smuggle that into this PR. Unsupported needs no key, so the question disappears.
The docs half of your point I did take. CometStaticInvoke.getUnsupportedReasons() now aggregates the per-function handlers' notes, since GenerateDocs only asks the serde registered for the expression class — without it the note reached the Iceberg guide but not the generated compatibility page. There is a test asserting it is reachable that way so it cannot silently drop out.
If you would rather have the opt-in after all, Incompatible is a one-line change once the config-name question is settled on its own.
| // No `write.distribution-mode`: Iceberg picks hash distribution for a partitioned table, | ||
| // which plans a shuffle and a local sort on the partition transforms. The string column is | ||
| // deliberately not a partition source: with it in the spec this test failed on the Linux | ||
| // CI runners with a missing data file, which does not reproduce locally, so the multi-byte |
There was a problem hiding this comment.
I think the missing-file failure does have an explanation, and it is the partition-path caveat already documented in iceberg-writes.md. iceberg-java percent-encodes partition directory names and values through URLEncoder in PartitionSpec.partitionToPath, and iceberg-rust writes them raw. The Linux CI runners run the JVM under a POSIX locale, so sun.jnu.encoding is ASCII and opening a local path containing 日本 or 😀 fails, which reads back as a missing data file. On macOS that encoding is fixed to UTF-8 by the launcher and a -Dsun.jnu.encoding override is ignored, which would be why it did not reproduce locally even when forced. Running the original version of this test under LANG=C on Linux should confirm it. If it does, this is a native-writer divergence with an operational effect rather than a test artifact, and worth filing against iceberg-rust's partition_to_path the same way you filed the other two.
There was a problem hiding this comment.
LLM-assisted: this reply and the changes it describes were written with Claude Code.
Your mechanism is right, and it is already fixed upstream — which is the part I had wrong when I said I had no cause.
iceberg-java escapes both field name and value through URLEncoder.encode in PartitionSpec.partitionToPath; iceberg-rust at our pinned rev formats {name}={human_string} raw. apache/iceberg-rust#2875 changed that to form_urlencoded::Serializer::append_pair on both sides and merged 2026-07-30 — one day after the rev we pin, 3d84c81 (2026-07-29). So the caveat in iceberg-writes.md is accurate for our pin and stale against upstream main.
Three things support the locale half without my having read the logs. [scans] runs in the amd64/rust container, which is Debian with no LANG, so sun.jnu.encoding resolves to ASCII. The failing set was exactly {4.0, 4.1}, which is exactly the set of jobs that reached the test — 3.4 and 3.5 aborted in beforeAll on the LocalDate encoder, and 4.2 skips the suite because icebergAvailable is false for isSpark42Plus. And @sunchao reproduced the filesystem half on Linux/JDK 21. I would still call the historical failure inferred rather than confirmed, per his point.
#5651 bumps the pin, so I am not doing it here. Once it lands the string column can go back into the write test's partition spec, and that run is what would actually confirm this.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
I agree with the decimal-specific Incompatible(Some(reason)) approach in the new discussion. It uses the existing default-fallback/explicit-opt-in mechanism without disabling other truncate types or decimal bucket. The existing P2 remains until that behavior is implemented.
One wiring detail: with the current registration, the outer CometStaticInvoke supplies the configuration name, so the opt-in would be spark.comet.expression.StaticInvoke.allowIncompatible. The generated documentation also queries that registered wrapper. A getIncompatibleReasons override only on the inner truncate handler would not surface the note unless the wrapper exposes or aggregates it. Keep the existing invalid-parameter and unsupported-type checks.
On the Unicode-path hypothesis, I independently reproduced the proposed filesystem mechanism in a standalone Linux/OpenJDK 21 probe. With LANG=LC_ALL=C, an existing raw 日本-😀 filename failed through FileInputStream and NIO, while the percent-encoded ASCII control opened. Under C.UTF-8, both opened. Adding -Dsun.jnu.encoding=UTF-8 under the C locale did not change the result on this installation.
That is component evidence, not confirmation of the original JDK 17 CI/Hadoop/Comet failure. Please capture the failing JVM's effective locale/path encoding and exact path/stack, with a UTF-8 control, before treating the historical cause as established. No new P1/P2 findings or full Spark/native run in this pass. Current-head CI has 57 successful checks, six skipped and eight still running.
Iceberg's `TruncateDecimal.invoke` returns a `Decimal` that can exceed the column's declared precision, and Spark turns it into null only when the row is materialized. An Arrow `Decimal128(precision, scale)` array has no encoding for that intermediate, so the kernel nulled it during evaluation, which changes what an enclosing predicate or hash sees. Report decimal inputs as `Unsupported` so they stay with Spark. `CometStaticInvoke.getUnsupportedReasons()` now aggregates the per-function handlers' notes, since `GenerateDocs` only asks the serde registered for the expression class and the note would otherwise reach the Iceberg guide but not the compatibility page. Also drops the claim that Comet and Spark agree on the sort key. `SortExec` orders rows with `RowOrdering.create`, which evaluates the sort expression per comparison rather than through an `UnsafeRowWriter`, so Spark's sort sees the oversized decimal. That was wrong in `iceberg.md` and in the kernel comment.
…benchmark Every case now runs over a no-null column and a column with one null in eight, and each case's output is compared row by row between the two engines over the same corpus that is then timed, before it is timed. `truncate` on a decimal is dropped since it no longer has a native path. `excludedRulesWith` becomes `protected` so the verification can use the same constant-folding exclusion the timed cases use.
|
LLM-assisted: this reply and the changes it describes were written with Claude Code. @sunchao the decimal P2 is resolved rather than mitigated: decimal The sort claim is dropped rather than narrowed. @jordepic corrected himself and you were right to push: Benchmark: every case now runs over a no-null column and a column with one null in eight, and each case's output is compared row by row between the two engines over the same corpus that is then timed, before it is timed. 40 cases, all equal. Apple M3 Max, JDK 17, Spark 4.1, release native build, 1,048,576 rows, best of five, as On isolating transform cost from the scan: I have not done that, and you are right that a shared Parquet input does not achieve it. Both arms read the same files with the same partitioning, so the delta is scan-plus-projection, and these ratios are a floor on the expression-level gain rather than a measurement of it. The kernel-level criterion benchmark is the isolated measurement and the query-level one is the end-to-end number; I have described them that way rather than adding a third arm, which would need a change to the shared benchmark base. |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
The hard decimal fallback addresses the existing P2 in the normal Spark system-function path, and the unsupported note now reaches the registered wrapper used by GenerateDocs. Other truncate types and decimal bucket remain admitted. I found a new Spark 3.x compilation regression in the updated benchmark, noted inline, so this follow-up is a comment rather than an approval.
The same-corpus equality checks and null variants address those parts of the benchmark request. Could you remove the remaining identical-scan and expression-gain lower-bound claims? Toggling Comet can change scan cost, so shared input files do not make the query ratio a lower bound on expression-only gain. Please retain the scan-plus-projection description and include the executed plans and warning/results output from the reported run.
This pass used exact source, retained evidence and current CI logs. I did not rerun Spark or the benchmarks locally. CI currently has 14 successful checks, six skipped, five running and four failures. The four failed jobs all stop at the same benchmark compilation error, not a Celeborn runtime failure.
| */ | ||
| private def verifyOutputsMatch(name: String, query: String): Unit = { | ||
| def collect(cometEnabled: Boolean): Array[Row] = | ||
| withSQLConf( |
There was a problem hiding this comment.
[P1] Keep the benchmark collection helper compatible with Spark 3.x
This method promises Array[Row], but the inherited Spark 3.5 SQLHelper.withSQLConf takes f: => Unit and returns Unit. Returning the wrapper directly therefore does not compile. Spark 4's generic result-returning helper masks this difference. The current Spark 3.5 CI job reports found: Unit, required: Array[org.apache.spark.sql.Row] at this line. The Spark 3.4 job and both Celeborn compatibility jobs fail on the same error, before their tests can run.
Could you capture the collected rows inside the configuration scope and return them afterward, or use an existing cross-version result-returning helper? Please verify test compilation on Spark 3.4 and 3.5 as well as the Spark 4 profile.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Rechecked 29f69db3 against a089669d. The existing Spark 3.x compilation P1 remains: the merge leaves the benchmark collection helper and its inheritance unchanged. Current CI reproduces the same found: Unit, required: Array[Row] error in the Spark 3.4/3.5 and both Celeborn compatibility jobs.
No new P1/P2 findings in this incremental review. At 23:34 UTC, CI had 12 successful, 4 failed, 6 running and 6 skipped checks. I reviewed the source/dependency increment and current CI logs, but did not rerun compilation, product tests or benchmarks.
…k 3.x Spark 3.4 and 3.5 declare `SQLHelper.withSQLConf` as `(pairs: (String, String)*)(f: => Unit): Unit`; only Spark 4 has the generic result-returning form. Returning the wrapper from a method typed `Array[Row]` therefore failed to compile on both 3.x profiles and on the two Celeborn jobs, before any test could run. Collect into a local inside the configuration scope and return it after the block. Also drop the claim that the two arms scan identically. Enabling Comet replaces the Parquet scan and the projection with native operators as well as the transform, so each ratio is a query-level scan-plus-projection result, not a measurement of the transform on its own; the criterion benchmark in native/spark-expr/benches is what isolates the kernel.
The multi-byte string column came out of the partitioned write test's spec because the test failed on the Linux CI runners with a missing data file, which never reproduced locally. The cause was iceberg-rust writing partition directory names raw where iceberg-java percent-encodes them, so a name containing multi-byte characters could not be reopened under the runners' ASCII platform charset. apache/iceberg-rust#2875 fixed that upstream and apache#5651 bumped the pin, which is now merged into this branch, so put the column back. The test then also covers the escaping. Dropping `truncate(1000, i64)` from the spec removes the reason for the Long.MinValue row filter as well, so the whole corpus is written again.
|
LLM-assisted: this reply and the changes it describes were written with Claude Code. @sunchao the P1 is fixed and both claims are gone. The compile break (f2f0c70). The claims. The scaladoc no longer says the two arms scan identically. It now says that enabling Comet also replaces the Parquet scan and the projection with native operators, that each ratio is therefore a query-level scan-plus-projection result, and that it does not isolate the transform. The sentence about the ratios being a floor on the expression-level gain is out of the PR description, which @jordepic had also asked me to rewrite in plainer English, so I rewrote the whole thing. What is left of the Parquet point is only what it should be: keeping the data in Parquet keeps the Iceberg reader out of the Spark arm and lets both arms read the same files. The run. I re-ran on the current head rather than reusing the numbers, so the table in the description is from this run. One reproduction note worth recording: my first attempt reported Comet at 0.1x-0.3x across the board, because Apple M3 Max, macOS 26.6.2, OpenJDK 17.0.10, Spark 4.1, Scala 2.13, These differ from the ones I posted before by at most 0.2x in either direction, which is run-to-run drift on a laptop with other work on it, not a change in the code. Warnings. None. The results file contains no Executed plans. Captured during the timed run by printing 40 cases, Spark plan vs Comet planEvery Comet plan is Full plansbucket(dec)Spark: Comet: truncate(str_dict)Spark: Comet: years(ts)Spark: Comet: months(date)Spark: Comet: days(ts, nulls)Spark: Comet: hours(ts)Spark: Comet: The other 34 pairs differ only in the function class, the column name and the expression ids. I can post them all if you want them verbatim. @jordepic on your third point: #5651 has landed and is merged into this branch, so the multi-byte string column is back in the partitioned write test's partition spec (8c6a064), together with the |
|
LLM-assisted: posted with Claude Code. The full CometIcebergSystemFunctionBenchmark-results.txt |
|
LLM-assisted: posted with Claude Code. @jordepic your diagnosis is confirmed. With the multi-byte string column back in the partition spec, all five Linux No failures on the run so far (48 passed, 18 still going). |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Rechecked 8c6a0645 against a089669d. The existing Spark 3.x compilation P1 is fixed by collecting inside withSQLConf and returning afterward. The corrected benchmark attribution and supplied plans/full results address the earlier request.
The restored partition-write test passed in Linux Spark 3.4, 3.5, 4.0 and 4.1 CI on a merge commit whose source tree equals this head. No remaining P1/P2 findings. At 02:01 UTC, CI had 62 successful, 7 skipped and 4 still-running checks, with no failures.
The benchmark timings remain author-reported whole-query results. I did not rerun local compilation, product tests or benchmarks.
|
Mertged. Thanks @sunchao |
|
Great job @andygrove ! We're really getting there! |
4 participants
Which issue does this PR close?
Closes #5635.
Rationale for this change
Iceberg exposes its partition transforms as SQL functions:
bucket,truncate,years,months,daysandhours. Spark binds each of them as aStaticInvokeon a per-type class underorg.apache.iceberg.spark.functions. None of those classes were inCometStaticInvoke's allowlist, so any plan that mentioned one of these functions fell back to Spark.The case that matters most is a write to a partitioned table. With the default
write.distribution-mode=hash, Iceberg asks Spark for a shuffle and a local sort keyed on the partition transforms. Both stayed on the JVM, so the native writer from #5361 declined its input for not being aCometNativeExec. The 5.5x number reported in #5361 was only reachable after settingwrite.distribution-mode=none. The same functions also turn up in row-level filters (#5259) and in transformed sort orders (#5339).What changes are included in this PR?
New native kernels in
native/spark-expr/src/iceberg_funcs/, registered as six scalar UDFs.They reproduce Iceberg's Java implementations rather than approximate them. There are three places where the obvious approach gives a different answer:
buckethashes the byte encoding from Appendix B of the Iceberg spec with standard Murmur3. Comet's existingspark_compatible_murmur3_hashcannot be reused: Spark mixes the trailing bytes one at a time and Guava packs them into a word, so the two disagree for every length that is not a multiple of four.truncateuses Java's wrappingintandlongarithmetic. iceberg-rust usesrem_euclid, which gives a different result from Java for widths above 2^30 (filed as Truncate transform diverges from Iceberg Java's TruncateUtil for widths above 2^30, and panics in debug builds at the integer minimum iceberg-rust#3141).yearsandmonthssplit the calendar with integer arithmetic instead ofchrono, whose range stops at year 262143 while Spark'sDATEcovers all ofi32. They also read the raw epoch value instead of using Arrow'sdate_part, which honours the timezone tag on the array; Iceberg is always UTC (filed as Year and Month transforms depend on the input array's timezone tag; Iceberg computes them in UTC iceberg-rust#3142). Those two differences are why these two kernels cannot delegate to iceberg-rust'sYearandMonth.Dictionary-encoded input is transformed once per distinct value and expanded through the keys.
On the Scala side,
serde/icebergFunctions.scalakeys its handlers on the fully qualified class name, because Iceberg is not on Comet's compile classpath.CometStaticInvokefalls through to that map and delegatesgetSupportLevelandgetUnsupportedReasonsto it. AnumBucketsorwidthargument that is not a positive integer literal reportsUnsupported, sobucket(0, x)still raises Iceberg's ownArithmeticException. Fallback reasons now name the declaring class, since every Iceberg function is calledinvoke.truncateon a decimal stays with Spark. Truncating a negative decimal grows its magnitude, so the result can need one more digit than the column's precision allows:truncate(10, v)on adecimal(18,4)value of-99999999999999.9999is-100000000000000.0000, which has 19 digits. Iceberg'sTruncateDecimal.invokehands that oversizedDecimalback unchanged, and Spark turns it into null only when the row is materialized. An ArrowDecimal128(p, s)array has no encoding for it, so a native kernel would have to null it during evaluation, and that changes what an enclosing predicate or hash sees. Decimaltruncatetherefore reportsUnsupported. Every othertruncateinput type, andbucketon decimals, runs natively.How are these changes tested?
CometIcebergSystemFunctionSuiteruns every function over every supported type with Comet on and off, so the reference values are Iceberg's own JVM classes evaluated by Spark. The corpus is seeded random data with nulls plus per-type boundary values: numeric extremes,decimal(38,10)extremes, the epoch and the microsecond before it, and surrogate pairs. The temporal comparisons are repeated under three session timezones. The suite also covers filters, a native sort, a native hash shuffle, the decimal fallback, and an end-to-endINSERTwith the default distribution mode. That last test asserts a native shuffle, aCometIcebergWriteExec, a row round trip, and Iceberg's own partition count matching what the JVM transforms compute; it partitions on the multi-byte string column, so it also exercises the partition-path escaping that #5651 picked up.iceberg_rust_transform_parityiniceberg_write.rspins each kernel against iceberg-rust'screate_transform_function. This matters because a partitioned write uses both: the sort key comes from these kernels and the partition values come from iceberg-rust. The clustered writer requires the two to agree, and when they do not the task fails at runtime withThe input is not sorted!. The parity test turns a future iceberg-rust bump that changes semantics into a build failure instead. The inputs it excludes are the real divergences, documented against apache/iceberg-rust#3141 and #3142.The Rust unit tests pin Appendix B's hash vectors, the minimal encoding produced by
BigInteger.toByteArray(), and Java's wrapping edge cases. Expected values at the domain extremes were produced by running Iceberg'sDateTimeUtilon a JDK 17 JVM. A separate test checks the integer calendar split againstchronoeverywherechronocan represent the date.Benchmarks
There are two, and they measure different things.
native/spark-expr/benches/iceberg_transforms.rsis a criterion benchmark of the kernels on their own: every transform over one array per supported type, with and without nulls, plus the dictionary-encoded string shape a Parquet scan produces for a partition column. This is the measurement that isolates the transform. It is what caught the dictionary path being slower than the plain path (bucket72.9µs to 3.0µs,truncate140.9µs to 35.4µs over 8192 rows with 8 distinct strings).CometIcebergSystemFunctionBenchmarkis the query-level comparison. With Comet off the transform runs through Iceberg's own JVM class, because Spark binds it as aStaticInvokeand codegen calls it once per row. Turning Comet on also replaces the Parquet scan and the projection with native operators, so each ratio below is a scan-plus-projection result and is not a measurement of the transform on its own.Every case runs over a column with no nulls and a column with one null in eight. Before a case is timed, both engines run it over the same corpus and the rows are compared one by one; all 40 cases matched. The benchmark also warns if the Comet plan is not fully native, or if the optimizer folded the expression away; neither warning fired.
Apple M3 Max, JDK 17, Spark 4.1, release native build, 1,048,576 rows, best of five, shown as
no-null / with-null:The executed plans and the full results file from that run are attached in a comment below.
Scaffolded with the
implement-comet-expressionandwire-datafusion-functionskills; the upstream check found nodatafusion-sparkimplementation of these transforms. Written with LLM assistance (Claude Code).