apache / apache/datafusion

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

Aperta
#23,974 11 commenti 0 reazioni 0 assegnatari Vedi su GitHub
Lingua principale
Rust
Stelle
9.3k
Fork
2.4k
Merge medio
3g 11h
PR unite (30g)
360

Descrizione

> 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
- [x] `SortMergeJoin`
- [x] `BitwiseSortMergeJoinStream` https://github.com/apache/datafusion/pull/23761 - moving to generators and simplifying the code
- [x] `MaterializingSortMergeJoinStream` - https://github.com/apache/datafusion/pull/23976
- [ ] `Aggregate`
- [x] `OrderedPartialAggregateStream` - https://github.com/apache/datafusion/pull/23951 - moving to generators, the code is already pretty simple
- [ ] `OrderedFinalAggregateStream` - https://github.com/apache/datafusion/pull/24008
- [ ] `PartialReduceHashAggregateStream ` - https://github.com/apache/datafusion/pull/24015
- [x] `PartialHashAggregateStream` - https://github.com/apache/datafusion/pull/24017
- [x] `FinalHashAggregateStream` - https://github.com/apache/datafusion/pull/24874
- [ ] `SingleHashAggregateStream` - https://github.com/apache/datafusion/pull/24016

# 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](https://www.dcs.ed.ac.uk/home/tz/phd/thesis/node20.htm)

Image

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

Guida per i contributori

Apri la guida per i contributori

Direzione di ricerca

Questo è un aggiornamento sullo stato piuttosto che un’attività circoscritta, e gran parte del lavoro è già rappresentata dalle pull request collegate. Leggi l’esempio prima-dopo in PR #23761, quindi esamina gli Aggregate streams non selezionati—OrderedFinalAggregateStream, PartialReduceHashAggregateStream e SingleHashAggregateStream—per comprendere l’ambito rimanente.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Valutazione

Stack tecnologico
rust
Ambito
data-engineering
Tipo di issue
Refactoring
Difficoltà
5/5
Tempo stimato
Più di una settimana
Stato di attività
Ferma
Chiarezza
Da chiarire
Idoneità per principianti
25/100

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.