Parquet Pushdown Filters: Way to keep the I/O pattern the same when pushdown filters are enabled
- 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?
- Related to https://github.com/apache/datafusion/issues/3463
- This is an idea for https://github.com/apache/datafusion/issues/20324
Namely, the idea is to avoid regressions when we turn on filter pushdown
We haven't been able to turn on predicate pushdown because it caused slowdowns. The slowdowns were caused by two reasons:
1. predicate pushdown changes the I/O patterns
2. Evaluating predicates is computationally slower in some cases (figuring out bitmasks, rows that passed, etc)
I think after the great work from @hhhizzz @haohuaijin and others upstream in arrow-rs/parquet we are close (if not already) at the point where it is computationally the same or faster
All that remains is the different I/O
The I/O pattern today is
1. Default: read all column data needed for both filtering and projections when reading a row group
2. Predicate Pushdown (`pushdown_filters`): progressive I/O: sequentially fetch the columns required to evaluate filter expressions, and then after filtering is complete, fetches columns needed to evaluate the projections. This stratgey (can) fetch fewer pages but results in more I/O requests (which can increase latency)
@Dandandan / @adriangb also noted on https://github.com/apache/datafusion/pull/23492#issuecomment-5297913079 that we can't prune I/O for files without a page index (notably the clickbench dataset) anyways , so other than peak buffer needs, there is no benefit to changing the I/O to read columns sequentially in that case
### Describe the solution you'd like
I would like is some way to turn on filter pushdown but **keep the I/O pattern the same as today** (aka read all required data pages for a Row Gorup) up front in one request
That way, once we sort out any lingering CPU slowdowns, we can turn on predicate pushdown by default and leave the I/O pattern as a policy choice (as the I/O policy will likely different for different I/O systems -- e.g. NVMe vs ObjectStore.
Note a small variation could be to automatically use one-shot I/O when the file has no offset/page index (since progressive fetch can't prune I/O there anyway), and use progressive I/O only when the index exists.
### Describe alternatives you've considered
So I think we should separate out the notions of "filter evaluation mechanics" from the I/O pattern.
One idea is to add a parquet configuration setting like 'progressive_io' (defaults to false, the behavior today). When true it would fetch data progressively, as happens today when filter_pushdown is enabled
So we would have a feature matrix like
| filter_pushdown | progressive_io | description |
|--------|--------|--------|
| false | false | Same as todays behavior -- one shot I/O |
| true | false | filter pushdown, but still keep same I/O (my proposed new default) |
| false | true | N/A (just a decode, so no progressive filtering)
| true | true | Progressive filter evaluation and I/O (what happens today when `filter_pushdown`) is enabled
### Additional context
_No response_
Contributor guide
Research direction
Start by reading linked issues 3463 and 20324, then review the discussion on pull request 23492 for existing context on Parquet pushdown and I/O behavior. Done means defining and implementing a policy that separates filter-pushdown mechanics from progressive versus one-shot I/O, with the requested default behavior and feature combinations documented or tested.
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
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 42/100