Consumer receives duplicate bound predicates when join mode is CollectLeft

[LiaCastaneda]

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

[UBarney]

take

[rkrishn7]

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?

Makes sense to me! The number of PartitionBounds entries should solely be a function of the left input's partitioning.

Though since we expect as many calls to report_bounds as the output partitioning of the join, I think we can either:

• Make it a no-op if there are more calls than this
• Collect partition bounds in a hashset keyed by partition so redundant updates have no effect

@UBarney saw you were working on this already. Let me know if you agree here or had other thoughts!

[UBarney]

• Collect partition bounds in a hashset keyed by partition so redundant updates have no effect

@rkrishn7

My implementation is similar to this, but I didn't use a HashSet.

I refine report_partition_bounds Logic in shared_bounds.rs:

• The partition parameter has been renamed to left_side_partition_id to clarify that it refers to the partition from the left (build) side.
• New deduplication logic has been added: it skips reporting bounds if the current left_side_partition_id is identical to the previously recorded one. This directly fixes the issue in CollectLeft mode, where all streams report the same left_side_partition_id of 0, thus preventing duplicate expressions.

https://github.com/apache/datafusion/pull/17551/files#diff-572ff28f0cfc7fac5657a15164403e2935d55b87cbc569670d58d117a7ed3595R277-R293