peter-toth commented on code in PR #58486:
URL: https://github.com/apache/spark/pull/58486#discussion_r3923010498
##########
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:
You are right, and I measured your shape. Written as a test, it returns
empty with the guard as it was, where it used to raise
`STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES`.
The guard is now your condition, `keys.nonEmpty || expressionsDescribeKeys`
per side. An empty side that nothing reduced reports the types its keys would
have held, so it answers for them and stays in the comparison. The
`createPartitioning` case I used to argue for the wider skip is, as you say,
both rarer and equally broken before this PR, so it does not pay for a lost
reducer check.
I did not verify the misrouting chain itself, since the fix is the same
whichever way that shape ends. What I did verify is that the check disappears,
which is enough to reject the wider guard. Neither the comment nor the
description claims anything is given up now.
Fixed in
https://github.com/apache/spark/pull/58486/commits/979f42093afad48c2322660a7b6548c16585b1e4
##########
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:
With the narrower guard both sentences hold again. A one-side reduce with an
empty unmarked target is still compared, so `keyDataTypes`' line about
`EnsureRequirements` refusing a disagreeing reducer is true, and so is the
`reducersBothWays` line. Both stay as they are.
The paragraph you commented on now says which side is left out, since that
is the part that changed.
Fixed in
https://github.com/apache/spark/pull/58486/commits/979f42093afad48c2322660a7b6548c16585b1e4
##########
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:
Added, close to your shape: `items(days(arrive_time))` over a `years(time)`
leg that an upstream inner join empties under the partition filter, with
`UnboundDaysFunctionWithToYearsReducerWithDateResult` as the contract-breaking
reducer. It asserts the error, so the intended answer is that it still throws.
That test is what measured the wider guard as wrong.
Fixed in
https://github.com/apache/spark/pull/58486/commits/979f42093afad48c2322660a7b6548c16585b1e4
##########
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:
Taken. With the narrower condition it reads:
```scala
val reducedDataTypes = if (!leftTypesDescribeKeys) {
rightReducedDataTypes
} else if (!rightTypesDescribeKeys || leftReducedDataTypes ==
rightReducedDataTypes) {
leftReducedDataTypes
} else {
throw
QueryExecutionErrors.storagePartitionJoinIncompatibleReducedTypesError(...)
}
```
Fixed in
https://github.com/apache/spark/pull/58486/commits/979f42093afad48c2322660a7b6548c16585b1e4
##########
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:
Taken, `bothTimestamps` is `Seq(ts2020, ts2021)` now and this site uses
`Seq(ts2021)`.
Fixed in
https://github.com/apache/spark/pull/58486/commits/979f42093afad48c2322660a7b6548c16585b1e4
##########
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:
Right, it sorts an empty sequence. Reworded to "has nothing to sort".
Fixed in
https://github.com/apache/spark/pull/58486/commits/979f42093afad48c2322660a7b6548c16585b1e4
##########
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:
Dropped.
Fixed in
https://github.com/apache/spark/pull/58486/commits/979f42093afad48c2322660a7b6548c16585b1e4
--
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]
