Repository navigation
fix: send a skipped shared subexpression to Spark where Spark's sharing depends on the plan - #7586
Draft
dwsmith1983 wants to merge 114 commits into
Draft
dwsmith1983 wants to merge 114 commits into
dwsmith1983 wants to merge 114 commits into
Conversation
…ull-guarded children Map lookups with float, collated or complex keys were declined by the serde without a dispatch fallback, so the whole projection fell back to Spark. Four array and map serdes reproduce NULL propagation with a guard that serializes the child twice, so a stateful child was evaluated twice natively and returned wrong answers. Mix the codegen dispatch fallback into the map lookups, and decline any nondeterministic child in size, array_append, arrays_zip and map_from_arrays through one shared gate so Spark's generated code evaluates it once. Closes apache#5580 Closes apache#5781
…d-nondeterministic-children
…in the dispatched fixtures
…e guard filters its evaluation
…e arrays_zip types With ANSI on, an item that raises on a row whose array is NULL raises in Spark but not on the native path, because the NULL guard skips the item there. The serde now reports that case as incompatible when the array is nullable, so it runs through the JVM codegen dispatcher by default and the native guard stays behind allowIncompatible. The divergence is recorded in the compatibility guide with its tracking issue, and a fixture pins the raise, the dispatch and the native path for a non-nullable array. The shared nondeterministic reason is now a full sentence for the generated guide, the arrays_zip type bullet names the element types the native kernel declines, and the arrays_zip fixture gets its comments in the right order plus a dispatched query over an array column.
…d-nondeterministic-children
An unsuffixed -0.0 is a decimal literal with no signed zero, so CAST(-0.0 AS DOUBLE) looked up +0.0 and found the key on the native path too. The lookups now use -0.0D and CAST(-0.0D AS FLOAT), which keep the sign, so the fixtures fail on results when the float key decline is missing. The contributor guide names the literal form as well.
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…ees bind alike Two arrays_zip calls over different columns bind to the same ordinals, so without a per-occurrence tag they shared one kernel and the second monotonically_increasing_id carried on from the first.
…d-nondeterministic-children
…d-nondeterministic-children # Conflicts: # docs/source/user-guide/latest/expressions.md # spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala # spark/src/main/scala/org/apache/comet/serde/maps.scala # spark/src/test/resources/sql-tests/expressions/map/element_at_map.sql # spark/src/test/resources/sql-tests/expressions/map/get_map_value.sql # spark/src/test/resources/sql-tests/expressions/map/map_from_arrays_dedup_policy.sql # spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…lacement can be NULL The generated code of a three-argument array_join evaluates the replacement first and skips the array and delimiter when it is NULL. The default rule only treated children after the first as skippable, so a subexpression shared through the array was missed: a dispatched lookup could return NULL where Spark, which evaluates the shared subexpression for every row, raises.
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…the dispatch gates With the item now evaluated natively on rows whose array is NULL, the array_append item gate is dropped. array_append keeps the rule of declining a nondeterministic operand only over a nullable array, and that case now goes to the JVM codegen dispatcher. The calendar-interval and shared-subexpression declines move into one GatedCodegenDispatchFallback trait that the six dispatching serdes mix in. A shared subexpression only counts in the dividend of a division or pmod, since Spark always evaluates the divisor. Fixtures fold the duplicated array_append files into array_append_eager_eval.sql, replace the extra nonzero tables with a divisor that cannot be zero, and move the folded-literal dispatch tests from CometMapExpressionSuite to SQL fixtures.
…d-nondeterministic-children
…ures Drop comments that restate the code, a query or its expectation, and shorten the rest to the Spark codegen behavior and fixture values they explain.
…d-nondeterministic-children
…d-nondeterministic-children # Conflicts: # spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala # spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala
The serde wraps decimal arithmetic in CheckOverflow before the dispatcher sees it, so the decline only recognizes a shared decimal key when both sides are promoted. Without that, Comet returned rows where Spark raises DIVIDE_BY_ZERO on a NULL map.
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…ng depends on the plan Whether Spark evaluates a shared subexpression for every row depends on the operator. A projection, an aggregate's function inputs, shuffle keys, the TopK projection and a sort's first key share it as the setting says. A filter, a join, an aggregate's results and other operators share it only in some Spark versions or without whole-stage codegen. In those operators a NULL-guarded lookup whose skipped argument holds a shared subexpression now falls back to Spark instead of evaluating it for every row, which could raise where Spark returns rows.
This was referenced Nov 10, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Part of #6970. This is the first of three pull requests for it.
These build on #5867, which adds the shared-subexpression check they extend, so #5867 merges first. Order after it:
They target
main, so until #5867 merges each one also shows its commits, and the ones before it.Rationale for this change
#5867 tells the serdes which subexpressions an operator shares, so a native NULL guard does not skip one that Spark evaluates for every row. It treats every operator like a projection. Spark shares that way only in a projection, an aggregate's function inputs, shuffle keys, the TopK projection and the prefix of a sort's first key. In a filter, a join, an aggregate's results and the other operators, whether Spark shares depends on the version (4.2 added it to
FilterExec), on whole-stage codegen and on the join side. Evaluating a shared argument for every row there can raiseDIVIDE_BY_ZEROon a row where Spark returns a result.What changes are included in this PR?
withSharedSubexpressionstakes a requiredcertainflag. Projections, aggregate function inputs, shuffle keys, the TopK projection and a sort's first key are certain. Other operators and aggregate results are not.withNullShortCircuitcallers andGetArrayItemfall back to Spark when an argument they would skip holds a shared subexpression, instead of evaluating it for every row. Each lists the reason.How are these changes tested?
map_null_short_circuit_uncertain.sqlcovers a filter, a join condition and join keys, sorts, and an aggregate's inputs and results. Each new query fails on #5867's code or when the change behind it is undone. The join checks turn adaptive execution off because of #6442. A unit test covers the flag. The ORDER BY queries inmap_lookup_shared_subexpression_operators.sqlnow dispatch their lookups instead of falling back. The suites and TPC-DS plan stability pass on Spark 3.5 and 4.1, with no plan changes, and the new file passes on 4.2.