[EPIC] Faster performance for parquet predicate evaluation for non selective filters

[alamb]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.

• Related to Enable parquet filter pushdown (filter_pushdown) by default datafusion#3463 in DataFusion.

When evaluating filters on data stored in parquet, you can:

1. Use the with_row_filter API to apply predicates during the scan
2. Read the data and apply the predicate using the filter kernel afterwards

Currently, it is faster to use with_row_filter for some predicates and filter for others. In DataFusion we have a configuration setting to choose between the strategies (filter_pushdown, see apache/datafusion#3463) but that is a bad UX as it
means the user must somehow know which strategy to choose, but the strategy changes

In general the queries that are slower when with_row_filter is used:

1. The predicates are not very selective (e.g. they pass more than 1% of the rows)
2. The filters are applied to columns which are also used in the query result (e.g. the a filter column is also in the projection)

More Background:

The predicates are provides as a RowFilter (see docs for more details)

RowFilter applies predicates in order, after decoding only the columns required. As predicates eliminate rows, fewer rows from subsequent columns may be required, thus potentially reducing IO and decode.

Describe the solution you'd like

I would like the evaluation of predicates in RowFilter (aka pushed down predicates) to never be worse than decoding the columns first and then filtering them with the filter kernel

We have added a benchmark #7401, which hopefully can

cargo bench --all-features --bench arrow_reader_row_filter

Describe alternatives you've considered
This goal will likely require several changes to the codebase. Here are some options:

• Add benchmark for parquet reader with row_filter and project settings #7401
• arrow_reader_row_filter benchmark doesn't capture page cache improvements #7460
• Benchmark for filter+concat and take+concat into even sized record batches #7589
• Parquet decoder / decoded predicate / page / results Cache #7363
• Adaptive Parquet Predicate Pushdown Evaluation #5523
• Consider removing skip from RowSelector #7450
• RowSelection::and_then is slow #7458
• improve: reuse Arc<dyn Array> in parquet record batch reader. #4864

[alamb]

I just spoke with @XiangpengHao -- from my perspective the current status is:

1. Parquet decoder / decoded predicate / page / results Cache #7363: blocked on getting some benchmark results that show the decoded page cache improves performance; Then we can proceed / merge the page cache change
2. In paralell / then we can move on to working on a better representation for RowFilter (Adaptive Parquet Predicate Pushdown Evaluation #5523 / Consider removing skip from RowSelector #7450 / RowSelection::and_then is slow #7458)

[alamb]

Fascinatingly, Clickbench released a blog post recently about their parquet pushdown work

https://clickhouse.com/blog/clickhouse-and-parquet-a-foundation-for-fast-lakehouse-analytics

Possibly even more interesting is that they link to a master's thesis from Peter Boncz's group about how to quickly evaluate predicates during Parquet Decoding: https://homepages.cwi.nl/~boncz/msc/2018-BoudewijnBraams.pdf

This thesis directly addresses some of the work we are considering (though they only consider Selection masks (bitmask) and Selection Vector (selected indices)

[alamb]

@zhuqi-lucas has a great insight in #7454 -- namely that instead of a two pass algorithm (evaluate RowFilter to form a final RowSelection and then re-decode the filter) we can combine the filter application and decode steps (see #7454 (review))

The current flow goes something like:

1. A set of array readers is created for the filter columns, and uses the provided RowSelection (this captures prunning
   pages ).
2. The decoded batches are used to evaluate the RowFilter / ArrowPredicates, which produces a BooleanArray bitmap
3. The "final" RowSelection is created, by unioning the existing RowSelection with the BooleanArrays
4. A new set of array readers is created with the updated RowSelection

The current PR starts heading down a slightly modified flow, where the RowSelection and RowFilters are not combined.

I think a combined solution would look something like:

1. Create Decoders for filter columns and projection (only) columns

Decoding proceeds like:

1. read rows from initial RowSelection (reads a 8192 rows) from filter columns, if any
2. Apply any RowFilters on it (produces a BooleanArray)
3. repeat 1-2 until there are at least 8192 (batch size) rows that pass the filter. (This means we have Vec<BooleanArray> with 8192 1s and a Vec for each filter column that is also a projection column)
4. Then decode as maby RecordBatches from the projection (only) columns using the initial RowSelection)
5. Apply the filters to each array to form the final output batch (in projection columns)

[alamb]

@zhuqi-lucas and I have been working on various strategies / structures to make the filtering faster. I believe we now have evidence enough to proceed with a more sophisticated implementation

Specifically,

• @zhuqi-lucas has shown the hybrid Filter/RowSelection approach works well in Poc for adaptive parquet predicate pushdown(bitmap/range) with page cache(3 data pages) #7454
• I have shown the idea of reusing actual filter results works well in POC: Sketch out parquet cached filter result API #7513 (though I still need to work out how to limit memory usage more, potentially adaptively)

Thus my next steps will be:

1. Create a few refactoring PRs that gets the predicate code into shape

Perhaps then @zhuqi-lucas can help port the hybrid Filter/RowSelection to the ReadPlan to get better performance without changing any public interfaces