Repository navigation
Avoid consecutive RepartitionExec #18341
Description
Reproducer in sqllogictest: #18343
While experimenting with aggregation, I noticed cases where two RepartitionExec operators appear consecutively in the plan. This seems suboptimal and worth improving.
I’m using a dataset stored in both CSV and Parquet formats to test this behavior.
d_dkey,env,service,host
A,dev,log,ma
B,prod,log,ma
C,prod,log,vim
D,prod,trace,vimThe plan of for data in cvs file looks reasonable
EXPLAIN SELECT env, count(*) FROM dimension_csv GROUP BY env;
+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan | Projection: dimension_csv.env, count(Int64(1)) AS count(*) |
| | Aggregate: groupBy=[[dimension_csv.env]], aggr=[[count(Int64(1))]] |
| | TableScan: dimension_csv projection=[env] |
| physical_plan | ProjectionExec: expr=[env@0 as env, count(Int64(1))@1 as count(*)] |
| | AggregateExec: mode=FinalPartitioned, gby=[env@0 as env], aggr=[count(Int64(1))] |
| | CoalesceBatchesExec: target_batch_size=8192 |
| | RepartitionExec: partitioning=Hash([env@0], 16), input_partitions=16 |
| | AggregateExec: mode=Partial, gby=[env@0 as env], aggr=[count(Int64(1))] |
| | RepartitionExec: partitioning=RoundRobinBatch(16), input_partitions=1 |
| | DataSourceExec: file_groups={1 group: [[Users/hoabinhnga.tran/datafusion-optimal-plans/testdata/dimension1/dimension_1.csv]]}, projection=[env], file_type=csv, has_header=true |
| | |
+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+And this is its graphical plan
However, the plan for Parquet data doesn’t appear to push the round-robin repartition down far enough, resulting in two RepartitionExec operators placed back-to-back. This seems quite suboptimal and likely worth improving.
EXPLAIN SELECT env, count(*) FROM dimension_parquet GROUP BY env;
+---------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+---------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan | Projection: dimension_parquet.env, count(Int64(1)) AS count(*) |
| | Aggregate: groupBy=[[dimension_parquet.env]], aggr=[[count(Int64(1))]] |
| | TableScan: dimension_parquet projection=[env] |
| physical_plan | ProjectionExec: expr=[env@0 as env, count(Int64(1))@1 as count(*)] |
| | AggregateExec: mode=FinalPartitioned, gby=[env@0 as env], aggr=[count(Int64(1))] |
| | CoalesceBatchesExec: target_batch_size=8192 |
| | RepartitionExec: partitioning=Hash([env@0], 16), input_partitions=16 -- Repartition Hash |
| | RepartitionExec: partitioning=RoundRobinBatch(16), input_partitions=1 -- Repartition Round Robin |
| | AggregateExec: mode=Partial, gby=[env@0 as env], aggr=[count(Int64(1))] |
| | DataSourceExec: file_groups={1 group: [[Users/hoabinhnga.tran/datafusion-optimal-plans/testdata/dimension1/dimension_1.parquet]]}, projection=[env], file_type=parquet |
| | |
+---------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
Fix Proposal
I suggest we fix this by either pushing the round-robin repartition further down—similar to the plan for the CSV file above—or eliminating it entirely if it's unnecessary.
Advanced proposal
If we can determine that the input file is small, there's no need to repartition the data. In that case, we should use a single-step aggregate—like in the example below.
-- Option to keep single partition
set datafusion.execution.target_partitions = 1;
EXPLAIN SELECT env, count(*) FROM dimension_parquet GROUP BY env;
+---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan | Projection: dimension_parquet.env, count(Int64(1)) AS count(*) |
| | Aggregate: groupBy=[[dimension_parquet.env]], aggr=[[count(Int64(1))]] |
| | TableScan: dimension_parquet projection=[env] |
| physical_plan | ProjectionExec: expr=[env@0 as env, count(Int64(1))@1 as count(*)] |
| | AggregateExec: mode=Single, gby=[env@0 as env], aggr=[count(Int64(1))] |
| | DataSourceExec: file_groups={1 group: [[Users/hoabinhnga.tran/datafusion-optimal-plans/testdata/dimension1/dimension_1.parquet]]}, projection=[env], file_type=parquet |
| | |
Metadata
Metadata
Assignees
Labels
No labels
Activity
NGA-TRAN commentedon Oct 28, 2025
Some hints: I suggest starting by investigating when the two RepartitionExec nodes are introduced in the plan, and why the behavior differs between the CSV and Parquet cases.
Running EXPLAIN VERBOSE can help reveal when these nodes are added—during either the logical or physical optimization phases.
gene-bordegaray commentedon Oct 28, 2025
take
alamb commentedon Oct 28, 2025
Here is some related discussion:
PhysicalPlanoutput less confusing #93702010YOUY01 commentedon Oct 30, 2025
I think it might be related to the issue, there is a discord discussion for slow tpch q1 https://discord.com/channels/885562378132000778/1290751484807352412/1432863136612089959
From the plan, there is only one hash repartition operator, and no round-robin repartition. I believe they're necessary if the data source is not very small (and we have to push them further down as explained in this issue, otherwise they're useless)
The reason is the push downed filter
l_shipdate <= date '1998-09-02'may filter out most of the input parquet scanner partition.(Note below is just a imaginary case, I haven't verified that's the actual reason for slow Q1, but I think it's possible, so it's necessary to fix the
RepartitionExecissue)e.g. Let's say
datafusion.execution.target_partitions=4, so there are 4 parallel parquet scanner, also 4 parallel aggregate operator in the upstream.partition 1: l_shipdate has range [1990, 2000]
partition 2: l_shipdate has range [2000, 2005]
partition 3: l_shipdate has range [2005, 2010]
partition 4: l_shipdate has range [2010, 2020]
The consequence is partition 2,3,4 has no output data in the parquet reader, then only one partition is busy, and the available CPUs can't be fully utilized.
TPCH Q1 plan
2010YOUY01 commentedon Oct 30, 2025
I've checked on
tpch sf1on parquet generated withbenchmarks/script, running q1 underdatafusion.execution.target_partitions=14, withexplain analyze verboseon the query. And nearly half of the partitions are actually empty.I think fixing it can significantly speed up tpch q1.
Add reproducer for consecutive RepartitionExec (#18343)
alamb commentedon Nov 1, 2025
BTW I have filed a separate ticket to work on tpch q1 performance:
Add reproducer for consecutive RepartitionExec (apache#18343)
gene-bordegaray commentedon Nov 7, 2025
i have made a PR for this fix along with a write-up I plan to share with others about my deep dive into the physical optimizer. I have attached the long-version of my write up that goes into depth (meant to help others understand this part of the codebase) and the report just on this issue along with follow up work I the need for:
Full Report: The Physical Optimizer and Fixing Consecutive Repartitions In the Enforce Distribution Rule.pdf
Issue Report: Fixing Consecutive Repartitions In the Enforce Distribution Rule.pdf
If others think my document is useful I would love to share with community to help others understand this part of the codebase
NGA-TRAN commentedon Nov 7, 2025
@gene-bordegaray : Great analysis. I have read both
full reportthat includes your studying how physical rules work and the theissue reportthat only include the issueTo reviewers: If you are familiar with DataFusion already, you only need to read the Issue Report
This is the summary of the fix
CSV files that lack of statistics:
We always add Round Robin Repartition and the plan looks like this which is very reasonable to me (because of no stats)
Parquet files with
ExactstatisticsThere are 2 cases
The fix is exactly what we want.
@gene-bordegaray : Could you confirm if the summary above is accurate? If it is, this looks like the ideal fix.
NGA-TRAN commentedon Nov 7, 2025
And if you read the @gene-bordegaray 's doc, he also included the benchmark numbers that everything is running faster as expected per summary above
gene-bordegaray commentedon Nov 7, 2025
Yes this is. great summary, thank you Nga
fix: Eliminate consecutive repartitions (#18521)
Add reproducer for consecutive RepartitionExec (apache#18343)
fix: Eliminate consecutive repartitions (apache#18521)