perf: Improve Latency by Eagerly Fetching Small Scans
- Dominant language
- Rust
- Stars
- 9.3k
- Forks
- 2.4k
- Avg merge
- 3d 7h
- Merged PRs (30d)
- 344
Description
### 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_
Contributor guide
Research direction
Start with HashJoinExec and the proposed implementation in PR 24921; review how sequential scans, dynamic filters, and DataSources interact. Done means small scans can be fetched in parallel while larger scans wait for dynamic filters, with query latency behavior validated against the described object-store scenario.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- data-engineering, performance
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100