perf: Improve Latency by Eagerly Fetching Small Scans
- Lingua principale
- Rust
- Stelle
- 9.3k
- Fork
- 2.4k
- Merge medio
- 3g 11h
- PR unite (30g)
- 360
Descrizione
### Is your feature request related to a problem or challenge?
Currently, executing a deep join tree (`HashJoinExec`) is done sequentially. First table A is scanned, the hash table is being computed, dynamic filters are updated, and then table B is scanned. This is in general not a bad thing. The rows that become available during scanning table A may help us to significantly prune the scan on table B due to dynamic filters, speeding up the overall query.
However, when a deep join tree requires scanning many --- possibly small tables --- the sequential scans will lead to high latency in an object-store-based environment. First table A is scanned, after 150 ms table B is scanned, etc. Especially in graph query languages like SPARQL (which we use DataFusion for), deep join trees are quite common as the graph patterns get translated into a series of joins.
Ideally, we would like to execute queries by i) doing all small scans in parallel and ii) delaying larger scans until dynamic filters have been computed. The rationale is that, when the scans are small (e.g., a single request), dynamic filters likely will not improve the performance by a large margin (maybe except if the filter remove all data). Then, all small scans (and the first scan in the join tree) are fetched in parallel, while large scans wait until their dynamic filters have been computed, avoiding the problems of why https://github.com/apache/datafusion/pull/19761 is not yet the default behavior in DF.
### Describe the solution you'd like
Proposal in https://github.com/apache/datafusion/pull/24921
### Describe alternatives you've considered
One alternative is going with the "buffer probe sides" approach. I think it's more difficult to get it right, as the probe side could be anything. For example, this could be a complex join that happens to have favourable statistics at the top-level node but not at the intermediate results, resulting in the query executor to evaluate a complex subplan in parallel.
Another alternative would a general "prefetching" infrastructure for `DataSource`s where the `DataSource` decides whether to prefetch or not.
### Additional context
_No response_
Guida per i contributori
Apri la guida per i contributori
Direzione di ricerca
Inizia con HashJoinExec e l’implementazione proposta in PR 24921; esamina come interagiscono le scansioni sequenziali, i filtri dinamici e DataSources. Il lavoro è completato quando le scansioni piccole possono essere recuperate in parallelo mentre quelle più grandi attendono i filtri dinamici, e il comportamento della latenza delle query è stato convalidato rispetto allo scenario di object-store descritto.
Scritto dal modello di indicizzazione a partire dal testo della issue.
Valutazione
- Stack tecnologico
- rust
- Ambito
- data-engineering, performance
- Tipo di issue
- Funzionalità
- Difficoltà
- 5/5
- Tempo stimato
- Più di una settimana
- Stato di attività
- Ferma
- Chiarezza
- Abbastanza chiara
- Idoneità per principianti
- 35/100