Repository navigation
fix: deduplicate StringView/BinaryView buffer refs in CollectLeft Has... - #25716
Closed
mohitgurav20 wants to merge 21 commits into
Closed
mohitgurav20 wants to merge 21 commits into
mohitgurav20 wants to merge 21 commits into
Conversation
github-actions
Bot
added
physical-plan
Contributor
Author
|
Hi @alamb @kosiew @asolimando, I've formatted the code and rebased the branch onto latest All CI checks are now running against latest |
Contributor
Contributor
Highly agree, this is hard to read. @mohitgurav20 a prompt I find very useful: |
Contributor
Author
|
Hey all, I’ve fully updated the PR description following the suggested prompt to make it much easier to read (focusing on the user impact, adding the SQL MRE, and removing the heavy implementation jargon). I also went through the CI logs and fixed the compilation issue causing the 5 checks to fail (it just needed a quick .into() conversion from Vec to Arc<[Buffer]> for the new_unchecked initialization). Everything should build properly now. Let me know what you think! |
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #25716 +/- ##
==========================================
+ Coverage 82.77% 82.79% +0.01%
==========================================
Files 1147 1147
Lines 450909 451408 +499
Branches 450909 451408 +499
==========================================
+ Hits 373252 373745 +493
- Misses 54934 54935 +1
- Partials 22723 22728 +5 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Contributor
Author
|
Hey all, just a quick update! I've resolved the failing CI checks. The formatting has been aligned using cargo fmt, and I resolved the strict Clippy warning by explicitly using Arc::clone(&schema) instead of relying on .clone(). Everything is green now and this should be fully ready for review! Let me know if you need any further tweaks. |
Contributor
|
@mohitgurav20 |
Contributor
Author
kosiew
requested changes
Oct 2, 2026
kosiew
left a comment
kosiew
left a comment
Contributor
There was a problem hiding this comment.
Thanks for working on this. I found one correctness issue in the buffer deduplication that needs to be addressed before this can merge. I also left a suggestion to add BinaryView coverage.
| let mut has_duplicates = false; | ||
|
|
||
| for buf in data_buffers.iter() { | ||
| let addr = buf.as_ptr() as usize; |
Contributor
There was a problem hiding this comment.
Using only as_ptr() is not enough to identify an Arrow Buffer range because two valid slices can start at the same address but have different lengths. Please deduplicate only identical ranges, for example by using pointer plus length, or otherwise prove the retained buffer covers every remapped view, and add a regression where a shorter slice is seen before the longer buffer and the concatenated values are verified.
| } | ||
|
|
||
| #[test] | ||
| fn concat_build_batches_deduplicates_view_buffers() -> Result<()> { |
Contributor
There was a problem hiding this comment.
Could you add coverage for BinaryViewArray as well? The implementation has separate BinaryView dispatch and downcast logic, but the current regression only exercises StringViewArray.
mohitgurav20
force-pushed
the
fix/dedup-stringview-buffers-collectleft-joins
branch
from
October 3, 2026 09:42
f3bf727 to
124082a
Compare
Contributor
Author
|
@kosiew Good catch on the slicing edge case—using just the pointer address definitely wasn't safe enough there. I've just pushed a fix for this. The deduplication map now uses a (pointer_address, length) tuple as the key to properly identify exact buffer ranges. I also added the test coverage you requested: A new test specifically for BinaryViewArray deduplication. |
kosiew
requested changes
Oct 5, 2026
kosiew
left a comment
kosiew
left a comment
Contributor
There was a problem hiding this comment.
Thanks for the follow-up. The pointer-and-length key fixes the correctness issue, and the BinaryView coverage looks good. There is still one regression-test gap that needs to be addressed.
| Arc::new(Schema::new(vec![Field::new("s", DataType::Utf8View, true)])); | ||
|
|
||
| // Create a shorter slice that starts at the same address | ||
| let short_slice = base_array.slice(0, 1); |
Contributor
There was a problem hiding this comment.
This test does not actually cover buffers with the same pointer but different lengths. GenericByteViewArray::slice only slices the views, so please construct two arrays using differently sized Buffer::slice_with_length(0, ...) ranges with the shorter one first, then assert that values from the longer range still survive concatenation.
mohitgurav20
force-pushed
the
fix/dedup-stringview-buffers-collectleft-joins
branch
2 times, most recently
from
October 5, 2026 16:52
ce9ad51 to
0fb7157
Compare
mohitgurav20
force-pushed
the
fix/dedup-stringview-buffers-collectleft-joins
branch
from
October 5, 2026 16:53
0fb7157 to
9647de3
Compare
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing b104668 (b104668) to b104668 diff Run configurationrun benchmark tpcds
baseline:
ref: "b104668"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
changed:
ref: "b104668"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing b104668 (b104668) to b104668 diff Run configurationrun benchmark tpcds
baseline:
ref: "b104668"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
changed:
ref: "b104668"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
Contributor
Author
|
Thanks for the thorough review and suggestions! @kosiew I've pushed a commit that addresses all four of your inline suggestions: Clarified in the docs that deduplication uses the (pointer, length) pair. All tests are passing. Let me know if you need anything else! |
Contributor
|
run benchmark tpcds env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true" |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing fix/dedup-stringview-buffers-collectleft-joins (ca2623c) to 06aa131 (merge-base) diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing fix/dedup-stringview-buffers-collectleft-joins (ca2623c) to 06aa131 (merge-base) diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
Contributor
|
hi @mohitgurav20 |
Contributor
Author
|
Thanks for running that @kosiew — that result makes sense to me. TPC-DS at SF1 is 1GB of data, so everything sits comfortably in memory and the buffer explosion never really gets a chance to kick in. The fix targets a pretty specific pattern: multiple chained broadcast joins on the same StringView/BinaryView column, where each concat pass accumulates duplicate references to the same underlying buffers. At small scale that overhead is invisible. The issue description actually measured this with a single column growing to 1.5M buffer references where only 2 were distinct — that's the real scenario where you'd see the 7x memory and 10x latency impact. TPC-DS SF1 just doesn't stress that path. Happy to put together a focused micro-benchmark that directly reproduces the exponential growth, or run at SF10/SF100 if you'd prefer a more standard signal. |
Contributor
|
I think since @adriangb filed the original ticket, perhaps he can weigh in here I agree if we can't find some representative benchmark that shows the extra complexity improves the performance, we should probably not merge this code I don't think a focused micro benchmark is enough |
Contributor
Author
Contributor
|
run benchmark tpcds baseline:
ref: 06aa131
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
changed:
ref: ca2623c
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true" |
Contributor
|
run benchmark tpcds baseline:
ref: 06aa131
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
changed:
ref: ca2623c
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false" |
Contributor
|
run benchmark tpcds baseline:
ref: ca2623c
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
changed:
ref: ca2623c
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false" |
Contributor
|
run benchmark tpcds baseline:
ref: 06aa131
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
changed:
ref: 06aa131
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false" |
Contributor
|
run benchmark tpcds baseline:
ref: 06aa131
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
changed:
ref: 06aa131
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true" |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing 06aa131 (06aa131) to 06aa131 diff Run configurationrun benchmark tpcds
baseline:
ref: "06aa131"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
changed:
ref: "06aa131"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing ca2623c (ca2623c) to 06aa131 diff Run configurationrun benchmark tpcds
baseline:
ref: "06aa131"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
changed:
ref: "ca2623c"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing ca2623c (ca2623c) to ca2623c diff Run configurationrun benchmark tpcds
baseline:
ref: "ca2623c"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
changed:
ref: "ca2623c"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing ca2623c (ca2623c) to 06aa131 diff Run configurationrun benchmark tpcds
baseline:
ref: "06aa131"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
changed:
ref: "ca2623c"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing 06aa131 (06aa131) to 06aa131 diff Run configurationrun benchmark tpcds
baseline:
ref: "06aa131"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
changed:
ref: "06aa131"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing ca2623c (ca2623c) to 06aa131 diff Run configurationrun benchmark tpcds
baseline:
ref: "06aa131"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
changed:
ref: "ca2623c"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing ca2623c (ca2623c) to 06aa131 diff Run configurationrun benchmark tpcds
baseline:
ref: "06aa131"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"
changed:
ref: "ca2623c"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing 06aa131 (06aa131) to 06aa131 diff Run configurationrun benchmark tpcds
baseline:
ref: "06aa131"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
changed:
ref: "06aa131"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing 06aa131 (06aa131) to 06aa131 diff Run configurationrun benchmark tpcds
baseline:
ref: "06aa131"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
changed:
ref: "06aa131"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing ca2623c (ca2623c) to ca2623c diff Run configurationrun benchmark tpcds
baseline:
ref: "ca2623c"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "true"
changed:
ref: "ca2623c"
env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: "false"CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
Contributor
|
TPC-DS SF1, change in query time (negative = faster). Noise band: ±10%, the largest per-query difference in the main on vs main on control run. A cell with — is within the noise band. Queries that are within the noise band in all four runs are not shown.
Total for the control run: -0.6%. main = |
Contributor
Author
|
Thanks for running the benchmarks with both pushdown configurations, @adriangb! The numbers are clear — at SF1, it's a flat As agreed earlier with @alamb, without a clear representative payoff, adding this complexity to the hash join build path isn't justified. Closing this out. If this ever surfaces as a measurable bottleneck on larger scale factors (SF100+) or high batch counts, we can revisit whether handling it upstream in Arrow's |
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
6 participants
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 #25712
Rationale for this change
Queries with multiple chained
CollectLefthash joins onStringVieworBinaryViewcolumns can run 10x slower and use 7x more memory than expected. This happens even when
the underlying string data is tiny.
Why it happens: When Arrow's
concatkernel combines batches that share the samebuffer allocations, it appends each batch's buffer list verbatim without checking for
duplicates. If N batches each hold a reference to the same K buffers, the result carries
N × K references — all pointing to the same memory. Every subsequent
CollectLeftjoinmultiplies the count again. On TPC-DS SF1 Q64 with
pushdown_filters = truewe measureda single column growing from 1 to 1,587,600 buffer references while the distinct
allocation count stayed at 2.
The two user-visible symptoms are:
RecordBatchMemoryCounter::count_buffer_memory_sizewalks everyreference on each poll. With millions of duplicate refs, this becomes the dominant
CPU cost.
Arc<Buffer>reference vectors themselves occupy memory,and downstream
takeoperations carry the full bloated buffer list into the next join.On TPC-DS SF1 Q64:
pushdown_filters = falsepushdown_filters = trueWhat changes are included in this PR?
A single new step is inserted immediately after the build-side
concat_batchescall inconcat_build_batches(hash_join/exec.rs):deduplicate_view_array_buffers<T>— for a singleGenericByteViewArray, walksthe
data_bufferslist, identifies duplicates by raw pointer address, and rewritesthe 4-byte
buffer_indexinside each non-inline view descriptor to point into thededuplicated buffer vector. No string bytes are copied. The fast path (≤ 1 buffer, or
no duplicates) returns a cheap
clone()of the array reference.deduplicate_record_batch_view_buffers— calls the above for everyUtf8Viewand
BinaryViewcolumn in aRecordBatch. Columns of other types are passed throughwith
Arc::clone.The deduplication runs once, on the concatenated build batch, before it is handed to the
join probe loop. All subsequent
takeoperations on the build side then start from aclean buffer list.
What is the testing strategy for this PR?
A new unit test
concat_build_batches_deduplicates_view_buffers(in thetestsmoduleof
hash_join/exec.rs) constructs threeRecordBatches that share the sameStringViewArrayallocation, concatenates them throughconcat_build_batches, andasserts that the resulting column holds exactly one buffer reference instead of three.
It also asserts the expected row count to confirm no data was dropped or duplicated.
The existing hash-join test suite continues to pass without modification, confirming the
change is a no-op for non-view columns and for view columns that already have unique
buffers.
Are there any user-facing changes?
No API changes. The fix is entirely internal to
concat_build_batches. Users runningqueries with chained
CollectLeftjoins onStringVieworBinaryViewcolumns willsee lower memory usage and faster query times without any configuration change.