You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "peter-toth (via GitHub)" <gi...@apache.org> on 2026/09/01 11:39:48 UTC
[PR] [SPARK-59121][SQL] Fix wrong results when a storage-partitioned join reduces the partition keys of both sides [spark]
peter-toth opened a new pull request, #58447:
URL: https://github.com/apache/spark/pull/58447
### What changes were proposed in this pull request?
This builds on #58335, which makes a reduced key-grouped partitioning report a transform that describes its keys for the two reducer shapes where one exists. When both sides of the join reduce there is none: the keys are `r1(f1(x))` = `r2(f2(x))`, a third key space that neither side's transform names. `bucket(12, id)` joined to `bucket(8, id)` is the flagship example - `BucketFunction.reducer` hands both sides `BucketReducer(4)`, the keys become `id % 4`, and both sides keep reporting the transforms they were built from. That is the gap #58335 names and leaves open, and this PR closes it by saying so rather than by inventing an expression.
`TransformExpression` gains a fourth field, `reducedWith: Option[TransformFunctionId]`, which names the transform this one's keys were reduced together with. `KeyedShuffleSpec.reducersBothWays` is the only producer: in its both-sides-reduce branch it now reports `e1.reducedTogetherWith(e2)` instead of the bare `e1`. The marker holds no `Expression`, only a canonical name and a bucket count, so it is safe in a field canonicalization does not descend into.
Putting it on the expression rather than on `KeyedPartitioning` is what keeps the change small. Every site that derives a partitioning already carries the expressions along - `AliasAwareOutputExpression` projects them, `GroupPartitionsExec.outputPartitioning` re-reports them, `TransformExpression.withReference` re-targets them - so the marker is inherited, and dropped with the position it belongs to, without a line of new plumbing. `GroupPartitionsExec` needed no change at all.
Four sites then refuse to reason about such keys, and a fifth refusal falls out of expression equality:
- `KeyedShuffleSpec.isExpressionCompatible` does not compare marked keys by transform. Two of them are compatible when the same pair was reduced together, which is the pair the join produced; anything else has to shuffle.
- `KeyedShuffleSpec.canCreatePartitioning` does not shuffle another child onto marked keys, because that evaluates the reported expressions per row.
- `KeyedShuffleSpec.reducersBothWays` does not reduce marked keys a second time.
- `keysSatisfy` does not let marked keys satisfy an `OrderedDistribution`. An ordering is a claim about the key *values*, and nothing makes a reducer order-preserving. This one is a local guard: a marked position always carries a transform, and a SQL `ORDER BY` cannot name one.
- `UnionExec.comparePartitioning` compares the children's expressions with `semanticEquals`, so a marked child no longer merges with an unmarked sibling reporting the same transform. No change was needed there.
Clustering is the one thing a marked partitioning still satisfies, and that is sound: the keys remain a function of the same attributes, which is all `ClusteredDistribution` asks.
Two things promised in the review of SPARK-59120 (#58420) are now delivered, since the marker answers the question they approximated: `KeyedPartitioning.expressionsDescribeKeyShape` and its use in `canCreatePartitioning` are replaced by `expressionsDescribeKeys`, and the three-reader list in the `keyDataTypes` scaladoc is gone.
Keeping the type check as a second conjunct was considered and dropped. It is not sufficient - `bucket(12)` and `bucket(8)` reducing onto `bucket(4)` keep their `IntegerType` and pass it - and it is not necessary either: after a one-side reduce the reported expression is the target transform, and `EnsureRequirements` already refuses a reducer whose `resultType()` disagrees with the other side's key types, which are that transform's, with `STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES`. So the keys and the expressions agree at every reachable one-side reduce, and a both-sides reduce is marked. The unit test that pinned the type proxy is replaced by one that pins the marker, plus an assertion that a struct-typed key needs no special handling now that the gate does not compare types.
Two smaller things came with it. `TransformExpression.resolveFunctionCall()` refuses a marked expression, so `eval` throws on one instead of computing the un-reduced transform and misrouting the row. That is a local gate rather than a live check: every consumer of a reduced partitioning refuses it first, and the write path never sees one. It also replaces the copy of the rule that prepends the bucket count as a literal argument in `DistributionAndOrderingUtils` - both copies had to be touched here anyway, since the extractor grew a field. It builds a fresh expression per call, because the result can be stateful.
### Why are the changes needed?
Four failures, each one a test here, all measured on this PR's base. All of them need `spark.sql.sources.v2.bucketing.allowCompatibleTransforms.enabled`, which is off by default, since that is what admits a reducer at all.
1. Two reduces onto different key spaces look co-partitioned. Four tables holding ids 0 until 24, bucketed by 12 and 8 on one side and by 12 and 18 on the other: the first pair reduces onto `id % 4`, the second onto `id % 6`, and both keep reporting `bucket(12, id)`. Joining the two returns **8 of 24 rows** with no shuffle. This is also what the pairing in the marker is for: marking the keys without recording which pair produced them fixes everything else here and still returns those 8 rows.
2. Reducing keys twice. Three tables bucketed by 12, 8 and 6: the first join reduces onto `id % 4`, then the second derives a `bucket(6)` reducer from the reported `bucket(12, id)` and applies it to keys that are already reduced. `(id % 4) % 6` leaves the left keys alone while the right side moves to `id % 6`, so the query loses rows with no shuffle.
3. Reducing keys twice, the other way. Two `days`/`years` joins that each reduce both sides onto one year space, then joined to each other with different key sets on the two legs: the outer join derives reducers again and planning throws `ClassCastException: Long cannot be cast to Integer`.
4. Merging a reduced partitioning in a union. `(bucket12 JOIN bucket8) UNION ALL bucket12` reports `bucket(12, id)` on both sides of the union, so the two are merged although the join side's keys are `id % 4`. A `GROUP BY id` above it returns each id twice.
The fifth refusal, `canCreatePartitioning`, guards a hazard the first one opens up. Once a marked partitioning is no longer compatible with anything but its own pairing, `EnsureRequirements` takes the one-side-shuffle path instead, and there it would happily shuffle the other child onto the reduced keys - which places rows by evaluating `bucket(12, id)` against partitions laid out by `id % 4`. Measured with tables bucketed by 12, 8 and 2 and `v2BucketingShuffleEnabled`: with the clause removed the query returns **8 of 12 rows**.
### Does this PR introduce _any_ user-facing change?
Yes, it fixes the wrong results and the crash above. A query that used to reach one of them now shuffles instead, so its plan changes.
It also refuses two plans that happen to be correct today, both because nothing can tell the two cases apart from the outside.
The first is a second join onto an already reduced space. `bucket(12) JOIN bucket(8)` lands on `id % 4`; meeting a `bucket(4, id)` or `bucket(2, id)` table, `BucketReducer` would compose correctly, since those counts divide 4. Meeting a `bucket(6, id)` table it would not, and that is failure 2 above. The `Reducer` API cannot say which of the two it is, so both now shuffle: `bucket12 JOIN bucket8 JOIN bucket4` goes from 0 shuffles to 2, with the same rows. This only arises where both sides reduced, i.e. where `gcd(a, b) < min(a, b)`; a chain in which one bucket count divides the other takes the one-side path and is unaffected.
The second is two reduces that land on the same space through different pairings, e.g. `bucket(12)` with `bucket(8)` and `bucket(12)` with `bucket(20)`, both of which give `id % 4`. They are treated as different spaces and the join shuffles.
Both cost a shuffle, not a wrong answer. A follow-up can give them back by letting a `ReducibleFunction` name the transform it reduces onto, which makes this whole shape disappear rather than be refused.
Queries whose keys were not reduced on both sides are unaffected, and a join of two sides that were reduced together keeps its plan.
### How was this patch tested?
Ten new tests: six query tests in `KeyGroupedPartitioningSuite`, and one each in `TransformExpressionSuite`, `ShuffleSpecSuite`, `GroupPartitionsExecSuite` and `ProjectedOrderingAndPartitioningSuite`. One existing test changed, see below.
Four of the six query tests fail on the base commit with wrong rows or a crash, listed as failures 1 to 4 above. A fifth, `another side is not shuffled onto reduced keys`, returns the right rows there with no shuffle; this PR costs it one shuffle, and it is what pins the `canCreatePartitioning` clause. Each refusal was ablated, and each ablation fails exactly the tests written for it:
- `canCreatePartitioning`'s clause removed: `another side is not shuffled onto reduced keys` returns 8 of 12 rows.
- the same-pairing allowance in `isExpressionCompatible` made unconditional, i.e. the refusal taken too far: `two sides reduced onto the same keys still join without a shuffle` fails, and so does the existing `SPARK-56164: Reducers with different result types to original keys`.
- the pairing thrown away, so that any two marked keys count as one space: `two reduced partitionings are not compatible by their transforms` returns 8 of 24 rows. That is the ablation the test exists for, and it is why the marker records the pair rather than a bit.
- the `reducersBothWays` guard removed: `two sides reduced together are not reduced a second time` throws the `ClassCastException`. No other test in the suite reaches that guard - instrumenting it to throw on entry showed every one of them misses it - which is why the test is built the way it is, with different key sets on the two legs so that the join computes reducers at all.
`SPARK-56164` is the existing test for the both-sides-reduce shape, and it gained one assertion: that the join's own requirements still validate, i.e. that the two sides which were reduced together are still co-partitioned. That is the no-regression half of this change, and it belongs on the test that already describes the shape rather than in a copy of it.
The four unit tests cover what a query test states only indirectly: the pairing relation itself, the `canCreatePartitioning` refusal, that a node which reduces nothing (or reduces another position) inherits the marker from its child, and that a projection drops the marker with the position it belongs to. The last two hold by construction under this representation, which is exactly why they are worth keeping - an earlier design of this fix carried the marker on `KeyedPartitioning` and got both wrong.
Green: `ShuffleSpecSuite`, `DistributionSuite`, `TransformExpressionSuite`, `KeyGroupedPartitioningSuite`, `GroupPartitionsExecSuite`, `EnsureRequirementsSuite`, `ProjectedOrderingAndPartitioningSuite`, `ValidateRequirementsSuite`, `WriteDistributionAndOrderingSuite`, `PlannerSuite`, `UnionSuite`, `DataSourceV2Suite` - 488 tests. `dev/lint-scala` is clean.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59121][SQL] Fix wrong results when a storage-partitioned join reduces the partition keys of both sides [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #58447:
URL: https://github.com/apache/spark/pull/58447#issuecomment-5515249740
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/ed06d809b020ca5e7dfdb29f9559fca2355b03f8
- merged into branch-4.x https://github.com/apache/spark/commit/92b20aa40713d7504e37ce5b972b60f9cebbc327
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59121][SQL] Fix wrong results when a storage-partitioned join reduces the partition keys of both sides [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #58447:
URL: https://github.com/apache/spark/pull/58447#discussion_r3913547045
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1598,24 +1559,40 @@ case class KeyedShuffleSpec(
}
}
- private def isExpressionCompatible(left: Expression, right: Expression): Boolean =
- (left, right) match {
- case (_: LeafExpression, _: LeafExpression) => true
- case (left: TransformExpression, right: TransformExpression) =>
- if (SQLConf.get.v2BucketingPushPartValuesEnabled &&
- !SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled &&
- SQLConf.get.v2BucketingAllowCompatibleTransforms) {
- left.isCompatible(right)
- } else {
- left.isSameFunction(right)
- }
- case (_: AttributeReference, _: TransformExpression) |
- (_: TransformExpression, _: AttributeReference) =>
- SQLConf.get.v2BucketingPushPartValuesEnabled &&
- !SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled &&
- SQLConf.get.v2BucketingAllowCompatibleTransforms
- case _ => false
+ private def isExpressionCompatible(left: Expression, right: Expression): Boolean = {
+ if (TransformExpression.hasReducedKeys(left) || TransformExpression.hasReducedKeys(right)) {
+ // Reduced keys are in a key space that neither transform names, so comparing the transforms
+ // says nothing about whether the two sides are laid out the same way. The pair that was
+ // reduced together is laid out the same way, since its two sides came out of one reduce onto
+ // one key space. Anything else has to shuffle. That includes a pair that reduced onto the
+ // same space through a different pairing, which nothing here can tell apart, and an identity
+ // side, which holds raw values.
+ (left, right) match {
+ case (l: TransformExpression, r: TransformExpression) => l.hasSameReducedKeys(r)
Review Comment:
+1 to make the condition strict.
The information is already lost at reduce time. After a both-sides reduce the real key space is `r1(f1(x))`, which no transform names, and the current `Reducer` API cannot say which space it reduced onto. All the planner has is the reported original expression (e.g. `bucket(32, id)`).
If the gate were widened to allow deriving another reducer from the reported expression (i.e. reducing already-reduced keys again), it would admit not only the divisible, benign case above but also failure 2 of this PR: `bucket(12) JOIN bucket(8)` lands on `id % 4`, then joining `bucket(6)` derives `(id % 4) % 6` — the left keys stay put while the right side moves to `id % 6`, and the query loses rows silently.
This is not an implementation accident; it is mathematically undecidable from the reduced values: a key of `id % 4 = 0` can stand for `id = 0, 4, 8`, whose `id % 6` values are `0, 4, 2` respectively. Mapping from a reduced space to a non-divisible one is simply not well-defined. So unless the exact reduced space is known, "reduce again" is only sound in the divisible case, and the available information cannot tell divisible from non-divisible.
So `hasSameReducedKeys` admits only the one provably sound case: both sides came out of the same reduce (same pairing), which guarantees the same layout. Everything else has to shuffle. That includes:
1. A further reduce onto a divisor (the `32/12` vs `64/24` case above: `id % 4` vs `id % 8`).
2. Different pairings that happen to land on the same space (`12 JOIN 8` and `12 JOIN 20` both give `id % 4`).
3. Meeting an unreduced table that is already laid out on that space (`bucket(12) JOIN bucket(8)` reduced onto `id % 4`, joined to a raw `bucket(4, id)` table whose keys are already aligned).
The cost in all three is an extra shuffle, never a wrong answer. On the base these shapes ran with 0 shuffles — correct by accident in the divisible cases, losing rows in the non-divisible ones — and the API cannot keep only the good half, so tightening the gate wholesale was the only safe choice available.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59121][SQL] Fix wrong results when a storage-partitioned join reduces the partition keys of both sides [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on PR #58447:
URL: https://github.com/apache/spark/pull/58447#issuecomment-5509682394
Thank you @peter-toth and @dongjoon-hyun @ulysses-you!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59121][SQL] Fix wrong results when a storage-partitioned join reduces the partition keys of both sides [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #58447:
URL: https://github.com/apache/spark/pull/58447#issuecomment-5507429180
@dongjoon-hyun, @ulysses-you, I updated the PR and it is now ready for review.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59121][SQL] Fix wrong results when a storage-partitioned join reduces the partition keys of both sides [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #58447:
URL: https://github.com/apache/spark/pull/58447#discussion_r3916478242
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -596,47 +596,42 @@ case class KeyedPartitioning(
* The types the `partitionKeys` rows were built with. Anything reading those rows should take its
* types from here. It is a driver-side value, since `partitionKeys` is `@transient`.
*
- * They are the `expressionDataTypes` unless a reducer has rewritten the keys. With
- * `v2BucketingAllowCompatibleTransforms` a storage-partitioned join reduces one or both sides'
- * keys onto a common key space, while the partitioning keeps reporting the expressions it was
- * built from. Joining an `identity(ts)`-partitioned table to a `years(ts)`-partitioned one leaves
- * `IntegerType` year values under a `TimestampType`-declared expression. With no key at all the
- * expressions are all there is, and there is no row to read or to place.
+ * They differ from the `expressionDataTypes` in two cases. A join that reduced both sides' keys
+ * onto a key space no transform names leaves a marked expression whose type can be anything, see
+ * `expressionsDescribeKeys`. A one-side reduce keeps them equal, because the expression the
+ * partitioning then reports is the target transform and `EnsureRequirements` refuses a reducer
+ * whose result type disagrees with it. `KeyedShuffleSpec.createPartitioning` is the other case.
+ * It puts the other child's expressions over these keys with no reducer in sight, so a struct
+ * field can be named differently on the two sides. With no key at all the expressions are all
+ * there is, and there is no row to read or to place.
Review Comment:
Thank you, this reproduces. I measured it both ways, on `days`/`years` tables where one leg's two sides hold disjoint keys, so the intersect empties it.
On this PR the query fails with `STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES`. On the base commit the same query fails with `ClassCastException: Long cannot be cast to Integer`, from applying a reducer to already reduced values. So the shape is broken either way and nothing here regresses.
One correction to the framing. The untruthful fallback is pre-existing, it arrived with SPARK-59120. The failure at the reduced-types check becomes reachable through this PR, because once no reducer is derived for an already reduced pair both sides read `keyDataTypes` directly. Before that the reducers were computed and both sides reported the reducer's type, so the check passed and the query died later.
Filed as SPARK-59176 with the repro and the two ways to fix it. The scaladoc now points at it, and I have started on it in a separate branch.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala:
##########
@@ -122,27 +147,78 @@ case class TransformExpression(
Option(res)
}
+ /**
+ * The unordered pair of transforms whose reduce produced this expression's keys, which is what
+ * identifies the key space they landed in. Reducing is symmetric, so the two sides of one reduce
+ * carry the same pair.
+ */
+ private def reducedKeySpace: Option[Set[TransformFunctionId]] =
+ reducedWith.map(partner => Set(functionId, partner))
+
+ /**
+ * Whether this and `other` describe the same reduced key space, i.e. whether the same pair of
+ * transforms was reduced together to produce both.
+ *
+ * Two reduces that happen to land on the same space through different pairings are not
+ * recognised as one. For instance `bucket(12)` with `bucket(8)` and `bucket(12)` with
+ * `bucket(20)` both reduce onto `id % 4`. The [[Reducer]] API does not name the space it reduces
+ * onto, so the pairing is all there is to compare.
+ */
+ def hasSameReducedKeys(other: TransformExpression): Boolean =
+ reducedWith.isDefined && reducedKeySpace == other.reducedKeySpace
+
+ /** Records that this expression's keys were reduced together with `other`'s. */
+ def reducedTogetherWith(other: TransformExpression): TransformExpression =
+ copy(reducedWith = Some(other.functionId))
+
override def dataType: DataType = function.resultType()
override protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): Expression =
copy(children = newChildren)
- private lazy val resolvedFunction: Option[Expression] = this match {
- case TransformExpression(scalarFunc: ScalarFunction[_], arguments, Some(numBuckets)) =>
- Some(V2ExpressionUtils.resolveScalarFunction(scalarFunc,
- Seq(Literal(numBuckets)) ++ arguments))
- case TransformExpression(scalarFunc: ScalarFunction[_], arguments, None) =>
+ /**
+ * Builds the scalar function call this transform stands for, with the bucket count prepended as
+ * a literal argument. `None` when the bound function is not a [[ScalarFunction]], and also when a
+ * join reduced this expression's keys, since then the call no longer computes them. Evaluating it
+ * to place a row would send the row to a partition it does not belong to. That second arm is a
+ * local gate. No caller reaches it today, because every consumer of a reduced partitioning
+ * refuses it first, and the write path never sees one.
+ *
+ * A fresh expression per call. The result can be stateful ([[ApplyFunctionExpression]]), so a
+ * caller that puts it in a plan must not share one instance across two positions.
+ */
+ def resolveFunctionCall(): Option[Expression] = function match {
+ case scalarFunc: ScalarFunction[_] if reducedWith.isEmpty =>
+ val arguments = numBucketsOpt.fold(children)(n => Literal(n) +: children)
Some(V2ExpressionUtils.resolveScalarFunction(scalarFunc, arguments))
case _ => None
}
- override def eval(input: InternalRow): Any = {
- resolvedFunction match {
- case Some(fn) => fn.eval(input)
- case None => throw QueryExecutionErrors.cannotEvaluateExpressionError(this)
- }
+ /**
+ * Memoised for `eval`, which runs per row. Safe to reuse only because it stays inside this
+ * expression, unlike the call `resolveFunctionCall` hands to a caller that plans with it.
+ */
+ private lazy val evaluableFunctionCall: Option[Expression] = resolveFunctionCall()
Review Comment:
Agreed, and both mechanisms are there. `QueryExecution.cloneWithFreshStatefulExpressions` maps every node's expressions through `freshCopyIfContainsStatefulExpression()` before optimization, and `ExpressionsEvaluator.prepareExpressions` does the same per evaluator. So the hazard the split guarded against cannot arise.
Collapsed back into one `lazy val resolvedFunction`, un-private, with the `reducedWith` guard, and both caveat comments are gone.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59121][SQL] Fix wrong results when a storage-partitioned join reduces the partition keys of both sides [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth closed pull request #58447: [SPARK-59121][SQL] Fix wrong results when a storage-partitioned join reduces the partition keys of both sides
URL: https://github.com/apache/spark/pull/58447
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59121][SQL] Fix wrong results when a storage-partitioned join reduces the partition keys of both sides [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #58447:
URL: https://github.com/apache/spark/pull/58447#issuecomment-5515259560
Thank you everyone for the review!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59121][SQL] Fix wrong results when a storage-partitioned join reduces the partition keys of both sides [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #58447:
URL: https://github.com/apache/spark/pull/58447#discussion_r3914916069
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -596,47 +596,42 @@ case class KeyedPartitioning(
* The types the `partitionKeys` rows were built with. Anything reading those rows should take its
* types from here. It is a driver-side value, since `partitionKeys` is `@transient`.
*
- * They are the `expressionDataTypes` unless a reducer has rewritten the keys. With
- * `v2BucketingAllowCompatibleTransforms` a storage-partitioned join reduces one or both sides'
- * keys onto a common key space, while the partitioning keeps reporting the expressions it was
- * built from. Joining an `identity(ts)`-partitioned table to a `years(ts)`-partitioned one leaves
- * `IntegerType` year values under a `TimestampType`-declared expression. With no key at all the
- * expressions are all there is, and there is no row to read or to place.
+ * They differ from the `expressionDataTypes` in two cases. A join that reduced both sides' keys
+ * onto a key space no transform names leaves a marked expression whose type can be anything, see
+ * `expressionsDescribeKeys`. A one-side reduce keeps them equal, because the expression the
+ * partitioning then reports is the target transform and `EnsureRequirements` refuses a reducer
+ * whose result type disagrees with it. `KeyedShuffleSpec.createPartitioning` is the other case.
+ * It puts the other child's expressions over these keys with no reducer in sight, so a struct
+ * field can be named differently on the two sides. With no key at all the expressions are all
+ * there is, and there is no row to read or to place.
Review Comment:
nit: the two cases described here can overlap, and the doc does not say what happens then. A both-sides-reduced (marked) partitioning can end up with zero `partitionKeys`, e.g. an inner join with `V2_BUCKETING_PARTITION_FILTER_ENABLED` whose two sides hold disjoint keys, so `mergeAndDedupPartitionKeys(intersect = true)` yields `Nil`. Then `keyDataTypes` falls back to `expressionDataTypes`, i.e. the un-reduced transform's type (`DateType` for `days`), while the sibling leg of the same pairing reports the reducer's type (`LongType`). `EnsureRequirements` compares the two at the reduced-types check without consulting `expressionsDescribeKeys`, since `reducersBothWays` returns `(None, None)`, and throws `STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES` for a correct (empty) plan.
This is pre-existing on the base commit, so not a blocker. But since this PR rewrites the doc, it would be worth one sentence noting that the no-key fallback is not truthful for a marked expression, or a `SPARK-` reference if you prefer to track it separately. Carrying the reducer's `resultType()` in the marker would fix it properly.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala:
##########
@@ -122,27 +147,78 @@ case class TransformExpression(
Option(res)
}
+ /**
+ * The unordered pair of transforms whose reduce produced this expression's keys, which is what
+ * identifies the key space they landed in. Reducing is symmetric, so the two sides of one reduce
+ * carry the same pair.
+ */
+ private def reducedKeySpace: Option[Set[TransformFunctionId]] =
+ reducedWith.map(partner => Set(functionId, partner))
+
+ /**
+ * Whether this and `other` describe the same reduced key space, i.e. whether the same pair of
+ * transforms was reduced together to produce both.
+ *
+ * Two reduces that happen to land on the same space through different pairings are not
+ * recognised as one. For instance `bucket(12)` with `bucket(8)` and `bucket(12)` with
+ * `bucket(20)` both reduce onto `id % 4`. The [[Reducer]] API does not name the space it reduces
+ * onto, so the pairing is all there is to compare.
+ */
+ def hasSameReducedKeys(other: TransformExpression): Boolean =
+ reducedWith.isDefined && reducedKeySpace == other.reducedKeySpace
+
+ /** Records that this expression's keys were reduced together with `other`'s. */
+ def reducedTogetherWith(other: TransformExpression): TransformExpression =
+ copy(reducedWith = Some(other.functionId))
+
override def dataType: DataType = function.resultType()
override protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): Expression =
copy(children = newChildren)
- private lazy val resolvedFunction: Option[Expression] = this match {
- case TransformExpression(scalarFunc: ScalarFunction[_], arguments, Some(numBuckets)) =>
- Some(V2ExpressionUtils.resolveScalarFunction(scalarFunc,
- Seq(Literal(numBuckets)) ++ arguments))
- case TransformExpression(scalarFunc: ScalarFunction[_], arguments, None) =>
+ /**
+ * Builds the scalar function call this transform stands for, with the bucket count prepended as
+ * a literal argument. `None` when the bound function is not a [[ScalarFunction]], and also when a
+ * join reduced this expression's keys, since then the call no longer computes them. Evaluating it
+ * to place a row would send the row to a partition it does not belong to. That second arm is a
+ * local gate. No caller reaches it today, because every consumer of a reduced partitioning
+ * refuses it first, and the write path never sees one.
+ *
+ * A fresh expression per call. The result can be stateful ([[ApplyFunctionExpression]]), so a
+ * caller that puts it in a plan must not share one instance across two positions.
+ */
+ def resolveFunctionCall(): Option[Expression] = function match {
+ case scalarFunc: ScalarFunction[_] if reducedWith.isEmpty =>
+ val arguments = numBucketsOpt.fold(children)(n => Literal(n) +: children)
Some(V2ExpressionUtils.resolveScalarFunction(scalarFunc, arguments))
case _ => None
}
- override def eval(input: InternalRow): Any = {
- resolvedFunction match {
- case Some(fn) => fn.eval(input)
- case None => throw QueryExecutionErrors.cannotEvaluateExpressionError(this)
- }
+ /**
+ * Memoised for `eval`, which runs per row. Safe to reuse only because it stays inside this
+ * expression, unlike the call `resolveFunctionCall` hands to a caller that plans with it.
+ */
+ private lazy val evaluableFunctionCall: Option[Expression] = resolveFunctionCall()
Review Comment:
nit: I think the fresh-per-call `resolveFunctionCall()` plus this memoised twin can collapse back into the single `lazy val resolvedFunction` that was here before, just un-`private`d and with the `reducedWith.isEmpty` guard. The sharing hazard the two comments describe does not arise: the only external caller, `DistributionAndOrderingUtils.resolveTransformExpression`, visits each node once inside `expr.transform`, and every `TransformExpression` in a write distribution/ordering is a distinct instance built by `V2ExpressionUtils.toCatalyst` (distribution and ordering are converted by separate calls). Even if one instance did land in two positions, `QueryExecution.cloneWithFreshStatefulExpressions` and `ExpressionsEvaluator.prepareExpressions` already fresh-copy stateful expressions per node and per evaluator. One member and no caveat comments would do the same job.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org