Repository navigation
Late materialization when LIMIT prunes heavily. #23263
Description
Activity
Also this optimization a strictly based on the fact that the number of output rows are small, I want to understand that would it be beneficial to take over this complexity as this seems to be a very common query pattern.
Hi @RatulDawar , happy to help with this if needed!
Hi @RatulDawar , happy to help with this if needed!
Sure !!
Since this largely depends on statistics here is what I propose.
ymbols (uniform per-cell cost):Symbol Meaning R rows before filtering r rows after filtering C columns projected ( SELECT *)c columns required in phase 1 (filter cols + sort key + row ids) Scan cost
Before optimization — filter after wide decode:
scan_before = R × CAfter optimization — narrow scan, then wide fetch for survivors:
scan_after = R × c + r × CWhen optimization wins
scan_before > scan_after R × C > R × c + r × C R × (C - c) > r × CSame condition, rearranged:
r / R < (C - c) / C (selectivity < column savings fraction) R / r > C / (C - c)Break-even (costs equal):
r* = R × (C - c) / C Optimization wins when r < r*Here it's possible to determine R, C and c. But r seems a bit hard, maybe we can start with just LIMIT queries as there we have r avaliable.
@saadtajwar what do you think ?
@RatulDawar thank you! Sorry for the delayed response, I will try to look at this in the afternoon when I get home from work in a few hours and let you know my thoughts!
Hmm okay, @RatulDawar looking at this now - going to have to ask a few newbie questions haha:
- Am I correct in my understanding in that the problem is, for this query, DataFusion at the very start of physical execution (in the
DataSourceExecnode) immediately does projection of all 105 columns in the table - however projection can be deferred until much later, because only really need theURLcolumn for the filter &EventTimefor the sort & limit, and we want DataFusion to after all of those expensive computations perform the projection of all columns? - If I'm understanding the above correctly, your proposal is to start by completing a first phase (filter/sort/limit on URL + EventTime + row IDs in this case) and a second phase that fetches the remaining columns only for the final rows, starting with LIMIT queries where r is bounded?
Please let me know if there's anything I'm missing - happy to start helping you out with implementation/continue digging if needed!
- Am I correct in my understanding in that the problem is, for this query, DataFusion at the very start of physical execution (in the
If I'm understanding the above correctly, your proposal is to start by completing a first phase (filter/sort/limit on URL + EventTime + row IDs in this case) and a second phase that fetches the remaining columns only for the final rows, starting with LIMIT queries where r is bounded?
Here I meant that since it's difficult to detect the final surviving rows after filter we can begin with queries that have limits.
@alamb can you guide us here if this would be a correct direrction or not ?I think @zhuqi-lucas has been working in this area recently -- perhaps he has some time to review
Reacted by Qi ZhuI think @zhuqi-lucas has been working in this area recently -- perhaps he has some time to review
Thanks @alamb , this is really a good direction.
Thanks for the deep-dive @RatulDawar — the framing is right. The 0.9s narrow-column baseline is effectively the theoretical lower bound: the filter/sort/limit phase only needs the filter and sort keys to decide which rows survive, so its cost should be independent of the SELECT list width. The 4s vs 0.9s gap is exactly the wide-decode work being paid for rows that never make it into the final 10.
Before jumping straight to a full late-materialization rewrite, a couple of orthogonal knobs to layer in first — the remaining gap tells us how much LM actually buys:
-
datafusion.execution.parquet.pushdown_filters = true(default isfalse). This enables ParquetRowFilter, which decodes the filter column (URL) first, appliesLIKE '%google%', and only decodes the other 104 columns for surviving rows within each row group. For ~1% match rate this eliminates the wide-decode on ~99% of rows. -
TopK dynamic row-group pruning (#22450, merged) — the TopK threshold on
EventTimeis pushed back to the scan as aDynamicFilter, so row groups whoseEventTimerange is fully below the current top-10 threshold get skipped. Applies automatically forSortExec + fetchover Parquet.
Splitting the 3.1s gap (4.0s − 0.9s):
Wasted work Rows Recovered by Wide decode of filter-rejected rows ~99M pushdown_filters=trueWide decode of filter-survived but LIMIT-rejected rows ~1M Late materialization So the sequence would be: turn on (1) and (2), re-measure Q23. If we land at ~1.5s, LM is worth ~0.6s. If we land at ~1.0s, LM buys ~0.1s — probably not worth the architectural weight. Concrete numbers before committing to the design.
If we do proceed with LM, your direction is exactly what DuckDB does —
src/optimizer/late_materialization.cpp(PR #15692) rewritesSELECT * FROM t ORDER BY x LIMIT NintoSELECT * FROM t WHERE rowid IN (SELECT rowid FROM t ORDER BY x LIMIT N), withrowidfor Parquet being(file_index, file_row_number). Once rewritten, the semi-join predicate becomes aTableFilteronfile_row_numberand inherits all existing row-group / page-index / RowFilter pushdown for free.DataFusion doesn't yet expose
file_row_numberas a projectable virtual column, so that plumbing is the first prerequisite. I've been working across all the moving parts here — Parquet reader /ParquetOpener,DataSourceExec, the physical optimizer, and the dynamic-filter path from #22450 — so I'm happy to drive the whole thing end-to-end:file_row_numbervirtual column exposure, the optimizer rewrite rule, plumbing thefile_row_number IN (...)filter into the Parquet access plan, and the cost heuristic. Would be great to have you and @saadtajwar as reviewers and on integration testing / cost-model validation — let's confirm the pushdown numbers first, then I'll start opening PRs.Given the scope (Parquet reader plumbing → virtual column plumbing → physical optimizer rule → cost heuristic → benchmarks), I'd suggest promoting this issue to an EPIC with sub-issues for each work item — similar to #23036 (Sort Pushdown) and #22715 (GroupValuesColumn coverage). Happy to help draft the sub-issue breakdown once we agree on the direction. cc @alamb for visibility.
-
Ran Q23 locally with
pushdown_filters=true(built from latestmain):default: 4173 ms avg (3 iters) --pushdown: 318 ms avg (3 iters)~13x speedup, and
SELECT *with pushdown is actually faster than the narrowSELECT "EventTime", "URL"baseline you reported (~900ms). ParquetRowFilteralready does within-row-group late materialization (filter column decoded first, wide columns only for surviving rows), so most of the gap is recovered by just enabling it.Reacted by Nathan, Ratul Dawar and Huang Qiweidatafusion.execution.parquet.pushdown_filters = true (default is false). This enables Parquet RowFilter, which decodes the filter column (URL) first, applies LIKE '%google%', and only decodes the other 104 columns for surviving rows within each row group. For ~1% match rate this eliminates the wide-decode on ~99% of rows.
The local run on this looks very promising, but this will benefit for filter only cases right ?
Suppose a querySELECT * FROM hits ORDER BY "EventTime" LIMIT 10;This won't help with the filter pushdown but yes the TopK dynamic row group pruning should be really helpful here and I think since you tested with master branch that also was used here.
Then we should set this property as default right ? Any reason not to ? This should be helpful in many other cases too not just wide selections as this completely skips rows, right ?@ratuldawar11 Ran the no-WHERE variant too:
SELECT * FROM hits ORDER BY "EventTime" LIMIT 10;
default: 409 ms avg (569 / 342 / 316) --pushdown: 279 ms avg (305 / 270 / 261)You were right — #22450 does most of the heavy lifting here (default is already 409ms).
pushdown_filters=trueadds a modest ~30% on top, because the runtimeEventTime > thresholdfrom TopK also gets applied as aRowFilterinside surviving row groups.On the "why not just default it to
true" — there's an active EPIC tracking exactly this: #20324 "Fix performance regressions when enabling parquet filter pushdown (late materialization)". Q23 gets ~9x faster (alamb's Feb 2025 numbers), but ~18 other ClickBench queries regress by 5-55% (Q2, Q6, Q7, Q10, Q11, Q12, Q13, Q14, Q19, Q24, Q25, Q26, Q27, Q28, Q37, Q40, Q41, Q42). Root cause is typically low-selectivity filters where theRowFilterbookkeeping outweighs the wide-decode savings. #20325 tracks Q10 specifically. Also see #22883 "[EPIC] Adaptive predicate evaluation" for a broader direction.So the trade-off is real — we'd want to close the regressions first, or make the pushdown decision adaptive per-filter (based on estimated selectivity).
And updated, i reproduced the benchmark result here:
#23369 (comment)I don't understand what this ticket is proposing
As @zhuqi-lucas points out, we know that enabling filter_pushdown for Q23 makes it go a lot faster (due to the late materialization you are pointing out). I think this epic is tracking
While the flag is called "filter_pushdown" in my mind it is really a form of late materialization (within the parquet reader -- @hhhizzz wrote a blog about this: https://arrow.apache.org/blog/2025/12/11/parquet-late-materialization-deep-dive/
So if you want to help figure out ways to get filter_pushdown on for real, that sounds like a great idea
@alamb I wasn't aware about the blog + the EPIC, bad on my end should have look through properly and will keep in mind next time, apologies. I wanted to propose late materialisation but since it's already being tracked I think this ticket can be closed.
There was a bit of a difference on the LM that I was proposing, instead of filtering I was thinking it to be like LM when the ratio of result/scan is very high. Which can be in the cases like TopK / heaving filtering, which is alsready been taken care of.As for turning on filter_pushdown by default, would definitely want to help in that. But this ticket can be closed since its already being tracked.
Reacted by Andrew Lamb- added a commit that references this issue
on Jul 7, 2026 @alamb @zhuqi-lucas closing this, apologies again.
hanks @RatulDawar
Currently ClickBench Q23 is a major outliner due to the fact that datafusion doesn't do late materialization of projected columns.
A query like this needs to scan all the columns only after all the redundant rows are prunned from output.
Some numbers that I was able to get locally showing this regression in play.
SELECT *)SELECT "EventTime", "URL")EXPLAIN ANALYZEmetrics (SELECT *)DataSourceExec
projectionoutput_rowsoutput_bytesrow_groups_pruned_statisticsrow_groups_pruned_bloom_filterpage_index_pages_prunedWhat duckdb does in this scenario is that it materialises the projection rows that are not needed for any computation or maybe needed in later steps as late as possilble.
But this late materialization comes at the cost of requiring a self join later on to get the other columns back, and got self join's we need row tagging.
In datafusion we can tag a row uniquely with
file_row_index(),input_file_name().But then if rows are not dynamically filtered on the basis of above two columns it will again lead to a full table scan for probe side of the join.