Hash join buffering on probe side #19761
Open
+973
−10
Conversation
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
github-actions
bot
added
optimizer
This was referenced Jan 12, 2026
Contributor
Author
|
run benchmarks |
|
🤖 Hi @gabotechs, thanks for the request (#19761 (comment)). |
github-actions
bot
added
the
documentation
Contributor
Author
|
run benchmarks |
|
🤖 |
|
🤖: Benchmark completed Details
|
Contributor
Author
|
run benchmark tpcds tpch10 |
|
🤖 |
|
Benchmark script failed with exit code 1. Last 10 lines of output: Click to expand |
Contributor
Author
|
run benchmark tpch10 |
|
🤖 |
|
🤖: Benchmark completed Details
|
Contributor
Author
|
run benchmark tpch |
|
🤖 |
|
🤖: Benchmark completed Details
|
Contributor
Author
|
run benchmark tpcds |
|
🤖 |
|
Benchmark script failed with exit code 1. Last 10 lines of output: Click to expand |
Contributor
Author
|
🤔 the tpcds benchmark command seems broken |
Contributor
|
Big picture question: could this be part of the HashJoinExec code? |
Contributor
Author
Potentially yes, although I do see value in having this be its own node, that way we can:
This is actually something we (DataDog) have had for a long time, we have a similar |
gabotechs
force-pushed
the
hash-join-buffering-on-probe-side
branch
2 times, most recently
from
6fa7c76 to
139cf50
Compare
Contributor
Author
|
run benchmark tpcds tpch |
|
🤖 |
|
🤖: Benchmark completed Details
|
|
🤖 |
|
🤖: Benchmark completed Details
|
Contributor
Author
|
I think for this to be optimal further complex logic will need to be introduced so that buffering can properly interact with dynamic filtering. However, even if we decide to not buffer eagerly in case there's a dynamic filter below a I want to propose moving forward with this, but with buffering disabled by default. That way, users can decide to opt into it until we find a smarter rule of when to apply buffering automatically based on the query shape and where are dynamic filters placed. The steps I want to propose are:
@Dandandan what do you think? |
github-merge-queue bot
pushed a commit
that referenced
this pull request
Jan 26, 2026
## Which issue does this PR close? <!-- We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123. --> - Closes #. ## Rationale for this change <!-- Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed. Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes. --> Prerequisite for the following PRs: - #19760 - #19761 Even if the api on the `MemoryPool` does not require `&mut self` for growing/shrinking the reserved size, the api in `MemoryReservation` does, making simple implementations irrepresentable without synchronization primitives. For example, the following would require a `Mutex` for concurrent access to the `MemoryReservation` in different threads, even though the `MemoryPool` doesn't: ```rust let mut stream: SendableRecordBatchStream = SendableRecordBatchStream::new(); let mem: Arc<MemoryReservation> = Arc::new(MemoryReservation::new_empty()); let mut builder = ReceiverStreamBuilder::new(10); let tx = builder.tx(); { let mem = mem.clone(); builder.spawn(async move { while let Some(msg) = stream.next().await { mem.try_grow(msg.unwrap().get_array_memory_size()); // ❌ `mem` is not mutable tx.send(msg).unwrap(); } }); } builder .build() .inspect_ok(|msg| mem.shrink(msg.get_array_memory_size())); // ❌ `mem` is not mutable ``` ## What changes are included in this PR? <!-- There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR. --> Make the methods in `MemoryReservation` require `&self` instead of `&mut self` for allowing concurrent shrink/grows from different tasks for the same reservation. ## Are these changes tested? <!-- We typically require tests for all PRs in order to: 1. Prevent the code from being accidentally broken by subsequent changes 2. Serve as another way to document the expected behavior of the code If tests are not included in your PR, please explain why (for example, are they covered by existing tests)? --> yes, by current tests ## Are there any user-facing changes? Users can now safely call methods of `MemoryReservation` from different tasks without synchronization primitives. This is a backwards compatible API change, as it will work out of the box for current users, however, depending on their clippy configuration, they might see some new warnings about "unused muts" in their codebase. <!-- If there are user-facing changes then we may require documentation to be updated before approving the PR. --> <!-- If there are any breaking changes to public APIs, please add the `api change` label. -->
gabotechs
force-pushed
the
hash-join-buffering-on-probe-side
branch
from
139cf50 to
3bd48fc
Compare
github-actions
bot
removed
core
gabotechs
force-pushed
the
hash-join-buffering-on-probe-side
branch
2 times, most recently
from
41e24b7 to
5a4f11f
Compare
gabotechs
force-pushed
the
hash-join-buffering-on-probe-side
branch
from
5a4f11f to
d2551e7
Compare
Contributor
Author
|
I want to propose moving forward with #19760 and this PR with the config parameter disabled by default, as it shows performance improvements too god to dismiss (see PR description for updated benchmarks results). cc @adriangb as it's tangentially related with dynamic filters and @Dandandan as you seem to have interest in improving join performance. |
Contributor
|
run benchmarks |
|
🤖 |
|
🤖: Benchmark completed Details
|
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
5 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?
It does not close any issue, but it's related to:
Rationale for this change
This is a PR from a batch of PRs that attempt to improve performance in hash joins:
It adds the new
BufferExecnode at the top of the probe side of hash joins so that some work is eagerly performed before the build side of the hash join is completely finished.Why should this speed up joins?
In order to better understand the impact of this PR, it's useful to understand how streams work in Rust: creating a stream does not perform any work, progress is just made if the stream gets polled.
This means that whenever we call
.execute()on anExecutionPlan(like the probe side of a join), nothing happens, not even the most basic TCP connections or system calls are performed. Instead, all this work is delayed as much as possible until the first poll is made to the stream, losing the opportunity to make some early progress.This gets worst when multiple hash joins are chained together: they will get executed in cascade as if they were domino pieces, which has the benefit of leaving a small memory footprint, but underutilizes the resources of the machine for executing the query faster.
Note
Even if this shows overall performance improvement in the benchmarks, it can show performance degradation on queries with dynamic filters, so hash join buffering is disabled by default, and users can opt in.
Follow up work will be needed in order to make this interact well with dynamic filters.
What changes are included in this PR?
Adds a new
HashJoinBufferingphysical optimizer rule that will idempotently placeBufferExecnodes on the probe side of has joins:Are these changes tested?
yes, by existing tests
Are there any user-facing changes?
Not by default, users can now opt in to this feature with the hash_join_buffering_capacity config parameter.
Results
Note
Note that a small number of TPC-DS queries have regressed, this is because with eager buffering they do not benefit from dynamic filters as much. This is the main reason for leaving this config parameter disabled by default until we have a proper way for interacting with the dynamic filters inside the BufferExec node.