peter-toth commented on code in PR #59863:
URL: https://github.com/apache/spark/pull/59863#discussion_r4411054861
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -924,15 +926,27 @@ case class EnsureRequirements(
// them. Each side stamps why it is left ungrouped (`UngroupingOrigin`),
so a reader of the
// finished plan judges the pair on the layouts it holds; the stamp also
decides the node's
// routing (`GroupPartitionsExec.distributePartitions`).
- (
- GroupPartitionsExec(rawLeft, leftSpec.joinKeyPositions,
- Some(mergedPartitionKeys), leftReducers,
+ def regroupOrIdentity(
+ child: SparkPlan,
+ spec: KeyedShuffleSpec,
+ reducers: Option[Seq[Option[KeyReducer]]],
+ replicates: Boolean): SparkPlan = {
+ val gpe = GroupPartitionsExec(child, spec.joinKeyPositions,
+ Some(mergedPartitionKeys), reducers,
ungroupingOrigin = Option.when(applyPartialClustering)(
- if (replicateLeftSide) REPLICATED_FOR_JOIN else SPLIT_FOR_JOIN)),
- GroupPartitionsExec(rawRight, rightSpec.joinKeyPositions,
- Some(mergedPartitionKeys), rightReducers,
- ungroupingOrigin = Option.when(applyPartialClustering)(
- if (replicateRightSide) REPLICATED_FOR_JOIN else SPLIT_FOR_JOIN)))
+ if (replicates) REPLICATED_FOR_JOIN else SPLIT_FOR_JOIN))
+ // A spread or replicate pairing keeps its nodes even when the
grouping is the identity:
+ // a re-run peels a local sort only through a `GroupPartitionsExec`,
and that peel is how
+ // the statistics-based replicate-side choice is reproduced.
+ if (!applyPartialClustering &&
Review Comment:
**Finding 1.** This keeps the node only when partial clustering was
*applied*. But the replicate-side choice above runs, and reads the statistics,
in one more case: partial clustering is enabled, and the statistics pick a side
the join type cannot replicate. Lines 871-880 then decline,
`applyPartialClustering` stays false, and an identity side loses its node.
The reply in
[r4408058934](https://github.com/apache/spark/pull/59863#discussion_r4408058934)
rules this case out ("with the config on but replication not applicable there
is no choice to reproduce"). That holds only when neither side can replicate,
i.e. `FullOuter`. For `LeftOuter`, `RightOuter`, `LeftSemi`, `LeftAnti`,
`ExistenceJoin` and `LeftSingle` there is a choice. A whole-plan re-run no
longer reproduces it:
1. The re-run is `ConvertSortMergeJoinToShuffledHashJoin` after converting
another join in the plan, or a force-applied `OptimizeSkewedJoin`.
2. `keepArrivedPairing` is off under the config (line 642).
3. `peelGroupPartitions` keeps `SortExec(scan)`, because no node sits under
it (line 1069).
4. That `SortExec` has no `logicalLink`, so the choice falls back to
partition counts (lines 853-863).
5. The counts can pick the other side, which this join type does allow.
6. Partial clustering is applied after all, on the side the statistics say
is larger.
I ran the shape of the new SPARK-58996 test with AQE on, swapping `JOIN` for
`LEFT JOIN` / `LEFT SEMI JOIN`. "Re-run" means the UNION's other branch
converts to a shuffled hash join (the `100m` local-map threshold). The values
are `distributePartitions` of the storage-partitioned join's
`GroupPartitionsExec`s, left = `pc_small`, right = `pc_large`:
| join | re-run | master `8e8f37f` | this PR `5c97c1c` | with the change
below |
|---|---|---|---|---|
| `JOIN` | no / yes | `false` / `true` | `false` / `true` | `false` / `true`
|
| `LEFT JOIN` | no | `false` / `false` | `false` / no node | `false` /
`false` |
| `LEFT JOIN` | yes | `false` / `false` | **`true` / `false`** | `false` /
`false` |
| `LEFT SEMI JOIN` | yes | `false` / `false` | **`true` / `false`** |
`false` / `false` |
On the two bold rows the join also goes from 3 partitions to 5. The `LEFT
JOIN` re-run plan on this PR, trimmed:
```
SortMergeJoin [id#38L], [id#40L], LeftOuter
:- Sort [id#38L ASC NULLS FIRST], false, 0
: +- GroupPartitions ExpectedPartitionKeys: 3 DistributePartitions: true
SortedMerge: false
: +- Project [id#38L]
: +- BatchScan testcat.ns.pc_small[id#38L]
+- GroupPartitions ExpectedPartitionKeys: 3 DistributePartitions: false
SortedMerge: false
+- Sort [id#40L ASC NULLS FIRST], false, 0
+- Project [id#40L]
+- Filter isnotnull(id#40L)
+- BatchScan testcat.ns.pc_large[id#40L]
```
The first pass left `Sort(pc_large)` with no node. The re-run built the
replicating node over it.
The rows match on every arm. With AQE off there is no re-run, and this PR
plans like master.
So this is a performance regression against master, and planning is not
idempotent across the rule's re-runs. Master and the first pass both decline
partial clustering here, because the side the statistics pick cannot replicate
for the join type. The re-run instead replicates `pc_large`, the side the
statistics say is larger, and reads it once per split of the matching
`pc_small` key. The config's doc promises the statistics decide: "Spark will
pick the side with less data size based on table statistics"
(`SQLConf.scala:2663`).
Keeping the node whenever the choice ran, applied or not, fixes it:
```scala
var applyPartialClustering = false
// Whether a replicate side was chosen, from statistics or partition
counts. A re-run has to
// make the same choice even when it was then declined for the join
type.
var replicateSideChosen = false
...
} else {
replicateSideChosen = true
// The statistics and the original partition keys come from the
pre-alignment plans,
...
if (!replicateSideChosen &&
gpe.outputPartitioning == child.outputPartitioning) {
```
With that change:
- The table's last column matches master.
- `KeyGroupedPartitioningSuite` and `EnsureRequirementsSuite` pass.
- The motivating join still converts to a shuffled hash join.
- The skip still applies with partial clustering off, and for `FullOuter`
with it on.
Please also:
- Add the `LEFT JOIN` variant of the new test, expecting `false` on both
sides. It fails on this PR, where the left side distributes after the re-run,
and passes with the change.
- Update the comment above and the description's "This is the exact
condition rather than the config" paragraph.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -10759,26 +10891,19 @@ class KeyGroupedPartitioningSuite
val plan = df.queryExecution.executedPlan
assert(ValidateRequirements.validate(plan),
s"the executed plan must satisfy every operator's required
distribution:\n$plan")
- // Locate the marked side's node: no reducers, and output partition
i holds exactly input
- // partition i. That is weaker than `identityGrouping`, which also
asks for one output per
- // input, so the assertions below are what pin the claim and the
three keys.
- val identityGpe = collectAllGroupPartitions(plan).find { g =>
- g.reducers.isEmpty && g.groupedPartitions.zipWithIndex.forall {
- case ((_, inputIndices), outputIndex) =>
- inputIndices.lengthCompare(1) == 0 && inputIndices.head ==
outputIndex
- }
- }
- assert(identityGpe.isDefined,
- s"expected a reducer-free identity GroupPartitionsExec,
got:\n$plan")
- identityGpe.get.outputPartitioning match {
- case k: KeyedPartitioning =>
- assert(k.mayContainUnknownPartitionKeys,
- "the identity grouping must keep the unknown-keyed claim")
- assert(k.numPartitions === 3,
- s"the marked side keeps all three of its declared
keys:\n$plan")
- case other =>
- fail(s"expected a KeyedPartitioning output, got $other")
- }
+ // The marked side's alignment is the pass-through, so no node is
built for it: the
+ // claim has to reach the second join on the side itself. Assert on
that join, since the
+ // first join's keyed shuffle already carries a marked three-key
member of its own.
+ val secondJoin = collect(plan) { case j: SortMergeJoinExec => j
}.head
+ assert(collectAllShuffles(plan).size == 1,
+ s"only the first join shuffles:\n$plan")
+ val marked = collect(secondJoin.left) { case p: SparkPlan =>
p.outputPartitioning }
+ .flatMap(PartitioningCollection.representativeOf)
+ .filter(_.mayContainUnknownPartitionKeys)
Review Comment:
**Finding 3.** This still reads the whole subtree.
`collect(secondJoin.left)` walks down into the first join and its one-side
keyed shuffle of `pt`. The comment just above says that shuffle "already
carries a marked three-key member of its own", so
`marked.exists(_.numPartitions == 3)` holds whatever the second join's left
input reports. I listed the nodes it accepts here: `InputAdapter,
WholeStageCodegen (4), Project, SortMergeJoin, InputAdapter, WholeStageCodegen
(3), Sort, InputAdapter, Exchange`. That is the gap from
[r4404496280](https://github.com/apache/spark/pull/59863#discussion_r4404496280).
The rework in `94d5b1a` moved the starting point to the second join but kept
the walk. Reading the input's own partitioning closes it:
```suggestion
val marked =
PartitioningCollection.representativeOf(secondJoin.left.outputPartitioning)
.filter(_.mayContainUnknownPartitionKeys)
```
`marked` becomes an `Option`, so the `exists` below it is unchanged. With it
the test passes on this PR, on both partition-filter arms.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1158,6 +1164,55 @@ class KeyGroupedPartitioningSuite
testWithCustomersAndOrders(customers_partitions, Array.empty, 2, 0)
}
+ test("SPARK-60618: Skip the pass-through GroupPartitionsExec in a
storage-partitioned " +
+ "join's alignment") {
+ withSQLConf(
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ // `items` reports more partition keys than the join uses, so the
aligned layout has to be
+ // projected onto the join keys. The shuffled side is laid out on that
same projection, so
+ // its `GroupPartitionsExec` would only re-state the keys its shuffle
already has and must
+ // not be built.
+ SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key ->
"true") {
+ withTable(s"testcat.ns.$items", s"testcat.ns.$purchases") {
+ createTable(items, itemsColumns, Array(identity("id"),
identity("name")))
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ s"(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ s"(1, 'aa', 30.0, cast('2020-01-02' as timestamp)), " +
+ s"(3, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ s"(4, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+ createTable(purchases, purchasesColumns, Array.empty)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ s"(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ s"(1, 89.0, cast('2020-01-03' as timestamp)), " +
+ s"(3, 19.5, cast('2020-02-01' as timestamp)), " +
+ s"(5, 26.0, cast('2023-01-01' as timestamp)), " +
+ s"(6, 50.0, cast('2023-02-01' as timestamp))")
+
+ val df = sql(s"SELECT i.id, i.name, i.price AS purchase_price, p.price
AS sale_price " +
+ s"FROM testcat.ns.$items i JOIN testcat.ns.$purchases p ON i.id =
p.item_id")
+ checkAnswer(df, Seq(
+ Row(1, "aa", 30.0, 42.0), Row(1, "aa", 30.0, 89.0),
+ Row(1, "aa", 40.0, 42.0), Row(1, "aa", 40.0, 89.0),
+ Row(3, "bb", 10.0, 19.5)))
+
+ val plan = df.queryExecution.executedPlan
+ // The one-side shuffle still happens; only the regrouping node over
it goes away.
+ assert(collectAllShuffles(plan).size == 1, s"expected one
shuffle:\n$plan")
+ assert(collect(plan) {
+ case g: GroupPartitionsExec
+ if g.child.isInstanceOf[ShuffleQueryStageExec] ||
+ g.child.isInstanceOf[AQEShuffleReadExec] => g
+ }.isEmpty, s"no GroupPartitionsExec should sit over a shuffle
stage:\n$plan")
+ // The source side keeps its node: `items` holds two splits for key
(1, 'aa'), and merging
+ // them is work the join needs.
+ assert(collect(plan) { case _: GroupPartitionsExec => true }.nonEmpty,
+ s"the source side should keep its grouping node:\n$plan")
Review Comment:
**Finding 4.** The test pins that no node sits over the stage. It does not
pin what the description says the change is for: the join converting to a
shuffled hash join. With
`spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold` at its default of 0,
no conversion is attempted.
I ran this test's tables and query with that threshold at `100m`. Master
plans one `SortMergeJoinExec` and two `GroupPartitionsExec`. This PR plans one
`ShuffledHashJoinExec` and one `GroupPartitionsExec`. So asserting the
conversion pins the benefit, and it fails on master. `LEFT OUTER`, `RIGHT
OUTER`, `FULL OUTER`, `LEFT SEMI` and `LEFT ANTI` convert too, with the same
rows as a run without SPJ.
Two smaller points in the same assertions:
- The source-side check matches a node anywhere in the plan.
- The `ShuffleQueryStageExec` / `AQEShuffleReadExec` pattern only exists
under AQE, which the test leaves at the session default.
Suggest adding `SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true"` and
`SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key -> "100m"` to
the `withSQLConf` above, and:
```suggestion
// Without the node over the stage the join converts to a shuffled
hash join.
val shj = collect(plan) { case j: ShuffledHashJoinExec => j }
assert(shj.size == 1 && collect(plan) { case j: SortMergeJoinExec =>
j }.isEmpty,
s"the join should convert to a shuffled hash join:\n$plan")
// The source side keeps its node: `items` holds two splits for key
(1, 'aa'), and merging
// them is work the join needs.
assert(collect(shj.head.left) { case _: GroupPartitionsExec => true
}.nonEmpty,
s"the source side should keep its grouping node:\n$plan")
```
With these changes the test passes on this PR.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -5416,6 +5477,67 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-58996: partially clustered join keeps its replicate-side choice
when the " +
+ "alignment is the identity") {
+ // The smaller side by statistics is replicated: `pc_small` holds five
splits in 200 bytes,
+ // `pc_large` four splits in 224. Both are bucketed, whose transform
derives no ordering, so
+ // the join's sort has to be added. The larger side ends up with one split
per bucket, so the
+ // alignment on it is the identity and the pass-through node has to stay:
a re-run peels a
+ // local sort only through a `GroupPartitionsExec`, and reading the
statistics from the scans
+ // again is what keeps the choice. Without the node the counts (five
against four) pick the
+ // other side.
Review Comment:
**Finding 6.** The numbers here don't match the data. `pc_large` gets three
rows, so it holds three splits, and `InMemoryBaseTable.estimateStatistics`
(12-byte header + 60-byte row) gives 216 bytes, not "four splits in 224". The
counts that would decide without the node are five against three, as the PR
description says:
```suggestion
// The smaller side by statistics is replicated: `pc_small` holds five
splits in 200 bytes,
// `pc_large` three splits in 216. Both are bucketed, whose transform
derives no ordering, so
// the join's sort has to be added. The larger side ends up with one
split per bucket, so the
// alignment on it is the identity and the pass-through node has to
stay: a re-run peels a
// local sort only through a `GroupPartitionsExec`, and reading the
statistics from the scans
// again is what keeps the choice. Without the node the counts (five
against three) pick the
// other side.
```
The test pins this PR's gate, so `SPARK-60618:` would be the usual prefix in
its name.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala:
##########
@@ -1757,7 +1753,9 @@ class EnsureRequirementsSuite extends SharedSparkSession {
// The fine pair's members are each their child's own layout, so no
projection is reported and
// the nodes only align to the merged keys. Taking the first agreeing
pair instead groups both
// sides on `a` alone, i.e. positions `Some(Seq(0))`.
- assert(groupPartitionsNodes(planned).map(_.joinKeyPositions) ===
Seq(None, None),
+ // A side whose alignment is the pass-through keeps its own layout and
gets no node, so the
+ // pair shows up as unprojected nodes only.
+ assert(groupPartitionsNodes(planned).forall(_.joinKeyPositions.isEmpty),
Review Comment:
**Finding 5.** `forall` also holds for an empty list, so this no longer
requires the push branch to build anything. Nothing else in this test pins it:
the only other assertion is that nothing was shuffled. The pass-through side
legitimately has no node, but the other side always has one here, so the exact
shape can be asserted:
```suggestion
assert(groupPartitionsNodes(planned).map(_.joinKeyPositions) ===
Seq(None),
```
The same relaxation is in two more places:
- `EnsureRequirementsSuite.scala:2967` uses `forall(_ == Some(3))`. The
partition-count assertion right after it still pins the outcome, but `===
Seq(Some(3))` restores the old strength.
- The SPARK-41413 mismatch test reads through `plannedPartitioning` (line
1071), which falls back to the dummy plan's own partitioning. So all three of
its cases pass with no node on either side, while the removed pattern required
one. `assert(groupPartitionsNodes(smj).size == 1)` in each `case smj:
SortMergeJoinExec =>` keeps the requirement that the pruning side is regrouped.
All three replacements pass on this PR.
--
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]
