Repository navigation
feat: add support for comet cartesian join - #6938
coderfender wants to merge 3 commits into
Conversation
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
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
CartesianRDDto 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 emptyCrossJoinmessage, and delegate execution to DataFusion’sCrossJoinExec. 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.1confirms 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.enableddefaults totrue. Eligible products now buffer the left side in a non-spillable native operator. Setting this flag tofalserestores 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.
| let left = Arc::clone(&join_params.left.native_plan); | ||
| let right = Arc::clone(&join_params.right.native_plan); | ||
|
|
||
| let cross_join = Arc::new(CrossJoinExec::new(left, right)); |
There was a problem hiding this comment.
[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
left a comment
andygrove
left a comment
There was a problem hiding this comment.
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.
|
@andygrove, @sunchao , I created an upstream DF issue (apache/datafusion#26383) and disabled this join support by default (to be enabled explicitly by the user only). Please take a look whenever you get a chance. |
The memory regression is resolved now that native cartesian products are opt-in.
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
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 thatCartesianProductExeccarries 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.mdmoveCartesianProductExecout 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.
|
Benchmark results: |
DataFusion's CrossJoinExec buffers the whole left input and cannot spill, so a large left can OOM where Spark's spillable CartesianProduct succeeds (sunchao's P2). Default the flag to false until the native path can spill; add a regression test that a cross join stays on Spark by default. Part of apache#199.
b02d7a2 to
b57a0df
Compare
comphead
left a comment
comphead
left a comment
There was a problem hiding this comment.
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.
| withParquetTable(Seq((0, 100), (1, 101)), "r") { | ||
| // A non-equi inner join plans a CartesianProductExec that carries the predicate; Comet | ||
| // has no native post-product filter, so results must still match Spark via fallback. | ||
| checkSparkAnswer(sql("SELECT * FROM l JOIN r ON l._2 > r._2")) |
There was a problem hiding this comment.
This test cannot fail on the condition guard. Both sides are plain parquet scans, which report UnknownPartitioning(0) (CometNativeScanExec.outputPartitioning and FileSourceScanExec for non-bucketed scans), so the single-partition check rejects the product even if the op.condition.isDefined branch were removed. Could both sides use /*+ COALESCE(1) */ as in the native test, so the condition is the only reason left? A sql-tests/join/cartesian_product.sql fixture with -- Config: spark.comet.exec.cartesianProduct.enabled=true and both broadcast thresholds at -1 could then hold this as query expect_fallback(CartesianProduct with a join condition is not supported natively), next to the native case and the multi-partition, NULL, empty-side and over-8192-row cases andygrove listed. The default-off check needs the plan and can stay in Scala. I have not run the fixture.
| // filter, so fall back to Spark when a condition is attached. | ||
| if (op.condition.isDefined) { | ||
| Unsupported(Some("CartesianProduct with a join condition is not supported natively")) | ||
| } else if (op.left.outputPartitioning.numPartitions != 1 || |
There was a problem hiding this comment.
As far as I can tell, no plain file scan passes this check, since non-bucketed scans report UnknownPartitioning(0). Inputs qualify only after coalesce(1), repartition(1), a single-slice range or a SinglePartition exchange, which is what every benchmark case forces. Under AQE this rule runs before shuffle partitions are coalesced, so a shuffle that later shrinks to one partition is probably still rejected, though I have not run that. Could the config note and the operators.md entry say this, so users know when enabling it helps? Separately, for the large-left, small-right case, DataFusion 55.2 CrossJoinExec::build_batch emits one batch per left row, sized by the right batch, so 262144 x 64 produces 262144 batches of 64 rows. That probably explains why that case gains 3.1X against 18.6X for the square one. I read this from the source and did not measure it.
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?
CrossJoinoperator messageLimitations
CrossJoinExechas no post-product filter; Spark normally folds across-join predicate into a separate
FilterExec, and aCartesianProductExecthat stillcarries a condition falls back.
cannot inject an exchange here (
EnsureRequirementshas already run), while a cross product isinherently cross-partition. With more than one input partition we fall back to Spark's
CartesianProductExec(which crosses every left×right partition pair viaCartesianRDD)How are these changes tested?