ulysses-you commented on code in PR #58335:
URL: https://github.com/apache/spark/pull/58335#discussion_r3877548711
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1374,9 +1391,25 @@ case class KeyedShuffleSpec(
*
* @param other other key-grouped shuffle spec
*/
- def reducers(other: KeyedShuffleSpec): Option[Seq[Option[Reducer[_, _]]]] = {
+ def reducers(other: KeyedShuffleSpec): Option[Seq[Option[KeyReducer]]] = {
val results =
partitioning.expressions.zip(other.partitioning.expressions).map {
- case (e1: TransformExpression, e2: TransformExpression) =>
e1.reducers(e2)
+ case (e1: TransformExpression, e2: TransformExpression) =>
+ e1.reducers(e2).map { reducer =>
+ if (e2.reducers(e1).isEmpty) {
+ // Only this side reduces. The reducer contract is r(f1(x)) =
f2(x) where "=" matches
+ // both value and data type, so the reduced keys equal the target
transform applied to
+ // this side's child. Report that expression (re-targeted at this
side's attribute)
+ // instead of the un-reduced `e1`, whose type can differ from the
reduced keys.
+ val thisSideChild = e1.references.head
+ val reducedExpr = e2.transform { case _: AttributeReference =>
thisSideChild }
+ KeyReducer(reducer, reducedExpr.asInstanceOf[TransformExpression])
+ } else {
+ // Both sides reduce: the reduced keys are r1(f1(x)) = r2(f2(x)),
which no single
+ // transform describes. Keep reporting the original expression.
Known gap, tracked in
+ // the follow-up for SPARK-59045.
+ KeyReducer(reducer, e1)
Review Comment:
Thanks for the detailed analysis. We explored both fail-safe options and
decided to keep reporting the un-reduced expression until the follow-up, for
two reasons:
1. `return None` does not actually fall back to a shuffle. With no reducer
on either side, the reduced-types cross-check in `EnsureRequirements` compares
the two sides' *original* types (`DateType` vs `IntegerType` in your example)
and throws `storagePartitionJoinIncompatibleReducedTypesError` at planning
time. We verified this by running the SPARK-56164 query with a per-pair `None`
implementation - it failed with exactly that error. So `None` and raising the
dedicated error are indistinguishable here: the query cannot run either way,
and the error case is already covered by the existing cross-check without new
code.
2. `None` additionally opens a silent-wrong-results hole the dedicated error
does not have: if both transforms' original types are equal while the reducers'
result type differs (constructible by a third-party `ReducibleFunction` pair),
both sides return `None`, the cross-check passes on the equal original types,
and the two sides' un-reduced key spaces get merged without any reduction.
Since the reduction itself produces correct key values in this shape (only
the reported expression is mis-typed), we kept it and left the gap to the
follow-up, which carries the reduced data types on `KeyedPartitioning` and
fixes this shape properly.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -59,7 +58,7 @@ case class GroupPartitionsExec(
child: SparkPlan,
@transient joinKeyPositions: Option[Seq[Int]] = None,
@transient expectedPartitionKeys:
Option[Seq[(InternalRowComparableWrapper, Int)]] = None,
- @transient reducers: Option[Seq[Option[Reducer[_, _]]]] = None,
+ @transient reducers: Option[Seq[Option[KeyReducer]]] = None,
Review Comment:
Fixed in 943811539a8. `GroupPartitionsExec.doCanonicalize` now normalizes
the exprIds inside `reducedExpression` via `QueryPlan.normalizeExpressions`
against the child's output (the same pattern `BatchScanExec` uses for
`keyGroupedPartitioning`), so structurally identical SPJ subtrees with
value-equal reducers compare equal after canonicalization again. Added a test
that builds two `GroupPartitionsExec`s with differently-numbered exprIds and
value-equal `BucketReducer` instances and asserts their canonical forms are
equal.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -713,17 +714,18 @@ object KeyedPartitioning {
def reduceKeys(
keys: Seq[InternalRowComparableWrapper],
dataTypes: Seq[DataType],
- reducers: Seq[Option[Reducer[_, _]]]): (Seq[DataType],
Seq[InternalRowComparableWrapper]) = {
+ reducers: Seq[Option[KeyReducer]]):
+ (Seq[DataType], Seq[InternalRowComparableWrapper]) = {
val reducedDataTypes = dataTypes.zip(reducers).map {
- case (_, Some(reducer: Reducer[Any, Any])) => reducer.resultType()
+ case (_, Some(KeyReducer(reducer: Reducer[Any, Any], _))) =>
reducer.resultType()
Review Comment:
Agreed on the drift risk. We first tried asserting agreement when
constructing `KeyReducer`, but the in-tree `DaysToYearsReducerWithDateResult`
intentionally violates it (`SPARK-56046: Reducers with different result types`
depends on such a reducer reaching the cross-check error), so a hard assert
would break that test. Instead the agreement is now closed by construction / by
the existing check in 943811539a8: the identity-vs-transform reducer reports
the transform itself; the single-side branch reports the target transform,
whose type the reduced-types cross-check in `EnsureRequirements` validates
against the other side's target-transform-typed keys; and the both-sides-reduce
branch reports the original expression (known gap, see the other thread).
Comments in `reducersBothWays` spell this out.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1374,9 +1391,25 @@ case class KeyedShuffleSpec(
*
* @param other other key-grouped shuffle spec
*/
- def reducers(other: KeyedShuffleSpec): Option[Seq[Option[Reducer[_, _]]]] = {
+ def reducers(other: KeyedShuffleSpec): Option[Seq[Option[KeyReducer]]] = {
val results =
partitioning.expressions.zip(other.partitioning.expressions).map {
- case (e1: TransformExpression, e2: TransformExpression) =>
e1.reducers(e2)
+ case (e1: TransformExpression, e2: TransformExpression) =>
+ e1.reducers(e2).map { reducer =>
+ if (e2.reducers(e1).isEmpty) {
+ // Only this side reduces. The reducer contract is r(f1(x)) =
f2(x) where "=" matches
+ // both value and data type, so the reduced keys equal the target
transform applied to
+ // this side's child. Report that expression (re-targeted at this
side's attribute)
+ // instead of the un-reduced `e1`, whose type can differ from the
reduced keys.
+ val thisSideChild = e1.references.head
+ val reducedExpr = e2.transform { case _: AttributeReference =>
thisSideChild }
Review Comment:
Fixed in 943811539a8. The reducer construction now stores the raw
expression, and `GroupPartitionsExec.outputPartitioning` does the single
re-targeting at the use site (via the new `TransformExpression.withReference`
helper).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1374,9 +1391,25 @@ case class KeyedShuffleSpec(
*
* @param other other key-grouped shuffle spec
*/
- def reducers(other: KeyedShuffleSpec): Option[Seq[Option[Reducer[_, _]]]] = {
+ def reducers(other: KeyedShuffleSpec): Option[Seq[Option[KeyReducer]]] = {
val results =
partitioning.expressions.zip(other.partitioning.expressions).map {
- case (e1: TransformExpression, e2: TransformExpression) =>
e1.reducers(e2)
+ case (e1: TransformExpression, e2: TransformExpression) =>
+ e1.reducers(e2).map { reducer =>
+ if (e2.reducers(e1).isEmpty) {
Review Comment:
Fixed in 943811539a8. The two directions are now computed in one pass,
`KeyedShuffleSpec.reducersBothWays`, so each expression pair's `reducers`
lookup runs once per direction: the reverse lookup a direction used only for
the single-side probe is exactly the other direction's reducer.
`EnsureRequirements` now calls it once instead of calling `reducers` on both
specs.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -70,11 +69,29 @@ case class GroupPartitionsExec(
// There can be multiple `KeyedPartitioning`s in an output
partitioning of a join, but they
// can only differ in `expressions`; their `partitionKeys` reference
is shared (enforced by
// `PartitioningCollection`), so `groupedPartitions` is computed only
once.
+ // When reducers are applied, the reduced expressions are reported
instead of the original
+ // ones. For the identity-vs-transform and single-side-transform
reducers their data type
+ // matches the reduced partition keys by construction; for the
both-sides-reduce shape no
+ // single transform describes the keys (see
`KeyedShuffleSpec.reducers`).
val partitionKeys = groupedPartitions.map(_._1)
p.transform {
case k: KeyedPartitioning =>
val projectedExpressions =
joinKeyPositions.fold(k.expressions)(_.map(k.expressions))
- KeyedPartitioning(projectedExpressions, partitionKeys, isGrouped =
isGrouped)
+ val effectiveExpressions = reducers match {
+ case Some(exprs) =>
+ assert(projectedExpressions.length == exprs.length)
+ projectedExpressions.zip(exprs).map {
+ case (expr, Some(KeyReducer(_, reduced))) =>
+ // `reduced` was derived from the single spec that
`createKeyedShuffleSpec`
+ // picked (`collectFirst`); re-target it at this
`KeyedPartitioning`'s own key
+ // attribute so that every `KeyedPartitioning` in a
collection keeps its own.
+ val attr = expr.references.head
+ reduced.transform { case _: AttributeReference => attr }
Review Comment:
Fixed in 943811539a8. Added `TransformExpression.withReference(attr)` in
catalyst and switched all the copies (the reducer construction in both
directions and the use site) to it, which also removes the `asInstanceOf` casts
at the call sites.
--
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]
