feat: add support for comet cartesian join

[coderfender]

Which issue does this PR close?

Closes #199

Rationale for this change

Add native support for cross join exec in comet

What changes are included in this PR?

1. Proto changes : new CrossJoin operator message
2. Wiring on the rust side

Limitations

1. No join condition : CrossJoinExec has no post-product filter; Spark normally folds a
   cross-join predicate into a separate FilterExec, and a CartesianProductExec that still
   carries a condition falls back.
2. Both inputs are a single partition : Comet runs one native plan per Spark partition and
   cannot inject an exchange here (EnsureRequirements has already run), while a cross product is
   inherently cross-partition. With more than one input partition we fall back to Spark's
   CartesianProductExec (which crosses every left×right partition pair via CartesianRDD)

How are these changes tested?

[sunchao]

Reviewed the full eight-file diff from 474b5ece841a7fc6dff37ae04bd6787e654d7be7 to 5db5de8f4844951a501ef7bba5d54ef15f263305. The PR remains open and non-draft. One introduced P2 issue was reproduced: native Cartesian execution can exhaust memory where Spark succeeds. There are no existing reviews, conversation comments, inline comments, or review threads.

Summary

• Prior state and problem: Cartesian products previously remained in Spark, using CartesianRDD to cross partition pairs and spillable buffering for the right input. This PR adds native execution for a restricted subset.
• Design approach: Register CometCartesianProductExec, serialize an empty CrossJoin message, and delegate execution to DataFusion’s CrossJoinExec. Conversion requires condition-free, single-partition inputs.
• Correctness: The partition and condition guards preserve Spark’s product semantics. Focused tests matched Spark for nulls, nested values, count-only output, empty input, and multiple partitions with AQE enabled and disabled. The memory-pressure case fails natively despite succeeding through Spark, as detailed in the finding. No other introduced P1/P2 correctness issue was identified.
• Compatibility analysis: Compared Spark sources for 3.4.3, 3.5.9, 4.0.4, 4.1.3, and 4.2.0. All preserve left-then-right output columns and provide spillable Cartesian buffering. Spark 4.1 adds byte-based spill thresholds. Source comparison against branch-1.1 confirms Cartesian execution previously stayed in Spark. Native routing is intended, but losing bounded-memory execution is an unintended regression.
• Key design decisions: Requiring one partition per input correctly avoids treating a zipped partition pair as a distributed Cartesian product. Falling back when a condition is attached prevents silently dropping predicates.
• Implementation sketch: The change adds the public enable flag, Scala operator and serde registration, protobuf field 123, native planner dispatch, tracing name, existing join metrics, and two join-suite tests.
• Performance: DataFusion retains and concatenates the entire left input before producing output. This introduces a demonstrated memory requirement even when the right input contains one row. No throughput benchmark was available, so this review makes no speedup claim.
• Design: Reusing DataFusion keeps the implementation small and understandable. However, partition eligibility alone does not establish that replacing Spark’s spillable operator is safe.
• Abstraction & complexity: The existing CometBinaryExec, serde, join-parameter helper, and metrics abstractions fit this implementation. No abstraction or complexity issue meeting the P1/P2 bar was identified.
• Behavioral changes worth calling out: spark.comet.exec.cartesianProduct.enabled defaults to true. Eligible products now buffer the left side in a non-spillable native operator. Setting this flag to false restores Spark execution and makes the reproduced query succeed.
• Suggested improvements: Preserve bounded-memory execution before enabling this path by default. Implement spillable buffering or retain explicit opt-in until that behavior exists, and add the reproduced memory-pressure comparison as a regression test.

Routed skills: review-comet-pr, review-comet-expression-pr, review-comet-ffi-pr, and review-comet-memory-pr, plus the contributor operator guide and relevant subsystem documentation.

Exact-head CI: 25 checks succeeded and 18 were skipped, with no failed or pending checks. Linux native build, Rust tests, Spark 4.1 Comet suites, and TPC-H/TPC-DS verification passed. The execution job ran 1,628 tests successfully, including both new Cartesian tests. Spark SQL suites, Iceberg suites, macOS, and benchmark checks were skipped.

Validation: Built the exact-head JVM sources in a disposable copy and ran a two-test suite on Spark 4.1.3/JDK 21 using the SHA-verified native artifact from this head’s CI. Both tests passed, including the test asserting the reproduced native memory failure. DataFusion 55.2.0 source was verified against the lockfile checksum. No local native rebuild, full Spark SQL/Iceberg suite, other Spark-version runtime matrix, or throughput benchmark was run. The checkout remains unchanged, and nothing was published.

[sunchao]

[P2] [P2] Preserve bounded-memory execution before enabling native Cartesian products

CrossJoinExec::new(left, right) collects the entire left input and propagates a refused memory reservation without spilling. Spark’s UnsafeCartesianRDD instead streams the left input and uses spillable buffering for the right. Consequently, the single-partition guard admits queries that previously succeeded but now fail by default. A million-row left range crossed with a one-row right range reproduced this under a 4 MiB off-heap pool: Spark returned the expected sum, while native execution failed at its 2 MiB fair share. Could this preserve spillable execution, or remain opt-in until it can, with a regression covering this case?

Evidence: Executed a disposable Review6938Suite against exact-head JVM sources and CI native artifact 11841794909 on Spark 4.1.3/JDK 21. Configuration: spark.memory.offHeap.enabled=true, spark.memory.offHeap.size=4m, spark.comet.convert.range.enabled=true, both broadcast thresholds -1, and spark.sql.adaptive.enabled=false. Query: SELECT sum(l.id) FROM range(0, 1000000, 1, 1) l CROSS JOIN range(0, 1, 1, 1) r. With spark.comet.exec.cartesianProduct.enabled=false, the executed plan contained Spark CartesianProduct and returned 499999500000. With the flag enabled, the plan contained CometCartesianProduct and failed with Resources exhausted: Additional allocation failed for CrossJoinExec, can spill: false, and Failed to acquire 65632 bytes ... already holds 2034592 bytes ... fair limit is 2097152 bytes. DataFusion 55.2.0 load_left_input confirms unconditional accumulation through reservation.try_grow(batch_size)?. Validation log: /tmp/pr6938-current-validation/repro-localhost.log.

[andygrove]

Thanks for picking up #199. Reusing DataFusion's CrossJoinExec keeps the change small, and the condition and single-partition guards are the right instincts. I have a few things I would like to see addressed before this merges.

Memory. Spark's CartesianProductExec buffers its right side in a spillable buffer. CrossJoinExec buffers the whole left side instead. In DataFusion 55.2.0 load_left_input does reservation.try_grow(batch_size)?, so exhaustion becomes a query error and never a spill. It then calls concat_batches, which makes a second copy that is not reserved, so peak memory is about twice what the pool sees. The sunchao review reproduced a query that fails natively and succeeds in Spark. Spark only plans a Cartesian product when neither side can be broadcast, so this is the case where buffering a whole side matters most. Since the config defaults to true, a user upgrading from 1.1 can go from a working query to a failed task. Could this stay opt-in until it can spill, or fall back when the left side may be large? The reproduced query would make a good regression test.

Benefit. Requiring a single partition on both sides narrows the native path a lot, because most products have many partitions on each side. Do you have a case where this is faster than Spark? A benchmark in the style of CometCrossJoinBenchmark would show it.

Comment accuracy. The comments in getSupportLevel, the CrossJoin proto message and the planner say that Spark folds a cross-join predicate into a separate FilterExec. CartesianProductExec carries condition itself, which is exactly what the fallback test relies on. Could you reword those comments?

Spark SQL suites. This touches the planner, the serde and a native operator, and the Spark SQL suites were skipped in CI. Could you apply the run-spark-sql* labels or run dev/local-ci.sh spark sql_core-1? I ask because #6907 had to mark SPARK-26138 IgnoreComet after a join started running natively with a different row order.

Tests. The fallback test uses checkSparkAnswer only, so it would still pass if the query stopped planning a CartesianProductExec. Could it assert that the plan contains one? Could we also add a case where a multi-partition input falls back, and cases with nulls, an empty side, and a right side of more than 8192 rows?

Docs. docs/source/user-guide/latest/operators.md lists CartesianProductExec under "Not currently planned: rare and expensive, with little acceleration benefit". Could you update that line and add CartesianProductExec to the supported joins table?

I am requesting changes for the memory regression. I will re-review once it is addressed.