grorge123 commented on PR #5526:
URL:
https://github.com/apache/datafusion-comet/pull/5526#issuecomment-5955478262
@sunchao thank you, you're right. Routing that `coalesce` to the dispatcher
made your two-column query worse than main, so "none is made worse here" in my
previous comment was wrong for it. The branch is rebased onto today's main
(`351facd26`), and the fix is in the last commit:
- Each occurrence of a dispatched expression that contains a Catalyst
`Nondeterministic` node (`monotonically_increasing_id`, `rand`, `uuid`, ...)
now gets its own kernel. Before serializing such a tree, the serde wraps it in
`DispatchOccurrence` with a per-occurrence id, and the dispatcher strips the
wrapper before compiling. The compiled class is still shared, but each
occurrence gets its own kernel instance and state. Your query is now a
regression test (`CometCodegenSuite`, "identical non-deterministic dispatched
expressions keep their own state"), together with two identical
`map(monotonically_increasing_id(), NULL)` columns. Both fail without the
change, over 64 rows in batches of 8.
- While the fix was under review, two more dispatcher problems in the same
area came up. Both are also on main, but this branch makes them reachable
through more shapes, so the dispatcher now refuses them and the operator falls
back to Spark:
- A non-deterministic user function (an `asNondeterministic()` ScalaUDF,
or a non-deterministic `Invoke`, `StaticInvoke` or V2 function) or a reflected
method (`reflect`, whose static target can hold JVM-wide state). Spark keeps
the state in one object that every call shares and advances it row by row.
Kernels that evaluate one expression over a whole batch either copy that object
or consume it column by column. For example, `nextCount('x'), nextCount('y')`
with a stateful Java UDF returned `(0, 0)` where Spark returns `(0, 1)`. These
UDFs ran through the dispatcher on main and now fall back, which
`scala_java_udfs.md` documents.
- A scalar subquery inside a dispatched expression (`my_udf((SELECT max(x)
FROM t))`). The native plan is serialized before the operator starts its
subqueries, so the shipped copy failed with "Subquery ... has not finished". A
subquery beside the dispatched expression is unaffected.
Each refusal has a test that fails without it.
- #6458 landed in the meantime and reconciles native IF branches itself,
naming struct fields after the THEN branch. So I dropped my change that
serialized IF as a one-branch `CASE WHEN`, and the struct branch casts now
apply to `CASE WHEN` and `coalesce` only. The IF fixtures in `if_expr.sql` now
run against the upstream fix.
Smaller changes in the same commit: a `date_trunc` fixture for the
interpreted-format gate, the `hash.sql` NullType witness split into one query
per function, the round-robin NullType rule in the contributor shuffle guides,
and the `IF` / `coalesce` audit entries.
**Known issues, updated**
The first item in my previous list (identical non-deterministic expressions
sharing a kernel) is fixed; the rest are unchanged. One more, also on main:
Spark skips the right operand of a binary expression when the left one is NULL,
but native evaluation computes both, so a stateful right operand advances on
rows Spark skips. `IF(id = 0, NULL, id) + monotonically_increasing_id()` over
ids 0..3 in one partition returns `NULL, 2, 4, 6` where Spark returns `NULL, 1,
3, 5`, and `arrays_overlap` and `array_contains` behave the same way. This
branch adds one more shape that reaches it (an `array<void>` left operand
produced by the dispatcher, which main ran in Spark). Fixing it means changing
how native binary expressions evaluate, which is outside this PR.
**Testing**
With the native library built in release mode, on the commit before its last
two changes (the `reflect` refusal and today's rebase):
- Spark 4.1: `CometSqlFileTestSuite` (608), `CometNullTypeCompositionSuite`
(28), `CometArrayExpressionSuite` (70), `CometExpressionSuite` (175),
`CometTemporalExpressionSuite` (37), `CometMapExpressionSuite` (32),
`CometAggregateSuite` (128), `CometJoinSuite` (65), `CometExecSuite` (154),
`CometJsonExpressionSuite` (8), `CometCodegenSuite` (108),
`CometCodegenSourceSuite` (67), `CometShuffleSuite` (48),
`DisableAQECometShuffleSuite` (48), `CometNativeShuffleSuite` (58),
`CometNativePositionalRoundRobinSuite` (11), `CometNativeCastSuite` (187),
`UtilsSuite` (11), `GenerateDocsSuite` (4), the native unit tests for list
literals and `CASE`, `cargo clippy`, `cargo fmt` and `spotless:check`.
- Spark 3.5: `CometSqlFileTestSuite` (608), `CometNullTypeCompositionSuite`
(28), `CometArrayExpressionSuite` (69), `CometExpressionSuite` (170),
`CometAggregateSuite` (126), `CometCodegenSuite` (107) and
`CometNativeShuffleSuite` (52).
- Spark 3.4: `CometCodegenSuite` (107).
After the `reflect` refusal: Spark 4.1 `CometCodegenSuite` (108),
`CometCodegenSourceSuite` (67), `CometNullTypeCompositionSuite` (28) and
`CometSqlFileTestSuite` (608), and `CometCodegenSuite` on Spark 3.5 and 3.4
(107 each). Then on the final commit, rebased onto `351facd26`: Spark 4.1
`CometCodegenSuite` (108), `CometSqlFileTestSuite` (610),
`CometNativeCastSuite` (190), `CometNullTypeCompositionSuite` (28),
`GenerateDocsSuite` (4) and `spotless:check`, and Spark 3.5 `CometCodegenSuite`
(107).
All passed, and each new test fails with its fix reverted. CI on the
previous head is still waiting for workflow approval.
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]
