Repository navigation
Conversation
Spark gives a join the type Cross when the query asks for a cross join explicitly, with CROSS JOIN in SQL or Dataset.crossJoin, and its join operators run Cross exactly as they run Inner. Comet's sort-merge, hash and nested loop join serdes matched only Inner, so these joins fell back with "Unsupported join type Cross". Map InnerLike to the native Inner join type in all three. Spark's SPARK-26138 test takes LIMIT 3 of a cross join with no ORDER BY. That join now runs as a native nested loop join, which emits rows in a different order than Spark, so ignore the test under Comet in every Spark diff. Closes apache#6903.
Times Spark, Comet with the join left in Spark, and Comet for sort-merge, shuffled hash, broadcast hash and broadcast nested loop joins of type Cross, at 10 and 100 matches per probe row. Before timing, each case checks that every arm planned the join it claims, that the Comet arm is fully native, and that the arms return the same sum.
Member
Author
|
The benchmark shows that when rows match many rows, the native sort-merge join is slower than what Comet does today with the join left in Spark: twice the time at 100 matches per row, and 2.7x at 1000. That isn't specific to Profiling put most of the native time in DataFusion's sort-merge join recording its matches one buffered row at a time, so I opened apache/datafusion#26277 to append them in bulk. With a prototype of that change on DataFusion 55.2.0, measured in the same runs:
That cuts the native join's CPU time by 39% at 100 matches per row and 46% at 1000. What's left at very high fan-out is mostly building the output batches. The join gathers every column of both sides because it has no built-in projection, and its batch coalescer copies them again (apache/datafusion#25515). I'll keep this PR in draft until the fix reaches Comet, then rerun |
haohuaijin
pushed a commit
to haohuaijin/arrow-datafusion
that referenced
this pull request
Oct 20, 2026
## Which issue does this PR close? - Part of apache#20910. ## Rationale for this change A sort-merge join in which each row matches many rows on the other side spends most of its time recording the matches. For every buffered row of the key group, `pair_streamed_row_with_group` calls `append_output_pair`, which looks up the current chunk and appends one value to each of two `UInt64Builder`s, and then advances the scan by one row. In DataFusion Comet, which runs DataFusion 55.2.0, a sort-merge join of 1M rows in which every row matches 100 rows (105M joined rows feeding a `sum`) spent 46% of its active CPU in these two functions, against 14% in the `take` that builds the output batches. ## What changes are included in this PR? - `pair_streamed_row_with_group` pairs the streamed row with the rest of the scanning buffered batch at once, capped by the room left in the output batch. The new `BufferedData::scanning_remaining` gives those rows, and `BufferedData::scanning_advance_by` moves the scan past them. - The new `StreamedBatch::append_output_pairs` appends a run of one row, as in a one-to-one join, directly. Longer runs go to `append_output_range`, which extends the chunk's streamed indices with `repeat_n` and its buffered indices with the range of rows, and appends their validity with one `append_n_non_nulls`. It is kept out of line so that the one-row path doesn't pay for its setup. - `StreamedJoinedChunk` keeps its indices in `Vec<u64>`s, as hash join does, with a `NullBufferBuilder` for the buffered nulls, instead of in `UInt64Builder`s. That takes the per-row null and capacity checks out of the bulk append, and `freeze_streamed` builds the arrays with `UInt64Array::try_new`, which skips the `ArrayData` that `PrimitiveBuilder::finish` builds for every chunk (apache/arrow-rs#11717). - The chunk lookup is now `chunk_for`, which keeps only its check inline and starts a new chunk in a cold `start_chunk`. With that, one-to-one joins are slightly faster than on `main`. - `append_output_pair`'s only remaining caller passed a null buffered index, so it is now `append_null_joined_pair`. `scanning_advance` is replaced by `scanning_advance_by(1)`. The pairs, their order and the chunks they go to are unchanged, so freezing, filtering and spilling see the same indices as before. ## What is the testing strategy for this PR? - The existing sort-merge join tests cover this path for every join type, including those that run with batch sizes of 1, 2 and 50 and those that spill. - The join fuzz tests, which compare sort-merge and hash join results across join types, pass (`cargo test -p datafusion --features extended_tests --test fuzz -- join_fuzz`). Most of their key groups hold one row, though, so only 3–5% of their appends are runs of more than one row, and none is cut by the output batch. The new low-cardinality cases, `test_{inner,left,right,full}_join_1k_low_cardinality` and their `_filtered` variants, draw the keys from 25 values. In each of them about 30% of the appends are runs of more than one row, 14,000–19,000 runs are cut by the output batch, and 51,000–56,000 end at a buffered batch boundary with the group continuing in the next one. - New `join_inner_group_spans_buffered_and_output_batches`, next to `join_with_group_spanning_batches_rescanned_per_streamed_row`, which uses one-row buffered batches. In the new test the key group spans buffered batches of 3, 2 and 1 rows and the batch size is 4, so runs of several rows end both at the end of a buffered batch and when the output batch fills. - New benchmark cases `inner_1to100`, in which every streamed row matches 100 buffered rows, and `left_1to100_filtered`, which joins the same inputs with a filter, so that the pairs go through deferred filtering. `sort_merge_join` benchmark against `main`, the median of seven interleaved runs of each binary on an Apple M3 Max: | Case | `main` | This PR | Change | | --- | --- | --- | --- | | `inner_1to1` | 1.484 ms | 1.427 ms | −3.9% | | `inner_1to10` | 12.89 ms | 8.755 ms | −32.1% | | `inner_1to100` (new) | 24.89 ms | 15.98 ms | −35.8% | | `left_1to100_filtered` (new) | 53.56 ms | 44.59 ms | −16.8% | | `left_1to1_unmatched` | 1.485 ms | 1.428 ms | −3.8% | | `left_semi_1to10` | 0.607 ms | 0.607 ms | 0.0% | | `left_anti_partial` | 0.763 ms | 0.764 ms | +0.1% | | `inner_group_spans_buffered_batches/512` | 4.052 ms | 2.981 ms | −26.4% | | `inner_group_spans_buffered_batches/64` | 5.286 ms | 3.929 ms | −25.7% | | `inner_group_spans_buffered_batches/8` | 15.52 ms | 12.41 ms | −20.1% | The semi and anti cases don't use this stream, and they moved by 0.1% at most. Keeping the bulk path and the chunk creation inline instead made the one-to-one cases about 5% slower than `main`. Keeping the indices in `UInt64Builder`s, with a loop of `append_value` for a run's buffered indices, made the cases with several matches per row 7–20% slower. In DataFusion Comet (apache/datafusion-comet#6907), a prototype of the bulk path on 55.2.0 cut the native sort-merge join's CPU time by 39% at 100 matches per row and by 46% at 1000. At 100 matches per row that took it from twice the time of Spark's whole-stage-codegen sort-merge join to the same time. ## Are there any user-facing changes? No. Sort-merge joins in which rows match many rows run faster.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Closes #6903.
Rationale for this change
Spark gives a join the type
Crosswhen the query asks for a cross join explicitly, withCROSS JOINin SQL orDataset.crossJoin, and Spark's join operators runCrossexactly as they runInner. Comet's join serdes matched onlyInner, so a cross join that Spark planned as a sort-merge, shuffled hash, broadcast hash or broadcast nested loop join fell back withUnsupported join type Cross, while the same join written asJOIN ... ONran natively.A production job on Spark 3.5 hit this. A cross join with an equi-key filter became a
SortMergeJoinof typeCrossthat produced 110 billion rows in Spark, and its stage took two thirds of the job's task time.What changes are included in this PR?
InnerLike(InnerandCross) to the nativeInnerjoin type. The keys and condition carry over unchanged, so the native plan is the one the equivalent inner join gets.InnerLikewith either build side, in its support check and its conversion, as it already did forInner.Cross, andLeftSinglefrom Spark 4.0.SPARK-26138 Pushdown limit through InnerLike when condition is emptytestIgnoreCometin all five Spark diffs, regenerated from the Spark tags after checking that each committed diff round-trips unchanged. The test takesLIMIT 3oft1 CROSS JOIN t2with noORDER BYand expects the rows Spark returns. The join now runs as a native nested loop join, which emits rows in a different order, so Comet returns(0,0), (1,0), (2,0)where Spark returns(0,0), (0,1), (0,2). Both are valid answers.mainalready returns Comet's rows forFROM t1, t2 LIMIT 3, anInnerjoin; the test passed only because theCrossjoin fell back.CometCrossJoinBenchmark, which times each of the four join strategies on a join of typeCrossin three arms: Spark, Comet with the join left in Spark (the plan Comet made before this change), and Comet.Native cross nested loop joins now also have the problem described in #6898, which inner ones already have: the join reports the streamed side's order without keeping it.
CartesianProductExec, a cross join with no broadcast side, still falls back (#199).How are these changes tested?
New tests in
CometJoinSuite:CROSS JOIN ... WHERE tbl_a._2 = tbl_b._1, with and without a residual condition, underSHUFFLE_HASH,MERGEandBROADCASThints.CROSS JOINwith a broadcast hint on either side, with no condition and with a non-equi condition, which Spark plans as a broadcast nested loop join.Each test checks the answer against Spark, checks that the plan is fully native, and checks that the native join has the type
Cross, so a plan in which Spark choseInnercan't make it pass. With the serde change reverted, all four fail withUnsupported join type CrossorJoin type Cross is not supported.Locally,
CometJoinSuitepasses on Spark 3.4, 3.5 and 4.1. The other Comet tests that use a cross join pass on Spark 3.4 and 4.1:CometFloatSemanticsSuite, the AQE tests inCometExecSuite, aCometCodegenSuitetest, and two SQL file tests.For Spark's SQL tests, I went through every cross join in the
sql/coreandsql/hivetests of Spark 3.5.9 and 4.1.3 with the diffs applied, and compared the cross joins in 3.4.3, 4.0.4 and 4.2.0 against 4.1.3.SPARK-26138is the only test whose outcome changes. The others already accept Comet's join classes, are in suites ignored under Comet, or check only results. I haven't run Spark's SQL suites locally.Benchmark
CometCrossJoinBenchmarkjoins a 1M-row probe table in which every row meets 10 or 100 rows on the other side, 10.5M or 105M joined rows, and sums a column from each side so that only one row leaves the native plan. Before timing, each case checks that every arm planned the join it claims with the join typeCross, that the Comet arm is fully native, that Spark's join produced the expected number of rows, and that the three arms return the same sum.Best time in ms, the median of three runs with the range in brackets. Apple M3 Max with a load average of 5 to 10 from other work,
local[5], Spark 4.1, release build:Comet is 1.1x to 1.8x faster than Spark in every case except the sort-merge join at fan-out 100. Most of that comes from the native scans and shuffles: at fan-out 10, the two Comet arms are within 13% of each other.
At fan-out 100, the native shuffled and broadcast hash joins take 6% and 19% less time than the joins left in Spark. The native nested loop join takes 14% longer, and the native sort-merge join twice as long. Spark fuses the merge and the aggregate into one generated loop and never materializes the 105M joined rows, while the native join builds Arrow batches for all of them. This cost isn't specific to
Cross: anInnersort-merge join of the same shape took 410 ms natively, against 427 ms forCrossin the same run. It is the pattern described in #4614. Cross joins with an equi-key filter are often many-to-many, so they are likely to hit it, and the job that prompted #6903 had a join of this kind.