fix: ensure a maximum of buffer_len RecordBatches are cached in spawn_buffered#23560
Merged
Conversation
rluvaton
changed the title
buffer_len RecordBatches are cached in spawn_buffered
Member
|
run benchmarks |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing fix-23559 (cc4945a) to 79692c1 (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing fix-23559 (cc4945a) to 79692c1 (merge-base) diff using: clickbench_partitioned File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing fix-23559 (cc4945a) to 79692c1 (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 Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
rluvaton
reviewed
Jul 14, 2026
| use std::sync::atomic::{AtomicUsize, Ordering}; | ||
|
|
||
| let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)])); | ||
| let buffer_size = 1; |
Member
There was a problem hiding this comment.
can you have the same test but with buffer_size 2 as well? (just extract the test code and get an argument for buffer_size)
rluvaton
reviewed
Jul 14, 2026
Comment on lines
+606
to
+607
| // With reserve(), the producer acquires a channel slot before pulling from | ||
| // input, so at most buffer_size batches can exist in memory simultaneously. |
Member
There was a problem hiding this comment.
this (reserve call) refer to current code that can become outdated and then this comment will be confusing.
rluvaton
reviewed
Jul 14, 2026
Comment on lines
+200
to
206
| while let Ok(permit) = sender.reserve().await { | ||
| // Receiver dropped when query is shutdown early (e.g., limit) or error, | ||
| // no need to return propagate the send error. | ||
| match input.next().await { | ||
| Some(item) => permit.send(item), | ||
| None => break, | ||
| } |
Member
There was a problem hiding this comment.
Please add the comment for the reserve here rather than in the test code.
something like
/// We call `reserve` (which wait to have capacity of at least 1 message in the channel buffer) **before** polling from input
/// since if we first polled from input and then wait for the channel to have capacity for the message we will essentially hold `buffer` (in the channel) + 1 (waiting for the channel to have capacity)
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
3 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?
Fixes: #23559
Rationale for this change
See the issue description
What changes are included in this PR?
Change spawn_buffered's behavior to ensure a maximum of
bufferRecordBatches are produced before any consumer polls the stream.Are these changes tested?
Yes
Are there any user-facing changes?
No