Repository navigation
[Status Update] Simplifing Streams to be more textbook-like and have less state while keeping the same perf #23974
Description
Activity
I would like to try one of them to learn more about streams while trying to simplify it. Are you open to help on this?
@buraksenn sure, try to pick something simple but complex enough that the added cost of async won't be too much
Reacted by Burak ŞenThanks. I've looked into this a bit and will start working on
SingleHashAggregateStreamif you've no objections. Otherwise I can pick up what you advise too.Reacted by Raz Luvaton- added a commit that references this issue
on Jul 30, 2026 Hey @rluvaton - could I take a stab at
CrossJoinStream?Sure
@rluvaton thanks! I'll have a PR up for it in the next week or so :)
- added a commit that references this issue
on Aug 12, 2026 - added a commit that references this issue
on Aug 18, 2026 @rluvaton / @2010YOUY01 - going to take
NestedLoopJoinStreamnext if that's OK!@rluvaton / @2010YOUY01 - going to take
NestedLoopJoinStreamnext if that's OK!Thank you! I have a few thoughts you could consider. I’m only around 70% confident in them, so please only use them as suggestion.
- For the complexity in NLJ, we might want to keep explicit state management, with the current state represented as an
enum, and let that coexist with the generator pattern. My reasoning is:
a) We probably need a state-transition diagram to fully understand the implementation anyway, and structuring the code similarly makes it easier to reason about.
b) Explicit states make the entry and exit conditions for each state visible, which may make the implementation safer. - We could split the regular joins (inner, left, right, full) and the semi/anti joins into separate streams. They are really different relational operations, and different optimizations tend to apply to each. The current approach combines them and therefore needs several flags/configurations to route the internal logic, which adds complexity. We have already made a similar split in sort-merge join.
This second point may be slightly outside the scope of this project, but since we are already doing a fairly large refactor, I wanted to mention it briefly.
- For the complexity in NLJ, we might want to keep explicit state management, with the current state represented as an
@rluvaton / @2010YOUY01 - going to take
NestedLoopJoinStreamnext if that's OK!Thank you! I have a few thoughts you could consider. I’m only around 70% confident in them, so please only use them as suggestion.
- For the complexity in NLJ, we might want to keep explicit state management, with the current state represented as an
enum, and let that coexist with the generator pattern. My reasoning is:
a) We probably need a state-transition diagram to fully understand the implementation anyway, and structuring the code similarly makes it easier to reason about.
b) Explicit states make the entry and exit conditions for each state visible, which may make the implementation safer. - We could split the regular joins (inner, left, right, full) and the semi/anti joins into separate streams. They are really different relational operations, and different optimizations tend to apply to each. The current approach combines them and therefore needs several flags/configurations to route the internal logic, which adds complexity. We have already made a similar split in sort-merge join.
This second point may be slightly outside the scope of this project, but since we are already doing a fairly large refactor, I wanted to mention it briefly.
Thanks for the detailed thoughts @2010YOUY01 ! This definitely makes sense to me - it seems like it might be a bit easier to split this into separate PRs of:
- First separating regular joins from semi/anti/mark joins (preserving the existing polling behavior/spilling/state machines)
- Then converting the streams to generators independently
If that makes sense to you then I'll go ahead and start on implementation :)
- For the complexity in NLJ, we might want to keep explicit state management, with the current state represented as an
- First separating regular joins from semi/anti/mark joins (preserving the existing polling behavior/spilling/state machines)
- Then converting the streams to generators independently
If you decide to split the semi/anti/mark joins, perhaps we can
PR1: Keep existing implementation, and implement a new stream for semi/anti/mark joins directly with the generator pattern
if !standard_join: SemiAntiStream else: NestedLoopStreamPR2: Implement the standard join similarly, and delete the legacy implementation
If the generator pattern produces overall simpler solution, then directly implement it this way might be easier 🤔, but anyway this is still a full rewrite, can be a little tricky, we can try and see how to split the work down to smaller PRs further.
BTW, we could open a new issue and continue the discussion there.
- added a commit that references this issue
on Sep 2, 2026 - added a commit that references this issue
on Sep 6, 2026 - added a commit that references this issue
on Sep 25, 2026 - added a commit that references this issue
on Oct 3, 2026 - added a commit that references this issue
on Oct 6, 2026
Status:
SortPreservingMergeStreamSortPreservingMergeStreamto use generators instead of state machine #23407 - Moving to generatorsSortMergeJoinBitwiseSortMergeJoinStreamchore: refactor SortMergeJoin bitwise stream to generators and simplify to be textbook like as possible #23761 - moving to generators and simplifying the codeMaterializingSortMergeJoinStream- chore: refactorMaterializingSortMergeJoinStreaminto generators and simplify code to be textbook like as possible #23976AggregateOrderedPartialAggregateStream- chore(ordered-partial-aggregate): moveOrderedPartialAggregateStreamto generators for readability #23951 - moving to generators, the code is already pretty simpleOrderedFinalAggregateStream- perf: Improve output materializing speed forOrderedFinalAggregateStream#25639PartialReduceHashAggregateStream- chore(PartialReduceHashAggregateStream): convert to async generators and cleanup #24015PartialHashAggregateStream- chore: convertPartialHashAggregateStreamto async generators and cleanup #24017FinalHashAggregateStream- chore: convertFinalHashAggregateStreamto async generators and cleanup #24874SingleHashAggregateStream- chore(SingleHashAggregateStream): refactor to async generator implementation #24016Background
So, I start seeing that some code that on the surface should be dead simple in textbook form in reality the main entry point and the whole state handling is really complex
For example SortMergeJoin, on the surface the algorithm is simple
Algorithm
Taken from Sort-Merge Joins
But due to all the following reasons:
the actual implementation is really complex which means that doing any sort of PR there is hard to review or to understand
Tradeoffs
So I started with moving some stuff to async generators that solve some of the problems while adding others.
problem that it is solving:
Problems that it is creating:
elapsed_computetimer betweeni.
yields - since you shouldn't count the time that the parent has done work between calling youii.
awaits on child streams - because you don't want to count the child timeiii.
awaits on stuff that you doasynclike reading a spill file - because even though the on the surface this is your work, you might overcount the elapsed time and the spill reading finish earlier but you did not woke up (up for debate)awaits so the data that you hold should be reserved forObservedStreamto track end time and output batches/rows/etcAlternative solutions
We already have
RecordBatchReceiverStreamBuilderbut the flow of data is different - i.e. it is push based and not pull based which means:Initial Progress
I started the first PR in
SortPreservingMergeStreamto use generators instead of state machine #23407and @pepijnve kindly extracted a trimmed down version for the async generators from
genawaitercrate that I previously used in that PR and tokioasync-streamcrate in:All the discussion for why we have our own implementation and also why not having macro implementation can be found in:
SortPreservingMergeStreamto use generators instead of state machine #23407 (comment) - for why not using external crate or macroSortPreservingMergeStreamto use generators instead of state machine #23407 - why not using macro in the pr descriptionthe first rewrite PR allowed for having the code rewritten to match simpler form: