feat: add support for comet cartesian join - #6938

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.

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.

Thanks @coderfender, and thanks for filing apache/datafusion#26383. Making this opt-in resolves the memory regression, so I'm dismissing my earlier review. A few things from that review are still open before I can approve.

• Could you rebase onto main, since the branch conflicts now?
• Could the comments in getSupportLevel, the proto and the planner say that CartesianProductExec carries the condition itself?
• Could the fallback test assert that the plan contains a CartesianProductExec, with added cases for a multi-partition input, nulls, an empty side and a right side over 8192 rows?
• Could operators.md move CartesianProductExec out of "Not currently planned" and into the supported joins table, noting that it's opt-in?

I'd also still like to see a case where this is faster than Spark, since both sides have to be a single partition. A small benchmark in the style of CometCrossJoinBenchmark would answer that.

Making the operator opt-in settles the memory concern, and the benchmark shows a real gain for eligible inputs. andygrove's requests on the FilterExec wording in the comments and on operators.md still look open at this head. Two comments on what the tests and eligibility actually reach.

The default-off gate and the single-partition check address my earlier concern about enabling the non-spillable cross join for general inputs. I have one correction to the published eligibility rule.

The update covers the one-bucket Parquet case from my previous comment. I found one more scan that can run the native join, and a mismatch with the operator table’s legend.

Reviewed e063de9 across correctness, resources, design, tests and documentation. The compatibility marker and single-partition wording now match the opt-in implementation. One new SQL test fails under the AQE=false matrix entry, confirmed in the current fork CI and with a local Spark 4.1.3 partitioning probe.

Reviewed 0ecd27c with five independent scopes. The REBALANCE regression now lives in an AQE-only fixture, while the general Cartesian-product fixture still runs with AQE both enabled and disabled. I checked the exact-head fork report: all three fixture executions passed within 2306 successful expression tests. The updated docs and single-partition guard still agree. No remaining actionable issue found.

Thanks @coderfender. Making this opt-in settles the memory concern, the benchmark shows a real gain for eligible inputs, and the single-partition and condition guards look right to me. I compared against branch-1.2 and nothing changes with the flag at its default. Approving. Two small nits, both optional.

Nit, optional: I read CrossJoinStream in DataFusion 55.2.0. It emits one output batch per left row for each right batch, so the output is right-batch-major. Spark's UnsafeCartesianRDD is left-row-major. With more than one left row and a right side over 8192 rows, as in the cp_large case, I think the row order differs from Spark. I have not run it. Does the query comparison in cartesian_product.sql ignore row order? If it does, could the operators.md row mention that the row order of a native product can differ from Spark's? That would help anyone with a LIMIT and no ORDER BY above a product.

Nit, optional: could the COMET_EXEC_CARTESIAN_PRODUCT_ENABLED note link apache/datafusion#26383? Then it is clear what has to land before the default can change.

Thanks for the updates. Both points from my last review are resolved: the condition fallback query now coalesces both sides, so the condition guard is the only thing left to reject it, and the docs and config note state the one-partition rule. I also looked at the new AllTuples override. It is satisfied by any one-partition input, and since nothing reruns EnsureRequirements on Comet nodes it only feeds AQE's ValidateRequirements, which is what the AQE fixture needs.

One thing about the description, since it becomes the squash commit message. It still says Spark folds a cross-join predicate into a separate FilterExec, it does not mention that the native path is opt-in, and "How are these changes tested?" is empty. Closes #199 would also close the general issue while multi-partition inputs and spilling are still open. Could it say "Part of #199", or link a follow-up issue for the rest?