Repository navigation
Improve Performance (Polars 2.0 Call to Action) #26169
Description
Is your feature request related to a problem or challenge?
@ritchie46 posted a blog https://pola.rs/posts/release-polars-2/ recently that provides benchmark numbers for Polars 2.0 that make Datafusion look quite bad:
I suspect it is telling that
Describe the solution you'd like
I would like the performance of DataFusion to be better (obviously)
Describe alternatives you've considered
My strong suspicion is that the difference is largely caused by the fact that TPC-H and TPC-DS require good join ordering for high performance, something that DataFusion has been historically weak at
I also think this problem is also likely due to the same thing:
Polars and both DuckDB versions completed all queries. DataFusion timed out on TPC-DS q72 (and once on q67) and ran out of memory on TPC-H q18 on c7a.4xlarge; those queries are excluded from the results above for all engines.
Another corroborating piece of evidence is that the post does not mention ClickBench https://benchmark.clickhouse.com/ where we have focused a huge amount of effort on benchmarking and the difference is much less pronounced. See for example the results here:
Additional context
No response
Willingness to contribute
None
Activity
alamb commentedon Oct 9, 2026
Some related work
alamb commentedon Oct 9, 2026
Some other thoughts:
devanbenz commentedon Oct 10, 2026
Running the benchmark locally on an M3 macbook: https://github.com/pola-rs/polars-2.0-benchmark
Raw data:
datafusion_tpch_100.txt
polars_tpch_100.txt
The slowest DataFusion query is q18.
DataFusion faster:
Biggest gaps (DataFusion ÷ Polars):
Closest (Polars still ahead):
devanbenz commentedon Oct 10, 2026
Here are flamegraphs for q18, q17, q11, q07, and q02:
flamegraphs.zip
tschwarzinger commentedon Oct 11, 2026
Would be interesting to see whether some non-default options (e.g., filter pushdown #3463 ) can help. Perhaps I'll investigate a bit tomorrow.
adriangb commentedon Oct 11, 2026
Should we start by adding their benchmarks to our suite, maybe even as part of the standard benchmark runs?
adriangb commentedon Oct 11, 2026
Never mind, it is our standard benchmarks. So I guess like Andrew said more focus on TPC-H / TPC-DS.
alamb commentedon Oct 11, 2026
Do we have an explain analyze for q18?
It would shock me if the problem wasn't with the join order that is selected or that our HashJoin algorithm could be much better
tschwarzinger commentedon Oct 11, 2026
Yeah I looked at it today and IMO the join order was the major issue. I don't have the explain plan available now but I can provide it tomorrow if no one beats me to it.
I wanted to check whether there are some "lower" hanging fruits for improvements.
tschwarzinger commentedon Oct 11, 2026
Ok I've started my PC before saying something wrong.
Query:
Polars (streaming):
Polars (in-memory):
DataFusion:
So, we seem to compute the join of all (
customerxorderxlineitem), most of which are thrown away at right semi join which filters out all small (sum(l_quantity) < 300) orders.Enabling Pushdown Filters:
I think this hides part of the join ordering problems. For example, the second
lineitemsscan only returns4.37 Kelements, thus avoiding large intermediate join results. Still, join order is still a problem. Furthermore, the processing time in the scans increases by quite some time, eliminating most of the benefits of the join improvements (e.g.,row_pushdown_eval_time=16.61sin the last scan). It seems like DF can't prune any of the row groups:files_ranges_pruned_statistics=24 total → 24 matched, row_groups_pruned_statistics=524 total → 524 matched, row_groups_pruned_bloom_filter=524 total → 524 matched,. Not sure how to get this info in Polars.devanbenz commentedon Oct 11, 2026
@alamb After running the main branch of datafusion against polars I do see slight performance improvements, specifically Q18 looks a lot better.
Biggest gaps:
Closest:
Total is 2.15x (75.1s vs 34.9s)
In my previous comment I was running the exact benchmark the polars team published, which was 54.0.0.
The flamegraphs I have attached above are all based on main FYI.
Here's some explain analyzes for q17,q02,q11 since those were all ~4x slower than polars.
datafusion_q02.txt
datafusion_q11.txt
datafusion_q17.txt
tschwarzinger commentedon Oct 12, 2026
Common Subplan Elinimation also seems to make a difference in some queries where the
lineitemtable is scanned multiple times: #8777devanbenz commentedon Oct 12, 2026
Taking a look at Q17 without
filter_pushdown:AggregateExecis taking up most of the query compute time.time_calculating_group_ids=34.98sSo looks like without the pushdown we need to group every part key for 600m rows. Looking at the joins, they discard most of it -- so for this plan, the join ordering seems fine?After enabling pushdown, the
AggregateExectakes very little cpu time. All the CPU time goes in to theDynamicFilterwithin the scan:Enabling pushdown shaves off a bit of time.
DynamicFilterPhysicalExpr::evaluatetaking up most of the flamegraph.filter_arraymemory movement:Could be a problem with the shape of the predicate that gets built for
DynamicFilterPhysicalExprsince it looks like the THEN cases are taking up most of the computation time:datafusion/datafusion/physical-expr/src/expressions/case.rs
Lines 877 to 884 in 6663e43
adriangb commentedon Oct 12, 2026
The
DynamicFilterPhysicalExprissues seems like same issue we've been tracking for a while: it depends on the selectivity of the join, and we have no mechanism to neable or disable the filter pushdown dynamic.devanbenz commentedon Oct 12, 2026
I have an experiment at #26249 where instead of a
CASEstatement to route probe rows to the hash tables of a partitioned hash join, it uses a specialized expression that routes by row index.CASEcopies the batch's columns to isolate each branch's rows, this new expression only ever touches the hashes making sure we don't need to perform thememmoves seen as a bottleneck in my profiles above. It appears to improve performance overall for most TPC-H queries, with Q17 showing a huge improvement:alamb commentedon Oct 14, 2026
I had claude draw me a picture of the q17 plans
DataFusion (SF100, measured)
Note: Times are elapsed_compute summed over 14 partitions.
DuckDB v1.5.6 (SF1, measured, extrapoated to SF100 ("≈")
Difference
it seems as if one problem is that our dynamic filter isn't being used during the scan 🤔
tschwarzinger commentedon Oct 15, 2026
For me, when running
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS=true cargo run --profile release-nonlto --bin dfbench -- tpch --debug --query 17 --format parquet --iterations 1 --path ./benchmarks/data/tpch_sf10/the filters are applied but no pruning is possible.The last
DataSourceExecon line item has huge dynamic filters and the following metrics:Interesting things. The full output is in the attached file:
output_rows=61.38 K, elapsed_compute=14.46s, evaluating the filters works but takes a long time. This is what feat(perf): compute routing and lookup hashes once #26249 is trying to improve upon.files_ranges_pruned_statistics=24 total → 24 matched, row_groups_pruned_statistics=524 total → 524 matched, page_index_pages_pruned=3.14 K total → 3.14 K matched. This could be because the relevant lineitems of the parts are actually distributed over all data pages or because theCASE hash_repartition % 24 WHEN ...type of expressions that the multi-partition hash join produces is not suited for pruning. duckdb also seems to read the entire tablein: 6.00 M out: 5,986,865(at SF1).DELIM_SCANI think the actual difference comes from i) high dynamic filter evaluation time and ii) possibly the approach that duckdb takes with the
DELIM_SCAN. As far as I understand it, the delim scan reuses the part ids after filtering to join on the the line items. There it just has to build and probe a hash table with 2K entries instead of the 61.38 K entries hash table that we have in the top of the join tree which already joined the lineitems with the parts. In theory, the hash table look ups should be done in constant time but maybe the larger tables have significant effect on cache misses etc. such that the performance could be a different. One would have to measure though.output.log
tschwarzinger commentedon Oct 15, 2026
Running
DATAFUSION_OPTIMIZER_HASH_JOIN_INLIST_PUSHDOWN_MAX_DISTINCT_VALUES=0 DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS=true cargo run --profile release-nonlto --bin dfbench -- tpch --query 17 --format parquet --iterations 1 --p ath ./benchmarks/data/tpch_sf10/with the patch from #26249 does put us close to polars (0.41s for polars, 0.5s for datafusion) on my machine.devanbenz commentedon Oct 15, 2026
@tschwarzinger
I think this is a likely case. My current assumption (which could be wrong!) Is because we are joining on
l_partkeyand it's unordered across all pages we cannot prune properly.Omega359 commentedon Oct 15, 2026
I had codex do an investigation on this, take it with a grain or two of salt:
Q17 investigation completed: dynamic filters reject 99.90% of rows when Parquet row filtering is enabled; costly partitioned CASE evaluation and inflated cardinality estimates explain poor performance. SF10 benchmarks validated results, with shared build tables reducing median time from 657 to 276 ms. No Q17 fix was implemented; PR #26249 remains untested locally.Q17.md
adriangb commentedon Oct 15, 2026
How would folks feel about disabling hash join dynamic filters by default? I would like to work on making them cheaper to evaluate and dynamically deciding when they are worth evaluating or not, but until then it seems like they are hurting us on standard benchmarks even if they are beneficial to the right workloads.
IMO to match other systems we'll have to support late materialization. These hash join dynamic filter issues are one of the obstacles standing in the way of it: turning on filter pushdown makes queries slower because they are evaluated, as shown in #26169 (comment) and #26169 (comment)
Omega359 commentedon Oct 15, 2026
tschwarzinger commentedon Oct 15, 2026
I think you've got a better overview what it takes to make them more performant, so these are just my 2c. I would lean towards keeping them enabled by default because switching back will likely trigger a regression somewhere that could be difficult to resolve. I think with a faster implementation of the multi partition
hash_lookupexpression would go a long way.However, maybe rethinking the approach around
datafusion.optimizer.hash_join_inlist_pushdown_max_distinct_values(or its default value) could help. Maybe we should only useIN LISTif the build side across all partitions is small enough such that copying all build sides into a singleIN LISTexpression is inexpensive (i.e., no hash table routing necessary).I also think enabling it dynamically also makes sense, as computing the hashes and probing the table twice can incur overhead if we cannot save on decoding and aggregation (like in Q17). But I guess the cost model could be tricky to get right.
EDIT: ofc if we cannot get it to work, we can switch the default. But it'd suggest investing a bit into this approach before switching the defaults.
Sorry if I wasn't clear, my comment should say that enabling dynamic filters + filter pushdown with #26249 dramatically improves performance.
DATAFUSION_OPTIMIZER_HASH_JOIN_INLIST_PUSHDOWN_MAX_DISTINCT_VALUESjust disables the rule that replaces thehash_lookupexpression withIN LISTwhich then causes the current code in #26249 to not use the faster implementation.@Omega359 did you use
DATAFUSION_OPTIMIZER_HASH_JOIN_INLIST_PUSHDOWN_MAX_DISTINCT_VALUES=0for these measurements on the new branch? I think you need that for the other expression to trigger for smaller scale factors.Omega359 commentedon Oct 15, 2026
Yes — tested with
DATAFUSION_OPTIMIZER_HASH_JOIN_INLIST_PUSHDOWN_MAX_DISTINCT_VALUES=0, and confirmed that it activated the PR’s faster expression. However, the main on/off table in my previous comment used the default IN-list threshold.For PR #26249, at SF10 with 24 partitions and Parquet row filtering enabled:
The PR substantially helps when threshold zero activates its optimized path.