perf: bulk-append contiguous buffered runs in sort merge join#23673
perf: bulk-append contiguous buffered runs in sort merge join#23673andygrove wants to merge 1 commit into
Conversation
The materializing sort-merge-join stream advanced one output row at a time in join_partial. When a streamed row matches a run of buffered rows (high fan-out), each matched row cost a VecDeque index lookup in scanning_idx plus two more in scanning_advance, along with two individual builder appends. The matched buffered rows within a batch form a contiguous index range, so append the whole run at once via append_output_range (append_value_n + append_slice) and advance the scan cursor by the run length. A len == 1 fast path keeps the common one-row-per-key case free of scratch-buffer overhead. Benchmarks (sort_merge_join): inner join with 1:10 fan-out improves ~23% (13.28ms -> 10.18ms); 1:1 cases are neutral.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #23673 +/- ##
=======================================
Coverage 80.67% 80.67%
=======================================
Files 1086 1086
Lines 366800 366853 +53
Branches 366800 366853 +53
=======================================
+ Hits 295912 295962 +50
+ Misses 53260 53256 -4
- Partials 17628 17635 +7 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
run benchmark sort_merge_join |
|
Hi @andygrove, thanks for the request (#23673 (comment)). Only whitelisted users can trigger benchmarks. Allowed users: Dandandan, Fokko, Jefffrey, Omega359, Rachelint, adriangb, alamb, asubiotto, brunal, buraksenn, cetra3, codephage2020, coderfender, comphead, erenavsarogullari, etseidl, friendlymatthew, gabotechs, geoffreyclaude, grtlr, haohuaijin, jonathanc-n, kevinjqliu, klion26, kosiew, kumarUjjawal, kunalsinghdadhwal, liamzwbao, mbutrovich, mkleen, mzabaluev, neilconway, rluvaton, sdf-jkl, timsaucer, xudong963, zhuqi-lucas. File an issue against this benchmark runner |
|
@mbutrovich @comphead could you trigger the benchmark run for me? I'm not on the list. |
|
run benchmark sort_merge_join,tpch,tpcds |
|
Hi @comphead, thanks for the request (#23673 (comment)). Usage: Any benchmark name is accepted: Per-side configuration ( env:
# shared env is inherited by BOTH the build and the run, so build
# flags go here. Builds default to no debuginfo for speed; opt back
# in for hung-job gdb dumps and cap jobs to stay within memory:
CARGO_PROFILE_RELEASE_DEBUG: "1"
CARGO_BUILD_JOBS: "1"
baseline:
ref: v45.0.0
env:
# per-side env only reaches the benchmark run, not the build
DATAFUSION_RUNTIME_MEMORY_LIMIT: 1G
changed:
ref: v46.0.0
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: 2GFile an issue against this benchmark runner |
|
run benchmark sort_merge_join tpch tpcds |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing smj-bulk-range-append (a7fa061) to 1e58928 (merge-base) diff using: sort_merge_join File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing smj-bulk-range-append (a7fa061) to 1e58928 (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing smj-bulk-range-append (a7fa061) to 1e58928 (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
Benchmark for this request failed. Last 20 lines of output: Click to expandFile an issue against this benchmark runner |
|
run benchmark tpch tpcds env: |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing smj-bulk-range-append (a7fa061) to 1e58928 (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing smj-bulk-range-append (a7fa061) to 1e58928 (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
hmmm... benchmark results seem pretty mixed so far |
|
run benchmark tpch10 |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing smj-bulk-range-append (a7fa061) to 1e58928 (merge-base) diff using: tpch10 File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
Which issue does this PR close?
Rationale for this change
The
MaterializingSortMergeJoinStream(used for inner/left/right/full joins) produces output one row at a time injoin_partial. When a single streamed row matches a run of buffered rows — i.e. a high fan-out join — each matched row incurs:VecDequeindex lookup inscanning_idx(),VecDequeindex computations insidescanning_advance(), andUInt64Builder::append_valuecalls.The buffered rows that match within a single buffered batch always form a contiguous index range, so this per-row work is unnecessary overhead that scales with the number of output rows.
What changes are included in this PR?
StreamedBatch::append_output_range, which appends a contiguous run of buffered indices against the current streamed index in bulk (append_value_n+append_slice) instead of one pair at a time. A reusable scratch buffer (buffered_index_scratch) materializes the index range, and alen == 1fast path keeps the common one-row-per-key case free of scratch-buffer overhead.BufferedData::scanning_advance_by(n)andscanning_batch_remaining()so the scan cursor can advance by a whole run at once.join_partialnow computes the run length (capped by the remaining output-batch capacity) and appends the whole run in a single step. The FULL-join null-joined branch is likewise updated toextenda range rather than pushing per row.No behavior change — output is identical, only the way indices are accumulated changes.
Benchmark results (
cargo bench --bench sort_merge_join)inner_1to10(1:10 fan-out, 1M output rows)inner_1to1left_1to1_unmatchedThe semi/anti benchmarks use a different stream (
BitwiseSortMergeJoinStream) that is unchanged.Are these changes tested?
Covered by existing tests:
datafusion-physical-planpass.sort_merge_join.sltsqllogictest passes.Are there any user-facing changes?
No.