[SPARK-59272][SQL] Revalidate the storage-partitioned join pairing when a regrouped side drops the keyed claim

### What changes were proposed in this pull request?

`EnsureRequirements.checkKeyGroupCompatible` asks its two rebuilt children whether they still declare the same aligned partition key sequence before it commits to a storage-partitioned join, and declines when they do not.

The check is pairwise rather than a per-side `satisfies`, through `KeyLayout.describesSameKeys`, which compares the key rows and the key types.

### Why are the changes needed?

`GroupPartitionsExec` deliberately gives up a marked layout's keyed claim, reporting `UnknownPartitioning`, when the regrouping is not the identity: a marked side realigned onto differently-ordered merged keys, or a grouping that applies a reducer or a non-identity projection. That is the right answer to a claim the regrouping invalidates.

The gap is when it is answered. The give-up happens inside the node, which `checkKeyGroupCompatible` builds through `applyGroupPartitions` **after** it has decided the pairing is compatible, and nothing re-checks the rebuilt children. So a committed join can carry a child that no longer satisfies its required distribution. `ValidateRequirements` then rejects the whole stage, and every `AQEShuffleReadRule` and `OptimizeSkewedJoin` drops its result on a stage that does not validate, so partition coalescing, local read and skew join are all off for it.

A per-side recheck is not viable, and this is why the fix is pairwise. Partially clustered distribution deliberately leaves both rebuilt children ungrouped yet value-aligned, both onto one `mergedPartitionKeys` by `alignToExpectedKeys`. A `KeyedPartitioning` only satisfies a `ClusteredDistribution` once grouped, so a `satisfies` gate would reject that whole family. What the two sides owe each other is the key sequence `alignToExpectedKeys` constructs, each key repeated as many times as the merge expects, whichever side replicates, and that is what is asked here.

Nothing that was accepted before is refused on the `compatibleAsIs` path, where the children are the ones the pairing read. `KeyedShuffleSpec.isCompatibleWith` already ends in `describesSameKeys`, and all keyed members of a `PartitioningCollection` share one `KeyLayout`, so whichever member the spec matched on declares what the representative does.

### Does this PR introduce _any_ user-facing change?

**Yes**, one plan change. On the `SPARK-59050: SPJ: regrouping a marked layout must not keep the unknown-keyed claim` query the second join now takes a keyed one-side shuffle, three shuffles to four, and `ValidateRequirements.validate` on the executed plan goes from `false` to `true`. The alternative was not "no shuffle", it was "no shuffle and no AQE".

### How was this patch tested?

The `SPARK-59050` regrouping test already reaches this state, so it is the test: its expectation moves from three shuffles to four, and it now asserts `ValidateRequirements.validate` on the executed plan ahead of the shuffle count, because that names the reason the fourth shuffle exists. It is the only test in `KeyGroupedPartitioningSuite`'s 183 that the change moves.

**A differential sweep measures the rest.** 307200 generated storage-partitioned join plans, 40 partitioning shapes a side crossed with six join types over 32 configuration cells, run against master and against this change. Each plan is checked twice, that `ValidateRequirements` passes and that two sides whose shuffles were both skipped really declare the same key sequence. The cases are keyed and set-differenced, not just counted.

| | master | this change |
|---|---|---|
| plans that skip both shuffles yet disagree on the keys | 2024 | **0** |
| plans `ValidateRequirements` rejects | 6724 | 4700 |
| plans the planner cannot build at all | 29952 | 29952 |

**Zero cases are added to any of the three**, and the 2024 co-partitioning violations that go are exactly the same 2024 cases as the validation failures that go. That equality is the whole of this change in one number: every plan that skipped both shuffles on keys the two sides did not share was also a plan the validator rejected.

Green: `KeyGroupedPartitioningSuite`, `EnsureRequirementsSuite`, `ValidateRequirementsSuite`, `GroupPartitionsExecSuite`, `PlannerSuite`, `AdaptiveQueryExecSuite`, `JoinSuite` and the seven plan-stability suites, 1679 tests. `dev/lint-scala` is clean.

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code

Author: peter-toth

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala

@@ -794,7 +794,33 @@ case class EnsureRequirements(
         rightReducers, distributePartitions = applyPartialClustering && !replicateRightSide)
     }
 
-    if (compatibleAsIs || pushCommonValues) Some(Seq(newLeft, newRight)) else None
+    // The pairing is only worth committing to if both children still declare the same aligned key
+    // sequence once the grouping has been pushed into them. They can fail that. A
+    // `GroupPartitionsExec` gives up its keyed claim when it turns out to regroup a layout that
+    // pins undeclared rows to `hash(key) % numPartitions` (see
+    // `KeyLayout.mayContainUnknownPartitionKeys`), and only the node knows the permutation it
+    // performs, so that answer arrives after the pairing was chosen. Asking before returning is
+    // what keeps the join from skipping both shuffles for a child that no longer satisfies its
+    // distribution, which is a plan `ValidateRequirements` rejects and every AQE rule that needs a
+    // valid plan then refuses to touch.
+    //
+    // The check is pairwise, not a per-side `satisfies`. Partially clustered distribution leaves
+    // both children value-aligned yet not grouped on purpose, so a per-side gate would refuse that
+    // whole family. What both sides owe each other is the key sequence `alignToExpectedKeys`
+    // guarantees, each key repeated as many times as the merge expects, whichever side replicates.
+    // Through `KeyLayout.describesSameKeys`, which carries the reason the key types are compared as
+    // well as the rows.
+    //
+    // Nothing is refused that was accepted before this check on the `compatibleAsIs` path, where
+    // the children are the ones the pairing read: `KeyedShuffleSpec.isCompatibleWith` already ends
+    // in `describesSameKeys`, and all keyed members of a `PartitioningCollection` share one
+    // `KeyLayout`, so whichever member the spec matched on declares what the representative does.
+    def declaredLayout(plan: SparkPlan): Option[KeyLayout] =
+      PartitioningCollection.representativeOf(plan.outputPartitioning).map(_.layout)
+    Option.when((compatibleAsIs || pushCommonValues) &&
+      declaredLayout(newLeft).exists { left =>
+        declaredLayout(newRight).exists(left.describesSameKeys)
+      })(Seq(newLeft, newRight))
   }
 
   private def checkShufflePartitionIdPassThroughCompatible(

sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala

@@ -8247,13 +8247,22 @@ class KeyGroupedPartitioningSuite
           SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
         val df = sql(query)
         checkAnswer(df, expected)
-        // The regrouped marked side can no longer claim the hash-routing contract, so the final
-        // join re-shuffles it instead of storage-partitioning. Three one-side shuffles, all keyed
-        // with unknown partition keys: rt onto the union, rt2 onto ra2, and the final join's
-        // re-shuffle of the regrouped side. Before the fix the final join storage-partitioned
-        // (only the first two shuffles) and silently lost the id=5 row.
+        // The regrouped marked side can no longer claim the hash-routing contract. Four one-side
+        // shuffles, all keyed with unknown partition keys: rt onto the union, rs onto the marked
+        // side once the second join declines, rt2 onto ra2, and the final join's re-shuffle of the
+        // side that was regrouped. Before SPARK-59050 the final join storage-partitioned (only the
+        // first, third and fourth shuffles) and silently lost the id=5 row.
+        //
+        // The second join's shuffle is this fix's. The give-up happens inside the node, so it used
+        // to arrive after that join had already committed to the pairing and skipped both
+        // shuffles, leaving a plan `ValidateRequirements` rejects, which every AQEShuffleReadRule
+        // and OptimizeSkewedJoin then refuses to touch. The join now asks its two built children
+        // whether they still declare the same aligned keys, and declines when they do not. The
+        // validation assert comes first, because it names the reason the shuffle below exists.
+        assert(ValidateRequirements.validate(df.queryExecution.executedPlan),
+          "the executed plan must satisfy every operator's required distribution")
         assertShuffleMayContainUnknownPartitionKeys(df.queryExecution.executedPlan,
-          Seq(true, true, true))
+          Seq(true, true, true, true))
       }
     }
   }