From 4ac63be92cf191f878b6a3b9181ac272b46089cc Mon Sep 17 00:00:00 2001
From: Szehon Ho <szehon.apache@gmail.com>
Date: Fri, 28 Aug 2026 01:28:54 +0000
Subject: [PATCH 1/6] [SPARK-59068][SQL] Restore support for nested runtime
 filter attributes

---
 .../read/SupportsRuntimeFiltering.java        |   4 -
 .../read/SupportsRuntimeV2Filtering.java      |   4 -
 .../datasources/v2/DataSourceV2Relation.scala |  18 +--
 .../SupportsRuntimeCatalystFiltering.scala    |  12 +-
 .../InMemoryCatalystRuntimeFilterTable.scala  |  23 ++--
 .../catalog/InMemoryTableWithV2Filter.scala   |   3 +-
 ...taSourceV2CatalystRuntimeFilterSuite.scala | 105 +++---------------
 .../sql/connector/DataSourceV2SQLSuite.scala  |  40 ++++++-
 8 files changed, 67 insertions(+), 142 deletions(-)

diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeFiltering.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeFiltering.java
index 067202a362705..927d4a53e22fc 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeFiltering.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeFiltering.java
@@ -38,10 +38,6 @@ public interface SupportsRuntimeFiltering extends SupportsRuntimeV2Filtering {
    * <p>
    * 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.
    */
   NamedReference[] filterAttributes();
 
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeV2Filtering.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeV2Filtering.java
index 6b286f041b01e..63d43de167e17 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeV2Filtering.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeV2Filtering.java
@@ -51,10 +51,6 @@ public interface SupportsRuntimeV2Filtering extends Scan {
    * <p>
    * 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.
    */
   NamedReference[] filterAttributes();
 
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
index 7b2cfacb652fe..28a79aa98c7ca 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
@@ -210,7 +210,7 @@ case class DataSourceV2ScanRelation(
       case s: SupportsRuntimeCatalystFiltering => s.filterAttributes()
       case _ => Array.empty[NamedReference]
     }
-    resolveTopLevelFilterAttrs(filterAttrs)
+    resolveFilterAttrs(filterAttrs)
   }
 
   /**
@@ -223,21 +223,11 @@ case class DataSourceV2ScanRelation(
       case s: SupportsRuntimeCatalystFiltering => s.fullyPushedFilterAttributes()
       case _ => Array.empty[NamedReference]
     }
-    resolveTopLevelFilterAttrs(filterAttrs)
+    resolveFilterAttrs(filterAttrs)
   }
 
-  /**
-   * 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.
-   */
-  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.")
-    }
+  /** Resolves the given runtime-filter references against this relation's output. */
+  private def resolveFilterAttrs(filterAttrs: Array[NamedReference]): AttributeSet = {
     AttributeSet(V2ExpressionUtils.resolveRefs[Attribute](
       filterAttrs.toImmutableArraySeq, this))
   }
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala
index eaa4857f1abe3..f1af2fbea32de 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala
@@ -41,9 +41,6 @@ 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.
    */
   def filterAttributes(): Array[NamedReference]
 
@@ -62,10 +59,6 @@ trait SupportsRuntimeCatalystFiltering extends Scan {
    * an ANSI cast or overflow error, or on a nested access that the scan matches differently
    * against its partition layout. Declare an attribute only when the scan can evaluate every
    * predicate over it.
-   *
-   * 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.
    */
   def fullyPushedFilterAttributes(): Array[NamedReference] = Array.empty
 
@@ -76,9 +69,8 @@ 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.
+   * An expression may access nested fields. The scan is responsible for matching such accesses
+   * against its own partition layout.
    *
    * 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
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalystRuntimeFilterTable.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalystRuntimeFilterTable.scala
index 369e8123f1d7e..51757a946b46c 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalystRuntimeFilterTable.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalystRuntimeFilterTable.scala
@@ -23,7 +23,7 @@ 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.types.StructType
 import org.apache.spark.sql.util.CaseInsensitiveStringMap
@@ -90,29 +90,20 @@ 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).isDefined)
     }
 
     override def filterAttributes(): Array[NamedReference] = {
-      partitionAttrNames
-        .filter(name => restrictedFilterAttrs.forall(_.contains(name)))
-        .map(FieldReference.column)
+      partitionAttrs.filter(ref => restrictedFilterAttrs.forall(_.contains(ref.toString)))
     }
 
     // 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.toString))
     }
   }
 }
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala
index b59a81cc29092..7739b0b5bb843 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala
@@ -77,9 +77,8 @@ class InMemoryTableWithV2Filter(
     extends BatchScanBaseClass(_data, readSchema, tableSchema) with SupportsRuntimeV2Filtering {
 
     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).isDefined)
     }
 
     override def filter(filters: Array[Predicate]): Unit = {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
index c0a1faa032635..6ae090aa3a06c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
@@ -18,7 +18,7 @@
 package org.apache.spark.sql.connector
 
 import org.apache.spark.{SparkConf, SparkException}
-import org.apache.spark.sql.{AnalysisException, DataFrame, Row}
+import org.apache.spark.sql.{DataFrame, Row}
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{Add, AttributeReference, DynamicPruning, DynamicPruningExpression, EqualTo, Expression, GetStructField, GreaterThan, Literal, RLike}
 import org.apache.spark.sql.catalyst.plans.physical.KeyedPartitioning
@@ -244,34 +244,35 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession {
     }
   }
 
-  test("DPP filter on a nested partition field -> pushed with the nested access intact") {
+  test("DPP filter on a nested partition source -> pushed with the nested access intact") {
     val fact = s"$catalogName.fact_nested_dpp"
     val dim = s"$catalogName.dim_nested_dpp"
     withTable(fact, dim) {
-      sql(s"CREATE TABLE $fact (id INT, s STRUCT<part: INT>) USING $v2Source " +
-        "PARTITIONED BY (s.part)")
-      for (i <- 0 until 5) {
-        sql(s"INSERT INTO $fact VALUES ($i, named_struct('part', $i))")
-      }
-      sql(s"CREATE TABLE $dim (dim_id INT, dim_val STRING) USING $v2Source")
-      sql(s"INSERT INTO $dim VALUES (2, 'two')")
+      sql(s"CREATE TABLE $fact " +
+        "(id INT, derives STRUCT<toStr: STRING, other: STRING>) USING " +
+        s"$v2Source PARTITIONED BY (derives.toStr)")
+      sql(s"INSERT INTO $fact VALUES " +
+        "(1, named_struct('toStr', 'AA', 'other', 'a')), " +
+        "(2, named_struct('toStr', 'BB', 'other', 'b')), " +
+        "(3, named_struct('toStr', 'CC', 'other', 'c'))")
+      sql(s"CREATE TABLE $dim (value STRING, selected INT) USING $v2Source")
+      sql(s"INSERT INTO $dim VALUES ('AA', 0), ('BB', 1)")
 
       withDPPConf {
         val df = sql(
           s"""SELECT f.id FROM $fact f JOIN $dim d
-             |ON f.s.part = d.dim_id WHERE d.dim_val = 'two'""".stripMargin)
+             |ON f.derives.toStr = d.value WHERE d.selected = 1""".stripMargin)
         checkAnswer(df, Row(2))
 
         assertDPPRuntimeFilters(df)
-        // The scan only reports `s`, the struct holding the partition field, but the predicate it
-        // receives keeps the nested access, so it can still tell which partition to keep.
+        // The scan reports `derives.toStr`, and the predicate it receives keeps that nested access.
         val pushed = getPushedCatalystPredicates(df)
         assert(pushed.size === 1, s"expected a single pushed predicate, got $pushed")
         assert(pushed.head.exists(_.isInstanceOf[GetStructField]),
           s"expected the pushed predicate to keep the nested access, got ${pushed.head}")
 
         val batchScan = collectBatchScan(df)
-        assert(batchScan.inputPartitions.size === 5)
+        assert(batchScan.inputPartitions.size === 3)
         assert(batchScan.filteredPartitions.flatten.size === 1,
           s"expected 1 partition after pruning, got ${batchScan.filteredPartitions.flatten.size}")
       }
@@ -414,46 +415,6 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession {
     }
   }
 
-  test("filterAttributes that is not a top-level scan attribute") {
-    val tbl = s"$catalogName.tbl_unresolvable_attr"
-    withTable(tbl) {
-      sql(s"CREATE TABLE $tbl (id INT, part INT, s STRUCT<tz: STRING>) USING $v2Source " +
-        "PARTITIONED BY (part)")
-      sql(s"INSERT INTO $tbl VALUES (1, 1, named_struct('tz', 'a'))")
-
-      val scanRelation = sql(s"SELECT * FROM $tbl").queryExecution.optimizedPlan.collectFirst {
-        case r: DataSourceV2ScanRelation => r
-      }.getOrElse(fail("Expected a DataSourceV2ScanRelation"))
-
-      // An attribute the read schema does not carry, such as one pruned out of the projection.
-      val missing = intercept[AnalysisException] {
-        scanRelation.copy(scan = new MissingFilterAttributeScan).runtimeFilterAttrs
-      }
-      checkError(
-        exception = missing,
-        condition = "_LEGACY_ERROR_TEMP_1137",
-        parameters = Map("name" -> "missing", "outputStr" -> "id,part,s"))
-
-      // A nested reference is rejected up front, since `filterAttributes()` must return
-      // top-level read-schema attributes. This holds even over an int column that could never
-      // carry a nested field.
-      val nested = intercept[SparkException] {
-        scanRelation.copy(scan = new NestedFilterAttributeScan).runtimeFilterAttrs
-      }
-      assert(nested.getMessage.contains("must be a top-level attribute"),
-        s"expected the nested reference to be rejected, got ${nested.getMessage}")
-
-      // Over a struct it is rejected the same way, rather than widening to the struct column:
-      // accepting `s.tz` would make runtime filters over every field of `s` eligible, not just
-      // `s.tz`.
-      val struct = intercept[SparkException] {
-        scanRelation.copy(scan = new StructNestedFilterAttributeScan).runtimeFilterAttrs
-      }
-      assert(struct.getMessage.contains("must be a top-level attribute"),
-        s"expected the nested struct reference to be rejected, got ${struct.getMessage}")
-    }
-  }
-
   test("no runtime filter -> filter() is never called") {
     val tbl = s"$catalogName.tbl5"
     withTable(tbl) {
@@ -637,44 +598,6 @@ private class BothRuntimeFilteringInterfacesScan
   override def filter(expressions: Array[Expression]): Unit = {}
 }
 
-/** A scan declaring a filter attribute the read schema does not carry. */
-private class MissingFilterAttributeScan extends Scan with SupportsRuntimeCatalystFiltering {
-
-  override def readSchema(): StructType = new StructType().add("part", IntegerType)
-
-  override def filterAttributes(): Array[NamedReference] = Array(FieldReference("missing"))
-
-  override def filter(expressions: Array[Expression]): Unit = {}
-}
-
-/**
- * A scan breaking the rule that a filter attribute must be a top level read schema column: it
- * reports `part.nested` over the int column `part`, so it is rejected as a nested reference.
- */
-private class NestedFilterAttributeScan extends Scan with SupportsRuntimeCatalystFiltering {
-
-  override def readSchema(): StructType = new StructType().add("part", IntegerType)
-
-  override def filterAttributes(): Array[NamedReference] =
-    Array(FieldReference(Seq("part", "nested")))
-
-  override def filter(expressions: Array[Expression]): Unit = {}
-}
-
-/**
- * A scan breaking the same rule over a struct column: it reports `s.tz` where `s` is a struct.
- * The nested reference is rejected rather than widening to the struct column `s`.
- */
-private class StructNestedFilterAttributeScan extends Scan with SupportsRuntimeCatalystFiltering {
-
-  override def readSchema(): StructType =
-    new StructType().add("s", new StructType().add("tz", StringType))
-
-  override def filterAttributes(): Array[NamedReference] = Array(FieldReference(Seq("s", "tz")))
-
-  override def filter(expressions: Array[Expression]): Unit = {}
-}
-
 private case class KeyedInputPartition(key: Int) extends InputPartition with HasPartitionKey {
   override def partitionKey(): InternalRow = InternalRow(key)
 }
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
index 76c7902188ed3..68b1bccb8ebf9 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
@@ -5440,11 +5440,49 @@ class DataSourceV2SQLSuiteV1Filter
 }
 
 class DataSourceV2SQLSuiteV2Filter extends DataSourceV2SQLSuite {
-  import org.apache.spark.sql.catalyst.expressions.DynamicPruning
+  import org.apache.spark.sql.catalyst.expressions.{DynamicPruning, GetStructField}
   import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
 
   override protected val catalogAndNamespace = "testv2filter.ns1.ns2."
 
+  test("nested partition source column receives a DPP runtime filter") {
+    val fact = s"${catalogAndNamespace}fact_nested_runtime_filter"
+    val dim = s"${catalogAndNamespace}dim_nested_runtime_filter"
+    withTable(fact, dim) {
+      sql(s"CREATE TABLE $fact " +
+        "(id INT, derives STRUCT<toStr: STRING, other: STRING>) USING " +
+        s"$v2Format PARTITIONED BY (derives.toStr)")
+      sql(s"INSERT INTO $fact VALUES " +
+        "(1, named_struct('toStr', 'AA', 'other', 'a')), " +
+        "(2, named_struct('toStr', 'BB', 'other', 'b')), " +
+        "(3, named_struct('toStr', 'CC', 'other', 'c'))")
+      sql(s"CREATE TABLE $dim (value STRING, selected INT) USING $v2Format")
+      sql(s"INSERT INTO $dim VALUES ('AA', 0), ('BB', 1)")
+
+      withSQLConf(
+        SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+        SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+        SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "10") {
+        val df = sql(
+          s"""SELECT f.id FROM $fact f JOIN $dim d
+             |ON f.derives.toStr = d.value WHERE d.selected = 1""".stripMargin)
+        checkAnswer(df, Row(2))
+
+        val batchScans = collect(df.queryExecution.executedPlan) {
+          case b: BatchScanExec if b.runtimeFilters.nonEmpty => b
+        }
+        assert(batchScans.nonEmpty,
+          s"expected a scan with runtime filters, got ${df.queryExecution}")
+        val batchScan = batchScans.head
+        assert(batchScan.runtimeFilters.exists(_.exists(_.isInstanceOf[GetStructField])),
+          s"expected a runtime filter on derives.toStr, got ${batchScan.runtimeFilters}")
+        assert(batchScan.partitions.size === 3)
+        assert(batchScan.filteredPartitions.flatten.size === 1,
+          s"expected 1 partition after pruning, got ${batchScan.filteredPartitions.flatten.size}")
+      }
+    }
+  }
+
   test("SPARK-56467: scalar subquery filters on partition columns are pushed into runtimeFilters") {
     val tbl = s"${catalogAndNamespace}tbl"
     val dim = s"${catalogAndNamespace}dim"

From 6019cecc278916d4f0bd7677e77c84e0f32c813a Mon Sep 17 00:00:00 2001
From: Szehon Ho <szehon.apache@gmail.com>
Date: Fri, 28 Aug 2026 11:49:37 -0700
Subject: [PATCH 2/6] [SPARK-59068][SQL] Reject fully pushed nested runtime
 filter attributes

---
 .../datasources/v2/DataSourceV2Relation.scala | 15 ++++
 .../SupportsRuntimeCatalystFiltering.scala    |  4 +
 .../connector/catalog/InMemoryBaseTable.scala |  3 +-
 ...taSourceV2CatalystRuntimeFilterSuite.scala | 84 ++++++++++++++++++-
 .../sql/connector/DataSourceV2SQLSuite.scala  | 82 +++++++++---------
 5 files changed, 140 insertions(+), 48 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
index 28a79aa98c7ca..471b8f7b2ce1b 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
@@ -223,6 +223,21 @@ case class DataSourceV2ScanRelation(
       case s: SupportsRuntimeCatalystFiltering => s.fullyPushedFilterAttributes()
       case _ => Array.empty[NamedReference]
     }
+    resolveTopLevelFilterAttrs(filterAttrs)
+  }
+
+  /**
+   * Resolves fully pushed runtime-filter references against this relation's output. Nested
+   * references are rejected because [[AttributeSet]] keeps only their root attribute, which could
+   * make a predicate on a sibling field appear fully pushed and remove its post-scan evaluation.
+   */
+  private def resolveTopLevelFilterAttrs(filterAttrs: Array[NamedReference]): AttributeSet = {
+    filterAttrs.find(_.fieldNames.length > 1).foreach { ref =>
+      throw SparkException.internalError(
+        s"Fully pushed 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.")
+    }
     resolveFilterAttrs(filterAttrs)
   }
 
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala
index f1af2fbea32de..3b9e7562792c7 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala
@@ -59,6 +59,10 @@ trait SupportsRuntimeCatalystFiltering extends Scan {
    * an ANSI cast or overflow error, or on a nested access that the scan matches differently
    * against its partition layout. Declare an attribute only when the scan can evaluate every
    * predicate over it.
+   *
+   * 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.
    */
   def fullyPushedFilterAttributes(): Array[NamedReference] = Array.empty
 
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala
index d9540b8fff854..4c8ba2c7da714 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala
@@ -827,9 +827,8 @@ 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).isDefined)
     }
 
     override def filter(filters: Array[Filter]): Unit = {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
index 6ae090aa3a06c..6d2f76e602816 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
@@ -18,7 +18,7 @@
 package org.apache.spark.sql.connector
 
 import org.apache.spark.{SparkConf, SparkException}
-import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.{AnalysisException, DataFrame, Row}
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{Add, AttributeReference, DynamicPruning, DynamicPruningExpression, EqualTo, Expression, GetStructField, GreaterThan, Literal, RLike}
 import org.apache.spark.sql.catalyst.plans.physical.KeyedPartitioning
@@ -116,6 +116,49 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession {
     }
   }
 
+  test("nested fully pushed filter attribute -> rejected") {
+    val tbl = s"$catalogName.tbl_nested_fully_pushed"
+    withTable(tbl) {
+      sql(s"CREATE TABLE $tbl (id INT, s STRUCT<part: INT, other: INT>) USING $v2Source " +
+        "PARTITIONED BY (s.part) " +
+        "TBLPROPERTIES('fully-pushed-filter-attributes' = 's.part')")
+
+      val scanRelation = sql(s"SELECT * FROM $tbl").queryExecution.optimizedPlan.collectFirst {
+        case r: DataSourceV2ScanRelation => r
+      }.getOrElse(fail("Expected a DataSourceV2ScanRelation"))
+
+      // Ordinary nested filter attributes remain supported for partition pruning.
+      val runtimeFilterAttrs = scanRelation.runtimeFilterAttrs
+      assert(runtimeFilterAttrs.map(_.name).toSeq === Seq("s"))
+
+      // Fully pushed eligibility cannot use the root-only AttributeSet representation: doing so
+      // would make a predicate on `s.other` appear covered by the declaration of `s.part`.
+      val e = intercept[SparkException] {
+        scanRelation.fullyPushedRuntimeFilterAttrs
+      }
+      assert(e.getMessage.contains("must be a top-level attribute"),
+        s"expected the nested fully pushed reference to be rejected, got ${e.getMessage}")
+    }
+  }
+
+  test("missing filter attribute -> rejected") {
+    val tbl = s"$catalogName.tbl_missing_filter_attr"
+    withTable(tbl) {
+      sql(s"CREATE TABLE $tbl (id INT, part INT) USING $v2Source PARTITIONED BY (part)")
+
+      val scanRelation = sql(s"SELECT * FROM $tbl").queryExecution.optimizedPlan.collectFirst {
+        case r: DataSourceV2ScanRelation => r
+      }.getOrElse(fail("Expected a DataSourceV2ScanRelation"))
+      val e = intercept[AnalysisException] {
+        scanRelation.copy(scan = new MissingFilterAttributeScan).runtimeFilterAttrs
+      }
+      checkError(
+        exception = e,
+        condition = "_LEGACY_ERROR_TEMP_1137",
+        parameters = Map("name" -> "missing", "outputStr" -> "id,part"))
+    }
+  }
+
   test("non-deterministic predicate on fully pushed attributes -> evaluated after the scan") {
     val tbl = s"$catalogName.tbl_nondeterministic"
     val dim = s"$catalogName.dim_nondeterministic"
@@ -279,6 +322,29 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession {
     }
   }
 
+  test("sibling of a nested filter attribute remains evaluated after the scan") {
+    val fact = s"$catalogName.fact_nested_sibling"
+    val dim = s"$catalogName.dim_nested_sibling"
+    withTable(fact, dim) {
+      sql(s"CREATE TABLE $fact (id INT, s STRUCT<part: INT, other: INT>) USING $v2Source " +
+        "PARTITIONED BY (s.part)")
+      sql(s"INSERT INTO $fact VALUES " +
+        "(1, named_struct('part', 1, 'other', 10)), " +
+        "(2, named_struct('part', 1, 'other', 20)), " +
+        "(3, named_struct('part', 2, 'other', 30))")
+      sql(s"CREATE TABLE $dim (value INT) USING $v2Source")
+      sql(s"INSERT INTO $dim VALUES (10)")
+
+      val df = sql(s"SELECT id FROM $fact WHERE s.other = (SELECT max(value) FROM $dim)")
+      checkAnswer(df, Row(1))
+
+      // A nested reference currently contributes its root attribute to eligibility, so `s.other`
+      // may be routed to a scan that advertises `s.part`. It must remain above the scan unless
+      // eligibility becomes path-aware.
+      assertScalarSubqueryEvaluatedAfterScan(df, expected = true)
+    }
+  }
+
   test("scan implementing both runtime filtering interfaces -> rejected") {
     val tbl = s"$catalogName.tbl_both_interfaces"
     withTable(tbl) {
@@ -365,9 +431,9 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession {
       sql(s"CREATE TABLE $dim (val STRING) USING $v2Source")
       sql(s"INSERT INTO $dim VALUES ('tz1')")
 
-      // The scan declares the top-level struct column `s` as its filter attribute, so the
-      // predicate qualifies for pushdown even though it reaches into `s.tz`. Matching the nested
-      // access against the partition layout is left to the scan, which this fixture does.
+      // The scan declares the nested partition source `s.tz` as its filter attribute. The
+      // predicate arrives with the nested access intact, and matching it against the partition
+      // layout is left to the scan, which this fixture does.
       val df = sql(s"SELECT * FROM $tbl WHERE s.tz = (SELECT max(val) FROM $dim)")
       checkAnswer(df, Row(1, Row("tz1")))
 
@@ -598,6 +664,16 @@ private class BothRuntimeFilteringInterfacesScan
   override def filter(expressions: Array[Expression]): Unit = {}
 }
 
+/** A scan declaring a filter attribute that its read schema does not contain. */
+private class MissingFilterAttributeScan extends SupportsRuntimeCatalystFiltering {
+
+  override def readSchema(): StructType = new StructType().add("part", IntegerType)
+
+  override def filterAttributes(): Array[NamedReference] = Array(FieldReference("missing"))
+
+  override def filter(expressions: Array[Expression]): Unit = {}
+}
+
 private case class KeyedInputPartition(key: Int) extends InputPartition with HasPartitionKey {
   override def partitionKey(): InternalRow = InternalRow(key)
 }
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
index 68b1bccb8ebf9..46c42490b4f73 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
@@ -32,6 +32,7 @@ import org.apache.spark.sql.catalyst.{InternalRow, QualifiedTableName, TableIden
 import org.apache.spark.sql.catalyst.CurrentUserContext.CURRENT_USER
 import org.apache.spark.sql.catalyst.analysis.{CannotReplaceMissingTableException, NoSuchNamespaceException, TableAlreadyExistsException}
 import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType, CatalogUtils}
+import org.apache.spark.sql.catalyst.expressions.{DynamicPruning, GetStructField}
 import org.apache.spark.sql.catalyst.parser.ParseException
 import org.apache.spark.sql.catalyst.plans.logical.ColumnStat
 import org.apache.spark.sql.catalyst.statsEstimation.StatsEstimationTestBase
@@ -45,7 +46,7 @@ import org.apache.spark.sql.execution.FilterExec
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
 import org.apache.spark.sql.execution.columnar.InMemoryRelation
 import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelationWithTable}
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
+import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, DataSourceV2ScanRelation}
 import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
 import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
 import org.apache.spark.sql.internal.SQLConf.{PARTITION_OVERWRITE_MODE, PartitionOverwriteMode, V2_SESSION_CATALOG_IMPLEMENTATION}
@@ -144,6 +145,44 @@ abstract class DataSourceV2SQLSuite
     val relations = explain.split("\n").filter(_.contains("RelationV2"))
     assert(relations.nonEmpty && relations.forall(line => relationPattern.matches(line.trim)))
   }
+
+  test("nested partition source column receives a DPP runtime filter") {
+    val fact = s"${catalogAndNamespace}fact_nested_runtime_filter"
+    val dim = s"${catalogAndNamespace}dim_nested_runtime_filter"
+    withTable(fact, dim) {
+      sql(s"CREATE TABLE $fact " +
+        "(id INT, derives STRUCT<toStr: STRING, other: STRING>) USING " +
+        s"$v2Format PARTITIONED BY (derives.toStr)")
+      sql(s"INSERT INTO $fact VALUES " +
+        "(1, named_struct('toStr', 'AA', 'other', 'a')), " +
+        "(2, named_struct('toStr', 'BB', 'other', 'b')), " +
+        "(3, named_struct('toStr', 'CC', 'other', 'c'))")
+      sql(s"CREATE TABLE $dim (value STRING, selected INT) USING $v2Format")
+      sql(s"INSERT INTO $dim VALUES ('AA', 0), ('BB', 1)")
+
+      withSQLConf(
+        SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+        SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+        SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "10") {
+        val df = sql(
+          s"""SELECT f.id FROM $fact f JOIN $dim d
+             |ON f.derives.toStr = d.value WHERE d.selected = 1""".stripMargin)
+        checkAnswer(df, Row(2))
+
+        val batchScans = collect(df.queryExecution.executedPlan) {
+          case b: BatchScanExec if b.runtimeFilters.nonEmpty => b
+        }
+        assert(batchScans.nonEmpty,
+          s"expected a scan with runtime filters, got ${df.queryExecution}")
+        val batchScan = batchScans.head
+        assert(batchScan.runtimeFilters.exists(_.exists(_.isInstanceOf[GetStructField])),
+          s"expected a runtime filter on derives.toStr, got ${batchScan.runtimeFilters}")
+        assert(batchScan.partitions.size === 3)
+        assert(batchScan.filteredPartitions.flatten.size === 1,
+          s"expected 1 partition after pruning, got ${batchScan.filteredPartitions.flatten.size}")
+      }
+    }
+  }
 }
 
 class DataSourceV2SQLSuiteV1Filter
@@ -5440,49 +5479,8 @@ class DataSourceV2SQLSuiteV1Filter
 }
 
 class DataSourceV2SQLSuiteV2Filter extends DataSourceV2SQLSuite {
-  import org.apache.spark.sql.catalyst.expressions.{DynamicPruning, GetStructField}
-  import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
-
   override protected val catalogAndNamespace = "testv2filter.ns1.ns2."
 
-  test("nested partition source column receives a DPP runtime filter") {
-    val fact = s"${catalogAndNamespace}fact_nested_runtime_filter"
-    val dim = s"${catalogAndNamespace}dim_nested_runtime_filter"
-    withTable(fact, dim) {
-      sql(s"CREATE TABLE $fact " +
-        "(id INT, derives STRUCT<toStr: STRING, other: STRING>) USING " +
-        s"$v2Format PARTITIONED BY (derives.toStr)")
-      sql(s"INSERT INTO $fact VALUES " +
-        "(1, named_struct('toStr', 'AA', 'other', 'a')), " +
-        "(2, named_struct('toStr', 'BB', 'other', 'b')), " +
-        "(3, named_struct('toStr', 'CC', 'other', 'c'))")
-      sql(s"CREATE TABLE $dim (value STRING, selected INT) USING $v2Format")
-      sql(s"INSERT INTO $dim VALUES ('AA', 0), ('BB', 1)")
-
-      withSQLConf(
-        SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
-        SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
-        SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "10") {
-        val df = sql(
-          s"""SELECT f.id FROM $fact f JOIN $dim d
-             |ON f.derives.toStr = d.value WHERE d.selected = 1""".stripMargin)
-        checkAnswer(df, Row(2))
-
-        val batchScans = collect(df.queryExecution.executedPlan) {
-          case b: BatchScanExec if b.runtimeFilters.nonEmpty => b
-        }
-        assert(batchScans.nonEmpty,
-          s"expected a scan with runtime filters, got ${df.queryExecution}")
-        val batchScan = batchScans.head
-        assert(batchScan.runtimeFilters.exists(_.exists(_.isInstanceOf[GetStructField])),
-          s"expected a runtime filter on derives.toStr, got ${batchScan.runtimeFilters}")
-        assert(batchScan.partitions.size === 3)
-        assert(batchScan.filteredPartitions.flatten.size === 1,
-          s"expected 1 partition after pruning, got ${batchScan.filteredPartitions.flatten.size}")
-      }
-    }
-  }
-
   test("SPARK-56467: scalar subquery filters on partition columns are pushed into runtimeFilters") {
     val tbl = s"${catalogAndNamespace}tbl"
     val dim = s"${catalogAndNamespace}dim"

From 467323b126c99c47f8112593b28b2a5c1572073f Mon Sep 17 00:00:00 2001
From: Szehon Ho <szehon.apache@gmail.com>
Date: Fri, 28 Aug 2026 15:40:43 -0700
Subject: [PATCH 3/6] [SPARK-59068][SQL] Address nested runtime filtering
 review comments

---
 .../read/SupportsRuntimeFiltering.java        |  7 +++
 .../read/SupportsRuntimeV2Filtering.java      |  8 +++
 .../datasources/v2/DataSourceV2Relation.scala | 55 ++++++++++---------
 .../SupportsRuntimeCatalystFiltering.scala    |  7 ++-
 .../connector/catalog/InMemoryBaseTable.scala |  7 ++-
 .../InMemoryCatalystRuntimeFilterTable.scala  |  4 +-
 .../InMemoryRowLevelOperationTable.scala      |  5 +-
 .../catalog/InMemoryTableWithV2Filter.scala   |  4 +-
 ...wLevelOperationRuntimeGroupFiltering.scala | 30 +++++++---
 ...taSourceV2CatalystRuntimeFilterSuite.scala | 43 +++++++++++----
 ...lOperationCatalystRuntimeFilterSuite.scala | 31 +++++++++++
 ...rationCatalystRuntimeFilterSuiteBase.scala |  8 ++-
 .../RowLevelOperationSuiteBase.scala          |  9 +++
 13 files changed, 160 insertions(+), 58 deletions(-)

diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeFiltering.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeFiltering.java
index 927d4a53e22fc..2feee94326baf 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeFiltering.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeFiltering.java
@@ -38,6 +38,9 @@ public interface SupportsRuntimeFiltering extends SupportsRuntimeV2Filtering {
    * <p>
    * Spark will call {@link #filter(Filter[])} if it can derive a runtime
    * predicate for any of the filter attributes.
+   * <p>
+   * 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();
 
@@ -47,6 +50,10 @@ 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.
+   * <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
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeV2Filtering.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeV2Filtering.java
index 63d43de167e17..671a822096bbe 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeV2Filtering.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeV2Filtering.java
@@ -51,6 +51,9 @@ public interface SupportsRuntimeV2Filtering extends Scan {
    * <p>
    * Spark will call {@link #filter(Predicate[])} if it can derive a runtime
    * predicate for any of the filter attributes.
+   * <p>
+   * 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();
 
@@ -60,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
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
index 471b8f7b2ce1b..9de6e0d528acb 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
@@ -22,7 +22,7 @@ import java.util.{Collections, Optional, OptionalLong}
 import org.apache.spark.SparkException
 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
@@ -213,37 +213,28 @@ case class DataSourceV2ScanRelation(
     resolveFilterAttrs(filterAttrs)
   }
 
+  private lazy val declaredFullyPushedRuntimeFilterAttrs: Array[NamedReference] = scan match {
+    case s: SupportsRuntimeCatalystFiltering => s.fullyPushedFilterAttributes()
+    case _ => Array.empty
+  }
+
   /**
    * Resolved attributes for which a Catalyst runtime-filtering scan fully evaluates predicates.
    * Empty for a [[SupportsRuntimeV2Filtering]] scan, which keeps its post-scan filters.
    */
   lazy val fullyPushedRuntimeFilterAttrs: AttributeSet = {
     checkRuntimeFilteringInterfaces()
-    val filterAttrs = scan match {
-      case s: SupportsRuntimeCatalystFiltering => s.fullyPushedFilterAttributes()
-      case _ => Array.empty[NamedReference]
-    }
-    resolveTopLevelFilterAttrs(filterAttrs)
+    resolveFilterAttrs(declaredFullyPushedRuntimeFilterAttrs)
   }
 
   /**
-   * Resolves fully pushed runtime-filter references against this relation's output. Nested
-   * references are rejected because [[AttributeSet]] keeps only their root attribute, which could
-   * make a predicate on a sibling field appear fully pushed and remove its post-scan evaluation.
+   * 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"Fully pushed 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.")
-    }
-    resolveFilterAttrs(filterAttrs)
-  }
-
-  /** Resolves the given runtime-filter references against this relation's output. */
   private def resolveFilterAttrs(filterAttrs: Array[NamedReference]): AttributeSet = {
-    AttributeSet(V2ExpressionUtils.resolveRefs[Attribute](
+    AttributeSet(V2ExpressionUtils.resolveRefs[NamedExpression](
       filterAttrs.toImmutableArraySeq, this))
   }
 
@@ -295,12 +286,22 @@ 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 _: SupportsRuntimeCatalystFiltering =>
+        declaredFullyPushedRuntimeFilterAttrs.find(_.fieldNames.length > 1).foreach { ref =>
+          throw SparkException.internalError(
+            "Fully pushed runtime filter attribute " +
+            s"'${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.")
+        }
+      case _ =>
+    }
   }
 
   override def doCanonicalize(): DataSourceV2ScanRelation = {
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala
index 3b9e7562792c7..035e6d00517c1 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala
@@ -41,6 +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 resolve against the scan relation output when Spark builds it. Attributes
+   * pruned out of [[Scan.readSchema]] fail to resolve.
    */
   def filterAttributes(): Array[NamedReference]
 
@@ -73,8 +75,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. 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
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala
index 4c8ba2c7da714..633d9b85cf1d3 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala
+++ b/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)()
         }
@@ -828,7 +830,8 @@ abstract class InMemoryBaseTable(
 
     override def filterAttributes(): Array[NamedReference] = {
       partitioning.flatMap(_.references)
-        .filter(ref => readSchema.findNestedField(ref.fieldNames.toImmutableArraySeq).isDefined)
+        .filter(ref => readSchema.findNestedField(
+          ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined)
     }
 
     override def filter(filters: Array[Filter]): Unit = {
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalystRuntimeFilterTable.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalystRuntimeFilterTable.scala
index 51757a946b46c..8d94ca399e0a9 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalystRuntimeFilterTable.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalystRuntimeFilterTable.scala
@@ -25,6 +25,7 @@ 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.{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._
@@ -93,7 +94,8 @@ class InMemoryCatalystRuntimeFilterTable(
     /** 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).isDefined)
+        .filter(ref => readSchema.findNestedField(
+          ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined)
     }
 
     override def filterAttributes(): Array[NamedReference] = {
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala
index 6cb23d505f784..428a9215dcce0 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala
@@ -28,6 +28,7 @@ import org.apache.spark.sql.connector.expressions.filter.Predicate
 import org.apache.spark.sql.connector.read.{InputPartition, Scan, ScanBuilder}
 import org.apache.spark.sql.connector.write.{BatchWrite, DeltaBatchWrite, DeltaWrite, DeltaWriteBuilder, DeltaWriter, DeltaWriterFactory, LogicalWriteInfo, PhysicalWriteInfo, RequiresDistributionAndOrdering, RowLevelOperation, RowLevelOperationBuilder, RowLevelOperationInfo, SupportsDelta, Write, WriteBuilder, WriterCommitMessage}
 import org.apache.spark.sql.connector.write.RowLevelOperation.Command
+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.unsafe.types.UTF8String
@@ -310,9 +311,9 @@ class InMemoryRowLevelOperationTable private (
     with CatalystRuntimeFilteringScan {
 
     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)
     }
   }
 }
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala
index 7739b0b5bb843..fa50da6732bf2 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala
@@ -27,6 +27,7 @@ import org.apache.spark.sql.connector.expressions.{FieldReference, LiteralValue,
 import org.apache.spark.sql.connector.expressions.filter.{And, Predicate}
 import org.apache.spark.sql.connector.read.{InputPartition, Scan, ScanBuilder, SupportsRuntimeV2Filtering}
 import org.apache.spark.sql.connector.write.{LogicalWriteInfo, SupportsOverwriteV2, WriteBuilder, WriterCommitMessage}
+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._
@@ -78,7 +79,8 @@ class InMemoryTableWithV2Filter(
 
     override def filterAttributes(): Array[NamedReference] = {
       partitioning.flatMap(_.references)
-        .filter(ref => readSchema.findNestedField(ref.fieldNames.toImmutableArraySeq).isDefined)
+        .filter(ref => readSchema.findNestedField(
+          ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined)
     }
 
     override def filter(filters: Array[Predicate]): Unit = {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala
index ae7b432d72f5b..b0f857ac35899 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala
@@ -18,11 +18,11 @@
 package org.apache.spark.sql.execution.dynamicpruning
 
 import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, AttributeReference, DynamicPruningExpression, Expression, InSubquery, ListQuery, PredicateHelper, V2ExpressionUtils}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeMap, AttributeReference, DynamicPruningExpression, Expression, InSubquery, ListQuery, NamedExpression, PredicateHelper, V2ExpressionUtils}
 import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
 import org.apache.spark.sql.catalyst.optimizer.RewritePredicateSubquery
 import org.apache.spark.sql.catalyst.planning.{DeltaBasedRowLevelOperation, GroupBasedRowLevelOperation}
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, LogicalPlan, RowLevelWrite}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, LogicalPlan, Project, RowLevelWrite}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern.{REPLACE_DATA, WRITE_DELTA}
 import org.apache.spark.sql.connector.expressions.NamedReference
@@ -97,8 +97,9 @@ class RowLevelOperationRuntimeGroupFiltering(optimizeSubqueries: Rule[LogicalPla
         val relation = r.relation.copy(table = originalTable)
         val matchingRowsPlan = buildMatchingRowsPlan(write, relation, cond)
         val filterAttrsSeq = filterAttrs.toImmutableArraySeq
-        val buildKeys = V2ExpressionUtils.resolveRefs[Attribute](filterAttrsSeq, matchingRowsPlan)
-        val pruningKeys = V2ExpressionUtils.resolveRefs[Attribute](filterAttrsSeq, r)
+        val buildKeys =
+          V2ExpressionUtils.resolveRefs[NamedExpression](filterAttrsSeq, matchingRowsPlan)
+        val pruningKeys = V2ExpressionUtils.resolveRefs[NamedExpression](filterAttrsSeq, r)
         Filter(buildDynamicPruningCond(matchingRowsPlan, buildKeys, pruningKeys), r)
     }
     // optimize subqueries to rewrite them as joins and trigger job planning
@@ -143,13 +144,26 @@ class RowLevelOperationRuntimeGroupFiltering(optimizeSubqueries: Rule[LogicalPla
 
   private def buildDynamicPruningCond(
       matchingRowsPlan: LogicalPlan,
-      buildKeys: Seq[Attribute],
-      pruningKeys: Seq[Attribute]): Expression = {
+      buildKeys: Seq[NamedExpression],
+      pruningKeys: Seq[NamedExpression]): Expression = {
     assert(buildKeys.nonEmpty && pruningKeys.nonEmpty)
 
-    val buildQuery = Aggregate(buildKeys, buildKeys, matchingRowsPlan)
+    // Nested references resolve to aliases over field-extraction expressions. Materialize those
+    // aliases before aggregation so grouping-expression cleanup preserves the subquery schema.
+    val buildKeyAliases = buildKeys.collect { case alias: Alias => alias }
+    val buildPlan = if (buildKeyAliases.nonEmpty) {
+      Project(matchingRowsPlan.output ++ buildKeyAliases, matchingRowsPlan)
+    } else {
+      matchingRowsPlan
+    }
+    val buildKeyAttrs = buildKeys.map(_.toAttribute)
+    val buildQuery = Aggregate(buildKeyAttrs, buildKeyAttrs, buildPlan)
+    val pruningExprs = pruningKeys.map {
+      case alias: Alias => alias.child
+      case other => other
+    }
     DynamicPruningExpression(
-      InSubquery(pruningKeys, ListQuery(buildQuery, numCols = buildQuery.output.length)))
+      InSubquery(pruningExprs, ListQuery(buildQuery, numCols = buildQuery.output.length)))
   }
 
   private def buildTableToScanAttrMap(
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
index 6d2f76e602816..12e16016596fe 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
@@ -116,31 +116,39 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession {
     }
   }
 
-  test("nested fully pushed filter attribute -> rejected") {
+  test("nested fully pushed filter attribute -> rejected without a runtime filter") {
     val tbl = s"$catalogName.tbl_nested_fully_pushed"
     withTable(tbl) {
       sql(s"CREATE TABLE $tbl (id INT, s STRUCT<part: INT, other: INT>) USING $v2Source " +
         "PARTITIONED BY (s.part) " +
         "TBLPROPERTIES('fully-pushed-filter-attributes' = 's.part')")
 
-      val scanRelation = sql(s"SELECT * FROM $tbl").queryExecution.optimizedPlan.collectFirst {
-        case r: DataSourceV2ScanRelation => r
-      }.getOrElse(fail("Expected a DataSourceV2ScanRelation"))
-
-      // Ordinary nested filter attributes remain supported for partition pruning.
-      val runtimeFilterAttrs = scanRelation.runtimeFilterAttrs
-      assert(runtimeFilterAttrs.map(_.name).toSeq === Seq("s"))
-
-      // Fully pushed eligibility cannot use the root-only AttributeSet representation: doing so
-      // would make a predicate on `s.other` appear covered by the declaration of `s.part`.
       val e = intercept[SparkException] {
-        scanRelation.fullyPushedRuntimeFilterAttrs
+        sql(s"SELECT * FROM $tbl").queryExecution.executedPlan
       }
       assert(e.getMessage.contains("must be a top-level attribute"),
         s"expected the nested fully pushed reference to be rejected, got ${e.getMessage}")
     }
   }
 
+  test("nested filter attribute under a non-struct column -> rejected during resolution") {
+    val tbl = s"$catalogName.tbl_malformed_nested_filter_attr"
+    withTable(tbl) {
+      sql(s"CREATE TABLE $tbl (id INT, part INT) USING $v2Source PARTITIONED BY (part)")
+
+      val scanRelation = sql(s"SELECT * FROM $tbl").queryExecution.optimizedPlan.collectFirst {
+        case r: DataSourceV2ScanRelation => r
+      }.getOrElse(fail("Expected a DataSourceV2ScanRelation"))
+      val e = intercept[AnalysisException] {
+        scanRelation.copy(scan = new NestedFilterAttributeScan).runtimeFilterAttrs
+      }
+      checkError(
+        exception = e,
+        condition = "INVALID_EXTRACT_BASE_FIELD_TYPE",
+        parameters = Map("base" -> "\"part\"", "other" -> "\"INT\""))
+    }
+  }
+
   test("missing filter attribute -> rejected") {
     val tbl = s"$catalogName.tbl_missing_filter_attr"
     withTable(tbl) {
@@ -674,6 +682,17 @@ private class MissingFilterAttributeScan extends SupportsRuntimeCatalystFilterin
   override def filter(expressions: Array[Expression]): Unit = {}
 }
 
+/** A scan declaring a nested runtime-filter attribute beneath an integer column. */
+private class NestedFilterAttributeScan extends SupportsRuntimeCatalystFiltering {
+
+  override def readSchema(): StructType = new StructType().add("part", IntegerType)
+
+  override def filterAttributes(): Array[NamedReference] =
+    Array(FieldReference(Seq("part", "nested")))
+
+  override def filter(expressions: Array[Expression]): Unit = {}
+}
+
 private case class KeyedInputPartition(key: Int) extends InputPartition with HasPartitionKey {
   override def partitionKey(): InternalRow = InternalRow(key)
 }
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedRowLevelOperationCatalystRuntimeFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedRowLevelOperationCatalystRuntimeFilterSuite.scala
index 34b43e2354abc..ffe8211f0a796 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedRowLevelOperationCatalystRuntimeFilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedRowLevelOperationCatalystRuntimeFilterSuite.scala
@@ -19,6 +19,8 @@ package org.apache.spark.sql.connector
 
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.connector.catalog.InMemoryTable
+import org.apache.spark.sql.connector.expressions.LogicalExpressions.{identity, reference}
+import org.apache.spark.sql.connector.expressions.Transform
 import org.apache.spark.sql.connector.write.DeleteSummary
 
 class GroupBasedRowLevelOperationCatalystRuntimeFilterSuite
@@ -49,6 +51,35 @@ class GroupBasedRowLevelOperationCatalystRuntimeFilterSuite
     checkDeleteMetrics(numDeletedRows = 1, numCopiedRows = 1)
   }
 
+  test("delete runtime group filtering by a nested attribute") {
+    val schema = "pk INT NOT NULL, id INT, salary INT, " +
+      "dep STRUCT<name: STRING, region: STRING>"
+    createTable(schema, Array[Transform](identity(reference(Seq("dep", "name")))))
+    append(schema,
+      """{"pk":1,"id":1,"salary":300,"dep":{"name":"hr","region":"west"}}
+        |{"pk":2,"id":2,"salary":150,"dep":{"name":"software","region":"west"}}
+        |{"pk":3,"id":3,"salary":120,"dep":{"name":"hr","region":"east"}}
+        |""".stripMargin)
+
+    val executedPlan = executeAndKeepPlan {
+      sql(s"DELETE FROM $tableNameAsString WHERE salary IN (300, 400, 500)")
+    }
+    assertCatalystGroupFilter(
+      executedPlan,
+      expectedFilterAttrs = Seq("dep.name"),
+      expectedFilter = GroupFilter(
+        scanSchema = "salary INT, dep STRUCT<name: STRING>", groups = Seq("hr")),
+      expectedFilterRefs = Seq("dep"))
+
+    checkAnswer(
+      sql(s"SELECT * FROM $tableNameAsString"),
+      Row(2, 2, 150, Row("software", "west")) ::
+        Row(3, 3, 120, Row("hr", "east")) :: Nil)
+
+    checkReplacedPartitions(Seq("hr"))
+    checkDeleteMetrics(numDeletedRows = 1, numCopiedRows = 1)
+  }
+
   private def checkDeleteMetrics(numDeletedRows: Long, numCopiedRows: Long): Unit = {
     val t = catalog.loadTable(ident).asInstanceOf[InMemoryTable]
     val summary = t.commits.last.writeSummary.get.asInstanceOf[DeleteSummary]
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationCatalystRuntimeFilterSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationCatalystRuntimeFilterSuiteBase.scala
index 810131eb54cef..8b259266b32ec 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationCatalystRuntimeFilterSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationCatalystRuntimeFilterSuiteBase.scala
@@ -137,7 +137,8 @@ abstract class RowLevelOperationCatalystRuntimeFilterSuiteBase
   protected def assertCatalystGroupFilter(
       executedPlan: SparkPlan,
       expectedFilterAttrs: Seq[String],
-      expectedFilter: GroupFilter): Unit = {
+      expectedFilter: GroupFilter,
+      expectedFilterRefs: Seq[String] = Seq.empty): Unit = {
     val batchScans = collect(executedPlan) { case s: BatchScanExec => s }
     assert(batchScans.nonEmpty, "expected a batch scan for the row-level operation")
     val scan = catalystScan(batchScans.head)
@@ -147,11 +148,12 @@ abstract class RowLevelOperationCatalystRuntimeFilterSuiteBase
     val filterAttrs = scan.filterAttributes().map(_.fieldNames.mkString(".")).toSeq
     assert(filterAttrs === expectedFilterAttrs,
       s"expected the scan to declare $expectedFilterAttrs as filter attributes, got $filterAttrs")
+    val filterRefs = if (expectedFilterRefs.nonEmpty) expectedFilterRefs else expectedFilterAttrs
 
     batchScans.foreach { batchScan =>
       batchScan.runtimeFilters match {
         case Seq(DynamicPruningExpression(inSubquery: InSubqueryExec)) =>
-          assertGroupFilter(inSubquery, expectedFilterAttrs, expectedFilter)
+          assertGroupFilter(inSubquery, filterRefs, expectedFilter)
         case other => fail(s"expected a single dynamic pruning group filter, got $other")
       }
     }
@@ -163,7 +165,7 @@ abstract class RowLevelOperationCatalystRuntimeFilterSuiteBase
       s"expected each of the ${batchScans.size} scan node(s) to push the filter once, got $pushed")
     pushed.foreach {
       case inSubquery: InSubqueryExec =>
-        assertGroupFilter(inSubquery, expectedFilterAttrs, expectedFilter)
+        assertGroupFilter(inSubquery, filterRefs, expectedFilter)
       case other =>
         fail(s"expected the group filter pushed as an InSubqueryExec, got $other")
     }
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationSuiteBase.scala
index 21e27e9b6c1cb..c62cc48cffa1c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationSuiteBase.scala
@@ -117,8 +117,17 @@ abstract class RowLevelOperationSuiteBase
     createTable(columns)
   }
 
+  protected def createTable(schemaString: String, transforms: Array[Transform]): Unit = {
+    val columns = CatalogV2Util.structTypeToV2Columns(StructType.fromDDL(schemaString))
+    createTable(columns, transforms)
+  }
+
   protected def createTable(columns: Array[Column]): Unit = {
     val transforms = Array[Transform](identity(reference(Seq("dep"))))
+    createTable(columns, transforms)
+  }
+
+  protected def createTable(columns: Array[Column], transforms: Array[Transform]): Unit = {
     val tableInfo = new TableInfo.Builder()
       .withColumns(columns)
       .withPartitions(transforms)

From 1983e1593f3741f2211751bbe8bbea80e9f86386 Mon Sep 17 00:00:00 2001
From: Szehon Ho <szehon.apache@gmail.com>
Date: Fri, 28 Aug 2026 17:23:26 -0700
Subject: [PATCH 4/6] [SPARK-59068][SQL] Address follow-up runtime filtering
 review

---
 .../expressions/V2ExpressionUtils.scala       |  2 +-
 .../datasources/v2/DataSourceV2Relation.scala | 36 +++++++++++-----
 .../InMemoryCatalystRuntimeFilterTable.scala  |  6 ++-
 ...wLevelOperationRuntimeGroupFiltering.scala | 19 +++-----
 ...taSourceV2CatalystRuntimeFilterSuite.scala | 18 +++++---
 ...lOperationCatalystRuntimeFilterSuite.scala |  2 +-
 ...rationCatalystRuntimeFilterSuiteBase.scala | 43 ++++++++++++++++++-
 7 files changed, 90 insertions(+), 36 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala
index 702255e075743..349a59a1aaf8f 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala
+++ b/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))
   }
 
   /**
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
index 9de6e0d528acb..04a59ede40d7d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
@@ -20,6 +20,7 @@ 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, NamedExpression, SortOrder, V2ExpressionUtils}
@@ -39,7 +40,6 @@ import org.apache.spark.sql.connector.read.streaming.{Offset, SparkDataStream}
 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
 
 /**
@@ -205,6 +205,7 @@ case class DataSourceV2ScanRelation(
    */
   lazy val runtimeFilterAttrs: AttributeSet = {
     checkRuntimeFilteringInterfaces()
+    checkFullyPushedFilterAttrs()
     val filterAttrs = scan match {
       case s: SupportsRuntimeV2Filtering => s.filterAttributes
       case s: SupportsRuntimeCatalystFiltering => s.filterAttributes()
@@ -224,6 +225,7 @@ case class DataSourceV2ScanRelation(
    */
   lazy val fullyPushedRuntimeFilterAttrs: AttributeSet = {
     checkRuntimeFilteringInterfaces()
+    checkFullyPushedFilterAttrs()
     resolveFilterAttrs(declaredFullyPushedRuntimeFilterAttrs)
   }
 
@@ -234,8 +236,18 @@ case class DataSourceV2ScanRelation(
    * ordinary runtime-filter eligibility because Spark retains the post-scan predicate.
    */
   private def resolveFilterAttrs(filterAttrs: Array[NamedReference]): AttributeSet = {
-    AttributeSet(V2ExpressionUtils.resolveRefs[NamedExpression](
-      filterAttrs.toImmutableArraySeq, this))
+    val resolvedAttrs = filterAttrs.map { ref =>
+      try {
+        V2ExpressionUtils.resolveRef[NamedExpression](ref, this)
+      } catch {
+        case e: AnalysisException =>
+          throw SparkException.internalError(
+            s"Cannot resolve runtime filter attribute '${ref.fieldNames.mkString(".")}' " +
+            s"declared by ${scan.getClass.getName}.",
+            e)
+      }
+    }
+    AttributeSet(resolvedAttrs)
   }
 
   override val nodePatterns: Seq[TreePattern] = Seq(DATA_SOURCE_V2_SCAN_RELATION)
@@ -292,18 +304,20 @@ case class DataSourceV2ScanRelation(
         throw SparkException.internalError(
           "A scan must not implement both SupportsRuntimeV2Filtering and " +
           s"SupportsRuntimeCatalystFiltering, but ${scan.getClass.getName} implements both.")
-      case _: SupportsRuntimeCatalystFiltering =>
-        declaredFullyPushedRuntimeFilterAttrs.find(_.fieldNames.length > 1).foreach { ref =>
-          throw SparkException.internalError(
-            "Fully pushed runtime filter attribute " +
-            s"'${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.")
-        }
       case _ =>
     }
   }
 
+  private def checkFullyPushedFilterAttrs(): Unit = {
+    declaredFullyPushedRuntimeFilterAttrs.find(_.fieldNames.length > 1).foreach { ref =>
+      throw SparkException.internalError(
+        "Fully pushed runtime filter attribute " +
+        s"'${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.")
+    }
+  }
+
   override def doCanonicalize(): DataSourceV2ScanRelation = {
     this.copy(
       relation = this.relation.copy(
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalystRuntimeFilterTable.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalystRuntimeFilterTable.scala
index 8d94ca399e0a9..58f07307dcbe1 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalystRuntimeFilterTable.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalystRuntimeFilterTable.scala
@@ -99,13 +99,15 @@ class InMemoryCatalystRuntimeFilterTable(
     }
 
     override def filterAttributes(): Array[NamedReference] = {
-      partitionAttrs.filter(ref => restrictedFilterAttrs.forall(_.contains(ref.toString)))
+      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] = {
-      partitionAttrs.filter(ref => fullyPushedFilterAttrs.contains(ref.toString))
+      partitionAttrs.filter(ref => fullyPushedFilterAttrs.contains(ref.fieldNames.mkString(".")))
     }
   }
 }
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala
index b0f857ac35899..b17e224c4a5ad 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala
@@ -22,7 +22,7 @@ import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeMap
 import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
 import org.apache.spark.sql.catalyst.optimizer.RewritePredicateSubquery
 import org.apache.spark.sql.catalyst.planning.{DeltaBasedRowLevelOperation, GroupBasedRowLevelOperation}
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, LogicalPlan, Project, RowLevelWrite}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, LogicalPlan, RowLevelWrite}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern.{REPLACE_DATA, WRITE_DELTA}
 import org.apache.spark.sql.connector.expressions.NamedReference
@@ -148,22 +148,15 @@ class RowLevelOperationRuntimeGroupFiltering(optimizeSubqueries: Rule[LogicalPla
       pruningKeys: Seq[NamedExpression]): Expression = {
     assert(buildKeys.nonEmpty && pruningKeys.nonEmpty)
 
-    // Nested references resolve to aliases over field-extraction expressions. Materialize those
-    // aliases before aggregation so grouping-expression cleanup preserves the subquery schema.
-    val buildKeyAliases = buildKeys.collect { case alias: Alias => alias }
-    val buildPlan = if (buildKeyAliases.nonEmpty) {
-      Project(matchingRowsPlan.output ++ buildKeyAliases, matchingRowsPlan)
-    } else {
-      matchingRowsPlan
-    }
-    val buildKeyAttrs = buildKeys.map(_.toAttribute)
-    val buildQuery = Aggregate(buildKeyAttrs, buildKeyAttrs, buildPlan)
-    val pruningExprs = pruningKeys.map {
+    def unalias(expr: NamedExpression): Expression = expr match {
       case alias: Alias => alias.child
       case other => other
     }
+
+    val buildQuery = Aggregate(buildKeys.map(unalias), buildKeys, matchingRowsPlan)
     DynamicPruningExpression(
-      InSubquery(pruningExprs, ListQuery(buildQuery, numCols = buildQuery.output.length)))
+      InSubquery(pruningKeys.map(unalias),
+        ListQuery(buildQuery, numCols = buildQuery.output.length)))
   }
 
   private def buildTableToScanAttrMap(
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
index 12e16016596fe..eb14c51107074 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
@@ -139,13 +139,16 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession {
       val scanRelation = sql(s"SELECT * FROM $tbl").queryExecution.optimizedPlan.collectFirst {
         case r: DataSourceV2ScanRelation => r
       }.getOrElse(fail("Expected a DataSourceV2ScanRelation"))
-      val e = intercept[AnalysisException] {
+      val e = intercept[SparkException] {
         scanRelation.copy(scan = new NestedFilterAttributeScan).runtimeFilterAttrs
       }
+      val scanClass = classOf[NestedFilterAttributeScan].getName
       checkError(
         exception = e,
-        condition = "INVALID_EXTRACT_BASE_FIELD_TYPE",
-        parameters = Map("base" -> "\"part\"", "other" -> "\"INT\""))
+        condition = "INTERNAL_ERROR",
+        parameters = Map("message" ->
+          s"Cannot resolve runtime filter attribute 'part.nested' declared by $scanClass."))
+      assert(e.getCause.isInstanceOf[AnalysisException])
     }
   }
 
@@ -157,13 +160,16 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession {
       val scanRelation = sql(s"SELECT * FROM $tbl").queryExecution.optimizedPlan.collectFirst {
         case r: DataSourceV2ScanRelation => r
       }.getOrElse(fail("Expected a DataSourceV2ScanRelation"))
-      val e = intercept[AnalysisException] {
+      val e = intercept[SparkException] {
         scanRelation.copy(scan = new MissingFilterAttributeScan).runtimeFilterAttrs
       }
+      val scanClass = classOf[MissingFilterAttributeScan].getName
       checkError(
         exception = e,
-        condition = "_LEGACY_ERROR_TEMP_1137",
-        parameters = Map("name" -> "missing", "outputStr" -> "id,part"))
+        condition = "INTERNAL_ERROR",
+        parameters = Map("message" ->
+          s"Cannot resolve runtime filter attribute 'missing' declared by $scanClass."))
+      assert(e.getCause.isInstanceOf[AnalysisException])
     }
   }
 
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedRowLevelOperationCatalystRuntimeFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedRowLevelOperationCatalystRuntimeFilterSuite.scala
index ffe8211f0a796..4a7d3d1bcf18c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedRowLevelOperationCatalystRuntimeFilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedRowLevelOperationCatalystRuntimeFilterSuite.scala
@@ -69,7 +69,7 @@ class GroupBasedRowLevelOperationCatalystRuntimeFilterSuite
       expectedFilterAttrs = Seq("dep.name"),
       expectedFilter = GroupFilter(
         scanSchema = "salary INT, dep STRUCT<name: STRING>", groups = Seq("hr")),
-      expectedFilterRefs = Seq("dep"))
+      expectedFilterRefs = Some(Seq("dep")))
 
     checkAnswer(
       sql(s"SELECT * FROM $tableNameAsString"),
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationCatalystRuntimeFilterSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationCatalystRuntimeFilterSuiteBase.scala
index 8b259266b32ec..cccd8e052a4fe 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationCatalystRuntimeFilterSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationCatalystRuntimeFilterSuiteBase.scala
@@ -21,6 +21,8 @@ import org.apache.spark.sql.Row
 import org.apache.spark.sql.catalyst.expressions.DynamicPruningExpression
 import org.apache.spark.sql.catalyst.types.DataTypeUtils
 import org.apache.spark.sql.connector.catalog.{BufferedRows, InMemoryRowLevelOperationTable}
+import org.apache.spark.sql.connector.expressions.LogicalExpressions.{identity, reference}
+import org.apache.spark.sql.connector.expressions.Transform
 import org.apache.spark.sql.execution.InSubqueryExec
 import org.apache.spark.sql.execution.ReusedSubqueryExec
 import org.apache.spark.sql.execution.SparkPlan
@@ -121,6 +123,43 @@ abstract class RowLevelOperationCatalystRuntimeFilterSuiteBase
     }
   }
 
+  test("merge runtime group filtering by a nested attribute") {
+    withTempView("source") {
+      val schema = "pk INT NOT NULL, id INT, salary INT, " +
+        "dep STRUCT<name: STRING, region: STRING>"
+      createTable(schema, Array[Transform](identity(reference(Seq("dep", "name")))))
+      append(schema,
+        """{"pk":1,"id":1,"salary":100,"dep":{"name":"hr","region":"west"}}
+          |{"pk":2,"id":2,"salary":200,"dep":{"name":"hr","region":"east"}}
+          |{"pk":3,"id":3,"salary":300,"dep":{"name":"software","region":"west"}}
+          |""".stripMargin)
+
+      Seq(1, 2).toDF("pk").createOrReplaceTempView("source")
+
+      val executedPlan = executeAndKeepPlan {
+        sql(
+          s"""MERGE INTO $tableNameAsString t
+             |USING source s
+             |ON t.pk = s.pk
+             |WHEN MATCHED THEN
+             | UPDATE SET t.salary = t.salary + 1
+             |""".stripMargin)
+      }
+      assertCatalystGroupFilter(
+        executedPlan,
+        expectedFilterAttrs = Seq("dep.name"),
+        expectedFilter = GroupFilter(
+          scanSchema = "pk INT, dep STRUCT<name: STRING>", groups = Seq("hr")),
+        expectedFilterRefs = Some(Seq("dep")))
+
+      checkAnswer(
+        sql(s"SELECT * FROM $tableNameAsString"),
+        Row(1, 1, 101, Row("hr", "west")) ::
+          Row(2, 2, 201, Row("hr", "east")) ::
+          Row(3, 3, 300, Row("software", "west")) :: Nil)
+    }
+  }
+
   /**
    * Asserts the injected group filter down to its contents: the scan declares
    * `expectedFilterAttrs` in `filterAttributes`, every scan node carries one dynamic pruning
@@ -138,7 +177,7 @@ abstract class RowLevelOperationCatalystRuntimeFilterSuiteBase
       executedPlan: SparkPlan,
       expectedFilterAttrs: Seq[String],
       expectedFilter: GroupFilter,
-      expectedFilterRefs: Seq[String] = Seq.empty): Unit = {
+      expectedFilterRefs: Option[Seq[String]] = None): Unit = {
     val batchScans = collect(executedPlan) { case s: BatchScanExec => s }
     assert(batchScans.nonEmpty, "expected a batch scan for the row-level operation")
     val scan = catalystScan(batchScans.head)
@@ -148,7 +187,7 @@ abstract class RowLevelOperationCatalystRuntimeFilterSuiteBase
     val filterAttrs = scan.filterAttributes().map(_.fieldNames.mkString(".")).toSeq
     assert(filterAttrs === expectedFilterAttrs,
       s"expected the scan to declare $expectedFilterAttrs as filter attributes, got $filterAttrs")
-    val filterRefs = if (expectedFilterRefs.nonEmpty) expectedFilterRefs else expectedFilterAttrs
+    val filterRefs = expectedFilterRefs.getOrElse(expectedFilterAttrs)
 
     batchScans.foreach { batchScan =>
       batchScan.runtimeFilters match {

From 24ac1aaac301bb056016f1965f8dc2bd96512723 Mon Sep 17 00:00:00 2001
From: Szehon Ho <szehon.apache@gmail.com>
Date: Fri, 28 Aug 2026 17:28:22 -0700
Subject: [PATCH 5/6] [SPARK-59068][SQL] Add structured runtime filter
 attribute error

---
 .../resources/error/error-conditions.json     |  6 ++++
 .../sql/errors/QueryCompilationErrors.scala   | 14 ++++++++
 .../datasources/v2/DataSourceV2Relation.scala |  7 ++--
 ...taSourceV2CatalystRuntimeFilterSuite.scala | 32 +++++++++++++------
 4 files changed, 45 insertions(+), 14 deletions(-)

diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
index 5af823a3f24b2..555c73352d1a9 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -2260,6 +2260,12 @@
     ],
     "sqlState" : "KD010"
   },
+  "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE" : {
+    "message" : [
+      "The runtime filter attribute <attribute> reported by data source scan <scanClass> cannot be resolved against the scan read schema <readSchema>."
+    ],
+    "sqlState" : "KD000"
+  },
   "DATA_SOURCE_METADATA_SCHEMA_NOT_IMPLEMENTED" : {
     "message" : [
       "<class> does not implement metadataSchema."
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index ada71af43d3b5..5f4f34462e719 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -4654,6 +4654,20 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
     )
   }
 
+  def invalidDataSourceRuntimeFilterAttributeError(
+      attribute: Array[String],
+      scanClass: String,
+      readSchema: StructType,
+      cause: AnalysisException): AnalysisException = {
+    new AnalysisException(
+      errorClass = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE",
+      messageParameters = Map(
+        "attribute" -> toSQLId(attribute.toImmutableArraySeq),
+        "scanClass" -> scanClass,
+        "readSchema" -> toSQLType(readSchema)),
+      cause = Some(cause))
+  }
+
   def foundMultipleXMLDataSourceError(provider: String,
       sourceNames: Seq[String],
       externalSource: String): Throwable = {
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
index 04a59ede40d7d..fd887f380fece 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
@@ -37,6 +37,7 @@ import org.apache.spark.sql.connector.expressions.{FieldReference, NamedReferenc
 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
@@ -241,10 +242,8 @@ case class DataSourceV2ScanRelation(
         V2ExpressionUtils.resolveRef[NamedExpression](ref, this)
       } catch {
         case e: AnalysisException =>
-          throw SparkException.internalError(
-            s"Cannot resolve runtime filter attribute '${ref.fieldNames.mkString(".")}' " +
-            s"declared by ${scan.getClass.getName}.",
-            e)
+          throw QueryCompilationErrors.invalidDataSourceRuntimeFilterAttributeError(
+            ref.fieldNames, scan.getClass.getName, scan.readSchema(), e)
       }
     }
     AttributeSet(resolvedAttrs)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
index eb14c51107074..4a2d49c123359 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
@@ -139,16 +139,22 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession {
       val scanRelation = sql(s"SELECT * FROM $tbl").queryExecution.optimizedPlan.collectFirst {
         case r: DataSourceV2ScanRelation => r
       }.getOrElse(fail("Expected a DataSourceV2ScanRelation"))
-      val e = intercept[SparkException] {
+      val e = intercept[AnalysisException] {
         scanRelation.copy(scan = new NestedFilterAttributeScan).runtimeFilterAttrs
       }
       val scanClass = classOf[NestedFilterAttributeScan].getName
       checkError(
         exception = e,
-        condition = "INTERNAL_ERROR",
-        parameters = Map("message" ->
-          s"Cannot resolve runtime filter attribute 'part.nested' declared by $scanClass."))
-      assert(e.getCause.isInstanceOf[AnalysisException])
+        condition = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE",
+        parameters = Map(
+          "attribute" -> "`part`.`nested`",
+          "scanClass" -> scanClass,
+          "readSchema" -> "\"STRUCT<part: INT>\""),
+        sqlState = "KD000")
+      checkError(
+        exception = e.getCause.asInstanceOf[AnalysisException],
+        condition = "INVALID_EXTRACT_BASE_FIELD_TYPE",
+        parameters = Map("base" -> "\"part\"", "other" -> "\"INT\""))
     }
   }
 
@@ -160,16 +166,22 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession {
       val scanRelation = sql(s"SELECT * FROM $tbl").queryExecution.optimizedPlan.collectFirst {
         case r: DataSourceV2ScanRelation => r
       }.getOrElse(fail("Expected a DataSourceV2ScanRelation"))
-      val e = intercept[SparkException] {
+      val e = intercept[AnalysisException] {
         scanRelation.copy(scan = new MissingFilterAttributeScan).runtimeFilterAttrs
       }
       val scanClass = classOf[MissingFilterAttributeScan].getName
       checkError(
         exception = e,
-        condition = "INTERNAL_ERROR",
-        parameters = Map("message" ->
-          s"Cannot resolve runtime filter attribute 'missing' declared by $scanClass."))
-      assert(e.getCause.isInstanceOf[AnalysisException])
+        condition = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE",
+        parameters = Map(
+          "attribute" -> "`missing`",
+          "scanClass" -> scanClass,
+          "readSchema" -> "\"STRUCT<part: INT>\""),
+        sqlState = "KD000")
+      checkError(
+        exception = e.getCause.asInstanceOf[AnalysisException],
+        condition = "_LEGACY_ERROR_TEMP_1137",
+        parameters = Map("name" -> "missing", "outputStr" -> "id,part"))
     }
   }
 

From ec4fa1afd65993cd6c4419cef77144837d6c3693 Mon Sep 17 00:00:00 2001
From: Szehon Ho <szehon.apache@gmail.com>
Date: Fri, 28 Aug 2026 18:16:01 -0700
Subject: [PATCH 6/6] [SPARK-59068][SQL] Refine runtime filter attribute errors

---
 .../resources/error/error-conditions.json     | 14 +++-
 .../read/SupportsRuntimeFiltering.java        |  3 +
 .../sql/errors/QueryCompilationErrors.scala   | 36 ++++++++--
 .../datasources/v2/DataSourceV2Relation.scala | 31 ++++----
 .../SupportsRuntimeCatalystFiltering.scala    |  5 +-
 ...taSourceV2CatalystRuntimeFilterSuite.scala | 71 ++++++++++++++++---
 .../sql/connector/DataSourceV2SQLSuite.scala  | 12 ++--
 ...lOperationCatalystRuntimeFilterSuite.scala |  2 +-
 ...rationCatalystRuntimeFilterSuiteBase.scala | 32 ++++++---
 9 files changed, 163 insertions(+), 43 deletions(-)

diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
index 555c73352d1a9..dc6694e51eb47 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -2262,8 +2262,20 @@
   },
   "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE" : {
     "message" : [
-      "The runtime filter attribute <attribute> reported by data source scan <scanClass> cannot be resolved against the scan read schema <readSchema>."
+      "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" : {
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeFiltering.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeFiltering.java
index 2feee94326baf..c752c8c95d54f 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeFiltering.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeFiltering.java
@@ -53,6 +53,9 @@ public interface SupportsRuntimeFiltering extends SupportsRuntimeV2Filtering {
    * 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,
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index 5f4f34462e719..a662eee4ea41e 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -4654,18 +4654,44 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
     )
   }
 
-  def invalidDataSourceRuntimeFilterAttributeError(
+  def cannotResolveDataSourceRuntimeFilterAttributeError(
       attribute: Array[String],
+      method: String,
       scanClass: String,
-      readSchema: StructType,
+      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 = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE",
+      errorClass = s"DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.$errorSubClass",
       messageParameters = Map(
         "attribute" -> toSQLId(attribute.toImmutableArraySeq),
+        "method" -> method,
         "scanClass" -> scanClass,
-        "readSchema" -> toSQLType(readSchema)),
-      cause = Some(cause))
+        "relationOutput" -> toSQLType(relationOutput)),
+      cause = cause)
   }
 
   def foundMultipleXMLDataSourceError(provider: String,
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
index fd887f380fece..c13b44ab5fc6c 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
@@ -29,7 +29,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{ColumnStat, ExposesMetadataC
 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
@@ -202,7 +202,8 @@ 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()
@@ -212,7 +213,7 @@ case class DataSourceV2ScanRelation(
       case s: SupportsRuntimeCatalystFiltering => s.filterAttributes()
       case _ => Array.empty[NamedReference]
     }
-    resolveFilterAttrs(filterAttrs)
+    resolveFilterAttrs(filterAttrs, "filterAttributes()")
   }
 
   private lazy val declaredFullyPushedRuntimeFilterAttrs: Array[NamedReference] = scan match {
@@ -227,7 +228,8 @@ case class DataSourceV2ScanRelation(
   lazy val fullyPushedRuntimeFilterAttrs: AttributeSet = {
     checkRuntimeFilteringInterfaces()
     checkFullyPushedFilterAttrs()
-    resolveFilterAttrs(declaredFullyPushedRuntimeFilterAttrs)
+    resolveFilterAttrs(
+      declaredFullyPushedRuntimeFilterAttrs, "fullyPushedFilterAttributes()")
   }
 
   /**
@@ -236,14 +238,20 @@ case class DataSourceV2ScanRelation(
    * [[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 resolveFilterAttrs(filterAttrs: Array[NamedReference]): AttributeSet = {
+  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.invalidDataSourceRuntimeFilterAttributeError(
-            ref.fieldNames, scan.getClass.getName, scan.readSchema(), e)
+          throw QueryCompilationErrors.cannotResolveDataSourceRuntimeFilterAttributeError(
+            attribute = ref.fieldNames,
+            method = method,
+            scanClass = scan.getClass.getName,
+            relationOutput = fromAttributes(output),
+            cause = e)
       }
     }
     AttributeSet(resolvedAttrs)
@@ -309,11 +317,10 @@ case class DataSourceV2ScanRelation(
 
   private def checkFullyPushedFilterAttrs(): Unit = {
     declaredFullyPushedRuntimeFilterAttrs.find(_.fieldNames.length > 1).foreach { ref =>
-      throw SparkException.internalError(
-        "Fully pushed runtime filter attribute " +
-        s"'${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.")
+      throw QueryCompilationErrors.nestedDataSourceFullyPushedRuntimeFilterAttributeError(
+        attribute = ref.fieldNames,
+        scanClass = scan.getClass.getName,
+        relationOutput = fromAttributes(output))
     }
   }
 
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala
index 035e6d00517c1..ec34efed26252 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala
@@ -64,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
 
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
index 4a2d49c123359..11e7461bc7f14 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
@@ -123,11 +123,22 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession {
         "PARTITIONED BY (s.part) " +
         "TBLPROPERTIES('fully-pushed-filter-attributes' = 's.part')")
 
-      val e = intercept[SparkException] {
-        sql(s"SELECT * FROM $tbl").queryExecution.executedPlan
+      val df = sql(s"SELECT * FROM $tbl")
+      val scanClass = df.queryExecution.optimizedPlan.collectFirst {
+        case r: DataSourceV2ScanRelation => r.scan.getClass.getName
+      }.getOrElse(fail("Expected a DataSourceV2ScanRelation"))
+      val e = intercept[AnalysisException] {
+        df.queryExecution.executedPlan
       }
-      assert(e.getMessage.contains("must be a top-level attribute"),
-        s"expected the nested fully pushed reference to be rejected, got ${e.getMessage}")
+      checkError(
+        exception = e,
+        condition = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.NOT_TOP_LEVEL",
+        parameters = Map(
+          "attribute" -> "`s`.`part`",
+          "method" -> "fullyPushedFilterAttributes()",
+          "scanClass" -> scanClass,
+          "relationOutput" -> "\"STRUCT<id: INT, s: STRUCT<part: INT, other: INT>>\""),
+        sqlState = "KD000")
     }
   }
 
@@ -145,11 +156,12 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession {
       val scanClass = classOf[NestedFilterAttributeScan].getName
       checkError(
         exception = e,
-        condition = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE",
+        condition = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.CANNOT_RESOLVE",
         parameters = Map(
           "attribute" -> "`part`.`nested`",
+          "method" -> "filterAttributes()",
           "scanClass" -> scanClass,
-          "readSchema" -> "\"STRUCT<part: INT>\""),
+          "relationOutput" -> "\"STRUCT<id: INT, part: INT>\""),
         sqlState = "KD000")
       checkError(
         exception = e.getCause.asInstanceOf[AnalysisException],
@@ -172,11 +184,41 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession {
       val scanClass = classOf[MissingFilterAttributeScan].getName
       checkError(
         exception = e,
-        condition = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE",
+        condition = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.CANNOT_RESOLVE",
+        parameters = Map(
+          "attribute" -> "`missing`",
+          "method" -> "filterAttributes()",
+          "scanClass" -> scanClass,
+          "relationOutput" -> "\"STRUCT<id: INT, part: INT>\""),
+        sqlState = "KD000")
+      checkError(
+        exception = e.getCause.asInstanceOf[AnalysisException],
+        condition = "_LEGACY_ERROR_TEMP_1137",
+        parameters = Map("name" -> "missing", "outputStr" -> "id,part"))
+    }
+  }
+
+  test("missing fully pushed filter attribute -> identifies the declaring method") {
+    val tbl = s"$catalogName.tbl_missing_fully_pushed_filter_attr"
+    withTable(tbl) {
+      sql(s"CREATE TABLE $tbl (id INT, part INT) USING $v2Source PARTITIONED BY (part)")
+
+      val scanRelation = sql(s"SELECT * FROM $tbl").queryExecution.optimizedPlan.collectFirst {
+        case r: DataSourceV2ScanRelation => r
+      }.getOrElse(fail("Expected a DataSourceV2ScanRelation"))
+      val e = intercept[AnalysisException] {
+        scanRelation.copy(scan = new MissingFullyPushedFilterAttributeScan)
+          .fullyPushedRuntimeFilterAttrs
+      }
+      val scanClass = classOf[MissingFullyPushedFilterAttributeScan].getName
+      checkError(
+        exception = e,
+        condition = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.CANNOT_RESOLVE",
         parameters = Map(
           "attribute" -> "`missing`",
+          "method" -> "fullyPushedFilterAttributes()",
           "scanClass" -> scanClass,
-          "readSchema" -> "\"STRUCT<part: INT>\""),
+          "relationOutput" -> "\"STRUCT<id: INT, part: INT>\""),
         sqlState = "KD000")
       checkError(
         exception = e.getCause.asInstanceOf[AnalysisException],
@@ -700,6 +742,19 @@ private class MissingFilterAttributeScan extends SupportsRuntimeCatalystFilterin
   override def filter(expressions: Array[Expression]): Unit = {}
 }
 
+/** A scan declaring a fully pushed filter attribute that its relation output does not contain. */
+private class MissingFullyPushedFilterAttributeScan extends SupportsRuntimeCatalystFiltering {
+
+  override def readSchema(): StructType = new StructType().add("part", IntegerType)
+
+  override def filterAttributes(): Array[NamedReference] = Array(FieldReference("part"))
+
+  override def fullyPushedFilterAttributes(): Array[NamedReference] =
+    Array(FieldReference("missing"))
+
+  override def filter(expressions: Array[Expression]): Unit = {}
+}
+
 /** A scan declaring a nested runtime-filter attribute beneath an integer column. */
 private class NestedFilterAttributeScan extends SupportsRuntimeCatalystFiltering {
 
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
index 46c42490b4f73..cc003d440eb57 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
@@ -140,12 +140,6 @@ abstract class DataSourceV2SQLSuite
     }
   }
 
-  private def checkExplain(query: String, relationPattern: Regex): Unit = {
-    val explain = spark.sql(s"EXPLAIN EXTENDED $query").head().getString(0)
-    val relations = explain.split("\n").filter(_.contains("RelationV2"))
-    assert(relations.nonEmpty && relations.forall(line => relationPattern.matches(line.trim)))
-  }
-
   test("nested partition source column receives a DPP runtime filter") {
     val fact = s"${catalogAndNamespace}fact_nested_runtime_filter"
     val dim = s"${catalogAndNamespace}dim_nested_runtime_filter"
@@ -183,6 +177,12 @@ abstract class DataSourceV2SQLSuite
       }
     }
   }
+
+  private def checkExplain(query: String, relationPattern: Regex): Unit = {
+    val explain = spark.sql(s"EXPLAIN EXTENDED $query").head().getString(0)
+    val relations = explain.split("\n").filter(_.contains("RelationV2"))
+    assert(relations.nonEmpty && relations.forall(line => relationPattern.matches(line.trim)))
+  }
 }
 
 class DataSourceV2SQLSuiteV1Filter
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedRowLevelOperationCatalystRuntimeFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedRowLevelOperationCatalystRuntimeFilterSuite.scala
index 4a7d3d1bcf18c..becdde5aa2f4a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedRowLevelOperationCatalystRuntimeFilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedRowLevelOperationCatalystRuntimeFilterSuite.scala
@@ -69,7 +69,7 @@ class GroupBasedRowLevelOperationCatalystRuntimeFilterSuite
       expectedFilterAttrs = Seq("dep.name"),
       expectedFilter = GroupFilter(
         scanSchema = "salary INT, dep STRUCT<name: STRING>", groups = Seq("hr")),
-      expectedFilterRefs = Some(Seq("dep")))
+      expectedFilterPaths = Some(Seq(Seq("dep", "name"))))
 
     checkAnswer(
       sql(s"SELECT * FROM $tableNameAsString"),
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationCatalystRuntimeFilterSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationCatalystRuntimeFilterSuiteBase.scala
index cccd8e052a4fe..43ed7797fb5bc 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationCatalystRuntimeFilterSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationCatalystRuntimeFilterSuiteBase.scala
@@ -18,7 +18,7 @@
 package org.apache.spark.sql.connector
 
 import org.apache.spark.sql.Row
-import org.apache.spark.sql.catalyst.expressions.DynamicPruningExpression
+import org.apache.spark.sql.catalyst.expressions.{Attribute, CreateNamedStruct, DynamicPruningExpression, Expression, GetStructFieldObject}
 import org.apache.spark.sql.catalyst.types.DataTypeUtils
 import org.apache.spark.sql.connector.catalog.{BufferedRows, InMemoryRowLevelOperationTable}
 import org.apache.spark.sql.connector.expressions.LogicalExpressions.{identity, reference}
@@ -150,7 +150,7 @@ abstract class RowLevelOperationCatalystRuntimeFilterSuiteBase
         expectedFilterAttrs = Seq("dep.name"),
         expectedFilter = GroupFilter(
           scanSchema = "pk INT, dep STRUCT<name: STRING>", groups = Seq("hr")),
-        expectedFilterRefs = Some(Seq("dep")))
+        expectedFilterPaths = Some(Seq(Seq("dep", "name"))))
 
       checkAnswer(
         sql(s"SELECT * FROM $tableNameAsString"),
@@ -177,7 +177,7 @@ abstract class RowLevelOperationCatalystRuntimeFilterSuiteBase
       executedPlan: SparkPlan,
       expectedFilterAttrs: Seq[String],
       expectedFilter: GroupFilter,
-      expectedFilterRefs: Option[Seq[String]] = None): Unit = {
+      expectedFilterPaths: Option[Seq[Seq[String]]] = None): Unit = {
     val batchScans = collect(executedPlan) { case s: BatchScanExec => s }
     assert(batchScans.nonEmpty, "expected a batch scan for the row-level operation")
     val scan = catalystScan(batchScans.head)
@@ -187,12 +187,12 @@ abstract class RowLevelOperationCatalystRuntimeFilterSuiteBase
     val filterAttrs = scan.filterAttributes().map(_.fieldNames.mkString(".")).toSeq
     assert(filterAttrs === expectedFilterAttrs,
       s"expected the scan to declare $expectedFilterAttrs as filter attributes, got $filterAttrs")
-    val filterRefs = expectedFilterRefs.getOrElse(expectedFilterAttrs)
+    val filterPaths = expectedFilterPaths.getOrElse(expectedFilterAttrs.map(Seq(_)))
 
     batchScans.foreach { batchScan =>
       batchScan.runtimeFilters match {
         case Seq(DynamicPruningExpression(inSubquery: InSubqueryExec)) =>
-          assertGroupFilter(inSubquery, filterRefs, expectedFilter)
+          assertGroupFilter(inSubquery, filterPaths, expectedFilter)
         case other => fail(s"expected a single dynamic pruning group filter, got $other")
       }
     }
@@ -204,7 +204,7 @@ abstract class RowLevelOperationCatalystRuntimeFilterSuiteBase
       s"expected each of the ${batchScans.size} scan node(s) to push the filter once, got $pushed")
     pushed.foreach {
       case inSubquery: InSubqueryExec =>
-        assertGroupFilter(inSubquery, filterRefs, expectedFilter)
+        assertGroupFilter(inSubquery, filterPaths, expectedFilter)
       case other =>
         fail(s"expected the group filter pushed as an InSubqueryExec, got $other")
     }
@@ -222,10 +222,11 @@ abstract class RowLevelOperationCatalystRuntimeFilterSuiteBase
 
   private def assertGroupFilter(
       filter: InSubqueryExec,
-      expectedFilterAttrs: Seq[String],
+      expectedFilterPaths: Seq[Seq[String]],
       expectedFilter: GroupFilter): Unit = {
-    assert(filter.child.references.toSeq.map(_.name) === expectedFilterAttrs,
-      s"expected the group filter keyed on $expectedFilterAttrs, got ${filter.child}")
+    assert(fieldPaths(filter.child).contains(expectedFilterPaths),
+      s"expected the group filter keyed on ${expectedFilterPaths.map(_.mkString("."))}, " +
+        s"got ${filter.child}")
 
     // the second branch of a group-based UPDATE reuses the first branch's subquery, and
     // ReusedSubqueryExec is a leaf node, so unwrap it to reach the plan underneath
@@ -246,6 +247,19 @@ abstract class RowLevelOperationCatalystRuntimeFilterSuiteBase
       s"group filter must select the groups holding matching rows, got ${groups.mkString(", ")}")
   }
 
+  private def fieldPath(expr: Expression): Option[Seq[String]] = expr match {
+    case attr: Attribute => Some(Seq(attr.name))
+    case GetStructFieldObject(child, field) => fieldPath(child).map(_ :+ field.name)
+    case _ => None
+  }
+
+  private def fieldPaths(expr: Expression): Option[Seq[Seq[String]]] = expr match {
+    case struct: CreateNamedStruct =>
+      val paths = struct.valExprs.map(fieldPath)
+      Option.when(paths.forall(_.isDefined))(paths.flatten)
+    case _ => fieldPath(expr).map(Seq(_))
+  }
+
   /** Asserts no group filter was injected, e.g. because the scan does not read the group key. */
   protected def assertNoCatalystGroupFilter(executedPlan: SparkPlan): Unit = {
     val batchScan = collect(executedPlan) { case s: BatchScanExec => s }.head