mem: Cleanup resources of done streams immediately#22064
Open
EmilyMatt wants to merge 6 commits intoapache:mainfrom
Open
mem: Cleanup resources of done streams immediately#22064EmilyMatt wants to merge 6 commits intoapache:mainfrom
EmilyMatt wants to merge 6 commits intoapache:mainfrom
Conversation
rluvaton
reviewed
May 7, 2026
Comment on lines
1621
to
1622
| // No data at all — go directly to Done | ||
| self.left_exhausted = true; |
Member
There was a problem hiding this comment.
you forgot to update this as well
Contributor
Author
There was a problem hiding this comment.
No bc above it can be converted back to false.
I only placed it there bc if at the point I placed it it is set to true, it means the input is actually depleted.
rluvaton
reviewed
May 7, 2026
Comment on lines
+1614
to
+1618
| // If the left stream is fully exhausted, release its resources so the | ||
| // upstream pipeline can be torn down before we move on to probing. | ||
| if self.left_exhausted { | ||
| active.left_stream = None; | ||
| } |
Member
There was a problem hiding this comment.
maybe there are cases that left exhausted is true and left stream is not None, because what is the reason to have both exhausted flag and Option
rluvaton
reviewed
May 7, 2026
Comment on lines
+285
to
+287
| // Release the input pipeline's resources. | ||
| this.inner = | ||
| Box::pin(EmptyRecordBatchStream::new(Arc::clone(&this.schema))); |
Member
There was a problem hiding this comment.
I think to avoid bugs the schema for inner should be kept the same and not use this.schema (even when they are the same schema) because you only want to release the stream
Contributor
Author
There was a problem hiding this comment.
Yeah fair point, applied
rluvaton
reviewed
May 7, 2026
Contributor
Author
Applied |
Weijun-H
approved these changes
May 7, 2026
Comment on lines
+1531
to
+1533
| // Release the right input pipeline's resources. | ||
| let right_schema = self.right_stream.schema(); | ||
| self.right_stream = Box::pin(EmptyRecordBatchStream::new(right_schema)); |
Member
There was a problem hiding this comment.
nit:
extract as a helper function
Member
|
run benchmarks |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing resource-cleanup (e2e4372) to 3b634aa (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing resource-cleanup (e2e4372) to 3b634aa (merge-base) diff using: clickbench_partitioned File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing resource-cleanup (e2e4372) to 3b634aa (merge-base) diff using: tpch 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 |
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
4 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?
Rationale for this change
Reduces memory pressure, cleans up resources eagerly, and makes pools aware that operators are done by dropping their MemoryReservation and MemoryConsumers.
What changes are included in this PR?
Whenever a stream is polled and returns None(is depleted), drops that stream, or replaces it with EmptyRecordBatchStream.
Are these changes tested?
This should have no effect on logic, as the streams are already depleted.
Are there any user-facing changes?
No, users implementing their own memory pool can expect to see the consumer count drop whenever a stream is released, but that is well within parameters and I don't think is considered a change at all.