[SPARK-59068][SQL] Restore support for nested runtime filter attributes

common/utils/src/main/resources/error/error-conditions.json

@@ -2260,6 +2260,24 @@
     ],
     "sqlState" : "KD010"
   },
+  "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE" : {
+    "message" : [
+      "The runtime filter attribute <attribute> reported by `<method>` in data source scan <scanClass> is invalid for the scan relation output <relationOutput>."
+    ],
+    "subClass" : {
+      "CANNOT_RESOLVE" : {
+        "message" : [
+          "The attribute cannot be resolved."
+        ]
+      },
+      "NOT_TOP_LEVEL" : {
+        "message" : [
+          "The attribute must be top-level, but it is a nested reference."
+        ]
+      }
+    },
+    "sqlState" : "KD000"
+  },
   "DATA_SOURCE_METADATA_SCHEMA_NOT_IMPLEMENTED" : {
     "message" : [
       "<class> does not implement metadataSchema."

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

@@ -39,9 +39,8 @@ public interface SupportsRuntimeFiltering extends SupportsRuntimeV2Filtering {
    * Spark will call {@link #filter(Filter[])} if it can derive a runtime
    * predicate for any of the filter attributes.
    * <p>
-   * Each reference must be a top-level attribute present in {@link Scan#readSchema()}.
-   * Nested references and attributes pruned out of the read schema fail to resolve when
-   * Spark builds the scan relation.
+   * Each reference must resolve against the scan relation output when Spark builds it. Attributes
+   * pruned out of {@link Scan#readSchema()} fail to resolve.
    */
   NamedReference[] filterAttributes();
 
@@ -51,6 +50,13 @@ public interface SupportsRuntimeFiltering extends SupportsRuntimeV2Filtering {
    * The provided expressions must be interpreted as a set of filters that are ANDed together.
    * Implementations may use the filters to prune initially planned {@link InputPartition}s.
    * <p>
+   * Spark tracks runtime-filter eligibility by root attribute. If {@link #filterAttributes()}
+   * returns a nested reference, this method may receive a filter on another nested field under
+   * the same root. Implementations must inspect each filter and use only filters they can apply.
+   * Nested paths are encoded in a V1 {@link Filter} as unquoted dot-separated names such as
+   * {@code parent.child}. A top-level column whose name contains a dot remains quoted, such as
+   * {@code `parent.child`}.
+   * <p>
    * If the scan also implements {@link SupportsReportPartitioning}, it must preserve
    * the originally reported partitioning during runtime filtering. While applying runtime filters,
    * the scan may detect that some {@link InputPartition}s have no matching data, in which case

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

@@ -52,9 +52,8 @@ public interface SupportsRuntimeV2Filtering extends Scan {
    * Spark will call {@link #filter(Predicate[])} if it can derive a runtime
    * predicate for any of the filter attributes.
    * <p>
-   * Each reference must be a top-level attribute present in {@link Scan#readSchema()}.
-   * Nested references and attributes pruned out of the read schema fail to resolve when
-   * Spark builds the scan relation.
+   * Each reference must resolve against the scan relation output when Spark builds it. Attributes
+   * pruned out of {@link Scan#readSchema()} fail to resolve.
    */
   NamedReference[] filterAttributes();
 
@@ -64,6 +63,11 @@ public interface SupportsRuntimeV2Filtering extends Scan {
    * The provided expressions must be interpreted as a set of predicates that are ANDed together.
    * Implementations may use the predicates to prune initially planned {@link InputPartition}s.
    * <p>
+   * Spark tracks runtime-filter eligibility by root attribute. If {@link #filterAttributes()}
+   * returns a nested reference, this method may receive a predicate on another nested field under
+   * the same root. Implementations must inspect each predicate and use only predicates they can
+   * apply.
+   * <p>
    * If the scan also implements {@link SupportsReportPartitioning}, it must preserve
    * the originally reported partitioning during runtime filtering. While applying runtime
    * predicates, the scan may detect that some {@link InputPartition}s have no matching data, in

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

@@ -67,7 +67,7 @@ object V2ExpressionUtils extends SQLConfHelper with Logging {
       refs: Array[NamedReference],
       output: Seq[Attribute]): AttributeSet = {
     val plan = LocalRelation(output)
-    AttributeSet(resolveRefs[Attribute](refs.toImmutableArraySeq, plan))
+    AttributeSet(resolveRefs[NamedExpression](refs.toImmutableArraySeq, plan))
   }
 
   /**

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala

@@ -4654,6 +4654,46 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
     )
   }
 
+  def cannotResolveDataSourceRuntimeFilterAttributeError(
+      attribute: Array[String],
+      method: String,
+      scanClass: String,
+      relationOutput: StructType,
+      cause: AnalysisException): AnalysisException = {
+    invalidDataSourceRuntimeFilterAttributeError(
+      attribute, method, scanClass, relationOutput, "CANNOT_RESOLVE", Some(cause))
+  }
+
+  def nestedDataSourceFullyPushedRuntimeFilterAttributeError(
+      attribute: Array[String],
+      scanClass: String,
+      relationOutput: StructType): AnalysisException = {
+    invalidDataSourceRuntimeFilterAttributeError(
+      attribute,
+      "fullyPushedFilterAttributes()",
+      scanClass,
+      relationOutput,
+      "NOT_TOP_LEVEL",
+      None)
+  }
+
+  private def invalidDataSourceRuntimeFilterAttributeError(
+      attribute: Array[String],
+      method: String,
+      scanClass: String,
+      relationOutput: StructType,
+      errorSubClass: String,
+      cause: Option[AnalysisException]): AnalysisException = {
+    new AnalysisException(
+      errorClass = s"DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.$errorSubClass",
+      messageParameters = Map(
+        "attribute" -> toSQLId(attribute.toImmutableArraySeq),
+        "method" -> method,
+        "scanClass" -> scanClass,
+        "relationOutput" -> toSQLType(relationOutput)),
+      cause = cause)
+  }
+
   def foundMultipleXMLDataSourceError(provider: String,
       sourceNames: Seq[String],
       externalSource: String): Throwable = {

sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala

@@ -20,26 +20,27 @@ package org.apache.spark.sql.execution.datasources.v2
 import java.util.{Collections, Optional, OptionalLong}
 
 import org.apache.spark.SparkException
+import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation, NamedRelation, TimeTravelSpec}
 import org.apache.spark.sql.catalyst.catalog.{CatalogColumnStat, CatalogStatistics}
-import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, AttributeReference, AttributeSet, Expression, SortOrder, V2ExpressionUtils}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, AttributeReference, AttributeSet, Expression, NamedExpression, SortOrder, V2ExpressionUtils}
 import org.apache.spark.sql.catalyst.plans.QueryPlan
 import org.apache.spark.sql.catalyst.plans.logical.{ColumnStat, ExposesMetadataColumns, Histogram, HistogramBin, LeafNode, LogicalPlan, Statistics}
 import org.apache.spark.sql.catalyst.plans.logical.statsEstimation.EstimationUtils
 import org.apache.spark.sql.catalyst.streaming.{StreamingSourceIdentifyingName, Unassigned}
 import org.apache.spark.sql.catalyst.trees.TreePattern.{DATA_SOURCE_V2_RELATION, DATA_SOURCE_V2_SCAN_RELATION, TreePattern}
-import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes
+import org.apache.spark.sql.catalyst.types.DataTypeUtils.{fromAttributes, toAttributes}
 import org.apache.spark.sql.catalyst.util.{removeInternalMetadata, truncatedString, CharVarcharUtils}
 import org.apache.spark.sql.connector.catalog.{CatalogPlugin, FunctionCatalog, Identifier, SupportsMetadataColumns, Table, TableCapability, TableCatalog, V2TableUtil}
 import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.CatalogHelper
 import org.apache.spark.sql.connector.expressions.{FieldReference, NamedReference}
 import org.apache.spark.sql.connector.read.{Scan, Statistics => V2Statistics, SupportsReportStatistics, SupportsRuntimeV2Filtering}
 import org.apache.spark.sql.connector.read.colstats.{ColumnStatistics, Histogram => V2Histogram, HistogramBin => V2HistogramBin}
 import org.apache.spark.sql.connector.read.streaming.{Offset, SparkDataStream}
+import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.internal.connector.{SupportsRuntimeCatalystFiltering, V2StatisticsUtils}
 import org.apache.spark.sql.types.{DataType, StructType}
 import org.apache.spark.sql.util.CaseInsensitiveStringMap
-import org.apache.spark.util.ArrayImplicits._
 import org.apache.spark.util.Utils
 
 /**
@@ -201,16 +202,23 @@ case class DataSourceV2ScanRelation(
    * Resolved attributes that the scan declares for runtime filtering via
    * [[SupportsRuntimeV2Filtering.filterAttributes]] or
    * [[SupportsRuntimeCatalystFiltering.filterAttributes]]. Empty when the scan
-   * implements neither interface or exposes no attributes.
+   * implements neither interface or exposes no attributes. Accessing this value also validates
+   * attributes returned by [[SupportsRuntimeCatalystFiltering.fullyPushedFilterAttributes]].
    */
   lazy val runtimeFilterAttrs: AttributeSet = {
     checkRuntimeFilteringInterfaces()
+    checkFullyPushedFilterAttrs()
     val filterAttrs = scan match {
       case s: SupportsRuntimeV2Filtering => s.filterAttributes
       case s: SupportsRuntimeCatalystFiltering => s.filterAttributes()
       case _ => Array.empty[NamedReference]
     }
-    resolveTopLevelFilterAttrs(filterAttrs)
+    resolveFilterAttrs(filterAttrs, "filterAttributes()")
+  }
+
+  private lazy val declaredFullyPushedRuntimeFilterAttrs: Array[NamedReference] = scan match {
+    case s: SupportsRuntimeCatalystFiltering => s.fullyPushedFilterAttributes()
+    case _ => Array.empty
   }
 
   /**
@@ -219,27 +227,34 @@ case class DataSourceV2ScanRelation(
    */
   lazy val fullyPushedRuntimeFilterAttrs: AttributeSet = {
     checkRuntimeFilteringInterfaces()
-    val filterAttrs = scan match {
-      case s: SupportsRuntimeCatalystFiltering => s.fullyPushedFilterAttributes()
-      case _ => Array.empty[NamedReference]
-    }
-    resolveTopLevelFilterAttrs(filterAttrs)
+    checkFullyPushedFilterAttrs()
+    resolveFilterAttrs(
+      declaredFullyPushedRuntimeFilterAttrs, "fullyPushedFilterAttributes()")
   }
 
   /**
-   * Resolves the given runtime-filter references against this relation's output. Both runtime
-   * filtering interfaces require each reference to be a top-level attribute of the read schema,
-   * so a nested reference is rejected.
+   * Resolves runtime-filter references against this relation's output.
+   *
+   * [[AttributeSet]] reduces nested references to their root attributes. This is sufficient for
+   * ordinary runtime-filter eligibility because Spark retains the post-scan predicate.
    */
-  private def resolveTopLevelFilterAttrs(filterAttrs: Array[NamedReference]): AttributeSet = {
-    filterAttrs.find(_.fieldNames.length > 1).foreach { ref =>
-      throw SparkException.internalError(
-        s"Runtime filter attribute '${ref.fieldNames.mkString(".")}' declared by " +
-        s"${scan.getClass.getName} must be a top-level attribute of the scan read schema, " +
-        "but it is a nested reference.")
+  private def resolveFilterAttrs(
+      filterAttrs: Array[NamedReference],
+      method: String): AttributeSet = {
+    val resolvedAttrs = filterAttrs.map { ref =>
+      try {
+        V2ExpressionUtils.resolveRef[NamedExpression](ref, this)
+      } catch {
+        case e: AnalysisException =>
+          throw QueryCompilationErrors.cannotResolveDataSourceRuntimeFilterAttributeError(
+            attribute = ref.fieldNames,
+            method = method,
+            scanClass = scan.getClass.getName,
+            relationOutput = fromAttributes(output),
+            cause = e)
+      }
     }
-    AttributeSet(V2ExpressionUtils.resolveRefs[Attribute](
-      filterAttrs.toImmutableArraySeq, this))
+    AttributeSet(resolvedAttrs)
   }
 
   override val nodePatterns: Seq[TreePattern] = Seq(DATA_SOURCE_V2_SCAN_RELATION)
@@ -290,12 +305,23 @@ case class DataSourceV2ScanRelation(
     Statistics(sizeInBytes = conf.defaultSizeInBytes)
   }
 
-  private def checkRuntimeFilteringInterfaces(): Unit = scan match {
-    case _: SupportsRuntimeV2Filtering with SupportsRuntimeCatalystFiltering =>
-      throw SparkException.internalError(
-        "A scan must not implement both SupportsRuntimeV2Filtering and " +
-        s"SupportsRuntimeCatalystFiltering, but ${scan.getClass.getName} implements both.")
-    case _ =>
+  private def checkRuntimeFilteringInterfaces(): Unit = {
+    scan match {
+      case _: SupportsRuntimeV2Filtering with SupportsRuntimeCatalystFiltering =>
+        throw SparkException.internalError(
+          "A scan must not implement both SupportsRuntimeV2Filtering and " +
+          s"SupportsRuntimeCatalystFiltering, but ${scan.getClass.getName} implements both.")
+      case _ =>
+    }
+  }
+
+  private def checkFullyPushedFilterAttrs(): Unit = {
+    declaredFullyPushedRuntimeFilterAttrs.find(_.fieldNames.length > 1).foreach { ref =>
+      throw QueryCompilationErrors.nestedDataSourceFullyPushedRuntimeFilterAttributeError(
+        attribute = ref.fieldNames,
+        scanClass = scan.getClass.getName,
+        relationOutput = fromAttributes(output))
+    }
   }
 
   override def doCanonicalize(): DataSourceV2ScanRelation = {

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala

@@ -41,9 +41,8 @@ trait SupportsRuntimeCatalystFiltering extends Scan {
    * Returns attributes this scan can be filtered by at runtime.
    *
    * Spark will call [[filter]] if it can derive a runtime filter for any of these attributes.
-   * Each reference must be a top-level attribute present in [[Scan.readSchema]]. Nested
-   * references are rejected, and attributes pruned out of the read schema fail to resolve, when
-   * Spark builds the scan relation.
+   * Each reference must resolve against the scan relation output when Spark builds it. Attributes
+   * pruned out of [[Scan.readSchema]] fail to resolve.
    */
   def filterAttributes(): Array[NamedReference]
 
@@ -65,7 +64,10 @@ trait SupportsRuntimeCatalystFiltering extends Scan {
    *
    * Each reference must be a top-level attribute present in [[Scan.readSchema]]. Nested
    * references are rejected, and attributes pruned out of the read schema fail to resolve, when
-   * Spark builds the scan relation.
+   * Spark builds the scan relation. Spark cannot currently represent an individual fully pushed
+   * nested path. A scan must not return the root struct as a substitute unless it can fully
+   * evaluate predicates over every nested field, since Spark would remove their post-scan
+   * evaluation as well.
    */
   def fullyPushedFilterAttributes(): Array[NamedReference] = Array.empty
 
@@ -76,9 +78,9 @@ trait SupportsRuntimeCatalystFiltering extends Scan {
    * Implementations may use the expressions to prune initially planned
    * [[org.apache.spark.sql.connector.read.InputPartition]]s.
    *
-   * An expression may access nested fields of an attribute returned by [[filterAttributes]], as
-   * that attribute is required to be top-level. The scan is responsible for matching such
-   * accesses against its own partition layout.
+   * Spark tracks runtime-filter eligibility by root attribute. If [[filterAttributes]] returns a
+   * nested reference, an expression may access another nested field under the same root. The scan
+   * must match each access against its own partition layout and use only expressions it can apply.
    *
    * Spark may call this method more than once for the same scan instance: a plan can hold several
    * scan nodes sharing one scan (e.g. the two branches of a group-based UPDATE), and each pushes

sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala

@@ -770,7 +770,9 @@ abstract class InMemoryBaseTable(
     private def partitionAttributes: Seq[(Seq[String], AttributeReference)] = {
       partitioning.flatMap(_.references()).flatMap { ref =>
         val path = ref.fieldNames.toImmutableArraySeq
-        readSchema.findNestedField(path).orElse(tableSchema.findNestedField(path)).map {
+        val resolver = SQLConf.get.resolver
+        readSchema.findNestedField(path, resolver = resolver)
+          .orElse(tableSchema.findNestedField(path, resolver = resolver)).map {
           case (_, f) =>
             path -> AttributeReference(ref.fieldNames.mkString("."), f.dataType, f.nullable)()
         }
@@ -827,9 +829,9 @@ abstract class InMemoryBaseTable(
     var pushedFilters: Array[Filter] = Array.empty
 
     override def filterAttributes(): Array[NamedReference] = {
-      val scanFields = readSchema.fields.map(_.name).toSet
       partitioning.flatMap(_.references)
-        .filter(ref => scanFields.contains(ref.fieldNames.mkString(".")))
+        .filter(ref => readSchema.findNestedField(
+          ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined)
     }
 
     override def filter(filters: Array[Filter]): Unit = {

sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalystRuntimeFilterTable.scala

@@ -23,8 +23,9 @@ import InMemoryCatalystRuntimeFilterTable._
 
 import org.apache.spark.sql.connector.catalog.constraints.Constraint
 import org.apache.spark.sql.connector.distributions.{Distribution, Distributions}
-import org.apache.spark.sql.connector.expressions.{FieldReference, NamedReference, SortOrder, Transform}
+import org.apache.spark.sql.connector.expressions.{NamedReference, SortOrder, Transform}
 import org.apache.spark.sql.connector.read.{InputPartition, Scan, ScanBuilder}
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.StructType
 import org.apache.spark.sql.util.CaseInsensitiveStringMap
 import org.apache.spark.util.ArrayImplicits._
@@ -90,29 +91,23 @@ class InMemoryCatalystRuntimeFilterTable(
       .map(_.split(",").map(_.trim).toSet)
       .getOrElse(Set.empty)
 
-    /**
-     * The partition columns, each named by the top level read schema column it lives under, the
-     * form both interface methods require. Columns pruned out of the read schema are dropped,
-     * since neither method may name one. Examples:
-     *   - `PARTITIONED BY (part)` -> `"part"`
-     *   - `PARTITIONED BY (s.nested)` -> `"s"`, the struct column holding the partition field
-     */
-    private def partitionAttrNames: Array[String] = {
-      val scanFields = readSchema.fields.map(_.name).toSet
-      partitioning.flatMap(_.references()).map(_.fieldNames.head).distinct
-        .filter(scanFields.contains)
+    /** Partition source columns that are present in the scan read schema. */
+    private def partitionAttrs: Array[NamedReference] = {
+      partitioning.flatMap(_.references()).distinct
+        .filter(ref => readSchema.findNestedField(
+          ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined)
     }
 
     override def filterAttributes(): Array[NamedReference] = {
-      partitionAttrNames
-        .filter(name => restrictedFilterAttrs.forall(_.contains(name)))
-        .map(FieldReference.column)
+      partitionAttrs.filter { ref =>
+        restrictedFilterAttrs.forall(_.contains(ref.fieldNames.mkString(".")))
+      }
     }
 
     // Not intersected with `filterAttributes()`, so a table can declare a fully pushed attribute
     // that is not a filter attribute, a combination the interface forbids.
     override def fullyPushedFilterAttributes(): Array[NamedReference] = {
-      partitionAttrNames.filter(fullyPushedFilterAttrs.contains).map(FieldReference.column)
+      partitionAttrs.filter(ref => fullyPushedFilterAttrs.contains(ref.fieldNames.mkString(".")))
     }
   }
 }