sunchao commented on code in PR #58514:
URL: https://github.com/apache/spark/pull/58514#discussion_r3938579916
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -838,11 +838,18 @@ case class EnsureRequirements(
joinType: JoinType,
keyOrdering: Ordering[InternalRowComparableWrapper]):
Seq[InternalRowComparableWrapper] = {
val merged = if
(SQLConf.get.getConf(SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED)) {
+ // Rows only match within a key group, so a group no output row can come
from is dropped.
+ // Only equi-joins reach here: every SMJ/SHJ takes its keys from
`ExtractEquiJoinKeys`, which
+ // needs `joinKeys.nonEmpty`. So Cross matches on its keys like Inner.
Which side's rows
+ // survive follows `PushExtraPredicateThroughJoin`, plus LeftSingle.
joinType match {
- case Inner =>
+ // neither side keeps unmatched rows
+ case _: InnerLike | LeftSemi =>
mergeAndDedupPartitionKeys(leftPartitionKeys, rightPartitionKeys,
intersect = true)
Review Comment:
**[P2] Empty intersections lose reduced partition-key types**
Adding `LeftSemi` and `Cross` to intersection filtering exposes a planning
failure for nested storage-partitioned joins using type-changing partition
reducers.
A concrete case using the existing `days`/`years` test functions whose
reducers return `LongType`:
- A semi join between timestamps in **2020** and **2021** now retains zero
partition groups.
- Its partition metadata consequently forgets `LongType`:
`KeyedPartitioning.keyDataTypes` falls back to the original `days` transform's
`DateType` when no keys remain.
- A second join with matching **2022** timestamps retains a `LongType` group.
- A full outer join between those results should preserve the 2022 row, but
planning instead raises `STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES`.
This requires V2 bucketing, partition-value pushdown, partition filtering,
and compatible transforms enabled, with partial clustering disabled. MERGE
hints and AQE disabled give a stable physical shape.
Before this PR, the semi join retained the union of partition keys,
preserving their reduced types even when it emitted no rows. The underlying
limitation is documented as SPARK-59176 in
[`KeyedPartitioning.keyDataTypes`](https://github.com/apache/spark/blob/224c78b252c15287de85a7a2b869fa9dac0df361/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala#L613-L628),
but this change introduces another way to encounter it. The parent join's
reduced-type comparison fails at
[`EnsureRequirements.scala:568-575`](https://github.com/apache/spark/blob/224c78b252c15287de85a7a2b869fa9dac0df361/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala#L568-L575).
Please preserve reducer result types when filtering removes every key, and
add this nested-join regression test before extending intersection filtering.
**Validation:** source-traced; not reproduced locally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
