[SPARK-59289][SQL] Refuse to execute a grouping over a child it was not planned over

`GroupPartitionsExec.doExecute` and `doExecuteColumnar` compare `child.outputPartitioning` against
`childPartitioning` and throw an internal error when they differ, rather than coalescing the new
child's RDD on `grouping.partitions`, which indexes the child's partitions as of planning.

Giving up the keyed claim, which this PR already does, only gets `ValidateRequirements` to reject
such a plan. That is enough for the case it was written for, an `AQEShuffleReadExec` landing over a
keyed shuffle stage, because `AdaptiveSparkPlanExec.optimizeQueryStage` reverts a rule whose result
fails validation. It validates an `AQEShuffleReadRule`'s result and nothing else, so
`PlanAdaptiveDynamicPruningFilters`, an extension-injected query stage rule, or a preparation rule
after `EnsureRequirements` would not be caught. The base survived those by re-deriving the grouping
against the new child.

No rule is known to hand this node a child reporting a different partitioning, so the comparison is
free in practice and the throw is unreachable. It turns a silent wrong result into an explicit
error if that changes. The invariant test now asserts it for each of the three ways a child can
report something else.

Raised by dongjoon-hyun, who also pointed out that the comparison is by reference only while the
child hands back the same object: `PlanAdaptiveDynamicPruningFilters` rebuilds `BatchScanExec`, so
there it compares the partition keys by value and reads equal only while the connector returns the
same input partitions. That assumption is now in `outputPartitioning`'s doc, along with the
correction that the AQE revert does not cover every rule.

Author: peter-toth

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala

@@ -102,15 +102,19 @@ case class GroupPartitionsExec(
    * node a child reporting something else invalidates both, so the keyed claim is dropped and only
    * the physical output count is reported. `ValidateRequirements` then finds the parent's
    * distribution unsatisfied and `AdaptiveSparkPlanExec` reverts the rewrite, which is what an
-   * `AQEShuffleReadExec` landing over a keyed shuffle stage relies on. That revert is also what
-   * keeps `doExecute` off the changed child, since it still coalesces on the planning-time
-   * `grouping.partitions`. The base relied on the same revert, having re-derived to get there.
+   * `AQEShuffleReadExec` landing over a keyed shuffle stage relies on. Only an `AQEShuffleReadRule`
+   * is validated that way, so `checkChildStillMatches` covers the rest at execution.
    *
    * Re-deriving instead would apply a recipe chosen for one pairing to a different child, yielding
    * a claim nothing validated against the other side. It would also buy nothing today: no rule is
    * known to report a *different* `KeyedPartitioning`, so the reachable outcomes are an equal one,
    * where this check passes, and `UnknownPartitioning`, where re-deriving reports the same thing.
    *
+   * The comparison is by reference for a child that reports the same object, which the columnar and
+   * codegen wrappers do. `PlanAdaptiveDynamicPruningFilters` rebuilds `BatchScanExec`, so there it
+   * compares the partition keys by value, and reads equal only while the connector returns the same
+   * input partitions for the same scan.
+   *
    * Asked on the read rather than in `withNewChildInternal` because canonicalization rebuilds this
    * node over a canonicalized child whose partitioning cannot be read at all: a canonicalized
    * `BatchScanExec` throws from `reportedKeyedPartitioning`.
@@ -308,7 +312,24 @@ case class GroupPartitionsExec(
       sparkContext, executionId, driverAccumUpdates.toSeq)
   }
 
+  /**
+   * `grouping.partitions` indexes the child's partitions as of planning, so a child that no longer
+   * reports what this node was decided over would be coalesced on the wrong indices.
+   * `outputPartitioning` gives up its keyed claim in that case, which lets `ValidateRequirements`
+   * reject the plan, but only an `AQEShuffleReadRule`'s result is validated (see
+   * `AdaptiveSparkPlanExec.optimizeQueryStage`), so nothing catches a rewrite from any other rule.
+   * No rule is known to perform one; this makes the failure explicit rather than silent if one
+   * starts.
+   */
+  private def checkChildStillMatches(): Unit = {
+    if (child.outputPartitioning != childPartitioning) {
+      throw SparkException.internalError(
+        "GroupPartitionsExec's child no longer reports the partitioning it was planned over")
+    }
+  }
+
   override protected def doExecute(): RDD[InternalRow] = {
+    checkChildStillMatches()
     sendDriverMetrics()
     if (groupedPartitions.isEmpty) {
       sparkContext.emptyRDD
@@ -329,6 +350,7 @@ case class GroupPartitionsExec(
   override def supportsColumnar: Boolean = child.supportsColumnar && !usesSortedMerge
 
   override protected def doExecuteColumnar(): RDD[ColumnarBatch] = {
+    checkChildStillMatches()
     sendDriverMetrics()
     if (groupedPartitions.isEmpty) {
       sparkContext.emptyRDD

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExecSuite.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.execution.datasources.v2
 
-import org.apache.spark.SparkContext
+import org.apache.spark.{SparkContext, SparkException}
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{Ascending, Attribute, AttributeReference, SortOrder, TransformExpression}
@@ -668,6 +668,12 @@ class GroupPartitionsExecSuite extends SharedSparkSession {
         s"a child reporting $changed invalidates the grouping, so the claim goes")
       assert(rebuilt.plannedPartitioning === gpe.outputPartitioning,
         "what it was planned to report is still carried, it is just no longer reported")
+
+      // Giving up the claim is only half of it. `ValidateRequirements` rejects such a plan, but
+      // `AdaptiveSparkPlanExec.optimizeQueryStage` validates an `AQEShuffleReadRule`'s result and
+      // nothing else, so execution refuses rather than coalescing the new child on the old indices.
+      val e = intercept[SparkException](rebuilt.execute())
+      assert(e.getMessage.contains("no longer reports the partitioning it was planned over"))
     }
   }
 }