dongjoon-hyun commented on code in PR #58486:
URL: https://github.com/apache/spark/pull/58486#discussion_r3920731491
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -565,15 +565,24 @@ case class EnsureRequirements(
val (rightReducedDataTypes, rightReducedKeys) = rightReducers.fold(
(rightPartitioning.keyDataTypes, rightPartitioning.partitionKeys)
)(rightPartitioning.reduceKeys)
- val reducedDataTypes = if (leftReducedDataTypes ==
rightReducedDataTypes) {
- leftReducedDataTypes
- } else {
+ // The reduced types are the types of the key rows the merge below
sees, so only a side that
+ // has keys answers for them. `keyDataTypes` falls back to the
expressions' own types where
+ // there is no key, and after a reduce those do not describe the keys
(SPARK-59176).
+ // Skipping on an empty side is wider than that case. Where a reducer
supplied the
+ // types, the comparison was also checking the connector's
`Reducer.resultType()`
+ // against the paired transform, and that check is given up here. An
empty side has no
+ // row to misread, so a connector that breaks the contract loses a
message rather than
+ // correctness.
+ if (leftReducedKeys.nonEmpty && rightReducedKeys.nonEmpty &&
Review Comment:
The guard skips the comparison for any side with no keys, but the
`keyDataTypes` fallback is only untruthful for a marked partitioning
(`!expressionsDescribeKeys`). For an unmarked empty side the fallback is the
type its keys would have had, so the old comparison there was a valid
`Reducer.resultType()` check, and this drops it.
Shape: a one-side reduce `days(ts)` -> `years(ts)` where the `years` side is
a `GroupPartitionsExec` emptied by an upstream inner join under the partition
filter (unmarked, no keys), and a connector whose reducer returns `LongType`
against an `IntegerType` target. Before:
`STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES`. After: the check is
skipped, and with `LEFT OUTER` the `days` side reports an unmarked `years(ts)`
(`IntegerType`) over `LongType` key rows with `expressionsDescribeKeys ==
true`, so `canCreatePartitioning` accepts it. With `v2BucketingShuffleEnabled`,
`ShuffleExchangeExec` re-wraps those rows with `expressionDataTypes`;
`InterpretedHashFunction` hashes on the runtime value, so the stored `Long`
keys and the looked-up `Int` keys hash differently and
`KeyGroupedPartitioner.getPartition` falls back to `nonNegativeMod`. That is
misrouted rows rather than a lost message, so the comment above ("loses a
message rather than correctness", "no row to misread") understat
es what the PR description already concedes.
Would `keys.isEmpty && !partitioning.expressionsDescribeKeys` work instead?
It still fixes the test here (both legs are marked) and keeps the check. The
`createPartitioning` counter-case in the description needs a 0-partition spec
winning `bestSpec` plus a struct key differing only in field names, which seems
far rarer than a contract-breaking reducer and was equally broken before. If
the wide skip stays, a SPARK-59187 cross-reference here would mark it as
interim.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -612,17 +612,18 @@ case class KeyedPartitioning(
*
* The two cases can meet, and then the fallback is not truthful. A marked
partitioning can end up
Review Comment:
Two sentences outside this hunk now hold only when both sides have keys:
lines 606-608 ("A one-side reduce keeps them equal, because ...
`EnsureRequirements` refuses a reducer whose result type disagrees with it")
and the `reducersBothWays` doc at 1696-1698 ("A connector that violates the
contract ... fails the reduced-types check in `EnsureRequirements`").
`canCreatePartitioning` and `ShuffleExchangeExec` rely on that invariant
through `expressionsDescribeKeys`. Either narrow the guard so they stay true,
or qualify both.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -612,17 +612,18 @@ case class KeyedPartitioning(
*
* The two cases can meet, and then the fallback is not truthful. A marked
partitioning can end up
* with no key, for instance when `v2BucketingPartitionFilterEnabled`
intersects two sides that
Review Comment:
nit: "which no key of that partitioning would have held" restates "not
truthful" from the same sentence; the "no key row, no fact" reasoning and the
caller rule that follow are what carry the paragraph.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1099,17 +1111,44 @@ class KeyGroupedPartitioningSuite
SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true",
SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key -> "true") {
- val df = sql(reducedTsLegJoin)
+ val df = sql(reducedTsLegJoin())
- checkAnswer(df, Seq(
- Row(Timestamp.valueOf("2020-01-01 00:00:00")),
- Row(Timestamp.valueOf("2021-01-03 00:00:00"))))
+ checkAnswer(df, bothTimestamps)
val plan = stripAQEPlan(df.queryExecution.executedPlan)
assert(collectShuffles(plan).isEmpty, "should not add shuffle for any
of the three joins")
}
}
}
+ test("SPARK-59176: a leg reduced onto no key at all still joins") {
+ withReducedTsJoinLegs(bothRows, row2020, leg2YearsValues = Some(row2021)) {
+ // The second leg's two sides hold disjoint years, so the partition
filter intersects them to
+ // nothing and the leg reports a reduced partitioning with no key. The
reduced types then have
+ // to come from the first leg. The marked expressions still name the
un-reduced `days` and
+ // `years` transforms, whose types are not the `LongType` the reduced
keys hold.
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true",
+ SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key -> "true") {
+ // Both orders, since the side that has no key is the one to leave out
of the comparison.
+ // And both join types, since the inner join intersects the two key
sets to nothing and
Review Comment:
nit: `mergeAndDedupPartitions` sorts on every join type; the inner join
sorts an empty sequence rather than never sorting. "has nothing to sort" would
be accurate (the PR description says the same).
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1099,17 +1111,44 @@ class KeyGroupedPartitioningSuite
SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true",
SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key -> "true") {
- val df = sql(reducedTsLegJoin)
+ val df = sql(reducedTsLegJoin())
- checkAnswer(df, Seq(
- Row(Timestamp.valueOf("2020-01-01 00:00:00")),
- Row(Timestamp.valueOf("2021-01-03 00:00:00"))))
+ checkAnswer(df, bothTimestamps)
val plan = stripAQEPlan(df.queryExecution.executedPlan)
assert(collectShuffles(plan).isEmpty, "should not add shuffle for any
of the three joins")
}
}
}
+ test("SPARK-59176: a leg reduced onto no key at all still joins") {
Review Comment:
This covers the both-sides-marked shape only. The comment in
`EnsureRequirements` says the skip is wider than that, but nothing pins the
one-side reduce with an empty unmarked target. For example, the SPARK-56046
tables with `purchases(years(time))` joined to a third `years(time)` table
holding disjoint years under the partition filter, then
`items(days(arrive_time))` on top. That threw before this PR and returns empty
now; a test asserting whichever is intended would keep the guard from drifting.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -986,7 +999,7 @@ class KeyGroupedPartitioningSuite
SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true",
SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key -> "true") {
- checkAnswer(sql(reducedTsLegJoin),
Seq(Row(Timestamp.valueOf("2021-01-03 00:00:00"))))
+ checkAnswer(sql(reducedTsLegJoin()),
Seq(Row(Timestamp.valueOf("2021-01-03 00:00:00"))))
Review Comment:
nit: this is `bothTimestamps(1)`. Splitting into `ts2020`/`ts2021`
(`bothTimestamps = Seq(ts2020, ts2021)`) and using `Seq(ts2021)` here keeps it
symmetric with `row2021` on line 990.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -565,15 +565,24 @@ case class EnsureRequirements(
val (rightReducedDataTypes, rightReducedKeys) = rightReducers.fold(
(rightPartitioning.keyDataTypes, rightPartitioning.partitionKeys)
)(rightPartitioning.reduceKeys)
- val reducedDataTypes = if (leftReducedDataTypes ==
rightReducedDataTypes) {
- leftReducedDataTypes
- } else {
+ // The reduced types are the types of the key rows the merge below
sees, so only a side that
+ // has keys answers for them. `keyDataTypes` falls back to the
expressions' own types where
+ // there is no key, and after a reduce those do not describe the keys
(SPARK-59176).
+ // Skipping on an empty side is wider than that case. Where a reducer
supplied the
+ // types, the comparison was also checking the connector's
`Reducer.resultType()`
+ // against the paired transform, and that check is given up here. An
empty side has no
+ // row to misread, so a connector that breaks the contract loses a
message rather than
+ // correctness.
+ if (leftReducedKeys.nonEmpty && rightReducedKeys.nonEmpty &&
+ leftReducedDataTypes != rightReducedDataTypes) {
throw
QueryExecutionErrors.storagePartitionJoinIncompatibleReducedTypesError(
leftReducers = leftReducers,
leftReducedDataTypes = leftReducedDataTypes,
rightReducers = rightReducers,
rightReducedDataTypes = rightReducedDataTypes)
}
+ val reducedDataTypes =
Review Comment:
nit: the guard and this selection encode one rule in two statements that
have to stay in sync. A behaviour-identical single form:
```scala
val reducedDataTypes = if (leftReducedKeys.isEmpty) {
rightReducedDataTypes
} else if (rightReducedKeys.isEmpty || leftReducedDataTypes ==
rightReducedDataTypes) {
leftReducedDataTypes
} else {
throw
QueryExecutionErrors.storagePartitionJoinIncompatibleReducedTypesError(...)
}
```
--
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]
