Repository navigation
Consumer receives duplicate bound predicates when join mode is CollectLeft #17541
Description
Describe the bug
I see duplicated OR clauses on the DynamicPhysicalExpr I get in the consumer
for an execution plan like this:
ProjectionExec: expr=[c0@0 as c0, c1@1 as c1, c2@2 as c2]
CoalescePartitionsExec: fetch=5
CoalesceBatchesExec: target_batch_size=8192, fetch=5
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(c0@0, c32@32)]
CoalesceBatchesExec: target_batch_size=8192
FilterExec: c0@0 IS NOT NULL
DataSourceExec: partitions=1, partition_sizes=[1]
RepartitionExec: partitioning=RoundRobinBatch(16), input_partitions=1
CooperativeExec
DataSourceExec: partitions=1
The bounds predicates arrive as 16 identical conjuncts, 1 per (right) output partition it seems:
(
("c32" >= 'db-01' AND "c32" <= 'keb-03')
OR ("c32" >= 'db-01' AND "c32" <= 'keb-03')
OR ("c32" >= 'db-01' AND "c32" <= 'keb-03')
OR ("c32" >= 'db-01' AND "c32" <= 'keb-03')
OR ("c32" >= 'db-01' AND "c32" <= 'keb-03')
OR ("c32" >= 'db-01' AND "c32" <= 'keb-03')
OR ("c32" >= 'db-01' AND "c32" <= 'keb-03')
OR ("c32" >= 'db-01' AND "c32" <= 'keb-03')
OR ("c32" >= 'db-01' AND "c32" <= 'keb-03')
OR ("c32" >= 'db-01' AND "c32" <= 'keb-03')
OR ("c32" >= 'db-01' AND "c32" <= 'keb-03')
OR ("c32" >= 'db-01' AND "c32" <= 'keb-03')
OR ("c32" >= 'db-01' AND "c32" <= 'keb-03')
OR ("c32" >= 'db-01' AND "c32" <= 'keb-03')
OR ("c32" >= 'db-01' AND "c32" <= 'keb-03')
OR ("c32" >= 'db-01' AND "c32" <= 'keb-03')
)
This is probably related to this comment. I wrote some logic in the consumer node to dedup the predicates but it seems worth handling in DataFusion.
Following the code, in CollectLeft we derive the number of output predicates from the right side’s partition count. But iiuc CollectLeft collects the left into a single partition, so every right-side partition will see the same bounds in theory?
To Reproduce
No response
Expected behavior
No response
Additional context
No response
Activity
[-]Consumer receives duplicate predicates when join mode is CollectLeft[/-][+]Consumer receives duplicate bound predicates when join mode is CollectLeft[/+]UBarney commentedon Sep 13, 2025
take
rkrishn7 commentedon Sep 16, 2025
Makes sense to me! The number of
PartitionBoundsentries should solely be a function of the left input's partitioning.Though since we expect as many calls to
report_boundsas the output partitioning of the join, I think we can either:@UBarney saw you were working on this already. Let me know if you agree here or had other thoughts!
UBarney commentedon Sep 16, 2025
@rkrishn7
My implementation is similar to this, but I didn't use a HashSet.
I refine
report_partition_boundsLogic in shared_bounds.rs:https://github.com/apache/datafusion/pull/17551/files#diff-572ff28f0cfc7fac5657a15164403e2935d55b87cbc569670d58d117a7ed3595R277-R293