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: [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]
