andygrove opened a new pull request, #5638: URL: https://github.com/apache/datafusion-comet/pull/5638
## Which issue does this PR close? Closes #5635. ## Rationale for this change Iceberg's system functions (`bucket`, `truncate`, `years`, `months`, `days`, `hours`) bind as `StaticInvoke` calls on classes under `org.apache.iceberg.spark.functions`, and none of them were in `CometStaticInvoke`'s allowlist. Any plan that mentions one fell back to Spark. The case that matters most is a partitioned Iceberg table written with the default `write.distribution-mode` (`hash`): Spark plans a shuffle and a local sort keyed on the partition transforms, the shuffle stayed on the JVM, and the native writer from #5361 declined because its input was not a `CometNativeExec`. The 5.5x partitioned-write number in #5361 was only reachable after setting `write.distribution-mode=none` on the table. The same expressions show up in row-level filters and projections against hidden partitioning (Iceberg's `TestSystemFunctionPushDownInRowLevelOperations`, bucket 2 of #5259) and in transformed sort orders (#5339). ## What changes are included in this PR? Native kernels in a new `native/spark-expr/src/iceberg_funcs/` module, registered as registry UDFs (`iceberg_bucket`, `iceberg_truncate`, `iceberg_years`, `iceberg_months`, `iceberg_days`, `iceberg_hours`). They reproduce Iceberg's Java implementations rather than approximating them: - `bucket` hashes the spec's byte encoding (8-byte little-endian for int/long/date/timestamp, UTF-8 for strings, raw bytes for binary, minimal big-endian two's complement of the unscaled value for decimals) with standard 32-bit Murmur3 (seed 0) and masks the sign bit before the modulus. Comet's Spark-compatible murmur3 could not be reused: Spark mixes tail bytes one at a time, Guava packs them into one word, so the two disagree on every length that is not a multiple of four. - `truncate` uses Java's wrapping `int`/`long` arithmetic (including the `(v % w) + w` overflow for widths above 2^30 and the narrowing casts for tinyint/smallint), keeps decimal precision and scale, counts code points for strings, and bytes for binary. A negative decimal whose truncated value no longer fits the column's precision becomes null, matching what Spark's `UnsafeRowWriter` does with the JVM result. - `years`/`months`/`days`/`hours` work on the raw epoch values in UTC regardless of the session timezone (Iceberg's `DateTimeUtil` ignores it), floor before the epoch, and `days` returns a date to match `DaysFunction.resultType()`. They do not go through Arrow's `date_part`, which honors the array's timezone tag; iceberg-rust's transform kernels do, which is why those were not reused either. Scala side, a new `serde/icebergFunctions.scala` keys handlers on the fully qualified Iceberg class name (Iceberg is not on Comet's compile classpath), and `CometStaticInvoke` falls through to that map after its existing `(functionName, Class)` lookup and delegates `getSupportLevel` to the matched handler. A `numBuckets`/`width` that is not a positive integer literal reports `Unsupported`, so `bucket(0, x)` keeps raising Iceberg's own `ArithmeticException` on the JVM. The fallback reason for an unlisted static invoke now names the declaring class, since every Iceberg function is called `invoke`. Docs: a new "Iceberg system functions" section in the Iceberg user guide and a note in the writes guide. The new suite is added to both PR build workflows. The `implement-comet-expression` and `wire-datafusion-function` skills were used to scaffold the work; the upstream check found no `datafusion-spark` implementation of these transforms. ## How are these changes tested? - Rust unit tests pin the hash values from Appendix B of the Iceberg spec, `BigInteger.toByteArray()`'s minimal encoding for decimals, the spec's truncate examples plus Java's wrapping edge cases, and the temporal boundaries on both sides of the epoch under several timezone tags. - `CometIcebergSystemFunctionSuite` runs each function over every supported type with Comet on and off, so the reference values come from Iceberg's own `BucketFunction` / `TruncateFunction` / `DateTimeUtil` evaluated by Spark, over seeded random data with nulls plus the boundary values of each type (numeric extremes, `decimal(38,10)` extremes, the epoch and the microsecond before it, empty and multi-byte strings, surrogate pairs). The temporal comparison repeats under three session timezones. Further tests cover the functions in filters, in a native sort, in a native hash shuffle, and an end-to-end `INSERT` into a table partitioned by `bucket`/`truncate`/`days`/`months` with the default distribution mode, asserting a native shuffle, a `CometIcebergWriteExec`, a row round trip, and that Iceberg's partition count agrees with the JVM transforms. The decimal precision-overflow null case and an offset overflow in Arrow's `substring` kernel for a width of `Integer.MAX_VALUE` were both f ound by this comparison. - Support-level and fallback-message unit tests for the serde. -- 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]
