dongjoon-hyun commented on PR #58262: URL: https://github.com/apache/spark/pull/58262#issuecomment-5414478698
I ran a deep review of this PR (line-by-line scan, removed-behavior audit, cross-file tracing, plus reuse/simplification/efficiency angles, with each candidate adversarially verified against the checked-out tree). Overall the change looks solid: I verified that the old four-way classification maps faithfully onto the new two-bucket one (`nonGroupedSatisfies`, `AllTuples`, `UnspecifiedDistribution`, `OrderedDistribution`, and `requiredNumPartitions` paths all produce identical outcomes), that the inline and multi-child projection paths are mutually exclusive (no double projection), that re-applying the rule over an inserted `GroupPartitionsExec` is idempotent under AQE, and that using `needsGrouping.head` is safe given `PartitioningCollection`'s shared-`partitionKeys` invariant. The findings below are ordered by severity; the first three are correctness/design, the rest are cleanup. **1. Pre-existing crash carried into the rewritten function: `sliding(2)` MatchError for a single-partition `KeyedPartitioning` under `OrderedDistribution`** (confirmed) https://github.com/apache/spark/blob/8b74639cc949ab60fdef654e1318b6bc6f88dc7d/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala#L100-L102 `sliding(2)` on a 1-element `partitionKeys` yields a single window of size 1, which `case Seq(k1, k2)` cannot match, so planning throws `scala.MatchError`. Reachable: a v2 table partitioned by `identity(id)` whose rows all share one partition value reports a one-key `KeyedPartitioning`; with `spark.sql.sources.v2.bucketing.sorting.enabled=true`, `SELECT id FROM t ORDER BY id` (global `SortExec` requires `OrderedDistribution`, admitted via `areAllClusterKeysMatched`) reaches this branch. Not a regression of this PR — the line is carried over verbatim — but since the branch is being rewritten anyway, `case Seq(k1, k2, _*)` (or a size guard) plus a small test would close it here. No existing sorting test covers a single-partition table. **2. The co-partitioned `PartitioningCollection` gap is documented but remains a live wrong-results hole** (plausible, pre-existing) The new comment at L74-L76 accurately states that a co-partitioned child reporting a `PartitioningCollection` gets a projection from neither the inline path (suppressed by `isCoPartitioned`) nor the multi-child block (`withJoinKeyPositions` never matches a `ShuffleSpecCollection` best spec). The consequence is the same wrong-results class this PR fixes, just behind a rarer plan shape: e.g. `FlatMapCoGroupsInPandasExec` on key `i` over children reporting collections of KPs grouped on `(n, i)` — compatibility is judged on the projected specs while the children run unprojected. Since it's the identical bug class, it may deserve a JIRA now rather than an open-ended follow-up note. **3. `KeyedPartitioning.satisfies` still over-claims, so `ValidateRequirements` certifies the exact plans this PR deems wrong** (plausible, deferred by design) The compensation (`positions.isEmpty`) lives only inside `EnsureRequirements`; `ValidateRequirements.validateInternal` still goes through `satisfies`, and it is the safety gate in `AdaptiveSparkPlanExec.optimizeQueryStage` and `OptimizeSkewedJoin`. The pre-fix wrong plan (grouped `KP([id, name])` under `ClusteredDistribution([id])`, no `GroupPartitionsExec`) passes validation today, which also means no validator-based regression test can catch a reintroduction of this bug class. The PR description already flags this as a follow-up — agreed, but I'd track it with a JIRA for the same reason as (2). **4. The projected distinct count is computed twice per candidate partitioning** (confirmed) https://github.com/apache/spark/blob/8b74639cc949ab60fdef654e1318b6bc6f88dc7d/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala#L908-L913 `projectionCoalesces` computes `kp.projectKeys(positions)._2.distinct.size`, and `groupedNumPartitions` (L954-L955) recomputes the identical expression for the `requiredNumPartitions` check; the no-projection fallback there also misses the `isGrouped` shortcut `projectionCoalesces` has. `InternalRowComparableWrapper.hashCode` is uncached (interpreted Murmur3 per call), so with many partition keys this is two full projection + distinct passes where one suffices — and the two formulas can drift, desynchronizing the (load-bearing, tested) `requiredNumPartitions` gate from the count the inserted node actually produces. Having `projectionPositions` return the positions together with the resulting partition count would collapse both call sites. **5. Per-collection-member repetition of O(n) key passes** (confirmed) `split` runs `projectionPositions`/`projectionCoalesces` for every `PartitioningCollection` member, although the collection invariant guarantees all KP members share the same `partitionKeys` reference — an m-member collection does m sets of identical passes, and the cost also lands on the common no-op path (a satisfying partitioning forces `positions` merely to evaluate `positions.isEmpty`, where the pre-PR code traversed keys zero times). This runs per child per rule invocation, again per AQE stage. Memoizing the unprojected distinct count per `partitionKeys` reference (and the projected count per positions value) would remove the repetition. **6. Admission and projection are decided by different code — drift risk** (design) `splitKeyedPartitionings` admits a KP into `needsGrouping` via `KeyedPartitioning.groupedSatisfies`, but what to project is computed by the rule-local `projectionPositions` — now a third independent implementation of cluster-key coverage alongside `groupedSatisfies` and `KeyedShuffleSpec.keyPositions`. The two must agree per position, or the inserted `GroupPartitionsExec` produces a partitioning that doesn't satisfy the distribution it was inserted for (the PR's own `years(ts)` unit test demonstrates the failure mode for a naive derivation). If `groupedSatisfies` is ever widened (e.g. `GetStructField` references, compatible-transform matching) without mirroring `projectionPositions`, the result is the L898 assert firing at planning time or a silent mis-projection. Housing the position computation next to `groupedSatisfies` on `KeyedPartitioning` — one matcher feeding both answers — would make the drift impossible, and would also give the currently-defensive expression-level disjunct a non-speculative home. Fine as a follow-up given the fix is config-gated. **7. The duplicated-`PARTITION BY` test duplicates the whole fixture** (minor) The "window top-k over duplicated PARTITION BY key" test is a full copy of the preceding subset test with only `PARTITION BY id` → `PARTITION BY id, id`, and its comment concedes it cannot fail independently. Looping the window spec (`Seq("id", "id, id")`) inside one test keeps the pinned behavior at half the fixture and runtime cost. Refuted candidates, for the record: the `assert(positions.nonEmpty)` at L898 is provably unreachable for any KP satisfying `supportsExpressions` (every production producer preserves it); the `OrderedDistribution` comparator cannot bind a transform output to the wrong domain (`areAllClusterKeysMatched` requires 1:1 `semanticEquals` between partition expressions and `SortOrder` children); and `GroupPartitionsExec.outputPartitioning`'s collection rebuild cannot trip `PartitioningCollection`'s equal-count `require` in stock Spark (no producer emits a mixed KP + non-KP collection). Generated-by: Claude Fable 5 -- 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]
