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