andygrove opened a new issue, #5635:
URL: https://github.com/apache/datafusion-comet/issues/5635
### What is the problem the feature request solves?
Now that the native Iceberg writer has landed (#5361), it declines the case
most users will
actually hit: a partitioned table written with Iceberg's default
distribution mode.
`CometIcebergNativeWrite` sets `requiresNativeChildren = true`
(`spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala:43`),
so the
entire sub-plan feeding `IcebergWrite` has to be a `CometNativeExec`. For a
partitioned table with
no explicit `write.distribution-mode`, Iceberg's
`SparkWriteConf.defaultWriteDistributionMode()`
returns `HASH`, so Spark plans a `ShuffleExchangeExec` whose
`HashPartitioning` keys are the
partition transform expressions. Comet's native shuffle gate requires every
one of those to
serialize:
```scala
case HashPartitioning(expressions, _) =>
...
for (expr <- expressions) {
if (QueryPlanSerde.exprToProto(expr, inputs).isEmpty) {
reasons += s"unsupported hash partitioning expression: $expr"
}
}
```
(`CometShuffleExchangeExec.scala:473-484`, and again at `:610` for the
columnar-shuffle gate.)
Iceberg's system functions expose Spark's static magic method, so
`V2ExpressionUtils.resolveScalarFunction` binds them as `StaticInvoke`
rather than
`ApplyFunctionExpression`:
```
$ javap 'org.apache.iceberg.spark.functions.BucketFunction$BucketInt'
public static int invoke(int, int);
$ javap 'org.apache.iceberg.spark.functions.TruncateFunction$TruncateString'
public static org.apache.spark.unsafe.types.UTF8String invoke(int,
org.apache.spark.unsafe.types.UTF8String);
$ javap 'org.apache.iceberg.spark.functions.DaysFunction$DateToDaysFunction'
public static int invoke(int);
```
`CometStaticInvoke` dispatches on `(functionName, staticObject)` against a
fixed allowlist
(`spark/src/main/scala/org/apache/comet/serde/statics.scala:35-52`), and
none of the Iceberg
classes are in it. So `exprToProto` returns `None`, the exchange stays on
the JVM, the write's
child is not a `CometNativeExec`, and the native write declines with a
fall-back reason.
The benchmark in #5361 says as much: the 5.5x partitioned number was
measured with
`write.distribution-mode=none` plus `write.spark.fanout.enabled=true`,
chosen so all three
scenarios shared one plan shape. Under stock table settings a partitioned
append gets no
acceleration at all, so the headline win is currently reachable only by
reconfiguring the table.
The same gap costs us on the read and DML side. Iceberg's own
`TestSystemFunctionPushDownInRowLevelOperations` exercises `bucket`,
`truncate`, `years`,
`months`, `days`, and `hours` in copy-on-write `DELETE` / `UPDATE` / `MERGE`
predicates (it shows
up as bucket 2 in #5259), and #5339 wants the same transforms for reporting
Iceberg sort orders.
### Describe the potential solution
Implement the transforms natively and register them in
`CometStaticInvoke.staticInvokeExpressions`, keyed on the Iceberg function
classes. That is the
narrowest hook that fixes shuffle, filter, projection, and sort in one go,
since everything routes
through `exprToProto`.
The classes to cover, all under `org.apache.iceberg.spark.functions`:
| Function | Implementations |
| --- | --- |
| `BucketFunction` | `BucketInt`, `BucketLong`, `BucketString`,
`BucketBinary`, `BucketDecimal` |
| `TruncateFunction` | `TruncateTinyInt`, `TruncateSmallInt`, `TruncateInt`,
`TruncateBigInt`, `TruncateString`, `TruncateBinary`, `TruncateDecimal` |
| `YearsFunction` | `DateToYearsFunction`, `TimestampToYearsFunction`,
`TimestampNtzToYearsFunction` |
| `MonthsFunction` | `DateToMonthsFunction`, `TimestampToMonthsFunction`,
`TimestampNtzToMonthsFunction` |
| `DaysFunction` | `DateToDaysFunction`, `TimestampToDaysFunction`,
`TimestampNtzToDaysFunction` |
| `HoursFunction` | `TimestampToHoursFunction`,
`TimestampNtzToHoursFunction` |
`IcebergVersionFunction` is a zero-arg constant and not worth native support.
Semantics have to match Iceberg's spec exactly, not approximately:
- **bucket** is `(murmur3_32_x86(value) & Integer.MAX_VALUE) % numBuckets`,
over the byte encoding
in Appendix B of the Iceberg spec — 8-byte little-endian for int/long/date
(days)/timestamp
(micros), UTF-8 bytes for string, raw bytes for binary/fixed, minimal
big-endian two's-complement
unscaled bytes for decimal, 16-byte big-endian for UUID. Note that Iceberg
maps `uuid` to Spark's
`StringType`, which is the same mapping the write eligibility gate already
reasons about.
- **truncate** is `v - ((v % W) + W) % W` for integrals, the same on the
unscaled value for
decimals, a byte prefix for binary, and `UTF8String.substring(0, W)` for
strings — code points,
not bytes and not UTF-16 units.
- **years / months** are `ChronoUnit.between(EPOCH, value)`, so they are
calendar-aware and
relative to 1970 and go negative before it; **days / hours** are plain
floor division on the
epoch value.
- Spark binds these with `propagateNull = false`, so null handling belongs
to the implementation
rather than to the `StaticInvoke` wrapper. `BucketString.invoke` returns a
boxed `Integer`
precisely so it can return null.
Some of these may already exist in DataFusion or `datafusion-spark` in a
compatible form; worth
checking before writing Rust, per the `wire-datafusion-function` skill.
On testing: divergence here mostly fails loudly rather than silently, but
not always, so it is
worth being explicit about which is which. If Comet's bucket disagrees with
Iceberg's, rows of one
Iceberg partition get split across reducers, and the clustered (non-fanout)
writer rejects
out-of-order partitions — a task failure, not corruption. The silent case is
these functions in a
filter or projection, where a wrong hash is just a wrong answer. That argues
for per-type fuzz
comparison against Iceberg's own `Transforms` / `BucketFunction`
implementations rather than
hand-picked cases, which is the same bar #5339 sets for transformed sort
orders.
### Additional context
Small related diagnostic gap: the fall-back reason for an unlisted static
invoke is
`s"Static invoke expression: ${expr.functionName} is not supported"`
(`statics.scala:62-64`).
Every Iceberg system function has `functionName == "invoke"`, so a user
hitting this sees
`Static invoke expression: invoke is not supported` with nothing pointing at
Iceberg. Including
`expr.staticObject` in the message would make this self-diagnosing.
Related:
- #5361 — the native Iceberg writer, whose partitioned benchmark had to
disable hash distribution
- #5259 — split-operator CI failures, whose bucket 2 is Iceberg's
system-function pushdown tests
- #5339 — transformed sort orders (bucket / truncate / etc.) for Iceberg
reads
- #5121 — accelerate DataSource V2 writes
--
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]
