[SPARK-59721][SQL] Ignore reported partitioning and ordering that reference unresolvable columns

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

When a V2 scan reports a `KeyGroupedPartitioning` or an `ordering` that references a column the relation can't resolve, the query used to fail with `Unable to resolve X given [...]`. This PR makes `V2ScanPartitioningAndOrdering` check the columns referenced by the reported partition keys and ordering (recursively, including nested transform arguments) against the relation before converting them. If any can't be resolved, including a missing nested field, an ambiguous name, or an empty name, the rule logs a warning naming the relation, the scan class, and those columns, and ignores the report. The partitioning falls back to `UnknownPartitioning`, and the ordering is dropped.

This PR adds a `resolveRefOpt`, and makes `resolveRef` a thin wrapper around it. `resolveRefOpt` still throws for a missing nested field or an ambiguous name, so the rule treats those errors as unresolvable and includes the error condition in the warning. The `toCatalyst*` methods are unchanged.

Keys whose columns all resolve take the existing conversion path unchanged, so an unsupported expression type (e.g. `id + 1`) or a nested transform whose function can't be loaded still fails the query as before. Because the whole report is checked before any key is converted, a report that also references an unresolvable column falls back with a warning instead, e.g. `[id + 1, identity(missing)]`.

The Javadoc of `KeyGroupedPartitioning#keys()` and `SupportsReportOrdering#outputOrdering()` now says that Spark resolves the column references against the table columns, including columns pruned from the scan output, and ignores the whole report if any of them can't be resolved. The docs of `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` now mention that an ignored ordering can be replaced by one derived from the partition keys.

### Why are the changes needed?

Found while reviewing SPARK-58111 ([PR #55518](#55518)). This is a general, pre-existing bug in how `V2ScanPartitioningAndOrdering` handles reported partitioning and ordering, not something introduced by #55518.

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

Yes. A query against a data source whose scan reports a `KeyGroupedPartitioning` with a key that can't be resolved previously failed outright. It now succeeds, with partitioning degraded to `UnknownPartitioning` instead, and Spark logs a warning naming the relation, the scan class, and the columns that can't be resolved.

The same applies to an ambiguous or empty column name. A report that combines an unsupported expression with an unresolvable column, e.g. `missing + 1` or `[id + 1, identity(missing)]`, used to fail with `_LEGACY_ERROR_TEMP_3054` and now falls back with a warning. An unsupported expression over resolvable columns still fails as before.

If the scan keeps its partitioning but its reported ordering is ignored, Spark derives the ordering from the partition keys when `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` is on (the default).

### How was this patch tested?

New tests in `V2ExpressionUtilsSuite`:
- `resolveRefOpt` returns `None` for an unresolvable reference.
- `resolveRefOpt` throws `FIELD_NOT_FOUND` for a missing nested field and `AMBIGUOUS_REFERENCE` for an ambiguous reference.

New tests in `KeyGroupedPartitioningSuite`:
- A join against a table whose scan reports unresolvable partition keys succeeds, plans a shuffle instead of a storage-partitioned join, and logs a warning naming the relation, the scan class, and only the unresolvable columns. The cases are:
  - a `bucket` key, a nested transform key, and a multi-key partitioning with one bad key;
  - duplicate references, and two distinct unresolvable columns;
  - a missing nested field;
  - an unsupported expression next to or over an unresolvable column (`[id + 1, identity(missing)]`, `missing + 1`);
  - a connector-defined transform that hides the reference from `children()`.

  Resolvable keys (`id`, case-insensitive `ID`, and a connector-defined transform with an extra reference only in `children()`) keep the partitioning and plan a storage-partitioned join without a shuffle.
- A reported ordering that can't be resolved is ignored as a whole with a warning, and the scan still derives `id ASC` from its kept partitioning. The cases are a column that doesn't exist, one hidden from a connector-defined sort order's `children()`, an empty name, and a two-key ordering with one unresolvable key (`[id, missing]`). Resolvable orderings (`id`, case-insensitive `ID`, the metadata column `index`, and a connector-defined sort order with an extra reference only in `children()`) are kept without a warning.
- A reported partition key of an unsupported expression type (`id + 1`), or a nested transform whose function can't be loaded (`f(g(id))`), still fails the query with `_LEGACY_ERROR_TEMP_3054`.

New test in `DataSourceV2Suite`:
- An unresolvable ordering reported through connector-defined `SortOrder`, `Transform`, and `NamedReference` classes (`OrderAndPartitionAwareDataSource` and its Java counterpart) is ignored with a warning, and the scan relation keeps no ordering.

Also ran `V2ExpressionUtilsSuite`, `KeyGroupedPartitioningSuite`, `WriteDistributionAndOrderingSuite`, `DataSourceV2Suite`, `MergeSubplansSuite` (comment update only), `SQLConfSuite`, and `ProjectedOrderingAndPartitioningSuite` to confirm no regression in the partitioning/SPJ, ordering, and write paths this change touches.

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

Generated-by: Claude Code (Sonnet 5, Opus 5.5)
Verified manually by me.

Closes #58979 from anuragmantri/v2expr-opt-resolve-degrade.

Authored-by: Anurag Mantripragada <amantripragada@apple.com>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>
(cherry picked from commit 9de605f)
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

Author: anuragmantri

docs/sql-performance-tuning.md

@@ -708,7 +708,7 @@ The following SQL properties enable Storage Partition Join in different join que
       <td><code>spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled</code></td>
       <td>true</td>
       <td>
-        When enabled, Spark derives the output ordering of a V2 scan from its partition key expressions, if the source reports a keyed partitioning but no explicit ordering. All rows of such a partition share one key value, so the partition is trivially sorted by those expressions, and a sort Spark would otherwise add becomes unnecessary. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+        When enabled, Spark derives the output ordering of a V2 scan from its partition key expressions, if the source reports a keyed partitioning but no explicit ordering, or an ordering that Spark ignores because it references a column that cannot be resolved. All rows of such a partition share one key value, so the partition is trivially sorted by those expressions, and a sort Spark would otherwise add becomes unnecessary. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
       </td>
       <td>4.2.0</td>
     </tr>

sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsReportOrdering.java

@@ -35,6 +35,10 @@ public interface SupportsReportOrdering extends Scan {
 
   /**
    * Returns the order in each partition of this data source scan.
+   * <p>
+   * Spark resolves the column references in these sort orders against the table columns,
+   * including columns pruned from the scan output. If any of them cannot be resolved, Spark
+   * ignores the whole reported ordering and logs a warning.
    */
   SortOrder[] outputOrdering();
 }

sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/partitioning/KeyGroupedPartitioning.java

@@ -43,6 +43,10 @@ public KeyGroupedPartitioning(Expression[] keys, int numPartitions) {
 
   /**
    * Returns the partition transform expressions for this partitioning.
+   * <p>
+   * Spark resolves the column references in these expressions against the table columns,
+   * including columns pruned from the scan output. If any of them cannot be resolved, Spark
+   * ignores the whole reported partitioning and logs a warning.
    */
   public Expression[] keys() {
     return keys;

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala

@@ -46,15 +46,22 @@ import org.apache.spark.util.ArrayImplicits._
 object V2ExpressionUtils extends SQLConfHelper with Logging {
   import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
 
+  /**
+   * Variant of `resolveRef` that returns `None` if no attribute matches the reference. It still
+   * throws for a nested-field extraction error, such as a missing nested field, or an ambiguous
+   * reference.
+   */
+  private[sql] def resolveRefOpt(
+      ref: NamedReference, plan: LogicalPlan): Option[NamedExpression] = {
+    plan.resolve(ref.fieldNames.toImmutableArraySeq, conf.resolver)
+  }
+
   def resolveRef[T <: NamedExpression](ref: NamedReference, plan: LogicalPlan): T = {
-    plan.resolve(ref.fieldNames.toImmutableArraySeq, conf.resolver) match {
-      case Some(namedExpr) =>
-        namedExpr.asInstanceOf[T]
-      case None =>
-        val name = ref.fieldNames.toImmutableArraySeq.quoted
-        val outputString = plan.output.map(_.name).mkString(",")
-        throw QueryCompilationErrors.cannotResolveAttributeError(name, outputString)
-    }
+    resolveRefOpt(ref, plan).getOrElse {
+      val name = ref.fieldNames.toImmutableArraySeq.quoted
+      val outputString = plan.output.map(_.name).mkString(",")
+      throw QueryCompilationErrors.cannotResolveAttributeError(name, outputString)
+    }.asInstanceOf[T]
   }
 
   def resolveRefs[T <: NamedExpression](refs: Seq[NamedReference], plan: LogicalPlan): Seq[T] = {

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -2590,7 +2590,8 @@ object SQLConf {
     buildConf("spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled")
       .doc("When enabled, Spark derives output ordering from the partition key expressions of " +
         "a V2 data source that reports a KeyedPartitioning but does not report explicit ordering " +
-        "via SupportsReportOrdering. Within a single partition all rows share the same key " +
+        "via SupportsReportOrdering, or reports one that Spark ignores because it references a " +
+        "column that cannot be resolved. Within a single partition all rows share the same key " +
         s"value, so the data is trivially sorted by those expressions. Requires " +
         s"${V2_BUCKETING_ENABLED.key} to be enabled.")
       .version("4.2.0")

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtilsSuite.scala

@@ -21,7 +21,7 @@ import org.apache.spark.SparkFunSuite
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.plans.logical.LocalRelation
 import org.apache.spark.sql.connector.expressions._
-import org.apache.spark.sql.types.StringType
+import org.apache.spark.sql.types.{IntegerType, StringType, StructType}
 
 class V2ExpressionUtilsSuite extends SparkFunSuite {
 
@@ -37,4 +37,30 @@ class V2ExpressionUtilsSuite extends SparkFunSuite {
     }
     assert(exc.message.contains("v2Fun(a) ASC NULLS FIRST is not currently supported"))
   }
+
+  test("SPARK-59721: resolveRefOpt returns None for an unresolvable reference") {
+    val plan = LocalRelation(AttributeReference("a", StringType)())
+    assert(V2ExpressionUtils.resolveRefOpt(FieldReference("a"), plan).isDefined)
+    assert(V2ExpressionUtils.resolveRefOpt(FieldReference("missing"), plan).isEmpty)
+  }
+
+  test("SPARK-59721: resolveRefOpt throws for a missing nested field or an ambiguous reference") {
+    val structPlan =
+      LocalRelation(AttributeReference("s", new StructType().add("x", IntegerType))())
+    checkError(
+      exception = intercept[AnalysisException] {
+        V2ExpressionUtils.resolveRefOpt(FieldReference("s.missing"), structPlan)
+      },
+      condition = "FIELD_NOT_FOUND",
+      parameters = Map("fieldName" -> "`missing`", "fields" -> "`x`"))
+
+    val ambiguousPlan = LocalRelation(
+      AttributeReference("a", StringType)(), AttributeReference("a", StringType)())
+    checkError(
+      exception = intercept[AnalysisException] {
+        V2ExpressionUtils.resolveRefOpt(FieldReference("a"), ambiguousPlan)
+      },
+      condition = "AMBIGUOUS_REFERENCE",
+      parameters = Map("name" -> "`a`", "referenceNames" -> "[`a`, `a`]"))
+  }
 }

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

@@ -135,15 +135,15 @@ trait DataSourceV2ScanExecBase
 
   /**
    * Returns the output ordering for this scan. When the source reports ordering via
-   * `SupportsReportOrdering`, that ordering may reference columns pruned out of the scan output
-   * (see V2ScanPartitioningAndOrdering), while a consumer binds it against the output, e.g. the
-   * k-way merge of `GroupPartitionsExec`. Ordering is prefix-based, so the leading run of sort
-   * orders over the output is kept and the rest is dropped, except the sort orders on a partition
-   * key when the output partitioning is a `KeyedPartitioning`: each partition holds a single key,
-   * so those still hold. Otherwise, when the output partitioning is a `KeyedPartitioning` and
-   * `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` is on, each partition
-   * contains rows where the key expressions evaluate to a single constant value, so the data
-   * is trivially sorted by those expressions within the partition.
+   * `SupportsReportOrdering` and `V2ScanPartitioningAndOrdering` keeps it, that ordering may
+   * reference columns pruned out of the scan output, while a consumer binds it against the
+   * output, e.g. the k-way merge of `GroupPartitionsExec`. Ordering is prefix-based, so the
+   * leading run of sort orders over the output is kept and the rest is dropped, except the sort
+   * orders on a partition key when the output partitioning is a `KeyedPartitioning`: each
+   * partition holds a single key, so those still hold. Otherwise, when the output partitioning
+   * is a `KeyedPartitioning` and `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled`
+   * is on, each partition contains rows where the key expressions evaluate to a single constant
+   * value, so the data is trivially sorted by those expressions within the partition.
    */
   override def outputOrdering: Seq[SortOrder] = {
     (ordering, outputPartitioning) match {

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

@@ -17,11 +17,14 @@
 package org.apache.spark.sql.execution.datasources.v2
 
 import org.apache.spark.internal.Logging
-import org.apache.spark.internal.LogKeys.CLASS_NAME
+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.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern.DATA_SOURCE_V2_SCAN_RELATION
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
+import org.apache.spark.sql.connector.expressions.{Expression => V2Expression, NamedReference, SortOrder => V2SortOrder, Transform}
 import org.apache.spark.sql.connector.read.{SupportsReportOrdering, SupportsReportPartitioning}
 import org.apache.spark.sql.connector.read.partitioning.{KeyGroupedPartitioning, UnknownPartitioning}
 import org.apache.spark.util.ArrayImplicits._
@@ -47,9 +50,19 @@ object V2ScanPartitioningAndOrdering extends Rule[LogicalPlan] with Logging {
         if d.keyGroupedPartitioning.isEmpty =>
       val catalystPartitioning = scan.outputPartitioning() match {
         case kgp: KeyGroupedPartitioning =>
-          val partitioning = sequenceToOption(
-            kgp.keys().map(V2ExpressionUtils.toCatalystOpt(_, relation, relation.funCatalog))
-              .toImmutableArraySeq)
+          val keys = kgp.keys().toImmutableArraySeq
+          val unresolvedColumns = unresolvableColumns(keys, relation)
+          val partitioning = if (unresolvedColumns.nonEmpty) {
+            logWarning(
+              log"Spark ignores the KeyGroupedPartitioning reported by " +
+                log"${MDC(RELATION_NAME, relation.name)} (scan " +
+                log"${MDC(CLASS_NAME, scan.getClass.getName)}) because the partition key columns " +
+                log"cannot be resolved: ${MDC(COLUMN_NAMES, unresolvedColumns.mkString(", "))}.")
+            None
+          } else {
+            sequenceToOption(
+              keys.map(V2ExpressionUtils.toCatalystOpt(_, relation, relation.funCatalog)))
+          }
           // Keep the partitioning when at least one of its keys is still in the scan output: the
           // scan projects the pruned key positions away when reporting its physical output
           // partitioning (see DataSourceV2ScanExecBase.outputPartitioning). When no key survives,
@@ -75,12 +88,57 @@ object V2ScanPartitioningAndOrdering extends Rule[LogicalPlan] with Logging {
   private def ordering(plan: LogicalPlan) = plan.transformDownWithPruning(
       _.containsPattern(DATA_SOURCE_V2_SCAN_RELATION)) {
     case d @ ExtractV2ScanInfo(relation, scan: SupportsReportOrdering, _) =>
-      // The ordering is kept as reported, even where it references columns pruned out of the 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.
-      val ordering =
-        V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(), relation, relation.funCatalog)
-      d.copy(ordering = Some(ordering))
+      val reportedOrdering = scan.outputOrdering()
+      val unresolvedColumns = unresolvableColumns(reportedOrdering.toImmutableArraySeq, relation)
+      if (unresolvedColumns.nonEmpty) {
+        logWarning(
+          log"Spark ignores the ordering reported by ${MDC(RELATION_NAME, relation.name)} " +
+            log"(scan ${MDC(CLASS_NAME, scan.getClass.getName)}) because the ordering columns " +
+            log"cannot be resolved: ${MDC(COLUMN_NAMES, unresolvedColumns.mkString(", "))}.")
+        // Use None, not Some(Nil): only None lets DataSourceV2ScanExecBase.outputOrdering derive an
+        // ordering from a kept key-grouped partitioning, which does not depend on the dropped
+        // report.
+        d.copy(ordering = None)
+      } else {
+        // The ordering is kept as reported, even where it references columns pruned out of the
+        // 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.
+        val ordering =
+          V2ExpressionUtils.toCatalystOrdering(reportedOrdering, relation, relation.funCatalog)
+        d.copy(ordering = Some(ordering))
+      }
+  }
+
+  private def unresolvableColumns(
+      exprs: Seq[V2Expression],
+      relation: LogicalPlan): Seq[String] = {
+    // Walk the accessors the conversion reads, `Transform.arguments()` and
+    // `SortOrder.expression()`, instead of `V2Expression.references()` or `children()`: a connector
+    // can override those to disagree with the conversion, and `ApplyTransform.references()` returns
+    // only its top-level arguments, which would miss `f(g(missing))`. Other expressions, such as
+    // `GeneralScalarExpression`, which the conversion rejects, go through `children()`.
+    def collectReferences(expr: V2Expression): Seq[NamedReference] = expr match {
+      case ref: NamedReference => Seq(ref)
+      case t: Transform => t.arguments().toImmutableArraySeq.flatMap(collectReferences)
+      case s: V2SortOrder => collectReferences(s.expression())
+      case other => other.children().toImmutableArraySeq.flatMap(collectReferences)
+    }
+    exprs.flatMap(collectReferences).flatMap { ref =>
+      val parts = ref.fieldNames.toImmutableArraySeq
+      if (parts.isEmpty) {
+        // An empty name can be neither quoted nor resolved: both throw.
+        Some("<empty>")
+      } else {
+        val name = parts.quoted
+        try {
+          Option.when(V2ExpressionUtils.resolveRefOpt(ref, relation).isEmpty)(name)
+        } catch {
+          // A nested-field extraction error or an ambiguous reference throws instead of returning
+          // None.
+          case e: AnalysisException => Some(s"$name (${e.getCondition})")
+        }
+      }
+    }.distinct
   }
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala

@@ -980,12 +980,13 @@ class PlanMerger(
   // Only the reported EXPRESSIONS are compared, which is all a DataSourceV2ScanRelation carries;
   // the merged scan's split count and partition values can still differ from an input's, since it
   // may push a different best-effort filter and so prune differently. And a report the merged scan
-  // GAINS is not a degradation either. For partitioning that is because an input dropped its own
-  // only where none of its keys survived in that scan's output (V2ScanPartitioningAndOrdering's
-  // partitioning pass keeps the report whenever any key survives), not because the source stopped
-  // reporting; the ordering pass has no such guard, so an ordering report is
-  // never dropped by pruning and a gained one can only come from the source. Either way, keeping it
-  // is exactly the win this merge is after.
+  // GAINS is not a degradation either. Both V2ScanPartitioningAndOrdering passes drop a report
+  // that references a column the relation cannot resolve, but that does not depend on pruning.
+  // Beyond that, an input dropped its partitioning only where none of its keys survived in that
+  // scan's output (the partitioning pass keeps the report whenever any key survives), not because
+  // the source stopped reporting; the ordering pass has no such guard, so an ordering report is
+  // never dropped by pruning and a gained one can only come from the source. Either way, keeping
+  // it is exactly the win this merge is after.
   private def mergeDegradesReporting(
       merged: DataSourceV2ScanRelation,
       requiredKeyGroupedPartitioning: Seq[Expression],

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

@@ -24,6 +24,7 @@ import java.util.OptionalLong
 
 import scala.jdk.CollectionConverters._
 
+import org.apache.logging.log4j.Level
 import test.org.apache.spark.sql.connector._
 
 import org.apache.spark.{SparkException, SparkUnsupportedOperationException}
@@ -397,6 +398,34 @@ class DataSourceV2Suite extends SharedSparkSession with AdaptiveSparkPlanHelper
     }
   }
 
+  test("SPARK-59721: ignore an unresolvable ordering reported with connector-defined classes") {
+    Seq(
+      classOf[OrderAndPartitionAwareDataSource],
+      classOf[JavaOrderAndPartitionAwareDataSource]
+    ).foreach { cls =>
+      withClue(cls.getName) {
+        val df = spark.read
+          .option("partitionKeys", "i")
+          .option("orderKeys", "missing")
+          .format(cls.getName)
+          .load()
+        val logAppender = new LogAppender("unresolvable reported ordering")
+        withLogAppender(logAppender, level = Some(Level.WARN)) {
+          checkAnswer(df, Seq(Row(1, 4), Row(1, 5), Row(3, 5), Row(2, 6), Row(4, 1), Row(4, 2)))
+        }
+        val scanRelation = getScanRelation(df)
+        val expectedWarning =
+          s"Spark ignores the ordering reported by ${scanRelation.name} " +
+            s"(scan ${scanRelation.scan.getClass.getName}) because the ordering columns " +
+            "cannot be resolved: missing."
+        val warnings = logAppender.loggingEvents.map(_.getMessage.getFormattedMessage)
+          .filter(_.contains("columns cannot be resolved"))
+        assert(warnings.toSet == Set(expectedWarning), warnings)
+        assert(scanRelation.ordering.isEmpty)
+      }
+    }
+  }
+
   test ("statistics report data source") {
     Seq(classOf[ReportStatisticsDataSource], classOf[JavaReportStatisticsDataSource]).foreach {
       cls =>