fix: preserve join and generator semantics in plan identity

[ErikBPF]

Which issue does this PR close?

Closes #5824.

Rationale for this change

Exchange reuse can treat plans with different semantics as equivalent. Semi/anti join branches can return one branch twice, while ordinary/outer generator branches can lose the outer rows. A null probe also exposes a broadcast anti-join collision between NOT EXISTS and NOT IN.

What changes are included in this PR?

• Include joinType in equality and hashing for shuffled hash, broadcast hash, and sort-merge joins.
• Preserve GenerateExec.outer in generator identity and explain output.
• Derive the null-aware anti-join flag directly from Spark's operator and preserve it in broadcast join identity.
• Share post-operator exchange-reuse assertions through CometTestBase, including the existing aggregate regressions. AQE-enabled cases explicitly require an adaptive plan.

Native execution and protobuf schemas are unchanged. Hashing uses the repository's boxed-type ascription convention for Scala 2.12 compatibility.

How are these changes tested?

Added 16 parameterized regressions in the existing join and generator suites. They cover all three join strategies, nested/sliced arrays, explode/posexplode, null and empty arrays, null-aware anti joins, and AQE off/on. Tests compare Spark results and explicit expected rows, require native operators, and verify equivalent plans still reuse post-operator exchanges.

The regressions reproduced incorrect results before the fix. The shared assertion is also exercised by five existing aggregate canonicalization cases.

Local validation of the revised patch:

• Spark 4.1: full join, generator, and aggregate suites passed 193 tests, with 0 failures and 2 pre-existing ignored aggregate-metrics tests. Packaging passed.
• Focused 21-case runs passed on Spark 4.1 and Spark 3.5/Scala 2.12, with no failures or skips.
• Semantic and syntactic Scalafix, Spotless, Scalastyle, Apache RAT, Prettier, and suite-registration checks passed.

These are local results; GitHub CI is separate.

Focused command (21 cases across all four helper call sites):

./mvnw test -Pspark-4.1 -Dtest=none \
  '-Dsuites=org.apache.comet.exec.CometJoinSuite identity preserves exchange reuse,org.apache.comet.exec.CometGenerateExecSuite identity preserves exchange reuse,org.apache.comet.exec.CometAggregateSuite aggregate canonicalization preserves'

[andygrove]

Thanks for picking this up, and nice catch on the null-aware anti join. That third instance was not in the issue, and you are right that joinType alone does not separate it, since NOT EXISTS and NOT IN both arrive as LeftAnti with BuildRight, no condition, and identical children.

I checked this out locally and ran the regressions both ways. On your branch all 16 pass. With operators.scala reverted to main and your tests kept, all 16 fail, and every one of them fails on Results do not match for query rather than on a plan assertion. That is the strongest form these regressions could take. Full CometJoinSuite, CometGenerateExecSuite and CometAggregateSuite come out at 193 passed and 0 failed on your branch, so tightening equals is not costing any legitimate reuse.

I also swept every hand-written equals under spark/src/main/scala/org/apache/spark/sql/comet/ against its constructor parameter list. Nothing else is live after this PR, so the fix is complete. The only parameters left out anywhere are outputOrdering on the sort and join operators, and those are derived from fields you are now comparing.

Since this is the third pass over the same family (#5470 found resultExpressions, #5824 found joinType and outer, you found isNullAwareAntiJoin on top of that), I filed #5831 for the reflective guard rather than asking you to grow this PR, and #5832 for the contributor guide. The guide never mentions the override at all, and its CometFilterExec example leaves equals and hashCode out, so the documented copy-paste path produces the bug.

A few things I would like to sort out before this lands.

The reuse assertion block is now in four places, the one #5470 added to CometAggregateSuite plus the three here. Would you pull it into CometTestBase as something like assertExchangeReuseOver(plan, clue)(pf)? It would take about ten lines out of each test, and the next operator that hits this defect will want it too.

The #5470 tests you modelled these on assert reusedPlan.isInstanceOf[AdaptiveSparkPlanExec] when adaptive is true. Without it an AQE=true case that quietly planned without AQE would still pass and just duplicate the AQE=false case. Worth carrying over?

On the test names, the #5824 prefix is new for this tree. These are the only tests under spark/src/test that lead with an issue number, and the aggregate regressions in #5470 use plain descriptive names. Would you drop it? The issue link lives in the commit message.

One more that I could not leave inline because the line falls outside the diff. CometExplodeExec.stringArgs at operators.scala:1527 still does not carry outer, so explain renders explode and explode_outer identically. That opacity is part of why this was hard to spot in the first place, and Spark's own GenerateExec prints it since it uses the default stringArgs. CometExplodeExec appears in no plan stability golden file, so adding it is free. I would leave isNullAwareAntiJoin out of CometBroadcastHashJoinExec.stringArgs though, that one shows up in 156 golden plans and the churn is not worth it.

Last thing, CI has not run on this yet because the workflows need approval for a first time contributor. In the meantime I ran test-compile locally on -Pspark-3.5 -Pscala-2.12 and -Pspark-3.4 -Pscala-2.12, plus spotless and scalastyle, and all of them are clean.

[andygrove]

createExec reads the flag back out of the protobuf here, but doConvert has already derived it from the plan a few lines up, and every other argument in this call comes from op. Could createExec match on op the same way?

op match {
  case bhj: BroadcastHashJoinExec => bhj.isNullAwareAntiJoin
  case _ => false
}

That keeps one source of truth. The protobuf read also fails quietly rather than loudly: getHashJoin on an unset oneof returns the default instance, so if a future change stops setting hash_join on this path the field silently becomes false and the collision comes back. Your new test would catch that, so this is about directness more than a live bug.

[andygrove]

Small consistency point on the boxing, and the same applies to Boolean.box(isNullAwareAntiJoin) further down. This file ascribes the type rather than calling box, as in Objects.hashCode(output, limit: java.lang.Integer, child) at line 1308, and CometSampleExec and CometWindowGroupLimitExec do the same for Double and Int. Would outer: java.lang.Boolean read better here?

[ErikBPF]

Thanks @andygrove for the detailed feedback and for independently checking the red/green regressions and full suites.

Addressed all six points in e3c877e:

• Added assertExchangeReuseOver(plan, clue)(pf) to CometTestBase and used it at all four call sites, including the aggregate regressions.
• Added explicit AdaptiveSparkPlanExec assertions to the AQE-enabled join and generator cases.
• Removed the #5824 test-name prefixes and updated the PR's focused test command.
• Included outer in CometExplodeExec.stringArgs. Broadcast join explain output stays unchanged.
• Derived isNullAwareAntiJoin directly from the Spark operator in createExec.
• Used java.lang.Boolean type ascriptions for both hash inputs, matching the surrounding conventions.

Validation on the revised patch: 193 tests passed across the full Spark 4.1 join, generator, and aggregate suites, with 0 failures and 2 pre-existing ignored aggregate-metrics tests. All 21 focused cases passed on both Spark 4.1 and Spark 3.5/Scala 2.12. Packaging, semantic and syntactic Scalafix, Spotless, Scalastyle, RAT, Prettier, and suite-registration checks also passed locally.

[sunchao]

Correctness

Prior state and proposed change

Reviewed e3c877e6bcbf3cd737c02dba3995fa134154ed15 against 392da2ca7a99fd223644eff650406f61580eb46b. The base is an ancestor of this head. I found no new P1/P2 issue in the five changed files.

The existing hand-written identity omitted semantics that the protobuf already carried. Semi and anti joins could share the same output schema, and ordinary versus outer generators could share the same generator expression. A broadcast LeftAnti additionally needs isNullAwareAntiJoin to distinguish NOT EXISTS from NOT IN. Including these fields in both equality and hashing addresses those collisions. The new flags come directly from Spark's operator and survive the existing case-class copying and canonicalization path.

Spark compatibility and behavioral changes

I compared the affected behavior with the maintained Spark 3.5 and 4.0 sources. Outer generators preserve a row with null generator output for empty/null arrays. Null-aware anti joins treat a null probe differently from ordinary anti joins when the build side is nonempty. Neither output schema nor join type alone is sufficient to distinguish all of these cases. Spark also normalizes attribute nullability during canonicalization, so the explicit semantic fields are necessary.

Conditions, join keys, build side, ordered output attributes, generator expressions and children remain part of identity. Output construction and native serialization are unchanged. Spark's exchange, adaptive-stage and subquery reuse all key on canonicalized plans, so the fix applies at that common boundary. The generator explain output now exposes outer. Maintained Spark 3.4 and 4.1 sources were unavailable for this review. I do not claim source coverage for those versions.

Validation

The 16 new parameterized cases cover three join strategies, null-aware anti joins, explode/posexplode, struct-field and sliced-array inputs, null/empty arrays, and AQE off/on. Explicit null predicates and computed generator inputs avoid the optimizer filters that otherwise mask the collision. The positive controls require equal semantic hashes and reuse above the target operator, and the AQE cases require an adaptive plan. The shared helper also retains the five aggregate regression cases.

I independently ran 52 isolated assertions on each of Scala 2.12.20 and 2.13.18 using the exact extracted equality/hash and traversal methods with test stubs. They verify the base collisions, separation at this head, equal-object hash consistency, and rejection of reuse below the target operator. These are component probes, not Spark SQL or JNI execution. I did not run the full Comet suites. The author's revised-head local results and the earlier reviewer's red/green results remain separately reported evidence. GitHub CI still requires workflow approval. The successful label check is not test coverage.

Performance

The production change adds a bounded number of scalar comparisons and hash inputs to existing plan identity operations. It adds no per-row processing, new tree walk, or serialization work. Equal plans still have the same identity and hash, while plans with different semantics can no longer incorrectly share execution.

The new test helper performs an outer traversal and a child traversal for each reused exchange. That can revisit subtrees and allocate intermediate collections, but it is confined to the small test plans and preserves the previous aggregate assertion's algorithm. I found no material performance issue requiring a change. No runtime speedup or benchmark result is claimed.

Design

Keeping semantic fields explicitly on the Comet operators fits the existing design: canonicalization deliberately discards the original Spark plan and serialized block, so identity cannot rely on either to recover the missing flags. Reading the null-aware flag directly from the Spark operator also avoids coupling identity to protobuf defaults. Native execution and wire formats remain unchanged.

The patch is appropriately scoped to the known identity defects and their regressions. I found no additional design change necessary for this PR.

Abstraction & complexity

assertExchangeReuseOver earns its place in CometTestBase by consolidating four call sites while letting each specify the operator of interest. Its explicit descent into ReusedExchangeExec.child matters because that wrapper is a leaf for ordinary plan traversal. The inherited adaptive traversal handles AQE and query-stage wrappers. The helper does not introduce a production abstraction, and the focused patch remains straightforward to maintain.