[SPARK-59948][SQL] Strip aliases from a DSv2 reported ordering on a nested field

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

@@ -19,7 +19,7 @@ package org.apache.spark.sql.execution.datasources.v2
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.LogKeys.{CLASS_NAME, COLUMN_NAMES, RELATION_NAME}
 import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.expressions.V2ExpressionUtils
+import org.apache.spark.sql.catalyst.expressions.{AliasHelper, V2ExpressionUtils}
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern.DATA_SOURCE_V2_SCAN_RELATION
@@ -35,7 +35,7 @@ import org.apache.spark.util.collection.Utils.sequenceToOption
  * and ordering reported by data sources to their catalyst counterparts. Then, annotates the plan
  * with the partitioning and ordering result.
  */
-object V2ScanPartitioningAndOrdering extends Rule[LogicalPlan] with Logging {
+object V2ScanPartitioningAndOrdering extends Rule[LogicalPlan] with AliasHelper with Logging {
   override def apply(plan: LogicalPlan): LogicalPlan = {
     val scanRules = Seq[LogicalPlan => LogicalPlan] (partitioning, ordering)
 
@@ -104,8 +104,13 @@ object V2ScanPartitioningAndOrdering extends Rule[LogicalPlan] with Logging {
         // scan output: truncating it here would also drop the sort orders on a partition key past a
         // pruned column, which still hold. `DataSourceV2ScanRelation.doCanonicalize` and
         // `DataSourceV2ScanExecBase.outputOrdering` restrict it to the scan output instead.
+        // A nested sort key such as `s.x` resolves to a new `Alias(GetStructField(...))` on every
+        // run of this rule. Strip the alias so repeated runs give equal plans and, when the scan
+        // reads the whole struct, the ordering can satisfy a required ordering on the same field
+        // (see SPARK-60011).
         val ordering =
           V2ExpressionUtils.toCatalystOrdering(reportedOrdering, relation, relation.funCatalog)
+            .map(o => o.copy(child = trimAliases(o.child)))
         d.copy(ordering = Some(ordering))
       }
   }

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

@@ -4093,6 +4093,19 @@ class KeyGroupedPartitioningSuite
     }
   }
 
+  test("SPARK-59948: a scan's reported ordering on a nested field") {
+    val nestedItems = "nested_items"
+    createTable(nestedItems, Array(Column.create("id", LongType),
+      Column.create("s", new StructType().add("x", LongType))), Array.empty,
+      ordering = Array(sort(column("s.x"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST)))
+    sql(s"INSERT INTO testcat.ns.$nestedItems VALUES " +
+      "(1, named_struct('x', 1L)), (2, named_struct('x', 2L))")
+
+    val df = sql(s"SELECT * FROM testcat.ns.$nestedItems SORT BY s.x")
+    assert(collect(df.queryExecution.executedPlan) { case s: SortExec => s }.isEmpty)
+    checkAnswer(df, Seq(Row(1L, Row(1L)), Row(2L, Row(2L))))
+  }
+
   test("SPARK-47094: SPJ: Support compatible buckets") {
     val table1 = "tab1e1"
     val table2 = "table2"

sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala

@@ -2861,6 +2861,22 @@ class MergeSubplansSuite extends PlanTest {
       s"the merged scan should re-derive the reported ordering; got ${scans.head.ordering}")
   }
 
+  test("SPARK-59948: merge DSv2 scans reporting an ordering on a nested field") {
+    // Each scan derives its `s.x ASC` report separately. The nested key resolves to a fresh
+    // `Alias(GetStructField(...))` per derivation, so unless the alias is stripped the two
+    // reports never compare equal and the merge is declined.
+    val table = new TestV2Table(StructType(Seq(
+      StructField("s", StructType(Seq(StructField("x", IntegerType)))),
+      StructField("b", IntegerType), StructField("c", StringType))),
+      reportedOrderingCols = Seq("s.x"))
+    val q = testRelation.select(
+      ScalarSubquery(v2ScanReportingOn(table, Seq("s", "b")).groupBy()(sum($"b").as("sum_b"))),
+      ScalarSubquery(v2ScanReportingOn(table, Seq("s", "c")).groupBy()(sum($"c").as("sum_c"))))
+    val optimized = Optimize.execute(q.analyze)
+
+    assert(v2Scans(optimized).length == 1, s"the two scans should be fused into one:\n$optimized")
+  }
+
   test("SPARK-58549: do not merge when the rebuilt scan re-derives less than the combined " +
     "ordering") {
     // One input reports `a ASC`, the other `a ASC, c ASC`, so the combined requirement is the