fix: honor subexpression elimination off in the native null short-circuit

[dwsmith1983]

Which issue does this PR close?

Closes #6972.

Rationale for this change

canShortCircuitNulls keeps every argument of an expression eager when a later argument holds a subexpression the operator's expressions share, so an error Spark raises from the hoisted subexpression is raised in Comet too. It treated them as shared whatever spark.sql.subexpressionElimination.enabled says. With the setting off, Spark's projection hoists nothing, and its own null short-circuit skips the later argument on a row whose first argument is NULL, so Comet raised where Spark returns rows. SELECT m1[1 div i], m2[1 div i] with ANSI on, the setting off, NULL maps and i = 0 is one example.

Honoring the setting everywhere would be wrong the other way. Spark's whole-stage codegen aggregates share the subexpressions of their update expressions whatever the setting says (AggregateCodegenSupport.doConsumeWithoutKeys and the keyed paths in HashAggregateExec call subexpressionEliminationForWholeStageCodegen without checking it, on 3.5 through 4.1), while ProjectExec's codegen and the projections behind shuffle partition keys and TakeOrderedAndProject only do so when it is on.

What changes are included in this PR?

• withSharedSubexpressions takes an alwaysShared flag, default false. With the setting off the operator shares nothing unless it asked for alwaysShared. canShortCircuitNulls itself is unchanged.
• CometExecRule passes alwaysShared for aggregates (BaseAggregateExec). The native shuffle writer and TakeOrderedAndProject keep the default, matching Spark's UnsafeProjection.create.
• With the setting on nothing changes.

An aggregate that Spark runs without whole-stage codegen (object hash aggregate, sort aggregate without codegen, or codegen switched off) does not share with the setting off, but Comet still treats it as sharing, so it stays eager there as before: at worst it raises where Spark would not, never a different result.

#5867 changes the same part of QueryPlanSerde, so whichever lands second needs a small merge. Its decline for dispatched expressions reads the same shared set, so after both land a projection with the setting off stays dispatched again (Spark shares nothing there) while aggregates still decline; the three projection cases in its fixtures that run with the setting off then expect dispatch instead of a fallback.

How are these changes tested?

• QueryPlanSerdeSuite checks canShortCircuitNulls with the setting off, with and without alwaysShared, and outside any operator. The existing default-conf test covers the setting on, where alwaysShared changes nothing.
• New array_null_short_circuit_cse_off.sql and map_null_short_circuit_cse_off.sql run with ANSI on and the setting off. Projection, DISTRIBUTE BY, ORDER BY and TakeOrderedAndProject pairs that share a raising cast or division over NULL inputs run natively and return Spark's rows; each raised in Comet before this change, which also shows the shuffle cases run through the native shuffle writer. Ungrouped and grouped aggregate pairs that share the raising piece across two functions still raise like Spark, with the error class checked so a fallback to Spark cannot pass them, and an aggregate that shares the whole call (which skips the raising piece on its own) stays native. Both files pin whole-stage codegen, which the aggregate cases depend on.
• Dropping the setting check, leaving aggregates out of alwaysShared, or defaulting the flag to true each fails the cases it should.

Run on Spark 4.1 (the array and map SQL fixtures, QueryPlanSerdeSuite, CometMapExpressionSuite, CometArrayExpressionSuite, CometNativeShuffleSuite, CometTopKSuite), Spark 3.5 and 4.0 (the null short-circuit fixtures and QueryPlanSerdeSuite) and the strict-warnings compile.

[sunchao]

Reviewed the full five-file diff from 7d8d9b1544b734c85a93cca2cc9d5380ad70accd to 3ca36ab7b3cfb7cb332c7a5376600c0dbd141f52. The PR is not a draft. No introduced P1/P2 issues found within this review.

Summary

• Prior state and problem: Shared subexpressions prevented native null short-circuiting even with spark.sql.subexpressionElimination.enabled=false. Consequently, expressions over null arrays or maps could raise errors that Spark skips.
• Design approach: Make the shared-expression set respect Spark’s setting, with an explicit alwaysShared override for aggregate operators.
• Correctness: Checked Spark’s expression null handling, ProjectExec, UnsafeProjection, AggregateCodegenSupport, and HashAggregateExec across 3.4.3, 3.5.9, 4.0.4, 4.1.3, and 4.2.0. Projections honor the setting, while generated aggregate updates share subexpressions unconditionally. The change preserves that distinction. No introduced P1/P2 correctness issue was identified.
• Compatibility analysis: No configuration default, protobuf layout, or native interface changes. Compared the affected paths with branch-1.1 at 8ac6fccde53d96fec22184be287747eeae481cac, where these native calls evaluate arguments eagerly. Returning Spark’s null results for the repaired cases is an intended compatibility improvement.
• Key design decisions: BaseAggregateExec retains unconditional shared-expression analysis. Serialization outside a known operator scope remains conservative. The documented over-eager behavior of aggregates without codegen predates this change and is not worsened here.
• Implementation sketch: SharedSubexpressions.common returns Nil when elimination is disabled unless overridden. CometExecRule supplies the aggregate override. The added unit test covers configuration and override combinations, while ten SQL queries cover arrays, maps, shuffle keys, top-K projections, and aggregates.
• Performance: The disabled-setting path avoids constructing EquivalentExpressions. Configuration is consulted during serialization, not per row. Native short-circuit machinery is reused. No evidence-backed performance regression was identified, and no throughput benchmark was run.
• Design: The change is focused and keeps the Spark operator distinction at the conversion boundary. Shuffle and top-K callers inherit the appropriate default without separate implementations.
• Abstraction & complexity: One named boolean extends the existing scoped helper. Lazy computation and scope restoration remain intact. No additional abstraction or concrete simplification meeting the reporting bar was identified.
• Behavioral changes worth calling out: With elimination disabled, affected projections and shuffle keys can now return null-containing rows instead of raising from skipped arguments. Enabled-setting behavior and aggregate sharing remain unchanged.
• Suggested improvements: No additional improvement meeting the P1/P2 reporting bar was identified.

Routed skills: review-comet-pr, review-comet-expression-pr, and review-comet-shuffle-pr. Read AGENTS.md, relevant contributor guidance, and the discussion snapshot. The snapshot and refreshed GitHub discussion endpoints contained no existing reviews or comments.

Exact-head CI: The label check passed. Comet CI, CodeQL, and Check PR Title report action_required, awaiting workflow approval. There is no build or test verdict for this head.

Validation: A disposable Scala probe compiled the unchanged policy methods extracted from base and head against Spark 4.1.3. Generated UnsafeProjection execution confirmed that the invalid cast is skipped with elimination disabled and raises with it enabled. Aggregate override, shared whole-call handling, unknown scope, and nested scope restoration checks passed. git diff --check passed.

Validation limits: The reactor attempt to run QueryPlanSerdeSuite stopped before tests because this environment lacks javac. The focused probe does not replace a full Comet build. Native SQL fixtures and Spark’s SQL suite were not executed locally. The checkout remains unchanged.

[comphead]

The fix matches Spark's per-operator behaviour. One comment on the new unit test.

[comphead]

With the setting on, SharedSubexpressions.common is alwaysShared || enabled, so looping alwaysShared over Seq(false, true) adds no case for true. The shared assertion and the Seq(contains) assertion also repeat what canShortCircuitNulls skips an argument only where the shared subexpressions are known (line 106) already checks, since the default conf has elimination on. The new behaviour is the enabled = false block above. Could this block shrink to one alwaysShared = false case, or go?

[dwsmith1983]

Could this block shrink to one alwaysShared = false case, or go?

It went. With elimination on, alwaysShared doesn't change the shared set, and the remaining assertions (a shared cast blocks the skip, the cast alone allows it, unknown outside a scope) are the ones canShortCircuitNulls skips an argument only where the shared subexpressions are known already makes under the default conf. The new test keeps only the enabled = false block, and it still fails if SharedSubexpressions.common ignores the setting.