Allow co-partitioned `Partitioning::Range` inputs for symmetric hash joins

[gene-bordegaray]

Related:

• #22395
• #21992
• #23184
• #23451
• #23266

Is your feature request related to a problem or challenge?

#23451 makes partitioned SymmetricHashJoinExec require co-partitioned children, but compatible range-partitioned streams still take the conservative hash-repartition path. A partitioned symmetric hash join can process matching partition indexes locally when the inputs have compatible range layouts.

Example

SymmetricHashJoinExec: mode=Partitioned, join_type=Full, on=[(range_key@0, range_key@0)]
  StreamingTableExec: output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4)
  StreamingTableExec: output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4)

The compatible range inputs are already co-partitioned by partition index. The optimizer should not insert hash repartitions.

Describe the solution you'd like

Opt partitioned SymmetricHashJoinExec into the temporary compatible-range satisfaction path and provide a streaming test source that exposes range partitioning.

Covered join types:

• Supported SymmetricHashJoinExec join types in StreamJoinPartitionMode::Partitioned.

Acceptance criteria

• Compatible range/range inputs avoid hash repartitioning in partitioned mode.
• Incompatible split points, partition counts, sort options, or key expressions still trigger repartitioning.
• Tests cover inner and full joins, including matched and unmatched rows.
• The streaming test source exposes compatible and incompatible range layouts.

Additional context

This should follow the multi-child requirement model added in #23451 and precede the general consolidation in #23266.

[Udaytaneja]

Hi mantainers, I would like to work on this one . Please assign this to me

[Rich-T-kid]

Hi mantainers, I would like to work on this one . Please assign this to me

@Udaytaneja u can assign issues your interested in by commenting

take

on the issue and the GitHub bot will assign it to you 👍
example

[Udaytaneja]

take