[Status Update] Simplifing Streams to be more textbook-like and have less state while keeping the same perf

@pepijnve

If you wanna look at before and after example look at:

chore: refactor SortMergeJoin bitwise stream to generators and simplify to be textbook like as possible #23761

Status:

Background

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:

child return pending and have to keep state between calls
programming language as opposed to pseudo code
Flow is not linear as pseudo code since you can return only 1 value at a time
needing to handle spill
performance optimization

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:

Child return pending and have to keep state between calls
Flow is now linear since we can yield from the code and have the state and code in mind kept
Fewer lines - since less code needed for holding and managing the state
In some cases improve performance since going back to where you was in the state by making the function idompotent could be expensive

Problems that it is creating:

Timing is more annoying, you should pause the elapsed_compute timer between
i. yields - since you shouldn't count the time that the parent has done work between calling you
ii. awaits on child streams - because you don't want to count the child time
iii. awaits on stuff that you do async like 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)
Reserving memory is less intuitive since the code look linear but you hand off control between awaits so the data that you hold should be reserved for
You need to wrap the async generator with ObservedStream to track end time and output batches/rows/etc
can introduce performance problems since we are adding async machinery if not used with caution

Alternative solutions

We already have RecordBatchReceiverStreamBuilder but the flow of data is different - i.e. it is push based and not pull based which means:

more memory is being in the buffer (have at least 1 item pending
You produce data even if not needed
In case you are spawning blocked task to do all your work you can get to something like what was fixed here: Remove waits from blocking threads reading spill files. #15654

Initial Progress

I started the first PR in

chore: Simplifying SortPreservingMergeStream to use generators instead of state machine #23407

and @pepijnve kindly extracted a trimmed down version for the async generators from genawaiter crate that I previously used in that PR and tokio async-stream crate in:

Add minimal genarator-like stream implementation #23530

All the discussion for why we have our own implementation and also why not having macro implementation can be found in:

chore: Simplifying SortPreservingMergeStream to use generators instead of state machine #23407 (comment) - for why not using external crate or macro
chore: Simplifying SortPreservingMergeStream to use generators instead of state machine #23407 - why not using macro in the pr description

the first rewrite PR allowed for having the code rewritten to match simpler form:

chore: simplify SortPreservingMergeStream to be as textbook-like as possible #23702

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

@buraksenn sure, try to pick something simple but complex enough that the added cost of async won't be too much

Thanks. I've looked into this a bit and will start working on SingleHashAggregateStream if you've no objections. Otherwise I can pick up what you advise too.

chore: refactor `MaterializingSortMergeJoinStream` into generators an…

@rluvaton

Hey @rluvaton - could I take a stab at CrossJoinStream?

Sure

@rluvaton

@rluvaton thanks! I'll have a PR up for it in the next week or so :)

chore: refactor `MaterializingSortMergeJoinStream` into generators an…

@rluvaton

@rluvaton PR up #24291 ! Thanks!

refactor: CrossJoinStream (simplifying, less state, async generator p…

@rluvaton

@rluvaton / @2010YOUY01 - going to take NestedLoopJoinStream next if that's OK!

@rluvaton / @2010YOUY01 - going to take NestedLoopJoinStream next 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.

@2010YOUY01

@rluvaton / @2010YOUY01 - going to take NestedLoopJoinStream next 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 :)

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:
    NestedLoopStream

PR2: 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.

chore: convert `FinalHashAggregateStream` to async generators and cle…

chore: convert `PartialHashAggregateStream` to async generators and c…

perf: Improve output materializing speed for `OrderedFinalAggregateSt…

chore(`SingleHashAggregateStream`): refactor to async generator imple…

chore(`PartialReduceHashAggregateStream`): convert to async generator…

perf: Improve output materializing speed for `OrderedFinalAggregateSt…

chore(`SingleHashAggregateStream`): refactor to async generator imple…

chore(`PartialReduceHashAggregateStream`): convert to async generator…

Navigation Menu

Repository navigation

[Status Update] Simplifing Streams to be more textbook-like and have less state while keeping the same perf #23974

Background

Tradeoffs

Alternative solutions

Initial Progress

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

[Status Update] Simplifing Streams to be more textbook-like and have less state while keeping the same perf #23974

Description

Background

Tradeoffs

Alternative solutions

Initial Progress

Activity

buraksenn commented on Jul 29, 2026

rluvaton commented on Jul 29, 2026

buraksenn commented on Jul 29, 2026

saadtajwar commented on Aug 3, 2026

rluvaton commented on Aug 6, 2026

saadtajwar commented on Aug 7, 2026

saadtajwar commented on Aug 12, 2026

saadtajwar commented on Aug 18, 2026

2010YOUY01 commented on Aug 19, 2026

saadtajwar commented on Aug 20, 2026

2010YOUY01 commented on Aug 21, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions