peter-toth commented on code in PR #58552:
URL: https://github.com/apache/spark/pull/58552#discussion_r3976730655
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -719,16 +755,18 @@ case class KeyedPartitioning(
}
copy(
expressions = positions.map(expressions),
- partitionKeys = projectedKeys,
- isGrouped = !collapses && sourceOf.size == projectedKeys.length,
- isCollapsed = isCollapsed || collapses)
+ layout = KeyLayout(
Review Comment:
You are right, and it was an unintended behaviour change rather than a
judgement: `copy(...)` carried the marker, a fresh `KeyLayout` does not. Fixed
in
[`d839aee`](https://github.com/apache/spark/commit/d839aeeb3146f07d69b4cebda2b7ed9efdb2d9ff),
so the contract holds whatever positions a caller asks for and the
`createShuffleSpec` comment is true again.
Your reading of the reachability matches mine, and I would rather not have
this method depend on it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -572,58 +642,37 @@ case class CoalescedNullAwareHashPartitioning(
* }}}
*
* @param expressions Partition transform expressions (e.g., `years(col)`,
`bucket(10, col)`).
- * @param partitionKeys Partition keys wrapped in InternalRowComparableWrapper
for efficient
- * comparison and grouping. One per partition. Typically
in sorted order when
- * produced by a data source or `GroupPartitionsExec`,
but this is not
- * guaranteed after projection. May contain duplicates
when ungrouped.
- * @param isGrouped Whether partition keys are unique (no duplicates).
Computed on first
- * creation, then preserved through copy operations to avoid
recomputation.
- * @param isCollapsed Whether a projection or a reduction mapped keys that
were distinct in the
- * partitioning this one was derived from onto the same
key, so one key here can
- * stand for several of the original ones. Sticky. See "Key
Collapse" above for
- * what it gates and how it travels.
- * @param mayContainUnknownPartitionKeys Whether the data may contain rows
whose partition key is
- * not among the declared `partitionKeys`.
`KeyGroupedPartitioner`
- * routes such rows by a deterministic hash
when a side is
- * re-shuffled onto this partitioning (see
- * `KeyedShuffleSpec.createPartitioning`), so
co-location holds
- * for whole keys only: two marked
partitionings declaring the
- * same keys in the same order and using the
same partition
- * function per position still pair (equal
undeclared keys hash
- * to the same partition), but a row of an
undeclared key sits in
- * the partition of some other declared key
and need not be
- * co-located with rows sharing only a subset
of its columns.
- * `satisfies` and
`KeyedShuffleSpec.areKeysCompatible` therefore
- * accept a marked partitioning only for
full-key clustering,
- * never for a subset of its partition columns
and never for a
- * global ordering across several partitions.
Two carry rules:
- * (1) a node that changes the declared key
set must drop the
- * keyed partitioning, whether it coarsens it
(a key-dropping
- * projection, a key-changing reduction, a
join-key projection)
- * or expands it over a marked leg (a union,
where another leg
- * may declare exactly the key that leg holds
out-of-set);
- * (2) marker agreement is enforced by
`PartitioningCollection`:
- * the constructor requires it and
`fromPartitionings` normalizes
- * by OR, so members are uniformly marked or
unmarked and
- * consumers may read one member.
`ShuffledJoin`'s `InnerLike`
- * arm, the only site that meets a marked
input with an unmarked
- * one, clears the markers first. That is
precision, not the
- * guarantee: without it the OR would spread
the spurious marker
- * onto the accurate side.
+ * @param layout The partitions this one describes, which is everything about
them except the
+ * expressions naming them. See [[KeyLayout]].
*/
case class KeyedPartitioning(
expressions: Seq[Expression],
- @transient partitionKeys: Seq[InternalRowComparableWrapper],
- isGrouped: Boolean,
- isCollapsed: Boolean,
- mayContainUnknownPartitionKeys: Boolean = false)
- extends Expression with Partitioning with Unevaluable {
- override val numPartitions = partitionKeys.length
+ layout: KeyLayout) extends Expression with Partitioning with Unevaluable {
+ override val numPartitions = layout.partitionKeys.length
+
+ def partitionKeys: Seq[InternalRowComparableWrapper] = layout.partitionKeys
+ def isGrouped: Boolean = layout.isGrouped
+ def isCollapsed: Boolean = layout.isCollapsed
+ def mayContainUnknownPartitionKeys: Boolean =
layout.mayContainUnknownPartitionKeys
+
+ /** This partitioning over a changed layout, e.g.
`withLayout(_.copy(isGrouped = false))`. */
+ def withLayout(f: KeyLayout => KeyLayout): KeyedPartitioning = copy(layout =
f(layout))
override def children: Seq[Expression] = expressions
override def nullable: Boolean = false
override def dataType: DataType = IntegerType
+ /**
+ * Prints the layout's contents where the value object would print, which
keeps the plan string as
+ * it was before the layout held them. The list is curated rather than the
layout's own fields,
+ * for two reasons. `partitionKeys` has to be printed as a `Seq` for
`maxFields` to truncate it,
+ * and a partitioning can hold one key per split. And `dataTypes` has its
naming erased, so
+ * printing it would put a struct field name into a plan that appears
nowhere in the query. A
+ * field the layout grows is therefore a decision here.
+ */
+ override protected def stringArgs: Iterator[Any] =
Review Comment:
Also right, and the description was wrong to claim no plan string change.
Fixed in
[`d839aee`](https://github.com/apache/spark/commit/d839aeeb3146f07d69b4cebda2b7ed9efdb2d9ff)
by appending the marker, so the string is what it was before this PR and the
description is accurate as written.
I took the unconditional print rather than the `TransformExpression` shape.
Printing only when true keeps the marked case visible, but it changes the
string for every unmarked partitioning, which is the opposite of what the
description promises. The scaladoc now says that printing is the default for a
field the layout grows.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -2598,6 +2598,89 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59285: two legs whose struct field names differ are still
co-partitioned") {
+ // Both legs prune to nothing, and their key spaces differ only in a
struct field name, which
+ // `identity` carries into the key type. A reduce cannot bridge that,
since there is no reducer
+ // between two attributes, so calling the two sides incompatible leaves
nowhere to go: the join
+ // must keep taking them as one layout.
+ withTable("p1", "p2", "p3", "p4") {
+ createTable("p1", Array(Column.create("id", structA)),
Array(identity("id")))
+ sql("INSERT INTO testcat.ns.p1 VALUES (named_struct('a', 1))")
+ createTable("p2", Array(Column.create("id", structA)),
Array(identity("id")))
+ sql("INSERT INTO testcat.ns.p2 VALUES (named_struct('a', 2))")
+ createTable("p3", Array(Column.create("k", structB)),
Array(identity("k")))
+ sql("INSERT INTO testcat.ns.p3 VALUES (named_struct('b', 1))")
+ createTable("p4", Array(Column.create("k", structB)),
Array(identity("k")))
+ sql("INSERT INTO testcat.ns.p4 VALUES (named_struct('b', 2))")
+
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> "true") {
+ val df = sql(
+ """SELECT leg1.id, leg2.k FROM
+ | (SELECT p1.id AS id FROM testcat.ns.p1 JOIN testcat.ns.p2 ON
p1.id = p2.id) leg1
+ | JOIN
+ | (SELECT p3.k AS k FROM testcat.ns.p3 JOIN testcat.ns.p4 ON p3.k
= p4.k) leg2
+ | ON leg1.id = leg2.k
+ |""".stripMargin)
+ assert(collectShuffles(df.queryExecution.executedPlan).isEmpty,
+ "the two legs are taken as one layout, so neither is shuffled")
+ checkAnswer(df, Nil)
+ }
+ }
+ }
+
+ test("SPARK-59285: two sides whose partitions were all pruned are not one
layout") {
Review Comment:
Taken, in
[`d839aee`](https://github.com/apache/spark/commit/d839aeeb3146f07d69b4cebda2b7ed9efdb2d9ff).
Two asserts that are not on absence: `collectAllShuffles(plan).isEmpty`, which
holds as you expected, and a key-space pin.
The pin is on the topmost node that reports a keyed partitioning, not on the
whole plan. I tried the whole plan first and measured why it cannot work: the
identity legs below still report their own space, so the plan holds
`ArraySeq(IntegerType)` and `ArraySeq(LongType)`, one per node. Asking the
topmost node says the two legs agreed on the bucket space, which is the claim.
Together they close the vacuity you describe. The first fails if the plan
falls back to shuffles, the second if there is no keyed partitioning left to
report.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1687,6 +1744,13 @@ case class KeyedShuffleSpec(
case otherSpec @ KeyedShuffleSpec(otherPartitioning, otherDistribution, _)
=>
distribution.clustering.length == otherDistribution.clustering.length &&
numPartitions == other.numPartitions && areKeysCompatible(otherSpec) &&
+ // The key rows are compared at their types, since
`InternalRowComparableWrapper.equals`
+ // compares those first. Two empty key lists compare equal whatever
they describe, so the
+ // key space is asked separately: without that, a join between two
sides whose partitions
+ // were all pruned would call two different spaces one layout, and
+ // `ShuffledJoin.outputPartitioning` would then report both as
alternative descriptions of
+ // it. Where a key row exists this clause is implied.
+ partitioning.keyDataTypes == otherPartitioning.keyDataTypes &&
Review Comment:
Documented in
[`d839aee`](https://github.com/apache/spark/commit/d839aeeb3146f07d69b4cebda2b7ed9efdb2d9ff),
with your argument: an empty layout holds no row, so every claim over it is
vacuous, and a merge that later brings real rows under it rewrites each
member's expressions through `reducersBothWays` or a `GroupPartitionsExec`
first.
I left the same-function-per-position requirement out. It would have the
gate compare transforms rather than types, which is wider than this PR, and the
pair it would refuse cannot produce a wrong answer. Thank you for the probe.
--
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]
