ClickBench Q10 slows down when filter pushdown is enabled

[alamb]

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

[alamb]

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

[alamb]

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.

[alamb]

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.

[alamb]

Some random analysis I found while reviewing profiling:

Handling of Cached batches takes non trivial time

Without pushdown

With pushdown

[adriangb]

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.

[Dandandan]

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.

[adriangb]

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?

[alamb]

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.

[Dandandan]

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 ?

[Dandandan]

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 (?)