peter-toth commented on code in PR #58279:
URL: https://github.com/apache/spark/pull/58279#discussion_r3931930149
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -634,11 +642,16 @@ case class EnsureRequirements(
|""".stripMargin)
leftLink.get.stats.sizeInBytes < rightLink.get.stats.sizeInBytes
} else {
- // As a simple heuristic, we pick the side with fewer number of
partitions
- // to apply the grouping & replication of partitions
+ // As a simple heuristic, we pick the side with fewer number of
partitions to
+ // apply the grouping & replication of partitions. The counts
read the
+ // pre-alignment plans, for the same reason the statistics do:
on a re-run both
+ // aligned reports hold the same number of keys, so comparing
them decides nothing.
logInfo("Using number of partitions to determine which side of
join " +
"to fully cluster partition values")
- leftPartKeys.size < rightPartKeys.size
+
PartitioningCollection.numKeyedPartitions(unwrappedLeft.outputPartitioning)
Review Comment:
**Finding 12.** This implements [finding
10](https://github.com/apache/spark/pull/58279#discussion_r3924540492) and the
invariant it asked for holds now: the fallback reads one number per side, from
the pre-alignment plan, on every pass. It reaches that by changing what a first
pass compares, and that half is not stated anywhere.
Base's number is `leftSpec`/`rightSpec`'s key count, i.e. the *aligned*
report's. Finding 10 was about what that is on a re-run: equal on both sides,
so the comparison decides nothing. On a first pass it is a third thing. The
children loop has already wrapped a non-grouped child, so the report holds one
key per distinct value and base compared **distinct keys**. This head compares
**raw splits**, and the two differ for exactly the shape partial clustering is
about, a side with more than one split per key.
Measured on a bare first pass -- two `DummySparkPlan`s under a
`SortMergeJoinExec`, no `logicalLink`, so the fallback fires -- reading
`(left.distributePartitions, right.distributePartitions)` back off the planned
children:
| left | right | `0875765` | this head |
|---|---|---|---|
| 3 splits, 1 distinct key | 2 splits, 2 keys | `(false, true)` | `(true,
false)` |
| 4 splits, 2 distinct keys | 3 splits, 3 keys | `(false, true)` | `(true,
false)` |
| 2 splits, 2 keys | 3 splits, 3 keys | `(false, true)` | `(false, true)` |
I am not asking for the old number back. Splits are the better proxy, and
your new unit test's first arm is the case that shows it: one split against
three, and the base picks the three-split side to replicate. The two coherent
options are this head and reverting finding 10, and this head is the right one.
What is missing is one sentence in each of two places. The comment says "on
a re-run both aligned reports hold the same number of keys, so comparing them
decides nothing", which reads as if a first pass is untouched, and "Does this
PR introduce any user-facing change?" lists only the duplicated rows. This is a
wrong-results fix headed for `branch-4.2` and `branch-4.3`, so a plan change
riding along in a maintenance backport is worth naming:
```scala
// As a simple heuristic, we pick the side with fewer number
of partitions to
// apply the grouping & replication of partitions. The counts
read the
// pre-alignment plans, for the same reason the statistics do:
on a re-run both
// aligned reports hold the same number of keys, so comparing
them decides nothing.
// This also changes a first pass, which compared the aligned
report's distinct
// keys rather than the splits behind them.
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1133,6 +1133,14 @@ object PartitioningCollection {
case _ => None
}
+ /**
+ * The number of partitions of the [[KeyedPartitioning]] representing the
keyed members of
+ * `partitioning`, if any. Collections validate on construction that their
keyed members agree,
+ * so the representative's count stands for all of them.
+ */
+ def numKeyedPartitions(partitioning: Partitioning): Option[Int] =
Review Comment:
**Finding 14.** This is public, and its one caller is `EnsureRequirements`,
so `private[sql]` covers it. Worth narrowing because the method is a one-line
wrapper over `representativeOf`, which is `private[physical]` deliberately: as
written, the count of a keyed member becomes reachable from anywhere on the
classpath in one hop.
```scala
private[sql] def numKeyedPartitions(partitioning: Partitioning):
Option[Int] =
```
Separately, the scaladoc understates what holds. `require` at `:1049`
already forces *every* member to agree on `numPartitions`, keyed or not, so the
answer is `partitioning.numPartitions` whenever there is a keyed member at all
-- the representative only decides whether there is one. Saying that is shorter
than the agreement argument and it is the stronger fact.
--
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]
