Repository navigation
Support RowFilter in ParquetExec #3360
Description
Is your feature request related to a problem or challenge? Please describe what you are trying to do.
A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]
(This section helps Arrow developers understand the context and why for this feature, in addition to the what)
Describe the solution you'd like
A clear and concise description of what you want to happen.
arrow-rs has recently added the ability to do row-level filtering while decoding parquet files. This can dramatically reduce decoding and IO overhead when appropriately selective pruning predicates are pushed down to the table scan. We should support this in ParquetExec
Describe alternatives you've considered
A clear and concise description of any alternative solutions or features you've considered.
If a ParquetExec has a PruningPredicate it should be "compiled" to a vector of ArrowPredicate(s) and supplied as a RowFilter to the ParquetRecordBatchStream. We can implement this a couple of different ways:
- Take the pruning
Expr, create aPhysicalExprand then implement a singleArrowPredictaeFnwhich evaluates it. - Break apart conjunctions in the pruning
Exprand compile each to a separateArrowPredicateFnthat will be applied sequentially. We can either take the ordering as given or apply some heuristic to determine the ordering.
Some considerations:
- We need to be careful to avoid redundant computations. As filter predicates are pushed down in the optimizer you may end up with a predicate on the scan involving the output a scalar function and also a projection in a later stage involving the same computation. If you naively execute all filter predicates during the scan you may end up doing the same work twice.
- The filtering is not free. At minimum you may end up decoding the same data multiple times.
- Currently, if users have a custom
TableProviderthey can control what gets pushed down. Is that enough configurability? - Breaking the predicate into multiple filters can help in some situations if you can get the ordering correct (cheap filter first, expensive filters after) but if you get it wrong then it can be bad. This may be amplified as we add the ability to parallelize the column decoding (i.e. as the morsel-driven scheduler progresses).
Additional context
Add any other context or screenshots about the feature request here.
Activity
thinkharderdev commentedon Sep 5, 2022
@tustvold @Ted-Jiang I've been tinkering with this and can submit an RFC today or tomorrow for feedback. If you guys already have thought though we can discuss here :)
tustvold commentedon Sep 5, 2022
100% we should get this integrated 👍, awesome that you're working on this. Some miscellaneous thoughts:
Ted-Jiang commentedon Sep 5, 2022
@thinkharderdev Wow! So looking forward！ 💪
Nice write up! Thanks👍
I think one thing we should talk about , how to define the
non-selective predicates (expensive predicate).I think for now if we want to check wether is a predicate selective on no-sorted col , we need know the the result page number, so we need read
col-index. if we filter zero page, it will run slower than before.🤔Ted-Jiang commentedon Sep 5, 2022
maybe i will check how impala did this😂
thinkharderdev commentedon Sep 5, 2022
This isn't necessarily the case. Even if we don't prune any pages it can still be a pretty significant performance boost to skip decoding.
The general problem with selectivity is that we really don't have much to go on at the time we need to build the filters. We have parquet metadata but that isn't much :). I think the approach I'll go with for the draft PR is something like:
From there we can tweak it to include fancier hueristics (null counts, etc)
alamb commentedon Sep 5, 2022
Thank you @thinkharderdev @tustvold and @Ted-Jiang for driving this forwardf