metanil commented on code in PR #59760:
URL: https://github.com/apache/spark/pull/59760#discussion_r4458831331
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -2570,24 +2606,38 @@ 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. A join's keys share their types, so that
answers for what is built
+ // too. A cogroup's need not (SPARK-60782), and falling back to a hash
shuffle would pair its
+ // groups wrongly, so such a child fails here, while planning.
+ assert(newExpressions.forall {
Review Comment:
Good point, a user's query shouldn't read as a Spark bug. I split the two
cases:
- the assert stays for a caller that shuffles onto a layout without asking
`canEvaluatePartitioning`, which would be our bug;
- a key the layout's function doesn't take now throws a new
`STORAGE_PARTITION_JOIN_KEY_TYPE_MISMATCH` (sqlState 42K09, like
`STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES`). It names the transform,
its input types and the key's type, and suggests casting the keys to one type.
Your collation case is pinned too: the `EnsureRequirementsSuite` test now
runs both a `BIGINT` key and a `STRING` `COLLATE` `UTF8_LCASE` key with
`checkError`. The description now lists the collation case separately, since
master returned wrong groups there rather than failing. Both come from the same
root cause as SPARK-60782 (Pandas and Arrow cogroups don't coerce their keys),
so I'll add a note about collations there.
--
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]
