Row-level Parquet filter pushdown regresses uncorrelated-predicate scans (TPC-H Q6 +80%)

[andygrove]

What is the problem the feature request solves?

Follow-up to #3457 with measurements and root-cause analysis.

spark.comet.parquet.rowFilterPushdown.enabled (DataFusion's pushdown_filters, i.e. row-level RowFilter evaluation plus late materialization) still defaults to false. #4722 already made format-level pruning (row group statistics, page index, bloom filters) work by default, so this flag is now the only remaining piece of "filter pushdown" that is opt-in.

I benchmarked TPC-H Q1 and Q6 at SF100 to see whether it can be turned on by default. It cannot, and the reason is structural rather than a tuning problem.

Measurements

Setup: TPC-H SF100 local Parquet (lineitem = 600,037,902 rows, 25 GB, 32 files, 15,744 row groups), Spark 3.5.8 standalone, 2 executors x 8 cores, warm page cache, 10 iterations per config. Each config was run in both orders (pushdown off first, then on first) to control for cache-warming effects. Median of the last 5 iterations:

Query	off -> on (off ran first)	off -> on (on ran first)
Q1	4.074 -> 3.726 s (-8.5%)	3.804 -> 3.711 s (-2.4%)
Q6	0.843 -> 1.523 s (+80.7%)	1.029 -> 1.515 s (+47.2%)

Result hashes matched across all runs. The pushdown-on timings are stable in both orders (Q1 3.70-3.73 s, Q6 1.51-1.52 s); the off timings are what shift with cache state. Q1 is neutral to slightly positive, Q6 is a large reproducible regression.

Root cause

Scan metrics from the Spark event log, per iteration (600,037,902 rows):

Q6

Metric	pushdown off	pushdown on
Number of bytes scanned	5,971,436,225	5,971,436,225 (identical)
Scan output rows	600,037,902	11,422,456
Data decompression + decoding (summed over tasks)	7.89 s	20.02 s
Time evaluating row-level pushdown filters	~0	0.37 s
Row groups pruned by statistics	0 of 15,744	0 of 15,744

The filter itself works correctly: the scan emits 11.4M rows instead of 600M, the expected ~1.9% selectivity. But bytes scanned is byte-for-byte identical and decode time is 2.5x higher.

Late materialization can only avoid work when surviving rows are clustered. If every data page contains at least one surviving row, the reader still fetches and decompresses every page and skips only the final value-decode step. Q6's predicate (l_shipdate range AND l_discount range AND l_quantity < 24) is uncorrelated with physical row order, so ~1.9% selectivity is spread uniformly and no page is ever eliminated.

What Q6 pays with the flag on:

1. Full decode of the three filter columns (l_shipdate, l_discount, l_quantity) to evaluate the predicate. The predicate-cache metric confirms this exactly: 1,800,113,706 rows physically decoded per iteration = 3 filter columns x 600,037,902 rows.
2. Selection-mask construction and predicate-cache bookkeeping.
3. Selection-based decode of l_extendedprice, the only column late materialization can defer.
4. Re-evaluation of the same predicate in the CometFilter above the scan (see item 1 in "Possible work items").

against a baseline that simply decodes four columns straight through. That is strictly more work. The predicate cache is functioning (14.4B cached reads across 3 iterations), it just has nothing to skip.

Q6 has the worst possible shape for this feature: the filter columns are three of the four projected columns, so the theoretical upside is deferring one column, while the costs apply to all of them.

Q1 is the mild version of the same effect. Its predicate (l_shipdate <= date '1998-12-01' - interval '68 days') removes only 4.8M of 600M rows (0.8%), and decode goes 12.36 s -> 14.56 s summed over tasks. It comes out roughly neutral on wall clock.

Side observation

Statistics pruning eliminated 0 of 15,744 row groups and the page index filtered 0 rows, in every configuration, for both queries. The format-level pruning that #4722 enabled by default does nothing on this dataset for these queries, for the same clustering reason. This is a property of unsorted TPC-H data rather than a Comet defect, but it means neither Q1 nor Q6 can demonstrate the upside of #4722 either, and it is worth keeping in mind when using TPC-H to evaluate any pruning work.

Describe the potential solution

The flag should stay opt-in. Enabling it globally regresses uncorrelated-predicate scans on un-clustered fact tables, which is the common case.

Possible work items, roughly in order of value:

1. Drop the redundant CometFilter when every filter pushes down. native/core/src/parquet/parquet_exec.rs deliberately discards propagation.parent_pushdown_result because Spark's Filter above the scan re-evaluates every data filter. With the flag on, the predicate is therefore evaluated twice. Plumbing the per-filter PushedDown::Yes/No classification back to the JVM would let CometExecRule remove the redundant filter. This reduces the overhead but does not by itself turn Q6 into a win.

2. Decouple reorder_filters from the flag. jni_api.rs sets pushdown_filters and reorder_filters together. Upstream defaults reorder_filters to false; heuristic reordering has its own per-batch cost and can choose a worse order. These should be separately configurable so a regression can be attributed to the right one.

3. Expose max_predicate_cache_size. It is currently never set, so DataFusion uses None (the arrow-rs default). The cache is what prevents filter columns being decoded twice and is worth being able to tune and measure.

4. Make this a planner decision rather than a global flag. The feature wins when the projection is much wider than the filter columns and the predicate is selective, and loses when the filter columns dominate the projection or selectivity is uncorrelated with row order. A heuristic on projection width beyond the filter columns plus estimated selectivity would capture the upside without the Q6-shaped regressions.

Additional context

Measured on apache/main plus two unrelated aggregate-serde commits (no scan-path changes), DataFusion 54.0.0, arrow/parquet 58.3.0.

Benchmarks were run with benchmarks/tpc/run.py using a comet-rowfilter engine variant that sets spark.comet.parquet.rowFilterPushdown.enabled=true.

[andygrove]

Follow-up measurements. Items 2 and 3 in the description above are superseded by this comment — item 2 is disproved and item 3 is answered in the negative.

max_predicate_cache_size does not help

Two layers had to be fixed before this could be measured at all:

1. spark.comet.datafusion.* is gated behind spark.comet.exec.respectDataFusionConfigs, which defaults to false (CometConf.scala:839). Without it, the prefix is silently dropped. Confirmed by setting only spark.comet.datafusion.execution.parquet.pushdown_filters=true: the scan still emitted all 600,037,902 rows.
2. Even with that flag on, the value still never reaches the reader. get_options() builds TableParquetOptions::new() from scratch (parquet_exec.rs:218), while ParquetSource reads the setting from table_parquet_options.global (datafusion-datasource-parquet-54.0.0/src/source.rs:465). Filed separately as Native Parquet scan ignores session-level datafusion.execution.parquet.* options #4990.

With a local patch seeding table_parquet_options.global from the session config, and reorder_filters held at false, TPC-H SF100 Q6, median of the last 5 of 10 iterations:

max_predicate_cache_size	median	vs pushdown off (0.853 s)
0 (disabled)	2.360 s	+176.7%
100 MB (default)	2.290 s	+168.5%
1 GB	2.305 s	+170.2%

Verified genuinely applied this time: cache=0 produced no predicate-cache metrics at all, while the others reported 48.8B cached reads.

So the cache is worth roughly 3%, is already saturated at its 100 MB default, and raising it changes nothing. This is consistent with the root cause: the cache avoids re-decoding filter columns, whereas Q6's problem is that those columns must be fully decoded in the first place. Exposing the knob is still reasonable for tunability, but it is not a fix for this regression.

reorder_filters has no measurable effect

Same setup, default cache, Q6:

config	median
reorder_filters=true	2.302 s
reorder_filters=false	2.303 s

Item 2 above proposed decoupling reorder_filters on the theory that it might be responsible for part of the regression. On Q6 it makes no difference at all. Decoupling it may still be worth doing for clarity, but it should not be presented as a performance fix.

Caveat on the absolute numbers in this comment

The patch used to make max_predicate_cache_size reachable itself cost ~51% on the pushdown-on path (1.52 s -> 2.30 s on Q6). Reverting and re-measuring returned it to 1.517 s, so the attribution is solid, but the cause is unexplained: pushdown_filters was already effectively enabled via table_pushdown_enabled || config_pushdown_enabled (source.rs:740), and reorder_filters measures as a no-op. That is covered in #4990.

Every figure in this comment therefore carries that ~51% penalty and is only meaningful relative to the others in its own table. The headline regression for Q6 from an unpatched build remains 0.853 s -> 1.517 s (+78%), as in the description.

Net effect on the proposed work items

• Item 1 (drop the redundant CometFilter when all filters push down) is unaffected and remains the most valuable item.
• Item 2 (decouple reorder_filters) is disproved as a performance fix.
• Item 3 (expose max_predicate_cache_size) is answered: it does not help, and it is currently unreachable regardless (Native Parquet scan ignores session-level datafusion.execution.parquet.* options #4990).
• Item 4 (make this a planner decision on projection width and selectivity) is unaffected and, given the above, is now clearly the only route to enabling this by default.