metanil commented on code in PR #59760:
URL: https://github.com/apache/spark/pull/59760#discussion_r4463933090
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -2570,24 +2609,44 @@ case class KeyedShuffleSpec(
// `createPartitioning` replaces a transform's argument with the other
child's cluster key,
// which drops whatever surrounds the key, e.g. the `+ 1` of
`bucket(4, b + 1)` or the `.a`
// of `bucket(4, s.a)`. Such a spec's own child would then be laid out
differently from the
- // children shuffled onto it.
+ // children shuffled onto it. Whether Spark can evaluate it is
`canEvaluatePartitioning`.
case t: TransformExpression => argumentsAreColumns(t)
case _ => false
} &&
// Shuffling another child onto these keys evaluates the partition
expressions per row to
// decide where each row goes, and reduced keys are not what those
expressions compute.
partitioning.expressionsDescribeKeys
+ /** Whether Spark can evaluate each transform of this layout, see
`canEvaluate`. */
+ override lazy val canEvaluatePartitioning: Boolean =
partitioning.expressions.forall {
+ case t: TransformExpression => canEvaluate(t)
+ case _ => true
+ }
+
override def createPartitioning(clustering: Seq[Expression]): Partitioning =
{
assert(clustering.size == distribution.clustering.size,
"Required distributions of join legs should be the same size.")
-
val newExpressions = partitioning.expressions.zip(keyPositions).map {
case (te: TransformExpression, positionSet) =>
// `canCreatePartitioning` admits only a template over bare columns,
see there.
rebuiltOver(te, clustering(positionSet.head))
case (_, positionSet) => clustering(positionSet.head)
}
+ // The child shuffled onto this layout is placed by evaluating what is
built here, which would
+ // fail on an executor. A caller has to ask `canEvaluatePartitioning`
first, which answers for
+ // this layout's own keys, so a caller that does not is a bug.
+ assert(canEvaluatePartitioning,
+ s"Cannot shuffle a child onto a layout Spark cannot evaluate:
${partitioning.expressions}")
+ // A join's keys share their types, so that answers for what is built too.
A cogroup's need not
+ // (SPARK-60782), e.g. a `BIGINT` key or a string of another collation,
and falling back to a
+ // hash shuffle would pair its groups wrongly. So such a child fails here,
while planning, with
+ // an error that names the key's type.
+ newExpressions.foreach {
+ case t: TransformExpression if t.evaluableFunction.isEmpty =>
+ throw QueryExecutionErrors.storagePartitionJoinKeyTypeMismatchError(
+ t.function.name(), t.function.inputTypes().toSeq,
t.children.head.dataType)
Review Comment:
Thanks @dongjoon-hyun Yes, agree, I can do the same fallback making `orElse
private[sql]`
--
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]
