apache / apache/datafusion

perf: Improve Latency by Eagerly Fetching Small Scans

Open
#24,922 0 comments 0 reactions 0 assignees View on GitHub
enhancement
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.