apache / apache/datafusion-comet
[EPIC] Codegen-dispatch coverage audit: expressions that fall back to Spark where the JVM dispatcher would work
- Dominant language
- Scala
- Stars
- 1.3k
- Forks
- 373
- Avg merge
- 2d 4h
- Merged PRs (30d)
- 198
Description
## What / Why
Comet has a JVM codegen dispatcher (`CometScalaUDF.emitJvmCodegenDispatch`, `spark/src/main/scala/org/apache/comet/serde/CometScalaUDF.scala:70`) that compiles a Spark expression's own `doGenCode` output into a per-batch kernel reading and writing Arrow vectors directly. When a serde has no native path for some input, routing through the dispatcher keeps the whole operator inside Comet and still matches Spark byte for byte, instead of failing the enclosing projection back to Spark.
A serde opts in by mixing in `CodegenDispatchFallback` (`spark/src/main/scala/org/apache/comet/serde/CometExpressionSerde.scala:125`). `QueryPlanSerde` then tries the dispatcher for that serde's `Unsupported` and non-opt-in `Incompatible` results (`QueryPlanSerde.scala:941` and `:968`) before giving up.
Adoption is uneven. I swept all 261 expression serdes under `spark/src/main/scala/org/apache/comet/serde/` plus the version shims. 98 reach the dispatcher; 63 decline at least one case without one. This EPIC collects the cases where the mixin is applicable and worth adding, and — just as importantly — records the cases where it is *not* applicable, so they aren't re-litigated.
Several of the gaps are plain asymmetries against an already-dispatched sibling: `bround` dispatches but `round` on a double falls back; `to_unix_timestamp` dispatches but `unix_timestamp` on a string falls back; `to_json`, `from_csv` and `schema_of_csv` dispatch but `to_csv` never runs in Comet by default at all.
## What gates a dispatch
`CometBatchKernelCodegen.canHandle` (`spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegen.scala:119`) rejects `AggregateFunction`, `Generator`, `Unevaluable`, and any bound reference or output type outside `isSupportedDataType` (`:85`) — notably `NullType`, `ObjectType` and Variant. Everything else is admitted, including `CodegenFallback`, nondeterministic and stateful expressions, `HigherOrderFunction`, and subquery expressions.
Every item below was checked against that gate.
## Prerequisites
Both should land before the catch-all in the `StaticInvoke` item below.
- #5573 — guard the closure-serialize step in `emitJvmCodegenDispatch` so a non-serializable tree falls back cleanly instead of throwing at plan time
- #5574 — the dispatcher is unreachable from `convert`, so ~15 serdes that decline there never get a dispatch attempt (two of them already carry the mixin and silently lose it). Being addressed as a series of small per-serde moves (#5595, #5599). The one-line `.orElse` variant was prototyped and measured, and is **unsound**: rescuing at the failing node removes a fallback that native *ancestors* depended on for Spark compatibility, producing wrong answers. See the measurement on that issue.
## High value
Each of these is a small change, and most close a documented asymmetry against an already-dispatched sibling.
- #5575 — route unrecognized `StaticInvoke` and `Invoke` through the dispatcher. The widest-reaching item: it is a catch-all, not one expression. `docs/source/user-guide/latest/expressions.md` already records `encode` and non-hex `to_binary` falling back for exactly this reason.
- #5576 — `round` on float/double falls back, while `bround` already dispatches. Probably the highest-traffic single fallback in the sweep.
- #5577 — `unix_timestamp` on string input falls back, while `to_unix_timestamp` already dispatches. The string form is the common spelling.
- #5578 — `to_csv` never runs inside Comet by default at all: every path returns `Unsupported` or `Incompatible`. `to_json`, `from_csv` and `schema_of_csv` all dispatch. Also a docs bug, since the Implementation column claims "Native".
- #5579 — `lpad` / `rpad` with a non-literal `pad` argument. `lpad(name, 10, pad_col)` is an ordinary query shape.
- #5580 — map lookups with float, collated or complex keys (`map_col[key]`, `element_at`). The declines are correct analysis, which is exactly why the dispatcher rather than a native fix is the answer.
- #5581 — hash functions on decimal precision > 18, plus `sha2` with a non-literal `numBits`. Hits bucketing, partitioning and dedup paths.
## Worthwhile
- #5582 — array functions on binary and struct element types, via the shared `ArraysBase` type gate (seven serdes)
- #5583 — `arrays_zip` on map element types
- #5584 — `length` / `bit_length` / `octet_length` on binary input
- #5585 — `translate` falls back by default rather than dispatching like every comparable string function
- #5586 — `named_struct` with duplicate field names
- #5587 — `abs` on interval types
- #5588 — `timestamp_seconds` on decimal, byte and short input
- #5589 — `map_from_arrays` under `mapKeyDedupPolicy=LAST_WIN`, unlike `map_from_entries`
- #5590 — `map_sort` on non-scalar map key types
- #5591 — `next_day` and `levenshtein` on collated strings
## Related, but operator level rather than serde level
Not part of the original 261-serde sweep, and not a mixin change. Filed here because it is the same dispatcher and it came directly out of the #5575 work.
- #5710 — the `SerializeFromObject` / `MapElements` / `DeserializeToObject` sandwich that `ds.map(f)` produces falls back whole, and takes any aggregate or exchange above it down with it. No mixin can fix that: `ObjectType` is on one side of each operator, so neither is Arrow-representable in isolation. But `canHandle` only type-checks the root and the bound references, so the *fused* sandwich passes — which makes this a plan rewrite into `ProjectExec` that reaches the existing dispatcher with no proto or native change. This is the follow-up to the encoder and deserializer trees that #5575 deliberately scoped out.
## Considered and rejected — structurally impossible
These are recorded so the sweep doesn't get repeated on them:
- **All 25 aggregate serdes.** Blocked twice over: `canHandle` rejects `AggregateFunction` outright, and `CodegenDispatchFallback`'s self-type is `CometExpressionSerde[_]`, so it cannot be mixed into `CometAggregateExpressionSerde` at all. This covers `percentile` with an array of percentages, `approx_percentile` on non-numeric input, `collect_list` / `collect_set`, `bloom_filter_agg`, `avg` / `sum` on intervals, and the `bit_and` / `bit_or` / `bit_xor` family.
- **`CometHours` / `CometDays`** (`serde/datetime.scala:813`, `:850`). Verified against the Spark 4.0.1 bytecode: `Hours` and `Days` extend `PartitionTransformExpression`, which implements `Unevaluable`, so `canHandle` rejects them by construction.
- **`CometLiteral`, `CometAttributeReference`, `CometKnownFloatingPointNormalized`.** What they decline is exactly the set of types that cannot cross the Arrow FFI boundary — the same set `isSupportedDataType` rejects. The dispatcher would decline them again.
- **Null-element array cases.** The "null elements fall back" half of the `array_position` / `flatten` / `shuffle` notes in the compatibility guide is not dispatchable: `NullType` is absent from `isSupportedDataType`. Only the binary/struct half of those notes is actionable.
- **`CometSortOrder`.** Sort keys are ordering specifications consumed by the native Sort and Window operators, not value expressions, so there is nothing to dispatch.
- **`CometScalarFunction`'s ANSI guard** (`serde/CometScalarFunction.scala:30`). A developer mis-wiring check, not a runtime input case.
## Considered and deferred — marginal
Not filed, but recorded so the reasoning survives: negative-scale decimal in `CometCeil` / `CometFloor` / `CometRound` (only reachable with `spark.sql.legacy.allowNegativeScaleOfDecimal=true`); `CometToPrettyString` (only on the `df.show()` path); non-literal seeds in `CometRandStr` / `CometRand` / `CometRandn`; the deliberate all-foldable declines in `CometConcatWs` and `CometArrayPosition`, which are a handoff to `ConstantFolding` and should stay; and the unreachable sanity checks in `CometDivide`, `CometCheckOverflow`, `CometMakeDecimal`, `CometSize`, `CometPreciseTimestampConversion`, `CometLeft` and `CometRight`.
## Suggested sequencing
1. The two prerequisites. The closure-serialize guard in particular blocks the `StaticInvoke` catch-all.
2. The high-value items. Each is a small change and each closes a documented asymmetry against an already-dispatched sibling, which makes them easy to review.
3. The rest, in any order — they are independent and mostly good first issues.
The per-expression items below are all mixin changes: they report `Unsupported` from `getSupportLevel`, which routes through `dispatchIfFallback` *before* any ancestor has converted. That ordering is what makes them safe, and it is exactly what the rejected `.orElse` variant on #5574 gave up. None of them is blocked or made redundant by that issue.
## Verification note
This sweep is static analysis of the serde definitions, the dispatcher's `canHandle` gate, and the generated `docs/source/user-guide/latest/expressions.md`. Nothing here has been reproduced against a running cluster yet, so each issue should confirm the fallback with a test before the fix lands.
Contributor guide
Research direction
Start with CometBatchKernelCodegen.scala, CometExpressionSerde.scala, QueryPlanSerde.scala, and CometScalaUDF.scala to understand the dispatcher gate and fallback path. Follow the suggested prerequisites and high-value child issues, confirming each fallback with a test before changing the relevant serde mixin. Done means applicable cases are dispatched safely, documentation is corrected where noted, and structurally impossible or deferred cases remain recorded.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- scala
- Domain
- backend, compilers, testing-qa
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100