apache / apache/datafusion

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

Offen
#23,974 11 Kommentare 0 Reaktionen 0 zugewiesene Personen Auf GitHub ansehen
Vorherrschende Sprache
Rust
Sterne
9.3k
Forks
2.4k
Ø Merge
3 T. 11 Std.
Gemergte PRs (30 T.)
360

Beschreibung

> 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

Beitragsleitfaden

Beitragsleitfaden öffnen

Rechercherichtung

Dies ist eine Statusaktualisierung und keine abgegrenzte Aufgabe, und ein Großteil der Arbeit ist bereits durch die verknüpften Pull Requests abgebildet. Lies das Vorher-nachher-Beispiel in PR #23761 und überprüfe anschließend die nicht abgehakten Aggregate streams—OrderedFinalAggregateStream, PartialReduceHashAggregateStream und SingleHashAggregateStream—, um den verbleibenden Umfang zu verstehen.

Vom Indexierungsmodell aus dem Issue-Text verfasst.

Bewertung

Tech-Stack
rust
Bereich
data-engineering
Issue-Typ
Refactoring
Schwierigkeit
5/5
Geschätzter Aufwand
Über eine Woche
Aktivitätsstatus
Veraltet
Klarheit
Muss geklärt werden
Anfängerfreundlichkeit
25/100

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.