rluvaton opened a new issue, #23974: URL: https://github.com/apache/datafusion/issues/23974
> If you wanna look at before and after example look at: > - https://github.com/apache/datafusion/pull/23761 Status: - [x] `SortPreservingMergeStream` - [x] https://github.com/apache/datafusion/pull/23407 - Moving to generators - [x] https://github.com/apache/datafusion/pull/23702 - Simplifying the code - [ ] `SortMergeJoin` - [x] `BitwiseSortMergeJoinStream` https://github.com/apache/datafusion/pull/23761 - moving to generators and simplifying the code - [ ] `Aggregate` - [ ] `OrderedPartialAggregateStream` - https://github.com/apache/datafusion/pull/23951 - moving to generators, the code is already pretty simple Possible Candidates: - [ ] Aggregate - [ ] `SingleHashAggregateStream` - [ ] `OrderedFinalAggregateStream` - [ ] `PartialHashAggregateStream` - [ ] `PartialReduceHashAggregateStream` - [ ] `FinalHashAggregateStream` - [ ] Join - [ ] `SortMergeJoin` - [ ] `MaterializingSortMergeJoinStream` # 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 <details> <summary>Algorithm</summary> Taken from [Sort-Merge Joins](https://www.dcs.ed.ac.uk/home/tz/phd/thesis/node20.htm) <img width="555" height="904" alt="Image" src="https://github.com/user-attachments/assets/29d68c83-932d-4471-902b-b0545e372523"; /> </details> But due to all the following reasons: 1. child return pending and have to keep state between calls 2. programming language as opposed to pseudo code 3. Flow is not linear as pseudo code since you can return only 1 value at a time 4. needing to handle spill 5. 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: 1. Child return pending and have to keep state between calls 2. Flow is now linear since we can yield from the code and have the state and code in mind kept 3. Fewer lines - since less code needed for holding and managing the state 4. 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: 1. Timing is more annoying, you should pause the `elapsed_compute` timer between i. `yield`s - since you shouldn't count the time that the parent has done work between calling you ii. `await`s on child streams - because you don't want to count the child time iii. `await`s 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) 2. Reserving memory is less intuitive since the code look linear but you hand off control between `await`s so the data that you hold should be reserved for 3. You need to wrap the async generator with `ObservedStream` to track end time and output batches/rows/etc 4. 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: 1. more memory is being in the buffer (have at least 1 item pending 2. You produce data even if not needed 3. In case you are spawning blocked task to do all your work you can get to something like what was fixed here: https://github.com/apache/datafusion/pull/15654 ## Initial Progress I started the first PR in - #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: - #23530 All the discussion for why we have our own implementation and also why not having macro implementation can be found in: - https://github.com/apache/datafusion/pull/23407#discussion_r3554039843 - for why not using external crate or macro - #23407 - why not using macro in the pr description the first rewrite PR allowed for having the code rewritten to match simpler form: - https://github.com/apache/datafusion/pull/23702e -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
