andygrove opened a new pull request, #26277: URL: https://github.com/apache/datafusion/pull/26277
## Which issue does this PR close? - Part of #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, 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 `StreamedBatch::append_output_pairs` appends the streamed index with `append_value_n` and the buffered indices as a range with `append_trusted_len_iter`, and the new `BufferedData::scanning_advance_by` moves the scan past them. - `append_output_pair` and `append_output_pairs` share the chunk lookup through `chunk_for`. The pairs, their order and the chunks they go to are unchanged, so freezing, filtering and spilling see the same indices as before. Unmatched rows of outer joins still go through `append_output_pair`. ## What is the testing strategy for this PR? - The existing sort-merge join tests cover this path, including those that run with batch sizes of 1, 2 and 50 and those that spill. The join fuzz tests pass. - New `join_inner_group_spans_buffered_and_output_batches`: the buffered key group spans three batches and the batch size is 4, so runs end both at the end of a buffered batch and when the output batch fills. It checks that every pair comes out once, in buffered order. - New benchmark case `inner_1to100`, in which every streamed row matches 100 buffered rows. `sort_merge_join` benchmark against `main`, the median of five interleaved runs of each binary on an Apple M3 Max: | Case | `main` | This PR | Change | | --- | --- | --- | --- | | `inner_1to1` | 1.510 ms | 1.569 ms | +3.9% | | `inner_1to10` | 11.95 ms | 8.84 ms | −26.1% | | `inner_1to100` (new) | 23.02 ms | 16.04 ms | −30.3% | | `left_1to1_unmatched` | 1.513 ms | 1.565 ms | +3.4% | | `left_semi_1to10` | 0.611 ms | 0.608 ms | −0.4% | | `left_anti_partial` | 0.767 ms | 0.767 ms | 0.0% | | `inner_group_spans_buffered_batches/512` | 3.90 ms | 2.99 ms | −23.3% | | `inner_group_spans_buffered_batches/64` | 5.16 ms | 4.19 ms | −18.7% | | `inner_group_spans_buffered_batches/8` | 15.36 ms | 14.55 ms | −5.3% | The semi and anti cases don't use this stream, so they show the noise level. The one-to-one cases are about 4% slower, about 0.6 ns per row. Sending one-row runs through `append_output_pair` instead measured slower, not faster. In DataFusion Comet (apache/datafusion-comet#6907), a prototype of this change 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 is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
