peter-toth commented on PR #58659: URL: https://github.com/apache/spark/pull/58659#issuecomment-5700453515
Rebased onto master, and the review round is answered. @dongjoon-hyun @ulysses-you, thank you both. SPARK-59272 is out of this PR. It went in on its own as #58814, so this is SPARK-59289 alone, and that is the ticket the release note should hang off. @ulysses-you, you were right to ask for the two to be kept apart. ## @dongjoon-hyun **1. `withNewChildInternal` keeps a stale `grouping` / `outputPartitioning`.** Real, and fixed, though not the way you suggested. The node now carries `childPartitioning`, the child's partitioning it was decided over, and reports `plannedPartitioning` only while the child still reports that. Otherwise it reports `UnknownPartitioning`, which is the give-up it already performs for a marked layout it may not regroup. Re-deriving would be worse. The recipe was chosen for a pairing, so applying it to a different child yields a claim nothing validated against the other side, and a node cannot ask the other side. It also buys nothing today: no rule in the tree is known to report a *different* `KeyedPartitioning`, so the two reachable outcomes are an equal one, where the check passes, and `UnknownPartitioning`, where re-deriving reports exactly what this reports. The check is on the read rather than in `withNewChildInternal`, because canonicalization rebuilds this node over a canonicalized child and a canonicalized `BatchScanExec` throws from `reportedKeyedPartitioning`. Your second half, an AQE test, is in `KeyGroupedPartitioningSuite`, and it is honest about what it does not do. It runs the shape under AQE with every shuffle read rule on, but I could not get a rewrite applied to the keyed shuffle stage, and the reason looks structural: `CoalesceShufflePartitions` coalesces only a group whose leaves are all query stages, and `OptimizeShuffleWithLocalRead` needs the stage to be a broadcast join's own probe side. A storage-partitioned join always keeps one side as a scan, which is neither, and the shuffle-join-to-broadcast flip that would leave the stage as a probe side needs runtime statistics a scan never produces. Six plan shapes probed. So that test is coverage: it still passes with the gate removed, and its comment says so. The gate is pinned by `GroupPartitionsExecSuite`'s unit test, which does fail without it. **2. `satisfies` can now project the partition keys on every call.** You were right, and my first reading of your comment was wrong, so let me correct it: this is not a new question, it is the base's own check relocated. `resolveKeyedPartitioning` asked it behind a `mutable.Map` memo and behind a cheap full-coverage form tried across every member first. Moving it into `satisfies` lost both, and `satisfies` is asked by `ValidateRequirements` and every `AQEShuffleReadRule` application as well. Measured on a 307200-plan sweep: `numPartitionsProjectedOn` fires 0 times on master and 68568 times without a memo. Every one of those needs `allowJoinKeysSubsetOfPartitionKeys`, so the default configuration pays nothing. The answer is now memoized on the partitioning, by position set, through `TransientBestEffortLazyVal`, which takes the 68568 to 24. I also considered putting the check back in the rule, which would take it to 0 on that sweep. I did not, because the ordering guard that gets it to 0 only pays when a sibling member of the same collection has full coverage, and restoring it means the two-step find the single decision point just replaced. 24 projections in 307200 plans did not seem worth that. **3. The assert in `withJoinKeyPositions`.** Taken. The rebuild's condition is its match guard now, and an aligned node falls through to the wrapping case the helper already had, which is your suggestion. A planner crash was the wrong answer where a sound plan exists. A re-run is the path you named; the sweep applies the rule twice to every plan it builds and the assert never fired in 277844 re-applications, so the test sits at the helper. **4. `outputPartitioning` is the only non-`@transient` derived field.** It is `@transient` now, with everything else. What replaced it is two non-transient fields, `plannedPartitioning` and `childPartitioning`, and the class doc says why: they are what the node claims rather than how it arrived at the claim, which is how `ShuffleExchangeExec` holds its partitioning. ## @ulysses-you **`GroupPartitionsExec.scala:75`, the carried fields.** Same as dongjoon's first item, answered above. Your probe result matches what I measured. **`partitioning.scala:1357`, the filter feeding `ValidateRequirements`.** Taken exactly as you wrote it. `PartitioningCollection.createShuffleSpec` now filters on `maySatisfyAfterProjection`, which for a keyed member is `isGrouped && the required count matches && keysMaySatisfy`, so the admission set is the one `satisfies` gave before it became strict, and the new `ShuffleSpecSuite` case still has its fallback. There is a second test beside it for the member you ruled out, the ungrouped one, so the narrowing is pinned in both directions. One correction to my own comment there, found on a later pass: the filter is not *exactly* the old admission set. `isFunctionOfClusterKeys` accepts a partition expression that *is* a cluster key, which the reference-level test did not, so a `years(ts)` under a clustering naming `years(ts)` is newly kept. Nothing can co-partition on it, since its `keyPositions` entry is empty, so what it reaches is the `canCreatePartitioning` `forall` the comment's next paragraph already calls unreachable. The comment says that now instead of claiming parity. **`EnsureRequirementsSuite.scala:2686`, the idempotency ask.** Done, in `KeyGroupedPartitioningSuite`: six real SPJ queries, one per way the rule can plan one, including a partially clustered one so the re-plan path runs. Each asserts `apply(apply(p)) == apply(p)` and pins the shape it exercises, since idempotency over a plan the rule left alone proves nothing. Worth reporting: only the sixth needs `keepArrivedPairing`. With that term disabled the other five still pass. **`EnsureRequirements.scala:713`, the log inside `pushed`.** Moved after the gate, and it gained a counterpart, so a reader is told which way it went rather than only that a pushdown was attempted. **`EnsureRequirements.scala:1027`, `positions.map(_)`.** It reads `positions.map(old)` now, where `old` is the node's own positions bound by the `fold`. Same composition, named. **`GroupPartitionsExec.scala:408`, the `expectedPartitionKeys` doc.** Fixed, it says `expectedKeyCount`. **The sweep harness.** Attached, in https://github.com/apache/spark/pull/58814#issuecomment-5679504983. It is the same script, with the re-apply check added for the idempotency numbers above. **The scope note.** Taken, mostly. The `matchesClusterKeys` consolidation is reverted and filed for its own ticket; it was most of `partitioning.scala`'s diff and it dropped `AQEUtils` from this PR entirely. Only `isClusterKey` and `allClusterKeysAmong` remain, because the new code calls them. I kept the "operation key" to "cluster key" rename: it is 28 lines across four files, it makes the prose agree with `ClusteredDistribution.clustering`, and this change is master-only, so no cherry-pick pays for it. Would you still rather have it on its own? -- 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]
