grorge123 commented on PR #5526:
URL:
https://github.com/apache/datafusion-comet/pull/5526#issuecomment-5970568593
@sunchao thank you. You're right that the eager-argument divergence is newly
reachable here even though the native evaluation behind it is on main, so I've
fixed it in this PR. The branch is rebased onto `93d1189be`, and the fix is in
the new last commit.
The rule does not try to model which arguments Spark skips. Several attempts
at that kept missing cases: a divisor evaluated first, `ln` serializing its
argument twice, and kernels that serialization builds itself. Instead, it
relies on one fact about main. Main's dispatcher refused every NullType result,
so an operator whose native plan runs such a kernel fell back to Spark, and so
did the native operators above it up to a shuffle. An operator now falls back
when both of these hold:
- it evaluates a stateful expression (any Catalyst `Nondeterministic` node
such as `monotonically_increasing_id`, `rand` or `shuffle`, or a
non-deterministic Scala UDF, invoked method or V2 function, whatever its
arguments) that the dispatcher does not run as a whole; and
- its native plan runs a dispatcher kernel that computes a NullType value,
either as its result or anywhere in the tree it runs. A kernel whose payload
cannot be read counts as one.
For a kernel with a NullType result, this puts the operator back where main
ran it, in Spark. Checking the tree inside a kernel goes further than main did,
so a few shapes that main ran natively now fall back too (listed under the
costs below). The kernels are read from the serialized plans. That covers:
- the operator's own native block;
- the plan below broadcast, union and coalesce sinks, which exist only over
native plans;
- TakeOrderedAndProject's sort keys and projection, which it serializes when
it runs;
- the partitioning keys of a shuffle below the operator.
A column the dispatcher runs whole (`map(monotonically_increasing_id(),
NULL)`) stays native, because its kernel evaluates the whole Spark expression,
with its own state, row by row. A tree that serialization rebuilds (a decimal
promotion) loses that mark and falls back, which only adds fallback.
`spark_partition_id()` and the input-file expressions don't count, because they
hold no row state.
Review also found one non-NullType path of the same kind. Main dispatched a
relabel-only `array<date>` cast, but this PR made it Compatible, which moved a
non-deterministic child onto native evaluation. That shortcut now applies only
to a deterministic child.
Your witness and the other shapes are in
`misc/nulltype_stateful_argument.sql` and `misc/nulltype_stateful_shuffle.sql`
(a seeded `shuffle`, Spark 4.0+), on a one-partition Parquet table. They are
SQL file tests, so each query's answer is compared against Spark, and each
shape that should fall back also checks the reason. The other shapes are:
- a NullType value in another column;
- a NullType value in a lower projection;
- below a broadcast join and below a UNION ALL;
- `DIV`;
- `ln`;
- four TakeOrderedAndProject shapes;
- `DISTRIBUTE BY`;
- a typed kernel holding a NullType value;
- native controls.
`CometCodegenSuite` covers kernels that serialization builds itself (a
folded `map('k', NULL)` literal, a `CheckOverflow` decimal) with the default
optimizer. Locally, I removed each part of the rule in turn, and a fixture
failed every time. Those runs are not committed.
**What the rule costs.** It is conservative, so a few shapes leave native
execution where they produced correct results:
- a typed kernel main already dispatched whole, whose tree holds an
intermediate NullType value, next to a stateful expression;
- a set op over NullType elements, which this PR now dispatches, next to one.
Results stay correct in both. The composition suite's non-deterministic
sweep falls back more often, so its floor of natively executed cases drops from
110 to 60. Most of the cases it loses run a kernel with a NullType result
beside the stateful producer, which main ran in Spark; a few are the
typed-kernel shape above (the sweep's `filter(array(CAST(NULL AS int)), ...)`
producer).
**A correction to my earlier comment.** I wrote that the gate and the
coalescer's bypass share one predicate and that `UtilsSuite` runs the real
appender over every shape. Neither is accurate. There are two matching
predicates, `Utils.hasNullTypeUnderStruct` on Spark types and
`Utils.hasNullDirectlyUnderStruct` on the Arrow schema. `UtilsSuite` appends
only the shapes that coalesce, while the bypassed shapes never reach the
appender, so it cannot tell when an Arrow release fixes the hang.
**Testing**
With the native library built in release mode, on the commit before its last
change (the `shuffle` detection):
- Spark 4.1: `CometSqlFileTestSuite` (613), `CometNullTypeCompositionSuite`
(28), `CometCodegenSuite` (111), `CometNativeCastSuite` (190),
`CometCodegenSourceSuite` (67), `CometExpressionSuite` (175),
`CometArrayExpressionSuite` (70), `CometMapExpressionSuite` (32),
`CometTemporalExpressionSuite` (37), `CometJsonExpressionSuite` (8),
`CometAggregateSuite` (128), `CometExecSuite` (154), `CometJoinSuite` (65),
`CometNativeShuffleSuite` (58), `CometShuffleSuite` (48),
`DisableAQECometShuffleSuite` (48), `CometNativePositionalRoundRobinSuite`
(11), `UtilsSuite` (11), `GenerateDocsSuite` (4), the `datafusion-comet` native
unit tests (590) and the native `CASE` tests, `cargo fmt`, `cargo clippy` and
`spotless:check`.
- Spark 3.5: `CometSqlFileTestSuite` (613), `CometNullTypeCompositionSuite`
(28) and `CometCodegenSuite` (110).
- Spark 3.4: `CometCodegenSuite` (110).
On the final commit: Spark 4.1 `CometSqlFileTestSuite` (614),
`CometNullTypeCompositionSuite` (28), `CometCodegenSuite` (111),
`CometExecSuite` (154) and `CometExpressionSuite` (175); Spark 3.5
`CometSqlFileTestSuite` (614), `CometNullTypeCompositionSuite` (28) and
`CometCodegenSuite` (110); Spark 3.4 `CometCodegenSuite` (110).
All passed. Spark's own SQL suites have not been run locally.
Assisted-by: Claude Code (claude-opus-5-5)
--
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]
