[EPIC] Fix performance regressions when enabling parquet filter pushdown (late materialization)

TLDR

This epic tracks the known work needed to enable Parquet "filter pushdown" (late materialization) by default:

Enable parquet filter pushdown (filter_pushdown) by default #3463

When Parquet filter pushdown is turned on, some queries get (much) faster, but some queries get slower. We currently assume that we need to ensure we don't have any performance regressions when we enable it by default.

If you would like to help, the best thing you can do is to investigate the queries that are slower when we turn on pushdown and figure out how to make them faster (please file a ticket for this work, see #20325 for Q10). The likely involves profiling (docs here ) and then making code changes to fix the bottlenecks. I suspect many of the changes will be in the Parquet reader (in arrow-rs)

Leaderboard as of Feb 12. 2025

Highlighted improvements (the large gains are due to being able to evaluate
dynamic filters during the scan phase)

Benchmark clickbench_extended.json

│ QQuery 6 │  3977.38 ms │                   155.90 ms │ +25.51x faster │

Benchmark clickbench_partitioned.json

│ QQuery 23 │ 12166.40 ms │                  1289.23 ms │ +9.44x faster │

Here is the current queries that are slower when we turn on pushdown,

Benchmark clickbench_partitioned.json
┏━━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query     ┃        HEAD ┃ alamb_pushdown_and_arrow_58 ┃        Change ┃
┡━━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 2  │   134.10 ms │                   152.95 ms │  1.14x slower │
│ QQuery 6  │     6.54 ms │                    21.47 ms │  3.28x slower │
│ QQuery 7  │    55.50 ms │                    65.21 ms │  1.17x slower │
│ QQuery 10 │   333.30 ms │                   447.25 ms │  1.34x slower │
│ QQuery 11 │   392.77 ms │                   494.71 ms │  1.26x slower │
│ QQuery 12 │  1193.25 ms │                  1417.78 ms │  1.19x slower │
│ QQuery 13 │  1901.29 ms │                  2056.60 ms │  1.08x slower │
│ QQuery 14 │  1223.87 ms │                  1420.63 ms │  1.16x slower │
│ QQuery 19 │   121.77 ms │                   136.65 ms │  1.12x slower │
│ QQuery 24 │   216.43 ms │                   295.62 ms │  1.37x slower │
│ QQuery 25 │   453.43 ms │                   605.15 ms │  1.33x slower │
│ QQuery 26 │   211.98 ms │                   328.95 ms │  1.55x slower │
│ QQuery 27 │  2644.08 ms │                  2909.34 ms │  1.10x slower │
│ QQuery 28 │ 23267.40 ms │                 24763.39 ms │  1.06x slower │
│ QQuery 37 │    79.13 ms │                    87.01 ms │  1.10x slower │
│ QQuery 40 │    40.20 ms │                    56.08 ms │  1.40x slower │
│ QQuery 41 │    34.76 ms │                    44.99 ms │  1.29x slower │
│ QQuery 42 │    31.70 ms │                    34.01 ms │  1.07x slower │
└───────────┴─────────────┴─────────────────────────────┴───────────────┘

Full details for recent performance results I have when turning this on, see #20318:

Entire Results Benchmark

#20318 (comment)

Comparing HEAD and alamb_pushdown_and_arrow_58
--------------------
Benchmark clickbench_extended.json
--------------------
┏━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━┓
┃ Query    ┃        HEAD ┃ alamb_pushdown_and_arrow_58 ┃         Change ┃
┡━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━┩
│ QQuery 0 │  2424.01 ms │                  2264.27 ms │  +1.07x faster │
│ QQuery 1 │   896.58 ms │                   890.50 ms │      no change │
│ QQuery 2 │  1727.03 ms │                  1713.88 ms │      no change │
│ QQuery 3 │  1016.56 ms │                  1040.16 ms │      no change │
│ QQuery 4 │  2196.14 ms │                  2198.21 ms │      no change │
│ QQuery 5 │ 28540.31 ms │                 26699.31 ms │  +1.07x faster │
│ QQuery 6 │  3977.38 ms │                   155.90 ms │ +25.51x faster │
│ QQuery 7 │  2690.66 ms │                  2795.88 ms │      no change │
└──────────┴─────────────┴─────────────────────────────┴────────────────┘

#20318 (comment)

--------------------
Benchmark clickbench_partitioned.json
--------------------
┏━━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query     ┃        HEAD ┃ alamb_pushdown_and_arrow_58 ┃        Change ┃
┡━━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 0  │     2.54 ms │                     2.59 ms │     no change │
│ QQuery 1  │    49.67 ms │                    49.78 ms │     no change │
│ QQuery 2  │   134.10 ms │                   152.95 ms │  1.14x slower │
│ QQuery 3  │   155.24 ms │                   156.89 ms │     no change │
│ QQuery 4  │  1002.02 ms │                  1022.11 ms │     no change │
│ QQuery 5  │  1279.70 ms │                  1276.97 ms │     no change │
│ QQuery 6  │     6.54 ms │                    21.47 ms │  3.28x slower │
│ QQuery 7  │    55.50 ms │                    65.21 ms │  1.17x slower │
│ QQuery 8  │  1361.66 ms │                  1418.47 ms │     no change │
│ QQuery 9  │  1784.07 ms │                  1815.62 ms │     no change │
│ QQuery 10 │   333.30 ms │                   447.25 ms │  1.34x slower n│
│ QQuery 11 │   392.77 ms │                   494.71 ms │  1.26x slower │
│ QQuery 12 │  1193.25 ms │                  1417.78 ms │  1.19x slower │
│ QQuery 13 │  1901.29 ms │                  2056.60 ms │  1.08x slower │
│ QQuery 14 │  1223.87 ms │                  1420.63 ms │  1.16x slower │
│ QQuery 15 │  1146.25 ms │                  1193.73 ms │     no change │
│ QQuery 16 │  2462.82 ms │                  2471.52 ms │     no change │
│ QQuery 17 │  2466.41 ms │                  2369.56 ms │     no change │
│ QQuery 18 │  4990.28 ms │                  4544.09 ms │ +1.10x faster │
│ QQuery 19 │   121.77 ms │                   136.65 ms │  1.12x slower │
│ QQuery 20 │  1924.82 ms │                  1840.60 ms │     no change │
│ QQuery 21 │  2206.94 ms │                  2276.71 ms │     no change │
│ QQuery 22 │  3699.15 ms │                  3776.51 ms │     no change │
│ QQuery 23 │ 12166.40 ms │                  1289.23 ms │ +9.44x faster │
│ QQuery 24 │   216.43 ms │                   295.62 ms │  1.37x slower │
│ QQuery 25 │   453.43 ms │                   605.15 ms │  1.33x slower │
│ QQuery 26 │   211.98 ms │                   328.95 ms │  1.55x slower │
│ QQuery 27 │  2644.08 ms │                  2909.34 ms │  1.10x slower │
│ QQuery 28 │ 23267.40 ms │                 24763.39 ms │  1.06x slower │
│ QQuery 29 │   962.53 ms │                   964.69 ms │     no change │
│ QQuery 30 │  1262.47 ms │                  1253.99 ms │     no change │
│ QQuery 31 │  1315.46 ms │                  1273.97 ms │     no change │
│ QQuery 32 │  4319.41 ms │                  3864.61 ms │ +1.12x faster │
│ QQuery 33 │  5401.20 ms │                  4915.16 ms │ +1.10x faster │
│ QQuery 34 │  5862.38 ms │                  5446.49 ms │ +1.08x faster │
│ QQuery 35 │  1908.26 ms │                  1879.87 ms │     no change │
│ QQuery 36 │   189.23 ms │                   185.66 ms │     no change │
│ QQuery 37 │    79.13 ms │                    87.01 ms │  1.10x slower │
│ QQuery 38 │   115.09 ms │                    91.86 ms │ +1.25x faster │
│ QQuery 39 │   340.85 ms │                   286.45 ms │ +1.19x faster │
│ QQuery 40 │    40.20 ms │                    56.08 ms │  1.40x slower │
│ QQuery 41 │    34.76 ms │                    44.99 ms │  1.29x slower │
│ QQuery 42 │    31.70 ms │                    34.01 ms │  1.07x slower │
└───────────┴─────────────┴─────────────────────────────┴───────────────┘

#20318 (comment)

Comparing HEAD and alamb_pushdown_and_arrow_58
--------------------
Benchmark tpch_mem_sf1.json
--------------------
┏━━━━━━━━━━━┳━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query     ┃      HEAD ┃ alamb_pushdown_and_arrow_58 ┃        Change ┃
┡━━━━━━━━━━━╇━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 1  │ 106.36 ms │                   101.53 ms │     no change │
│ QQuery 2  │  33.05 ms │                    31.05 ms │ +1.06x faster │
│ QQuery 3  │  37.27 ms │                    35.90 ms │     no change │
│ QQuery 4  │  32.24 ms │                    31.09 ms │     no change │
│ QQuery 5  │  93.75 ms │                    81.95 ms │ +1.14x faster │
│ QQuery 6  │  21.07 ms │                    20.25 ms │     no change │
│ QQuery 7  │ 159.62 ms │                   146.90 ms │ +1.09x faster │
│ QQuery 8  │  42.98 ms │                    39.57 ms │ +1.09x faster │
│ QQuery 9  │ 111.61 ms │                   100.78 ms │ +1.11x faster │
│ QQuery 10 │  67.99 ms │                    63.06 ms │ +1.08x faster │
│ QQuery 11 │  18.50 ms │                    18.94 ms │     no change │
│ QQuery 12 │  52.30 ms │                    54.00 ms │     no change │
│ QQuery 13 │  49.77 ms │                    47.68 ms │     no change │
│ QQuery 14 │  14.97 ms │                    14.12 ms │ +1.06x faster │
│ QQuery 15 │  30.49 ms │                    29.21 ms │     no change │
│ QQuery 16 │  28.87 ms │                    27.64 ms │     no change │
│ QQuery 17 │ 145.35 ms │                   143.02 ms │     no change │
│ QQuery 18 │ 289.85 ms │                   269.10 ms │ +1.08x faster │
│ QQuery 19 │  40.12 ms │                    49.08 ms │  1.22x slower │
│ QQuery 20 │  56.72 ms │                    57.28 ms │     no change │
│ QQuery 21 │ 191.17 ms │                   189.19 ms │     no change │
│ QQuery 22 │  22.55 ms │                    22.26 ms │     no change │
└───────────┴───────────┴─────────────────────────────┴───────────────┘

How to run the benchmarks yourself

You can find the query text for the ClickBench queries here

ClickBench queries: https://github.com/apache/datafusion/tree/main/benchmarks/queries/clickbench/queries
ClickBench (extended) benchmarks: https://github.com/apache/datafusion/tree/main/benchmarks/queries/clickbench/extended

Download the data like this:

cd datafusion
./benchmarks/bench.sh data clickbench_partitioned

Then run the queries like this (adjusting the data location):

set datafusion.execution.parquet.binary_as_string = true; -- needed for ClickBench data
SET datafusion.execution.target_partitions = 1; -- set to 1 to reduce variability
create external table hits stored as parquet location '/home/ec2-user/datafusion/benchmarks/data/hits_partitioned';
QUERY_TEXT_HERE;
SET datafusion.execution.parquet.pushdown_filters = true;
QUERY_TEXT_HERE;

For example, to run Q10 on the partitioned ClickBench data:

set datafusion.execution.parquet.binary_as_string = true; -- needed for ClickBench data
SET datafusion.execution.target_partitions = 1; -- set to 1 to reduce variability
create external table hits stored as parquet location '/home/ec2-user/datafusion/benchmarks/data/hits_partitioned';

-- Q10 (default configuration, no pushdown)
SELECT "MobilePhoneModel", COUNT(DISTINCT "UserID") AS u FROM hits WHERE "MobilePhoneModel" <> '' GROUP BY "MobilePhoneModel" ORDER BY u DESC LIMIT 10;
SELECT "MobilePhoneModel", COUNT(DISTINCT "UserID") AS u FROM hits WHERE "MobilePhoneModel" <> '' GROUP BY "MobilePhoneModel" ORDER BY u DESC LIMIT 10;
SELECT "MobilePhoneModel", COUNT(DISTINCT "UserID") AS u FROM hits WHERE "MobilePhoneModel" <> '' GROUP BY "MobilePhoneModel" ORDER BY u DESC LIMIT 10;
SELECT "MobilePhoneModel", COUNT(DISTINCT "UserID") AS u FROM hits WHERE "MobilePhoneModel" <> '' GROUP BY "MobilePhoneModel" ORDER BY u DESC LIMIT 10;
SELECT "MobilePhoneModel", COUNT(DISTINCT "UserID") AS u FROM hits WHERE "MobilePhoneModel" <> '' GROUP BY "MobilePhoneModel" ORDER BY u DESC LIMIT 10;

-- Q10 enable with pushdown enabled
SET datafusion.execution.parquet.pushdown_filters = true;
SELECT "MobilePhoneModel", COUNT(DISTINCT "UserID") AS u FROM hits WHERE "MobilePhoneModel" <> '' GROUP BY "MobilePhoneModel" ORDER BY u DESC LIMIT 10;
SELECT "MobilePhoneModel", COUNT(DISTINCT "UserID") AS u FROM hits WHERE "MobilePhoneModel" <> '' GROUP BY "MobilePhoneModel" ORDER BY u DESC LIMIT 10;
SELECT "MobilePhoneModel", COUNT(DISTINCT "UserID") AS u FROM hits WHERE "MobilePhoneModel" <> '' GROUP BY "MobilePhoneModel" ORDER BY u DESC LIMIT 10;
SELECT "MobilePhoneModel", COUNT(DISTINCT "UserID") AS u FROM hits WHERE "MobilePhoneModel" <> '' GROUP BY "MobilePhoneModel" ORDER BY u DESC LIMIT 10;
SELECT "MobilePhoneModel", COUNT(DISTINCT "UserID") AS u FROM hits WHERE "MobilePhoneModel" <> '' GROUP BY "MobilePhoneModel" ORDER BY u DESC LIMIT 10;

Background:

What is "Parquet filter pushdown"?

In this context, "Parquet filter pushdown" refers to the ability of the query engine to push down filter predicates (e.g., WHERE clauses) into the Parquet file reader, allowing it to skip reading irrelevant data and thus improving query performance. You can read about the details in the A Practical Dive Into Late Materialization in arrow-rs Parquet Reads blog

As has been pointed out, this is a somewhat confusing name as parquet reader implements many other forms of filter pushdown (e.g. to prune files. row groups and pages) by default. "Parquet filter pushdown" is enabled in DataFusion by setting the configuration option datafusion.execution.parquet.pushdown_filters to true. See
the configuration docs:

name	default
datafusion.execution.parquet.pushdown_filters	false

(reading) If true, filter expressions are be applied during the parquet decoding operation to reduce the number of rows decoded. This optimization is sometimes called “late materialization”.

This refers to the ability of the query engine to push down filter predicates (e.g., WHERE clauses) into the Parquet file reader, allowing it to skip reading irrelevant data and thus improving query performance.

When this setting is on, DataFusion gives the predicates to the Parquet reader using the ArrowPredicate API.

See the implementation in DataFusion here:

datafusion/datafusion/datasource-parquet/src/row_filter.rs

Line 105 in aae3e0f

pub(crate) struct DatafusionArrowPredicate {

Specific query investigations:

We should detect constant true filters

Testing in #20395

Perhaps we should make it a bit more smart so it will do this at the row group level (instead of the file level).

ArrowPredicate could be extended with an always_true or is_active method or something, so the decoder knows to skip the predicate when it's not doing anything useful.

I filed #20325 to discuss what I have found with Q10

@andygrove

FYI @andygrove and @adriangb as you have asked about this and said maybe you are interested in helping

Also @Dandandan as I know you are all about performance as well

@alamb

Hey @alamb
I was looking into Q6 and found that the aggregate dynamic filters were being pushed down to RowFilter. We also start with zero selectivity which in turn qualifies more files for scanning than with filter pushdown turned off.

some explain analyze stats:

pushdown_rows_pruned=23.14 M (none without filter)
files_ranges_pruned_statistics=113 total → 31 matched (6 matched without filter)

perhaps for aggregate queries we could avoid evaluating rowgroups! Let me know what you think!

I'll have a draft PR for that up soon!

Hey @alamb I was looking into Q3 and found that the aggregate dynamic filters were being pushed down to RowFilter. We also start with zero selectivity which in turn qualifies more files for scanning than with filter pushdown turned off.

some explain analyze stats:
* `pushdown_rows_pruned=23.14 M` (none without filter)

* `files_ranges_pruned_statistics=113 total → 31 matched` (6 matched without filter)
perhaps for aggregate queries we could avoid evaluating rowgroups! Let me know what you think!

Q3 is SELECT AVG("UserID") FROM hits;

As far as I know no dynamic filters should be generated (at least after your fix in #20279). Q3 is also not one of the queries that gets slower.

ah sorry i meant to write Q6 (it's 3x slower)

Okay that makes more sense!

start with zero selectivity

This part makes sense. I'd be surprised if the overhead of a non-selective filter like "EventDate" > -inf AND "EventDate" < inf (or some other small/large bounds) made a significant impact, but it's possible.

in turn qualifies more files for scanning than with filter pushdown turned off.

This part doesn't make sense to me. How does having filters (even if low selectivity) allow more files / data through than not having them?

@adriangb

@adriangb dynamic filters when enabled start off with lit(true) for the first record batch. I guess in that short time when all rows are qualified and we have target_partitions > 1 other files are done reading. and we only start getting selective starting next batch. and may end up having many overheads of row filter evaluation (that's just my hypothesis)

@notashes

Okay yeah that's a valid hypothesis. I think things should be optimized enough that the overhead would not be that impactful, but maybe it is. I think it would be reasonable to some simplifier / check to discard filters that are always true.

That said I just tried running Q6:

set datafusion.execution.parquet.binary_as_string = true;
create external table hits stored as parquet location 'benchmarks/data/hits_partitioned';
explain analyze SELECT MIN("EventDate"), MAX("EventDate") FROM hits;

I'm getting:

ProjectionExec: expr=[15888 as min(hits.EventDate), 15917 as max(hits.EventDate)]

I.e. we don't even scan the data, we resolve it from statistics 🤔. Am I doing something wrong / different than you @notashes ?

@notashes

Am I doing something wrong / different than you @notashes ?

I think it might be that you are running throughdatafusion-cli. dfbench wraps these aggregate queries in a view to evade these optimizations!

@2010YOUY01

I've been able to reproduce something ~ similar:

set datafusion.execution.parquet.binary_as_string = true;
set datafusion.execution.parquet.pushdown_filters = false;

create external table hits stored as parquet location 'benchmarks/data/hits_partitioned';
CREATE VIEW hits_view AS SELECT "EventDate" + 0 as "EventDate" FROM hits;

set datafusion.optimizer.enable_aggregate_dynamic_filter_pushdown = false;
SELECT MIN("EventDate"), MAX("EventDate") FROM hits_view where "EventDate" > 0 AND "EventDate" < 99999999;

set datafusion.optimizer.enable_aggregate_dynamic_filter_pushdown = true;
SELECT MIN("EventDate"), MAX("EventDate") FROM hits_view;

+--------------------------+--------------------------+
| min(hits_view.EventDate) | max(hits_view.EventDate) |
+--------------------------+--------------------------+
| 15888                    | 15917                    |
+--------------------------+--------------------------+
1 row(s) fetched. 
Elapsed 0.026 seconds.

0 row(s) fetched. 
Elapsed 0.000 seconds.

+--------------------------+--------------------------+
| min(hits_view.EventDate) | max(hits_view.EventDate) |
+--------------------------+--------------------------+
| 15888                    | 15917                    |
+--------------------------+--------------------------+
1 row(s) fetched. 
Elapsed 0.135 seconds.

The filter for the second query is predicate=DynamicFilter [ CAST(EventDate@5 AS Int64) + 0 < 15888 OR CAST(EventDate@5 AS Int64) + 0 > 15917 ]

The cast() aside (which I think will be optimized away / I don't even know why it's there, the column is Int64 to begin with) I don't know why the dynamic filter version would be slower than the version with an arbitrary hard-coded filter. Maybe there's a lot of overhead to the dynamic filter wrapper?

But also: this is all without turning predicate pushdown on. So it seems that just having the aggregate dynamic filters on causes a regression for this query. cc @2010YOUY01. But then that means this also doesn't explain why the query gets slower with predicate pushdown turned on 🤔

@adriangb

do we need to update the filters for each batch? could we skip the bounds don't change?

datafusion/datafusion/physical-plan/src/aggregates/no_grouping.rs

Line 192 in f48dc72

filter_state.filter.update(predicate)?;

@adriangb i think it could be this. we rebuild the pruning predicates for each record batch.

datafusion/datafusion/pruning/src/file_pruner.rs

Lines 98 to 99 in f48dc72

    
           // If the generation hasn't changed, we can skip rebuilding the pruning 
        
           // predicate, which is an expensive operation involving expression analysis.

do we need to update the filters for each batch? could we skip the bounds don't change?

datafusion/datafusion/physical-plan/src/aggregates/no_grouping.rs

Line 192 in f48dc72

filter_state.filter.update(predicate)?;
@adriangb i think it could be this. we rebuild the pruning predicates for each record batch.

datafusion/datafusion/pruning/src/file_pruner.rs

Lines 98 to 99 in f48dc72

// If the generation hasn't changed, we can skip rebuilding the pruning
// predicate, which is an expensive operation involving expression analysis.

This certainly seems like it could be a smoking gun, good find!

As part of the epic, we should also look at the dynamic join filters which slows down the tpch/tpcds benchmarks with filter pushdown.

#20318 (comment)

As part of the epic, we should also look at the dynamic join filters which slows down the tpch/tpcds benchmarks with filter pushdown.

#20318 (comment)

I dug around and couldn't find any existing ticket that tracks this issue. Does one exist? if not I'll file one

As part of the epic, we should also look at the dynamic join filters which slows down the tpch/tpcds benchmarks with filter pushdown.
#20318 (comment)

I dug around and couldn't find any existing ticket that tracks this issue. Does one exist? if not I'll file one

See: #19858

@Dandandan

I see your point and agree. I think metrics based and parallelism fixes(biased towards morsel parallelism as per Hyper), they would be of great advantage and should still be pursued even if the heuristics approach is merged for quick wins.

And to be clear, I think DataFusion already has "morsel driven parallelism" for computation (aka the batches that flow through the plan)
What @Dandandan has done is to start working on making the Parquet IO scheduling more fine grained (and adaptively adjust at runtime to data skew)

Thanks for the clarification!

@darmie

@alamb @adriangb Curious to know your thoughts regarding #20417

My feeling is it's a special case of the general idea in #20363 and would prefer a system based on metrics and not heuristics (hopefully we can find a metric that encodes this case without special casing it). My current feeling is to focus on #20481 which fixes parallelism, data skew and gives us smaller units to adapt on for #20363 or similar. That said I would understand the view of "let's just merge the fix/improvement" that works and generalize later.

IMO we should •also• have some rules for row-filters to avoid fixed row-filters that will never reduce IO (at least when having no page index, perhaps we should add this as extra check on the PR @darmie ?)
Also when the remaining columns to read are really tiny, IO / decode savings are likely to be smaller than the extra overhead.

examples being:
select col from t where col = 1 <- not useful, can just as well apply the filter later
select col, col2 from t where col = 1 AND col = 2 <- only one of them is useful, the last one can not
select col, col2 from t where (col = 1 OR col = 2) <- not useful as it requires reading both columns

Seems removing the current "prefetching" helps fixing part of the query 6 performance regression as well (#20916):

│ QQuery 6  │        11.25 / 13.69 ±3.56 / 20.75 ms │           7.63 / 8.66 ±0.70 / 9.82 ms │ +1.58x faster │
``

This seems to show that in this case `open` happens too early (so it has to read+decode more than a more lazy approach).

This is perhaps something we can solve with the morsel-based planner as well.

One thing to note as well that the current filter pushdown can/will be a lot slower on object storage as well because of the (smaller) serial IO requests.

I added a change to "simulate" object storage in the benchmarks.

No pushdown:
│ QQuery 22 │ 3484.29 / 3628.68 ±141.45 / 3890.40 ms │ 3671.23 / 3774.84 ±117.72 / 3998.45 ms │ no change │

Pushdown:

│ QQuery 22 │ 9961.50 / 10190.13 ±278.27 / 10715.43 ms │ 9809.49 / 10222.74 ±390.94 / 10904.73 ms │ no change │

Yep that's what I expected 😢

We currently use some strategy to avoid this: group small / big columns, issue one request for small ones and one for the big ones. But I guess better prefetching would work even better.

Also I think morsels will help as well (but perhaps too small will hurt more here, I'll try that now).

fix: update filter predicates for min/max aggregates only if bounds c…

So for kicks I turned it on in Comet with TPC-DS SF1000. We still see pretty large regressions:

If I dig into the biggest regression on Q88 with Spark UI:

No row-level filtering/filter pushdown/late materialization:

Row-level filtering/filter pushdown/late materialization:

Even if we try to add an optimization to omit the CometFilter node when everything is pushed into the scan, the extra time in the scan doesn't offset eliding the CometFilter, so it's strictly slower. I'm not sure what next steps would be to help optimize this, but don't have cycles for it in the immediate future.

@xudong963

Our current hope is riding on multi-level (FilterExec, within the expression, and across expressions) adaptive predicate evaluation / filter evaluation. I think @xudong963 @zhuqi-lucas @hhhizzz @adriangb and @Dandandan and I are working on this (slowly)

[EPIC] Adaptive predicate evaluation #22883

I might have found an issue in Comet that would make this worse. It seems like the PhysicalExprAdapter in DF 53 introduces casts on type mismatches (even nullability) that Comet than replaces with a Spark-compatible SparkCast (not even sure that's correct behavior in Comet for Parquet pruning). However, SparkCast isn't recognized by PruningPredicate so I don't think we're getting metadata-level pruning with Comet (only partition-level, not row-group or page). So we might be hammering the row-filter logic harder than we need to if the coarse-grained filters aren't doing work first. If this is the case, it does mean all of those rows were hitting the CometFilter node anyway, so we're still net slower with row-level filters on, but let me revisit Comet performance after this PR merges:

apache/datafusion-comet#4730

However, SparkCast isn't recognized by PruningPredicate

I would love to avoid gotchas like this by making the code in DataFusion more general (and not checking on downcasts, for example)

perf(physical-expr): cache remapped expression in DynamicFilterPhysic…

Navigation Menu

Repository navigation

[EPIC] Fix performance regressions when enabling parquet filter pushdown (late materialization) #20324

TLDR

Leaderboard as of Feb 12. 2025

How to run the benchmarks yourself

Background:

What is "Parquet filter pushdown"?

Specific query investigations:

51 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	// If the generation hasn't changed, we can skip rebuilding the pruning
	// predicate, which is an expensive operation involving expression analysis.

[EPIC] Fix performance regressions when enabling parquet filter pushdown (late materialization) #20324

Description

TLDR

Leaderboard as of Feb 12. 2025

How to run the benchmarks yourself

Background:

What is "Parquet filter pushdown"?

Specific query investigations:

Activity

alamb commented on Feb 12, 2026

alamb commented on Feb 12, 2026

notashes commented on Feb 12, 2026

adriangb commented on Feb 12, 2026

notashes commented on Feb 12, 2026

adriangb commented on Feb 12, 2026

notashes commented on Feb 12, 2026

adriangb commented on Feb 12, 2026

notashes commented on Feb 12, 2026

adriangb commented on Feb 12, 2026

notashes commented on Feb 13, 2026

adriangb commented on Feb 13, 2026

Dandandan commented on Feb 13, 2026

alamb commented on Feb 13, 2026

Dandandan commented on Feb 14, 2026

51 remaining items

darmie commented on Mar 9, 2026

Dandandan commented on Mar 9, 2026

Dandandan commented on Mar 14, 2026

Dandandan commented on Mar 16, 2026

adriangb commented on Mar 16, 2026

Dandandan commented on Mar 16, 2026

Dandandan commented on Mar 16, 2026

mbutrovich commented on Jun 25, 2026

alamb commented on Jun 25, 2026

mbutrovich commented on Jun 25, 2026

alamb commented on Jun 25, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions