ClickBench Q10 slows down when filter pushdown is enabled

Is your feature request related to a problem or challenge?

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

When you run Q10 with predicate pushdown enabled (see #20324 for details of what that means) it goes more slowly:

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

Specifically, on my test I see it going 30% slower

Benchmark clickbench_partitioned.json
┏━━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query     ┃        HEAD ┃ alamb_pushdown_and_arrow_58 ┃        Change ┃
┡━━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 10 │   333.30 ms │                   447.25 ms │  1.34x slower │

You can repro it like this:

Repro Script

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;

When I run this with datafusion-cli -f /tmp/q.sql | grep Elapsed you can clearly see

Elapsed 0.001 seconds.
Elapsed 0.000 seconds.
Elapsed 0.063 seconds.
Elapsed 1.419 seconds. <-- First query run 
Elapsed 1.432 seconds.
Elapsed 1.387 seconds.
Elapsed 1.399 seconds.
Elapsed 1.388 seconds.
Elapsed 0.000 seconds. <---- turn on filter pushdown
Elapsed 1.690 seconds. <-- now the query runs 30% slower
Elapsed 1.671 seconds.
Elapsed 1.695 seconds.
Elapsed 1.695 seconds.
Elapsed 1.718 seconds.

Describe the solution you'd like

I want there to be no slowdown when I run Q10

Describe alternatives you've considered

No response

Additional context

No response

There are two major differences with the pushdown path:

The IO pattern is different (first the data needed for filtering is fetched, and then only the pages for the rows passing are fetched for the projection). Without pushdown all pages for all columns (both filter and projection are fetched)
The overhead of evaluating the filter, then selectively decoding only the rows that match for the projection column. Without pushdown, all columns are decoded and then a single filter pass is applied afterwards

Some theories I tested with codex
Tuning Thresholds for the adaptive row selection representation

Default pushdown is +24.7% slower than no pushdown for Q10 in this environment.
Tuning the threshold for (threshold=12 and no predicate cache) improves pushdown by 9.0% vs default pushdown.
Even tuned pushdown remains 13.4% slower than no pushdown for Q10.

Another thing I did was to look at the actual pattern of selected rows

You can see the overall predicate is quite selective and each selection is often a few rows (2, 3) but sometimes has many (e.g. 44, 34)

Selected runs are short (p50=3, p95=33); skipped runs are long (p50=233, p95=1358).

Here is a Q10 Filter Pattern Visualization

Predicate: "MobilePhoneModel" <> ''
Source sample: first 1,000,000 rows of hits_0.parquet

Rows: total=1000000 selected=19643 (1.964%) skipped=980357 (98.036%)
selected_runs=2438 avg_len=8.06 min=1 max=236
skipped_runs=2439 avg_len=401.95 min=1 max=5771

selected_run_len quantiles (rows): p50=3 p90=19 p95=33 p99=77
skipped_run_len quantiles (rows): p50=233 p90=980 p95=1358 p99=2297

Legend: S=selected run, .=skipped run
Scale: bar width ~= run_len/50 rows (min 1, cap 60)

First 120 runs:
   1 skipped  len=1065  rows=[1..1065] .....................
   2 selected len=2     rows=[1066..1067] S
   3 skipped  len=10    rows=[1068..1077] .
   4 selected len=44    rows=[1078..1121] S
   5 skipped  len=133   rows=[1122..1254] ..
   6 selected len=34    rows=[1255..1288] S
   7 skipped  len=8     rows=[1289..1296] .
   8 selected len=2     rows=[1297..1298] S
   9 skipped  len=152   rows=[1299..1450] ...
  10 selected len=6     rows=[1451..1456] S
  11 skipped  len=400   rows=[1457..1856] ........
  12 selected len=2     rows=[1857..1858] S
  13 skipped  len=1145  rows=[1859..3003] ......................
  14 selected len=26    rows=[3004..3029] S
  15 skipped  len=971   rows=[3030..4000] ...................
  16 selected len=6     rows=[4001..4006] S
  17 skipped  len=32    rows=[4007..4038] .
  18 selected len=4     rows=[4039..4042] S
  19 skipped  len=271   rows=[4043..4313] .....
  20 selected len=8     rows=[4314..4321] S
  21 skipped  len=20    rows=[4322..4341] .
  22 selected len=2     rows=[4342..4343] S
  23 skipped  len=166   rows=[4344..4509] ...
  24 selected len=24    rows=[4510..4533] S
  25 skipped  len=1512  rows=[4534..6045] ..............................
  26 selected len=8     rows=[6046..6053] S
  27 skipped  len=1283  rows=[6054..7336] .........................
  28 selected len=4     rows=[7337..7340] S
  29 skipped  len=398   rows=[7341..7738] .......
  30 selected len=8     rows=[7739..7746] S
  31 skipped  len=147   rows=[7747..7893] ..
  32 selected len=1     rows=[7894..7894] S
  33 skipped  len=375   rows=[7895..8269] .......
  34 selected len=36    rows=[8270..8305] S
  35 skipped  len=148   rows=[8306..8453] ..
  36 selected len=2     rows=[8454..8455] S
  37 skipped  len=26    rows=[8456..8481] .
  38 selected len=2     rows=[8482..8483] S
  39 skipped  len=303   rows=[8484..8786] ......
  40 selected len=38    rows=[8787..8824] S
  41 skipped  len=428   rows=[8825..9252] ........
  42 selected len=20    rows=[9253..9272] S
  43 skipped  len=42    rows=[9273..9314] .
  44 selected len=6     rows=[9315..9320] S
  45 skipped  len=1010  rows=[9321..10330] ....................
...
  85 skipped  len=803   rows=[17475..18277] ................
  86 selected len=6     rows=[18278..18283] S
  87 skipped  len=132   rows=[18284..18415] ..
  88 selected len=26    rows=[18416..18441] S
  89 skipped  len=170   rows=[18442..18611] ...
  90 selected len=4     rows=[18612..18615] S
  91 skipped  len=34    rows=[18616..18649] .
  92 selected len=3     rows=[18650..18652] S
  93 skipped  len=36    rows=[18653..18688] .
  94 selected len=2     rows=[18689..18690] S
  95 skipped  len=766   rows=[18691..19456] ...............
  96 selected len=6     rows=[19457..19462] S
  97 skipped  len=301   rows=[19463..19763] ......
  98 selected len=4     rows=[19764..19767] S
  99 skipped  len=517   rows=[19768..20284] ..........
 100 selected len=6     rows=[20285..20290] S
 101 skipped  len=25    rows=[20291..20315] .
 102 selected len=2     rows=[20316..20317] S
 103 skipped  len=749   rows=[20318..21066] ..............
 104 selected len=20    rows=[21067..21086] S
 105 skipped  len=688   rows=[21087..21774] .............
 106 selected len=6     rows=[21775..21780] S
 107 skipped  len=376   rows=[21781..22156] .......
 108 selected len=2     rows=[22157..22158] S
 109 skipped  len=309   rows=[22159..22467] ......
 110 selected len=2     rows=[22468..22469] S
 111 skipped  len=211   rows=[22470..22680] ....
 112 selected len=20    rows=[22681..22700] S
 113 skipped  len=902   rows=[22701..23602] ..................
 114 selected len=4     rows=[23603..23606] S
 115 skipped  len=38    rows=[23607..23644] .
 116 selected len=4     rows=[23645..23648] S
 117 skipped  len=60    rows=[23649..23708] .
 118 selected len=4     rows=[23709..23712] S
 119 skipped  len=1264  rows=[23713..24976] .........................
 120 selected len=10    rows=[24977..24986] S

Filter pattern (from prior analysis files):
- Sample in hits_0.parquet first 1,000,000 rows: selected 1.964%, skipped 98.036%.
- Selected runs are short (p50=3, p95=33); skipped runs are long (p50=233, p95=1358).
- Across hits_1..hits_99, pattern is not uniform:
  - selected_pct mean 5.599708%, sd 2.290834, min 0.2681% (hits_48), max 13.0079% (hits_17).
  - Conclusion: substantial cross-file variation; static thresholds are likely brittle.

Some random analysis I found while reviewing profiling:

Handling of Cached batches takes non trivial time

Without pushdown

With pushdown

There are two major differences with the pushdown path:

1. The IO pattern is different (first the data needed for filtering is fetched, and then only the pages for the rows passing are fetched for the projection). Without pushdown all pages for all columns (both filter and projection are fetched)

2. The overhead of evaluating the filter,  then selectively decoding only the rows that match for the projection column. Without pushdown, all columns are decoded and then a single filter pass is applied afterwards

I think a third one is that parallelism is different. FilterExec often sits on top of a RepartitionExec. Unless there are more files than cores (even if there are, some may be smaller/larger, etc.) or infra-file re-partitioning is turned on the parallelism is going to be lower with predicate pushdown than in a FilterExec.

Filter execution is also a bit different.

FilterExec and-combine multiple predicates and then filters all columns, whereas filter pushdown always filters a single array (so it can not cache the row indices / ranges).
Here apache/arrow-rs#9284 (comment) i have some good performance results by implementing/using fold for the iterators to speed things up for selective filters on single arrays:

filter i32 (kept 1/2)                                                         1.23     36.5±0.08µs        ? ?/sec    1.00     29.7±0.73µs        ? ?/sec
filter i32 high selectivity (kept 1023/1024)                                  1.16      8.7±0.36µs        ? ?/sec    1.00      7.5±0.23µs        ? ?/sec
filter i32 low selectivity (kept 1/1024)                                      1.34      2.3±0.00µs        ? ?/sec    1.00  1751.5±20.57ns        ? ?/sec

Not sure if this is really related to the slowdown, but it will help regardless and close the gap between the two ways of filtering.

Filter execution is also a bit different.

Would it be a reasonable experiment to take FilterExec and make it evaluate each predicate independently to see if we can reproduce the slowdown?

I feel the intuition is that if there are some very selective predicates evaluating those first, slicing and then evaluating the rest is beneficial because you immediately discard work. But maybe the overhead of slicing the data is high enough that this is not worth it unless the filter is very selective (and with contiguous ranges) and the other filters are very expensive?

I think a third one is that parallelism is different. FilterExec often sits on top of a RepartitionExec. Unless there are more files than cores (even if there are, some may be smaller/larger, etc.) or infra-file re-partitioning is turned on the parallelism is going to be lower with predicate pushdown than in a FilterExec.

This could be -- however, my tests I have set the parallelism to 1 via target_partitions so I think there are other things going on as well.

In addition to some of the overhead you already mentioned (CachedArrayReader / skips / filter + concat) that could be reduced, I think a lot is might be the IO pattern.

Some observations/thoughts when running target_partitions=cpu cores on my machine:

CPU usage is not super high for the parquet decoding / datafusion exec threads - ~50+% of it just waiting on data to arrive:

This can be demonstrated when doubling the parallelism (2x cores) on my machine, the query finishes ~20% faster due to being able to do more parallel IO / hiding IO latency.
In this case the IO pattern (skipping data) and the LocalFileSystem object store is also adding overhead (open / close / allocation / metadata) as it reads more smaller ranges, probably more context switching etc. going on:

I think it might help in this case to cache / remember the file handles (open / close is called many times for each range): Remove file-handle from object store GET operations arrow-rs-object-store#18 as ~25% of IO is spent on just that.
Tokio (with this query) spreads the IO out to ~32 blocking threads (and keeps them available / spreads the work on them) as it uses spawn_blocking in the object store (perhaps we can bring our own thread pool to reduce context switching) or use blocking IO as we're not doing anything else when waiting on the data. The 32 threads is because during metadata reading we use a concurrency of 32 (meta_fetch_concurrency).
Perhaps we can consider do more prefetching in parquet reader or also consider coalescing ranges for LocalFileSystem ?

So I think, in summary some things that could be tried out to optimize the IO side of things

Range coalescing for LocalFilesystem (Gemini advises 4kB - 64kB max gap) to do fewer system calls(?)
Implement Remove file-handle from object store GET operations arrow-rs-object-store#18 to do fewer open / close calls
Optimize allocations
Do something smarter for IO threads (?)

Another "problem" of this query is that the predicate is a string column (MobilePhoneModel) while the second column benefiting from the filter pushdown is a primitive column (userid) - so there is not much to gain IO and decoding-wise (and only overhead).

Just thinking out loud: I think what also would help most ClickBench queries with <> '' filters if instead of materializing the Parquet data before evaluating the filter, the filter could be evaluated during decoding.

We could implement Parquet decoders that push down e.g. scalar (in)equality that produces only the arrow array for the rows matching this predicate (e.g. <> '') and a boolean mask (when required) for the matching rows.

This would

save overhead of decoding / copy / converting rows to e.g. string views
save time post-filtering
save memory overhead to cache the (unfiltered) column

The downside mainly being is that it only works for the predicates implemented (e.g. equality / inequality), but it seems it would be general enough to be worth it.

That’s basically what Vortex do. It’s a lot of work, it basically means implementing operators / a mini query engine for each file format.

Yeah I was reading on Vortex.

I am not sure it is that much work though as we have the basic setup there already, I had some vibe coded PoC that does implement <> for byteview arrays and on dictionary encoded columns and it seems it improves the performance roughly ~10% in the arrow clickbench benchmark for various queries.

[Minor] Update object_store to 0.12.5 (#20378)

[Minor] Update object_store to 0.12.5 (apache#20378)

I am not sure it is that much work though as we have the basic setup there already, I had some vibe coded PoC that does implement <> for byteview arrays and on dictionary encoded columns and it seems it improves the performance roughly ~10% in the arrow clickbench benchmark for various queries.

I think the trick is that you need (potentially) specialized implementations for each operator for each encoding, which will become a large matrix ad increase both complexity and code size 🤔

That’s basically what Vortex do. It’s a lot of work, it basically means implementing operators / a mini query engine for each file format.

could you tell me how to query vertex file, I didn't find it in the datafusion document but found it at ClickBench.

I was referring to https://vortex.dev/

I am not really an expert in vortex

Here is the scripts used to gather their ClickBench results. Maybe it will contain some wisdom for you: https://github.com/ClickHouse/ClickBench/blob/main/datafusion-vortex-partitioned/run.sh

[Minor] Update object_store to 0.12.5 (apache#20378)

Navigation Menu

Repository navigation

ClickBench Q10 slows down when filter pushdown is enabled #20325

Is your feature request related to a problem or challenge?

Describe the solution you'd like

Describe alternatives you've considered

Additional context

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

ClickBench Q10 slows down when filter pushdown is enabled #20325

Description

Is your feature request related to a problem or challenge?

Describe the solution you'd like

Describe alternatives you've considered

Additional context

Activity

alamb commented on Feb 12, 2026

alamb commented on Feb 12, 2026

alamb commented on Feb 12, 2026

alamb commented on Feb 12, 2026

adriangb commented on Feb 12, 2026

Dandandan commented on Feb 12, 2026

adriangb commented on Feb 12, 2026

alamb commented on Feb 13, 2026

Dandandan commented on Feb 15, 2026

Dandandan commented on Feb 16, 2026

Dandandan commented on Feb 16, 2026

Dandandan commented on Feb 17, 2026

adriangb commented on Feb 17, 2026

Dandandan commented on Feb 17, 2026

alamb commented on Feb 26, 2026

l1t1 commented on Mar 16, 2026

alamb commented on Mar 16, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions