From 9a27aa001093926ad83ee835a81a2b850bdbc4d1 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Sat, 29 Aug 2026 08:58:59 -0700 Subject: [PATCH 1/8] [SPARK-59068][SQL][4.3] Restore support for nested runtime filter attributes --- .../resources/error/error-conditions.json | 18 ++ .../read/SupportsRuntimeFiltering.java | 12 +- .../read/SupportsRuntimeV2Filtering.java | 10 +- .../expressions/V2ExpressionUtils.scala | 2 +- .../sql/errors/QueryCompilationErrors.scala | 40 +++ .../datasources/v2/DataSourceV2Relation.scala | 74 +++-- .../SupportsRuntimeCatalystFiltering.scala | 18 +- .../connector/catalog/InMemoryBaseTable.scala | 73 ++++- .../InMemoryCatalystRuntimeFilterTable.scala | 30 +- .../InMemoryRowLevelOperationTable.scala | 5 +- .../catalog/InMemoryTableWithV2Filter.scala | 5 +- ...wLevelOperationRuntimeGroupFiltering.scala | 21 +- ...taSourceV2CatalystRuntimeFilterSuite.scala | 270 +++++++++++++++++- .../sql/connector/DataSourceV2SQLSuite.scala | 44 ++- ...lOperationCatalystRuntimeFilterSuite.scala | 31 ++ ...rationCatalystRuntimeFilterSuiteBase.scala | 69 ++++- .../RowLevelOperationSuiteBase.scala | 9 + 17 files changed, 649 insertions(+), 82 deletions(-) diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json index d8873f7dd4db6..57827f820dde8 100644 --- a/common/utils/src/main/resources/error/error-conditions.json +++ b/common/utils/src/main/resources/error/error-conditions.json @@ -2212,6 +2212,24 @@ ], "sqlState" : "KD010" }, + "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE" : { + "message" : [ + "The runtime filter attribute reported by `` in data source scan is invalid for the scan relation output ." + ], + "subClass" : { + "CANNOT_RESOLVE" : { + "message" : [ + "The attribute cannot be resolved." + ] + }, + "NOT_TOP_LEVEL" : { + "message" : [ + "The attribute must be top-level, but it is a nested reference." + ] + } + }, + "sqlState" : "KD000" + }, "DATA_SOURCE_METADATA_SCHEMA_NOT_IMPLEMENTED" : { "message" : [ " does not implement metadataSchema." 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..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 @@ -39,9 +39,8 @@ public interface SupportsRuntimeFiltering extends SupportsRuntimeV2Filtering { * Spark will call {@link #filter(Filter[])} if it can derive a runtime * predicate for any of the filter attributes. *

- * Each reference must be a top-level attribute present in {@link Scan#readSchema()}. - * Nested references and attributes pruned out of the read schema fail to resolve when - * Spark builds the scan relation. + * Each reference must resolve against the scan relation output when Spark builds it. Attributes + * pruned out of {@link Scan#readSchema()} fail to resolve. */ NamedReference[] filterAttributes(); @@ -51,6 +50,13 @@ public interface SupportsRuntimeFiltering extends SupportsRuntimeV2Filtering { * The provided expressions must be interpreted as a set of filters that are ANDed together. * Implementations may use the filters to prune initially planned {@link InputPartition}s. *

+ * 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`}. + *

* 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 6b286f041b01e..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 @@ -52,9 +52,8 @@ public interface SupportsRuntimeV2Filtering extends Scan { * Spark will call {@link #filter(Predicate[])} if it can derive a runtime * predicate for any of the filter attributes. *

- * Each reference must be a top-level attribute present in {@link Scan#readSchema()}. - * Nested references and attributes pruned out of the read schema fail to resolve when - * Spark builds the scan relation. + * Each reference must resolve against the scan relation output when Spark builds it. Attributes + * pruned out of {@link Scan#readSchema()} fail to resolve. */ NamedReference[] filterAttributes(); @@ -64,6 +63,11 @@ public interface SupportsRuntimeV2Filtering extends Scan { * The provided expressions must be interpreted as a set of predicates that are ANDed together. * Implementations may use the predicates to prune initially planned {@link InputPartition}s. *

+ * 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. + *

* 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/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/errors/QueryCompilationErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala index b618399388e6c..d659b100c7bd8 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 @@ -4640,6 +4640,46 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat ) } + def cannotResolveDataSourceRuntimeFilterAttributeError( + attribute: Array[String], + method: String, + scanClass: String, + relationOutput: StructType, + cause: AnalysisException): AnalysisException = { + invalidDataSourceRuntimeFilterAttributeError( + attribute, method, scanClass, relationOutput, "CANNOT_RESOLVE", Some(cause)) + } + + def nestedDataSourceFullyPushedRuntimeFilterAttributeError( + attribute: Array[String], + scanClass: String, + relationOutput: StructType): AnalysisException = { + invalidDataSourceRuntimeFilterAttributeError( + attribute, + "fullyPushedFilterAttributes()", + scanClass, + relationOutput, + "NOT_TOP_LEVEL", + None) + } + + private def invalidDataSourceRuntimeFilterAttributeError( + attribute: Array[String], + method: String, + scanClass: String, + relationOutput: StructType, + errorSubClass: String, + cause: Option[AnalysisException]): AnalysisException = { + new AnalysisException( + errorClass = s"DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.$errorSubClass", + messageParameters = Map( + "attribute" -> toSQLId(attribute.toImmutableArraySeq), + "method" -> method, + "scanClass" -> scanClass, + "relationOutput" -> toSQLType(relationOutput)), + cause = cause) + } + def foundMultipleXMLDataSourceError(provider: String, sourceNames: Seq[String], externalSource: String): Throwable = { 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 1c7a7abe4a8ad..09b90edc6ecd0 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,14 +20,15 @@ package org.apache.spark.sql.execution.datasources.v2 import java.util.{Collections, Optional, OptionalLong} import org.apache.spark.SparkException +import org.apache.spark.sql.AnalysisException import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation, NamedRelation, TimeTravelSpec} import org.apache.spark.sql.catalyst.catalog.{CatalogColumnStat, CatalogStatistics} -import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, AttributeReference, AttributeSet, Expression, SortOrder, V2ExpressionUtils} +import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, AttributeReference, AttributeSet, Expression, NamedExpression, SortOrder, V2ExpressionUtils} import org.apache.spark.sql.catalyst.plans.QueryPlan import org.apache.spark.sql.catalyst.plans.logical.{ColumnStat, ExposesMetadataColumns, Histogram, HistogramBin, LeafNode, LogicalPlan, Statistics} import org.apache.spark.sql.catalyst.plans.logical.statsEstimation.EstimationUtils import org.apache.spark.sql.catalyst.streaming.{StreamingSourceIdentifyingName, Unassigned} -import org.apache.spark.sql.catalyst.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 @@ -35,10 +36,10 @@ 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 -import org.apache.spark.util.ArrayImplicits._ import org.apache.spark.util.Utils /** @@ -198,17 +199,23 @@ case class DataSourceV2ScanRelation( * Resolved attributes that the scan declares for runtime filtering via * [[SupportsRuntimeV2Filtering.filterAttributes]] or * [[SupportsRuntimeCatalystFiltering.filterAttributes]]. Empty when the scan - * implements neither interface or exposes no attributes. + * implements neither interface or exposes no attributes. Accessing this value also validates + * attributes returned by [[SupportsRuntimeCatalystFiltering.fullyPushedFilterAttributes]]. */ lazy val runtimeFilterAttrs: AttributeSet = { checkRuntimeFilteringInterfaces() + checkFullyPushedFilterAttrs() val filterAttrs = scan match { case s: SupportsRuntimeV2Filtering => s.filterAttributes case s: SupportsRuntimeCatalystFiltering => s.filterAttributes() case _ => Array.empty[NamedReference] } - AttributeSet(V2ExpressionUtils.resolveRefs[Attribute]( - filterAttrs.toImmutableArraySeq, this)) + resolveFilterAttrs(filterAttrs, "filterAttributes()") + } + + private lazy val declaredFullyPushedRuntimeFilterAttrs: Array[NamedReference] = scan match { + case s: SupportsRuntimeCatalystFiltering => s.fullyPushedFilterAttributes() + case _ => Array.empty } /** @@ -217,12 +224,34 @@ case class DataSourceV2ScanRelation( */ lazy val fullyPushedRuntimeFilterAttrs: AttributeSet = { checkRuntimeFilteringInterfaces() - val filterAttrs = scan match { - case s: SupportsRuntimeCatalystFiltering => s.fullyPushedFilterAttributes() - case _ => Array.empty[NamedReference] + checkFullyPushedFilterAttrs() + resolveFilterAttrs( + declaredFullyPushedRuntimeFilterAttrs, "fullyPushedFilterAttributes()") + } + + /** + * 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 resolveFilterAttrs( + filterAttrs: Array[NamedReference], + method: String): AttributeSet = { + val resolvedAttrs = filterAttrs.map { ref => + try { + V2ExpressionUtils.resolveRef[NamedExpression](ref, this) + } catch { + case e: AnalysisException => + throw QueryCompilationErrors.cannotResolveDataSourceRuntimeFilterAttributeError( + attribute = ref.fieldNames, + method = method, + scanClass = scan.getClass.getName, + relationOutput = fromAttributes(output), + cause = e) + } } - AttributeSet(V2ExpressionUtils.resolveRefs[Attribute]( - filterAttrs.toImmutableArraySeq, this)) + AttributeSet(resolvedAttrs) } override def name: String = relation.name @@ -271,12 +300,23 @@ case class DataSourceV2ScanRelation( Statistics(sizeInBytes = conf.defaultSizeInBytes) } - private def checkRuntimeFilteringInterfaces(): Unit = scan match { - case _: SupportsRuntimeV2Filtering with SupportsRuntimeCatalystFiltering => - throw SparkException.internalError( - "A scan must not implement both SupportsRuntimeV2Filtering and " + - s"SupportsRuntimeCatalystFiltering, but ${scan.getClass.getName} implements both.") - case _ => + private def checkRuntimeFilteringInterfaces(): Unit = { + scan match { + case _: SupportsRuntimeV2Filtering with SupportsRuntimeCatalystFiltering => + throw SparkException.internalError( + "A scan must not implement both SupportsRuntimeV2Filtering and " + + s"SupportsRuntimeCatalystFiltering, but ${scan.getClass.getName} implements both.") + case _ => + } + } + + private def checkFullyPushedFilterAttrs(): Unit = { + declaredFullyPushedRuntimeFilterAttrs.find(_.fieldNames.length > 1).foreach { ref => + throw QueryCompilationErrors.nestedDataSourceFullyPushedRuntimeFilterAttributeError( + attribute = ref.fieldNames, + scanClass = scan.getClass.getName, + relationOutput = fromAttributes(output)) + } } override def doCanonicalize(): DataSourceV2ScanRelation = { 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 5e5900a211961..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 @@ -41,9 +41,8 @@ trait SupportsRuntimeCatalystFiltering extends Scan { * Returns attributes this scan can be filtered by at runtime. * * Spark will call [[filter]] if it can derive a runtime filter for any of these attributes. - * Each reference must be a top-level attribute present in [[Scan.readSchema]]. Nested - * references and attributes pruned out of the read schema fail to resolve when Spark builds - * the scan relation. + * Each reference must resolve against the scan relation output when Spark builds it. Attributes + * pruned out of [[Scan.readSchema]] fail to resolve. */ def filterAttributes(): Array[NamedReference] @@ -64,8 +63,11 @@ trait SupportsRuntimeCatalystFiltering extends Scan { * predicate over it. * * Each reference must be a top-level attribute present in [[Scan.readSchema]]. Nested - * references and attributes pruned out of the read schema fail to resolve when Spark builds - * the scan relation. + * references are rejected, and attributes pruned out of the read schema fail to resolve, when + * Spark builds the scan relation. Spark cannot currently represent an individual fully pushed + * nested path. A scan must not return the root struct as a substitute unless it can fully + * evaluate predicates over every nested field, since Spark would remove their post-scan + * evaluation as well. */ def fullyPushedFilterAttributes(): Array[NamedReference] = Array.empty @@ -76,9 +78,9 @@ trait SupportsRuntimeCatalystFiltering extends Scan { * Implementations may use the expressions to prune initially planned * [[org.apache.spark.sql.connector.read.InputPartition]]s. * - * An expression may access nested fields of an attribute returned by [[filterAttributes]], as - * that attribute is required to be top-level. The scan is responsible for matching such - * accesses against its own partition layout. + * Spark tracks runtime-filter eligibility by root attribute. If [[filterAttributes]] returns a + * nested reference, an expression may access another nested field under the same root. The scan + * must match each access against its own partition layout and use only expressions it can apply. * * Spark may call this method more than once for the same scan instance: a plan can hold several * scan nodes sharing one scan (e.g. the two branches of a group-based UPDATE), and each pushes 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 0b666ab473fcd..1e2132abc91df 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 @@ -29,7 +29,7 @@ import scala.collection.mutable.{ArrayBuffer, ListBuffer} import scala.jdk.CollectionConverters._ import org.apache.spark.sql.catalyst.InternalRow -import org.apache.spark.sql.catalyst.expressions.{AttributeReference, BindReferences, Cast, EvalMode, Expression => CatalystExpression, GenericInternalRow, JoinedRow, Literal, MetadataStructFieldWithLogicalName, Predicate => CatalystPredicate} +import org.apache.spark.sql.catalyst.expressions.{AttributeReference, BindReferences, Cast, EvalMode, Expression => CatalystExpression, GenericInternalRow, GetStructField, JoinedRow, Literal, MetadataStructFieldWithLogicalName, Predicate => CatalystPredicate} import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, ArrayData, CaseInsensitiveMap, CharVarcharUtils, DateTimeUtils, GenericArrayData, MapData, ResolveDefaultColumns} import org.apache.spark.sql.connector.catalog.constraints.Constraint import org.apache.spark.sql.connector.distributions.{Distribution, Distributions} @@ -733,17 +733,17 @@ abstract class InMemoryBaseTable( catalystPredicates ++= expressions val partAttrs = partitionAttributes if (partAttrs.isEmpty) return + val partAttrRefs = partAttrs.map(_._2) - val resolver = SQLConf.get.resolver expressions.foreach { expr => - val remapped = expr.transform { - case a: AttributeReference => - partAttrs.find(p => resolver(p.name, a.name)).getOrElse(a) + // Top down, so `s.part` is rewritten before its `s` child is considered. + val remapped = expr.transformDown { + case e => partitionAttrFor(e, partAttrs).getOrElse(e) } // Only evaluate expressions whose refs are all partition columns, so we can bind // against the partition key InternalRow (same approach as PartitionPredicateImpl). - if (remapped.references.forall(r => partAttrs.exists(_.exprId == r.exprId))) { - val bound = BindReferences.bindReference(remapped, partAttrs) + if (remapped.references.forall(r => partAttrRefs.exists(_.exprId == r.exprId))) { + val bound = BindReferences.bindReference(remapped, partAttrRefs) val pred = CatalystPredicate.createInterpreted(bound) self.data = self.data.filter { p => try { @@ -765,15 +765,60 @@ abstract class InMemoryBaseTable( /** Predicates recorded by [[filter]], for test assertions only. */ def pushedCatalystPredicates: Seq[CatalystExpression] = catalystPredicates.toSeq - /** AttributeReferences matching the partition-key InternalRow field order. */ - private def partitionAttributes: Seq[AttributeReference] = { + /** + * The `AttributeReference`s standing for the partition key InternalRow fields, in its field + * order, each paired with the name-part sequence of its partition column. The parts are kept + * unflattened so a quoted top-level column `a.b` (parts `Seq("a.b")`) stays distinct from a + * nested column `a`.`b` (parts `Seq("a", "b")`). Example: + * - `PARTITIONED BY (part, s.nested)` -> `(Seq("part"), AttributeReference(part))`, then + * `(Seq("s", "nested"), AttributeReference(s.nested))` + */ + private def partitionAttributes: Seq[(Seq[String], AttributeReference)] = { partitioning.flatMap(_.references()).flatMap { ref => - val name = ref.fieldNames.mkString(".") - readSchema.find(_.name == name).orElse(tableSchema.find(_.name == name)).map { f => - AttributeReference(f.name, f.dataType, f.nullable)() + val path = ref.fieldNames.toImmutableArraySeq + 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)() } }.toSeq } + + /** + * The partition key `AttributeReference` that `e` reads, or None if `e` reads no partition + * column. The path `e` reads is compared to each partition column's name parts component-wise + * with the resolver, so a quoted top-level column `a.b` cannot collide with a nested column + * `a`.`b`. Examples, under `PARTITIONED BY (part, s.nested)` where `nested` is field 0 of `s`: + * - `AttributeReference(part)` -> `AttributeReference(part)` + * - `GetStructField(AttributeReference(s), 0)` -> `AttributeReference(s.nested)` + * - `AttributeReference(s)` -> None if `s` itself is not a partition column, only `s.nested` + */ + private def partitionAttrFor( + e: CatalystExpression, + partAttrs: Seq[(Seq[String], AttributeReference)]): Option[AttributeReference] = { + val resolver = SQLConf.get.resolver + partitionKeyPath(e).flatMap { path => + partAttrs.collectFirst { + case (parts, attr) if parts.length == path.length && + parts.lazyZip(path).forall((part, name) => resolver(part, name)) => attr + } + } + } + + /** + * The name parts `e` reads, or None if it reads neither a column nor a struct field. Each + * `GetStructField` ordinal is the field's position in its parent struct. Examples: + * - `AttributeReference(a)` -> `Seq("a")`, the top level column a + * - `GetStructField(AttributeReference(a), 0)` -> `Seq("a", "b")`, the nested column a.b + * - `GetStructField(GetStructField(AttributeReference(a), 0), 0)` -> `Seq("a", "b", "c")` + */ + private def partitionKeyPath(e: CatalystExpression): Option[Seq[String]] = e match { + case a: AttributeReference => Some(Seq(a.name)) + case g: GetStructField => + partitionKeyPath(g.child).map(parent => parent :+ g.childSchema(g.ordinal).name) + case _ => None + } } case class InMemoryBatchScan( @@ -790,9 +835,9 @@ abstract class InMemoryBaseTable( var pushedFilters: Array[Filter] = Array.empty override def filterAttributes(): Array[NamedReference] = { - val scanFields = readSchema.fields.map(_.name).toSet partitioning.flatMap(_.references) - .filter(ref => scanFields.contains(ref.fieldNames.mkString("."))) + .filter(ref => readSchema.findNestedField( + ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined) } override def filter(filters: Array[Filter]): Unit = { 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 0f32028f20a94..52b808d55101b 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,6 +23,7 @@ import InMemoryCatalystRuntimeFilterTable._ import org.apache.spark.sql.connector.expressions.{NamedReference, 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._ @@ -74,23 +75,28 @@ class InMemoryCatalystRuntimeFilterTable( Option(InMemoryCatalystRuntimeFilterTable.this.properties.get(FilterAttributesKey)) .map(_.split(",").map(_.trim).toSet) + private val fullyPushedFilterAttrs: Set[String] = Option( + InMemoryCatalystRuntimeFilterTable.this.properties.get(FullyPushedFilterAttributesKey)) + .map(_.split(",").map(_.trim).toSet) + .getOrElse(Set.empty) + + /** Partition source columns that are present in the scan read schema. */ + private def partitionAttrs: Array[NamedReference] = { + partitioning.flatMap(_.references()).distinct + .filter(ref => readSchema.findNestedField( + ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined) + } + override def filterAttributes(): Array[NamedReference] = { - val scanFields = readSchema.fields.map(_.name).toSet - partitioning.flatMap(_.references()).filter { ref => - val name = ref.fieldNames.mkString(".") - scanFields.contains(name) && - restrictedFilterAttrs.forall(_.contains(name)) + 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] = { - val fullyPushedFilterAttrs = Option( - InMemoryCatalystRuntimeFilterTable.this.properties.get(FullyPushedFilterAttributesKey)) - .map(_.split(",").map(_.trim).toSet) - .getOrElse(Set.empty) - filterAttributes().filter { ref => - fullyPushedFilterAttrs.contains(ref.fieldNames.mkString(".")) - } + partitionAttrs.filter(ref => fullyPushedFilterAttrs.contains(ref.fieldNames.mkString("."))) } } 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 e9d73d0f9fe1e..84de623a91936 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 @@ -25,6 +25,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._ @@ -66,9 +67,9 @@ 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, 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 87139a3a20e15..2c3ac6710e58c 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,7 +18,7 @@ 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} @@ -95,8 +95,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 @@ -141,13 +142,19 @@ 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) + 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(pruningKeys, 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 63709e32ecd5b..b5b2ed2dee653 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,8 +18,8 @@ package org.apache.spark.sql.connector import org.apache.spark.{SparkConf, SparkException} -import org.apache.spark.sql.{DataFrame, Row} -import org.apache.spark.sql.catalyst.expressions.{Add, AttributeReference, DynamicPruning, DynamicPruningExpression, EqualTo, Expression, GreaterThan, Literal, RLike} +import org.apache.spark.sql.{AnalysisException, DataFrame, Row} +import org.apache.spark.sql.catalyst.expressions.{Add, AttributeReference, DynamicPruning, DynamicPruningExpression, EqualTo, Expression, GetStructField, GreaterThan, Literal, RLike} import org.apache.spark.sql.connector.catalog.{InMemoryCatalystRuntimeFilterTable, InMemoryTableCatalystRuntimeFilterCatalog} import org.apache.spark.sql.connector.expressions.{FieldReference, NamedReference} import org.apache.spark.sql.connector.expressions.filter.Predicate @@ -100,6 +100,117 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession { } } + 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) USING $v2Source " + + "PARTITIONED BY (s.part) " + + "TBLPROPERTIES('fully-pushed-filter-attributes' = 's.part')") + + 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 + } + checkError( + exception = e, + condition = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.NOT_TOP_LEVEL", + parameters = Map( + "attribute" -> "`s`.`part`", + "method" -> "fullyPushedFilterAttributes()", + "scanClass" -> scanClass, + "relationOutput" -> "\"STRUCT>\""), + sqlState = "KD000") + } + } + + 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 + } + val scanClass = classOf[NestedFilterAttributeScan].getName + checkError( + exception = e, + condition = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.CANNOT_RESOLVE", + parameters = Map( + "attribute" -> "`part`.`nested`", + "method" -> "filterAttributes()", + "scanClass" -> scanClass, + "relationOutput" -> "\"STRUCT\""), + sqlState = "KD000") + checkError( + exception = e.getCause.asInstanceOf[AnalysisException], + 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) { + 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 + } + val scanClass = classOf[MissingFilterAttributeScan].getName + checkError( + exception = e, + condition = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.CANNOT_RESOLVE", + parameters = Map( + "attribute" -> "`missing`", + "method" -> "filterAttributes()", + "scanClass" -> scanClass, + "relationOutput" -> "\"STRUCT\""), + 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, + "relationOutput" -> "\"STRUCT\""), + sqlState = "KD000") + checkError( + exception = e.getCause.asInstanceOf[AnalysisException], + 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" @@ -228,6 +339,64 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession { } } + 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, derives STRUCT) 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.derives.toStr = d.value WHERE d.selected = 1""".stripMargin) + checkAnswer(df, Row(2)) + + assertDPPRuntimeFilters(df) + // 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 === 3) + assert(batchScan.filteredPartitions.flatten.size === 1, + s"expected 1 partition after pruning, got ${batchScan.filteredPartitions.flatten.size}") + } + } + } + + 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) 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) { @@ -267,6 +436,69 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession { assert(collectBatchScan(df).runtimeFilters.isEmpty, "Expected no runtime filters for a column outside filterAttributes") assertPushedCatalystPredicates(df, 0) + assertScalarSubqueryEvaluatedAfterScan(df, expected = true) + } + } + + test("nested field of a filter attribute -> pushed with the nested access intact") { + val tbl = s"$catalogName.tbl_nested" + val dim = s"$catalogName.dim_nested" + withTable(tbl, dim) { + sql(s"CREATE TABLE $tbl (id INT, s STRUCT) USING $v2Source " + + "PARTITIONED BY (s.tz)") + for (i <- 0 until 3) { + sql(s"INSERT INTO $tbl VALUES ($i, named_struct('tz', 'tz$i'))") + } + sql(s"CREATE TABLE $dim (val STRING) USING $v2Source") + sql(s"INSERT INTO $dim VALUES ('tz1')") + + // 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"))) + + assertScalarSubqueryRuntimeFilters(df) + val pushed = getPushedCatalystPredicates(df) + assert(pushed.size === 1, s"expected a single pushed predicate, got $pushed") + val nestedAccesses = pushed.head.collect { case g: GetStructField => g } + assert(nestedAccesses.size === 1, + s"expected the pushed predicate to keep the nested access, got ${pushed.head}") + assert(nestedAccesses.head.childSchema.fieldNames.contains("tz")) + } + } + + test("dotted top-level and nested partition columns -> bound to the correct partition slot") { + val tbl = s"$catalogName.tbl_dotted_collision" + val dim = s"$catalogName.dim_dotted_collision" + withTable(tbl, dim) { + // Two partition columns whose dotted names collide: the quoted top-level column `x.y` and + // the nested field `x`.`y`. They carry different values in each row, so a predicate bound + // to the wrong slot would prune the wrong partitions. `x.y` is 3 exactly where `x`.`y` is + // 30, so binding a filter on the nested field to the top-level slot would find nothing. + sql(s"CREATE TABLE $tbl (id INT, `x.y` INT, x STRUCT) USING $v2Source " + + "PARTITIONED BY (`x.y`, x.y)") + for (i <- 0 until 5) { + sql(s"INSERT INTO $tbl VALUES ($i, $i, named_struct('y', ${i * 10}))") + } + sql(s"CREATE TABLE $dim (val INT) USING $v2Source") + sql(s"INSERT INTO $dim VALUES (30)") + + // Alias the table so `f.x.y` unambiguously reads the nested field, not the column `x.y`. + val df = sql(s"SELECT * FROM $tbl f WHERE f.x.y = (SELECT max(val) FROM $dim)") + checkAnswer(df, Row(3, 3, Row(30))) + + assertScalarSubqueryRuntimeFilters(df) + // The pushed predicate keeps the nested access, and the scan prunes to the single partition + // whose nested `x`.`y` is 30 rather than binding to the colliding top-level `x.y` slot. + 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.filteredPartitions.flatten.size === 1, + s"expected 1 partition after pruning, got ${batchScan.filteredPartitions.flatten.size}") } } @@ -398,3 +630,37 @@ 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 = {} +} + +/** 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 { + + 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 = {} +} 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..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 @@ -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} @@ -139,6 +140,44 @@ abstract class DataSourceV2SQLSuite } } + 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) 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}") + } + } + } + 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")) @@ -5440,9 +5479,6 @@ class DataSourceV2SQLSuiteV1Filter } class DataSourceV2SQLSuiteV2Filter extends DataSourceV2SQLSuite { - import org.apache.spark.sql.catalyst.expressions.DynamicPruning - import org.apache.spark.sql.execution.datasources.v2.BatchScanExec - override protected val catalogAndNamespace = "testv2filter.ns1.ns2." test("SPARK-56467: scalar subquery filters on partition columns are pushed into runtimeFilters") { 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..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 @@ -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" + 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", groups = Seq("hr")), + expectedFilterPaths = Some(Seq(Seq("dep", "name")))) + + 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..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,9 +18,11 @@ 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} +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" + 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", groups = Seq("hr")), + expectedFilterPaths = Some(Seq(Seq("dep", "name")))) + + 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 @@ -137,7 +176,8 @@ abstract class RowLevelOperationCatalystRuntimeFilterSuiteBase protected def assertCatalystGroupFilter( executedPlan: SparkPlan, expectedFilterAttrs: Seq[String], - expectedFilter: GroupFilter): Unit = { + expectedFilter: GroupFilter, + 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) @@ -147,11 +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 filterPaths = expectedFilterPaths.getOrElse(expectedFilterAttrs.map(Seq(_))) batchScans.foreach { batchScan => batchScan.runtimeFilters match { case Seq(DynamicPruningExpression(inSubquery: InSubqueryExec)) => - assertGroupFilter(inSubquery, expectedFilterAttrs, expectedFilter) + assertGroupFilter(inSubquery, filterPaths, expectedFilter) case other => fail(s"expected a single dynamic pruning group filter, got $other") } } @@ -163,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, expectedFilterAttrs, expectedFilter) + assertGroupFilter(inSubquery, filterPaths, expectedFilter) case other => fail(s"expected the group filter pushed as an InSubqueryExec, got $other") } @@ -181,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 @@ -205,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 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 117644782b47edb835de11109b5c22ef7a324c64 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Sun, 30 Aug 2026 00:16:32 -0700 Subject: [PATCH 2/8] [SPARK-59068][SQL] Preserve partition transform semantics in test connector --- .../connector/catalog/InMemoryBaseTable.scala | 50 ++++++++----------- ...taSourceV2CatalystRuntimeFilterSuite.scala | 36 +++++++++++++ 2 files changed, 58 insertions(+), 28 deletions(-) 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 1e2132abc91df..eeca328e116fd 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 @@ -29,7 +29,7 @@ import scala.collection.mutable.{ArrayBuffer, ListBuffer} import scala.jdk.CollectionConverters._ import org.apache.spark.sql.catalyst.InternalRow -import org.apache.spark.sql.catalyst.expressions.{AttributeReference, BindReferences, Cast, EvalMode, Expression => CatalystExpression, GenericInternalRow, GetStructField, JoinedRow, Literal, MetadataStructFieldWithLogicalName, Predicate => CatalystPredicate} +import org.apache.spark.sql.catalyst.expressions.{AttributeReference, BoundReference, Cast, EvalMode, Expression => CatalystExpression, GenericInternalRow, GetStructField, JoinedRow, Literal, MetadataStructFieldWithLogicalName, Predicate => CatalystPredicate} import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, ArrayData, CaseInsensitiveMap, CharVarcharUtils, DateTimeUtils, GenericArrayData, MapData, ResolveDefaultColumns} import org.apache.spark.sql.connector.catalog.constraints.Constraint import org.apache.spark.sql.connector.distributions.{Distribution, Distributions} @@ -733,18 +733,15 @@ abstract class InMemoryBaseTable( catalystPredicates ++= expressions val partAttrs = partitionAttributes if (partAttrs.isEmpty) return - val partAttrRefs = partAttrs.map(_._2) expressions.foreach { expr => // Top down, so `s.part` is rewritten before its `s` child is considered. val remapped = expr.transformDown { case e => partitionAttrFor(e, partAttrs).getOrElse(e) } - // Only evaluate expressions whose refs are all partition columns, so we can bind - // against the partition key InternalRow (same approach as PartitionPredicateImpl). - if (remapped.references.forall(r => partAttrRefs.exists(_.exprId == r.exprId))) { - val bound = BindReferences.bindReference(remapped, partAttrRefs) - val pred = CatalystPredicate.createInterpreted(bound) + // Evaluate expressions only when every reference maps to an identity partition-key slot. + if (remapped.references.isEmpty) { + val pred = CatalystPredicate.createInterpreted(remapped) self.data = self.data.filter { p => try { pred.eval(p.asInstanceOf[BufferedRows].partitionKey()) @@ -766,37 +763,34 @@ abstract class InMemoryBaseTable( def pushedCatalystPredicates: Seq[CatalystExpression] = catalystPredicates.toSeq /** - * The `AttributeReference`s standing for the partition key InternalRow fields, in its field - * order, each paired with the name-part sequence of its partition column. The parts are kept - * unflattened so a quoted top-level column `a.b` (parts `Seq("a.b")`) stays distinct from a - * nested column `a`.`b` (parts `Seq("a", "b")`). Example: - * - `PARTITIONED BY (part, s.nested)` -> `(Seq("part"), AttributeReference(part))`, then - * `(Seq("s", "nested"), AttributeReference(s.nested))` + * Identity partition columns paired with their bound partition-key slots. + * + * Only identity transforms expose a source path because their partition-key slot retains the + * source value. Name parts stay separate so a quoted top-level column `a.b` remains distinct + * from a nested column `a`.`b`. */ - private def partitionAttributes: Seq[(Seq[String], AttributeReference)] = { - partitioning.flatMap(_.references()).flatMap { ref => - val path = ref.fieldNames.toImmutableArraySeq - 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)() - } + private def partitionAttributes: Seq[(Seq[String], BoundReference)] = { + partitioning.zipWithIndex.flatMap { + case (IdentityTransform(ref), ordinal) => + val path = ref.fieldNames.toImmutableArraySeq + val resolver = SQLConf.get.resolver + readSchema.findNestedField(path, resolver = resolver) + .orElse(tableSchema.findNestedField(path, resolver = resolver)).map { + case (_, f) => path -> BoundReference(ordinal, f.dataType, f.nullable) + } + case _ => None }.toSeq } /** - * The partition key `AttributeReference` that `e` reads, or None if `e` reads no partition + * The partition-key slot that `e` reads, or None if `e` reads no identity partition * column. The path `e` reads is compared to each partition column's name parts component-wise * with the resolver, so a quoted top-level column `a.b` cannot collide with a nested column - * `a`.`b`. Examples, under `PARTITIONED BY (part, s.nested)` where `nested` is field 0 of `s`: - * - `AttributeReference(part)` -> `AttributeReference(part)` - * - `GetStructField(AttributeReference(s), 0)` -> `AttributeReference(s.nested)` - * - `AttributeReference(s)` -> None if `s` itself is not a partition column, only `s.nested` + * `a`.`b`. */ private def partitionAttrFor( e: CatalystExpression, - partAttrs: Seq[(Seq[String], AttributeReference)]): Option[AttributeReference] = { + partAttrs: Seq[(Seq[String], BoundReference)]): Option[BoundReference] = { val resolver = SQLConf.get.resolver partitionKeyPath(e).flatMap { path => partAttrs.collectFirst { 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 b5b2ed2dee653..bf4c56627fe11 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 @@ -502,6 +502,42 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession { } } + test("nested runtime filters preserve partition transform semantics and key ordinals") { + val tbl = s"$catalogName.tbl_nested_transforms" + val transformedDim = s"$catalogName.dim_nested_transformed" + val identityDim = s"$catalogName.dim_nested_identity" + withTable(tbl, transformedDim, identityDim) { + sql(s"CREATE TABLE $tbl (id INT, transformed STRUCT, " + + s"identity STRUCT) USING $v2Source " + + "PARTITIONED BY (truncate(transformed.part, 1), identity.part)") + sql(s"INSERT INTO $tbl VALUES " + + "(1, named_struct('part', 'AB'), named_struct('part', 'X')), " + + "(2, named_struct('part', 'CD'), named_struct('part', 'Y'))") + sql(s"CREATE TABLE $transformedDim (value STRING) USING $v2Source") + sql(s"INSERT INTO $transformedDim VALUES ('AB')") + sql(s"CREATE TABLE $identityDim (value STRING) USING $v2Source") + sql(s"INSERT INTO $identityDim VALUES ('Y')") + + val transformedDf = sql(s"SELECT id FROM $tbl " + + s"WHERE transformed.part = (SELECT max(value) FROM $transformedDim)") + checkAnswer(transformedDf, Row(1)) + assertScalarSubqueryRuntimeFilters(transformedDf) + val transformedScan = collectBatchScan(transformedDf) + assert(transformedScan.inputPartitions.size === 2) + assert(transformedScan.filteredPartitions.flatten.size === 2, + "a predicate on a transform source must not be evaluated against its partition key") + + val identityDf = sql(s"SELECT id FROM $tbl " + + s"WHERE identity.part = (SELECT max(value) FROM $identityDim)") + checkAnswer(identityDf, Row(2)) + assertScalarSubqueryRuntimeFilters(identityDf) + val identityScan = collectBatchScan(identityDf) + assert(identityScan.inputPartitions.size === 2) + assert(identityScan.filteredPartitions.flatten.size === 1, + "an identity partition source must bind to its original partition-key slot") + } + } + test("no runtime filter -> filter() is never called") { val tbl = s"$catalogName.tbl5" withTable(tbl) { From bf0c7272ae635aee69f2fc91ab7f658523d18ed2 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Mon, 31 Aug 2026 11:39:51 -0700 Subject: [PATCH 3/8] [SPARK-59068][SQL] Restrict test runtime filtering to identity transforms --- .../connector/catalog/InMemoryBaseTable.scala | 52 ++++++++-------- .../InMemoryCatalystRuntimeFilterTable.scala | 11 +++- .../catalog/InMemoryTableWithV2Filter.scala | 59 ++++++++++--------- ...taSourceV2CatalystRuntimeFilterSuite.scala | 24 ++++++++ .../sql/connector/DataSourceV2SQLSuite.scala | 34 +++++++++++ 5 files changed, 126 insertions(+), 54 deletions(-) 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 eeca328e116fd..db4ea5344ede8 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 @@ -229,6 +229,10 @@ abstract class InMemoryBaseTable( } } + protected def identityPartitionReferences: Array[NamedReference] = { + partitioning.collect { case IdentityTransform(ref) => ref } + } + private val UTC = ZoneId.of("UTC") private val EPOCH_LOCAL_DATE = Instant.EPOCH.atZone(UTC).toLocalDate @@ -715,11 +719,10 @@ abstract class InMemoryBaseTable( /** * Reference implementation of [[SupportsRuntimeCatalystFiltering.filter]] for the in-memory - * fixtures: records what was pushed, and for expressions referencing only partition columns - * binds them against the partition key and drops partitions that do not match. Binding and - * interpreting rather than pattern matching a fixed set of operators is what lets the fixture - * honor an arbitrary pushed expression, the same way `PartitionPredicateImpl` does. Mixing - * classes supply their own `filterAttributes()`. + * fixtures: records what was pushed, and binds expressions referencing only identity partition + * columns against the partition key to drop partitions that do not match. Interpreting the + * bound expression lets the fixture honor arbitrary pushed expressions. Mixing classes supply + * their own `filterAttributes()`. */ trait CatalystRuntimeFilteringScan extends SupportsRuntimeCatalystFiltering { self: BatchScanBaseClass => @@ -829,30 +832,31 @@ abstract class InMemoryBaseTable( var pushedFilters: Array[Filter] = Array.empty override def filterAttributes(): Array[NamedReference] = { - partitioning.flatMap(_.references) + identityPartitionReferences .filter(ref => readSchema.findNestedField( ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined) } override def filter(filters: Array[Filter]): Unit = { - if (partitioning.length == 1 && partitioning.head.references().length == 1) { - val ref = partitioning.head.references().head - filters.foreach { - case In(attrName, values) if attrName == ref.toString => - val matchingKeys = values.map { value => - if (value != null) value.toString else null - }.toSet - this.data = this.data.filter(partition => { - val rows = partition.asInstanceOf[BufferedRows] - rows.key match { - // null partitions are represented as Seq(null) - case Seq(null) => matchingKeys.contains(null) - case _ => matchingKeys.contains(rows.keyString()) - } - }) - - case _ => // skip - } + partitioning match { + case Array(IdentityTransform(ref)) => + filters.foreach { + case In(attrName, values) if attrName == ref.toString => + val matchingKeys = values.map { value => + if (value != null) value.toString else null + }.toSet + this.data = this.data.filter(partition => { + val rows = partition.asInstanceOf[BufferedRows] + rows.key match { + // null partitions are represented as Seq(null) + case Seq(null) => matchingKeys.contains(null) + case _ => matchingKeys.contains(rows.keyString()) + } + }) + + case _ => // skip + } + case _ => } } } 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 52b808d55101b..35a94ef5a1ef2 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 @@ -87,6 +87,13 @@ class InMemoryCatalystRuntimeFilterTable( ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined) } + /** Identity partition columns that are present in the scan read schema. */ + private def identityPartitionAttrs: Array[NamedReference] = { + identityPartitionReferences.distinct + .filter(ref => readSchema.findNestedField( + ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined) + } + override def filterAttributes(): Array[NamedReference] = { partitionAttrs.filter { ref => restrictedFilterAttrs.forall(_.contains(ref.fieldNames.mkString("."))) @@ -96,7 +103,9 @@ class InMemoryCatalystRuntimeFilterTable( // 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.fieldNames.mkString("."))) + identityPartitionAttrs.filter { ref => + fullyPushedFilterAttrs.contains(ref.fieldNames.mkString(".")) + } } } 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 84de623a91936..4076c63aea7d1 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 @@ -21,7 +21,7 @@ import java.util import org.scalatest.Assertions.assert -import org.apache.spark.sql.connector.expressions.{FieldReference, LiteralValue, NamedReference, Transform} +import org.apache.spark.sql.connector.expressions.{FieldReference, IdentityTransform, LiteralValue, NamedReference, Transform} 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} @@ -67,44 +67,45 @@ class InMemoryTableWithV2Filter( extends BatchScanBaseClass(_data, readSchema, tableSchema) with SupportsRuntimeV2Filtering { override def filterAttributes(): Array[NamedReference] = { - partitioning.flatMap(_.references) + identityPartitionReferences .filter(ref => readSchema.findNestedField( ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined) } override def filter(filters: Array[Predicate]): Unit = { - if (partitioning.length == 1 && partitioning.head.references().length == 1) { - val ref = partitioning.head.references().head - filters.foreach { - case p : Predicate if p.name().equals("IN") => - if (p.children().length > 1) { - val filterRef = p.children()(0).asInstanceOf[FieldReference].references.head - if (filterRef.toString.equals(ref.toString)) { - val matchingKeys = - p.children().drop(1).map(_.asInstanceOf[LiteralValue[_]].value.toString).toSet - data = data.filter(partition => { - val key = partition.asInstanceOf[BufferedRows].keyString() - matchingKeys.contains(key) - }) - } - } - case p : Predicate if p.name().equals("=") => - if (p.children().length == 2) { - val filterRef = p.children()(0).asInstanceOf[FieldReference].references.head - if (filterRef.toString.equals(ref.toString)) { - val matchingKey = p.children()(1).asInstanceOf[LiteralValue[_]].value - if (matchingKey != null) { + partitioning match { + case Array(IdentityTransform(ref)) => + filters.foreach { + case p : Predicate if p.name().equals("IN") => + if (p.children().length > 1) { + val filterRef = p.children()(0).asInstanceOf[FieldReference].references.head + if (filterRef.toString.equals(ref.toString)) { + val matchingKeys = + p.children().drop(1).map(_.asInstanceOf[LiteralValue[_]].value.toString).toSet data = data.filter(partition => { val key = partition.asInstanceOf[BufferedRows].keyString() - key == matchingKey.toString + matchingKeys.contains(key) }) - } else { - data = Seq.empty // NULL = anything is always false } } - } - case _ => // Ignore unsupported predicate types - } + case p : Predicate if p.name().equals("=") => + if (p.children().length == 2) { + val filterRef = p.children()(0).asInstanceOf[FieldReference].references.head + if (filterRef.toString.equals(ref.toString)) { + val matchingKey = p.children()(1).asInstanceOf[LiteralValue[_]].value + if (matchingKey != null) { + data = data.filter(partition => { + val key = partition.asInstanceOf[BufferedRows].keyString() + key == matchingKey.toString + }) + } else { + data = Seq.empty // NULL = anything is always false + } + } + } + case _ => // Ignore unsupported predicate types + } + case _ => } } } 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 bf4c56627fe11..1b9b6719f8a0f 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 @@ -100,6 +100,30 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession { } } + test("transformed partition source cannot be declared fully pushed") { + val tbl = s"$catalogName.tbl_transformed_fully_pushed" + val dim = s"$catalogName.dim_transformed_fully_pushed" + withTable(tbl, dim) { + sql(s"CREATE TABLE $tbl (id INT, part DATE) USING $v2Source " + + "PARTITIONED BY (days(part)) " + + "TBLPROPERTIES('fully-pushed-filter-attributes' = 'part')") + sql(s"INSERT INTO $tbl VALUES " + + "(1, DATE '2026-08-01'), (2, DATE '2026-08-02')") + sql(s"CREATE TABLE $dim (value DATE) USING $v2Source") + sql(s"INSERT INTO $dim VALUES (DATE '2026-08-02')") + + val df = sql(s"SELECT * FROM $tbl WHERE part = (SELECT max(value) FROM $dim)") + checkAnswer(df, Row(2, java.sql.Date.valueOf("2026-08-02"))) + + assertScalarSubqueryRuntimeFilters(df) + assertPushedCatalystPredicates(df, expected = 1) + assertScalarSubqueryEvaluatedAfterScan(df, expected = true) + val scan = collectBatchScan(df) + assert(scan.inputPartitions.size === 2) + assert(scan.filteredPartitions.flatten.size === 2) + } + } + test("nested fully pushed filter attribute -> rejected without a runtime filter") { val tbl = s"$catalogName.tbl_nested_fully_pushed" withTable(tbl) { 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 cc003d440eb57..7dd22878024bc 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 @@ -178,6 +178,40 @@ abstract class DataSourceV2SQLSuite } } + test("transformed nested partition source does not receive a DPP runtime filter") { + val fact = s"${catalogAndNamespace}fact_transformed_nested_runtime_filter" + val dim = s"${catalogAndNamespace}dim_transformed_nested_runtime_filter" + withTable(fact, dim) { + sql(s"CREATE TABLE $fact " + + "(id INT, derives STRUCT) USING " + + s"$v2Format PARTITIONED BY (truncate(derives.toStr, 1))") + 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 factScan = collect(df.queryExecution.executedPlan) { + case b: BatchScanExec if b.output.exists(_.name == "derives") => b + }.head + assert(factScan.runtimeFilters.isEmpty, + s"expected no runtime filters on a transform source, got ${factScan.runtimeFilters}") + assert(factScan.partitions.size === 3) + assert(factScan.filteredPartitions.flatten.size === 3) + } + } + } + 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")) From f464fadf7d69baedd89def06da8ef1cb5f4546c0 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Mon, 31 Aug 2026 12:20:44 -0700 Subject: [PATCH 4/8] [SPARK-59068][SQL] Validate fully pushed runtime filter attributes --- .../sql/connector/read/SupportsRuntimeFiltering.java | 6 +++--- .../datasources/v2/DataSourceV2Relation.scala | 12 ++++++++---- .../DataSourceV2CatalystRuntimeFilterSuite.scala | 4 ++-- 3 files changed, 13 insertions(+), 9 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 c752c8c95d54f..51531234a6a64 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,9 +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`}. + * Nested paths are encoded in a V1 {@link Filter} as dot-separated names, with each path part + * quoted as needed, such as {@code parent.`child.with.dot`}. A top-level column whose name + * contains a dot remains quoted, such as {@code `parent.child`}. *

* 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/execution/datasources/v2/DataSourceV2Relation.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala index 09b90edc6ecd0..f8b868e13c447 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 @@ -204,7 +204,7 @@ case class DataSourceV2ScanRelation( */ lazy val runtimeFilterAttrs: AttributeSet = { checkRuntimeFilteringInterfaces() - checkFullyPushedFilterAttrs() + resolvedFullyPushedRuntimeFilterAttrs val filterAttrs = scan match { case s: SupportsRuntimeV2Filtering => s.filterAttributes case s: SupportsRuntimeCatalystFiltering => s.filterAttributes() @@ -218,15 +218,19 @@ case class DataSourceV2ScanRelation( case _ => Array.empty } + private lazy val resolvedFullyPushedRuntimeFilterAttrs: AttributeSet = { + checkFullyPushedFilterAttrs() + resolveFilterAttrs( + declaredFullyPushedRuntimeFilterAttrs, "fullyPushedFilterAttributes()") + } + /** * 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() - checkFullyPushedFilterAttrs() - resolveFilterAttrs( - declaredFullyPushedRuntimeFilterAttrs, "fullyPushedFilterAttributes()") + resolvedFullyPushedRuntimeFilterAttrs } /** 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 1b9b6719f8a0f..57ee07515438e 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 @@ -206,7 +206,7 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession { } } - test("missing fully pushed filter attribute -> identifies the declaring method") { + test("missing fully pushed filter attribute -> rejected through runtime filter attrs") { 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)") @@ -216,7 +216,7 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession { }.getOrElse(fail("Expected a DataSourceV2ScanRelation")) val e = intercept[AnalysisException] { scanRelation.copy(scan = new MissingFullyPushedFilterAttributeScan) - .fullyPushedRuntimeFilterAttrs + .runtimeFilterAttrs } val scanClass = classOf[MissingFullyPushedFilterAttributeScan].getName checkError( From eb34c4661ecbd08d107b0d15a2e496ae0d40b3cd Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Thu, 3 Sep 2026 11:17:07 -0700 Subject: [PATCH 5/8] [SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures --- .../resources/error/error-conditions.json | 5 ++ .../read/SupportsRuntimeFiltering.java | 3 +- .../read/SupportsRuntimeV2Filtering.java | 3 +- .../sql/errors/QueryCompilationErrors.scala | 13 +++ .../datasources/v2/DataSourceV2Relation.scala | 66 ++++++++++----- .../SupportsRuntimeCatalystFiltering.scala | 7 +- .../connector/catalog/InMemoryBaseTable.scala | 44 +++++----- .../InMemoryRowLevelOperationTable.scala | 2 +- .../catalog/InMemoryTableWithV2Filter.scala | 55 +++++++------ .../datasources/v2/PushDownUtils.scala | 9 ++- .../dynamicpruning/PartitionPruning.scala | 8 +- ...wLevelOperationRuntimeGroupFiltering.scala | 20 ++--- ...taSourceV2CatalystRuntimeFilterSuite.scala | 81 +++++++++++++------ ...lOperationCatalystRuntimeFilterSuite.scala | 7 +- ...rationCatalystRuntimeFilterSuiteBase.scala | 45 +++++++---- 15 files changed, 229 insertions(+), 139 deletions(-) diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json index 57827f820dde8..9ec602506dcab 100644 --- a/common/utils/src/main/resources/error/error-conditions.json +++ b/common/utils/src/main/resources/error/error-conditions.json @@ -2222,6 +2222,11 @@ "The attribute cannot be resolved." ] }, + "NOT_IN_FILTER_ATTRIBUTES" : { + "message" : [ + "The attribute must also be returned by `filterAttributes()`." + ] + }, "NOT_TOP_LEVEL" : { "message" : [ "The attribute must be top-level, but it is a nested reference." 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 51531234a6a64..cdb1e4c842ae8 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 @@ -50,7 +50,8 @@ 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. *

- * Spark tracks runtime-filter eligibility by root attribute. If {@link #filterAttributes()} + * Spark currently 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 dot-separated names, with each path part 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 671a822096bbe..d43b3eab545a2 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 @@ -63,7 +63,8 @@ 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. *

- * Spark tracks runtime-filter eligibility by root attribute. If {@link #filterAttributes()} + * Spark currently 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. 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 d659b100c7bd8..d73a1a4548534 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 @@ -4663,6 +4663,19 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat None) } + def fullyPushedDataSourceRuntimeFilterAttributeNotFilterableError( + attribute: Array[String], + scanClass: String, + relationOutput: StructType): AnalysisException = { + invalidDataSourceRuntimeFilterAttributeError( + attribute, + "fullyPushedFilterAttributes()", + scanClass, + relationOutput, + "NOT_IN_FILTER_ATTRIBUTES", + None) + } + private def invalidDataSourceRuntimeFilterAttributeError( attribute: Array[String], method: 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 f8b868e13c447..27e07ca4a4e32 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 @@ -25,7 +25,7 @@ import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation, NamedRelat import org.apache.spark.sql.catalyst.catalog.{CatalogColumnStat, CatalogStatistics} 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.{ColumnStat, ExposesMetadataColumns, Histogram, HistogramBin, LeafNode, LocalRelation, LogicalPlan, Statistics} import org.apache.spark.sql.catalyst.plans.logical.statsEstimation.EstimationUtils import org.apache.spark.sql.catalyst.streaming.{StreamingSourceIdentifyingName, Unassigned} import org.apache.spark.sql.catalyst.types.DataTypeUtils.{fromAttributes, toAttributes} @@ -205,12 +205,13 @@ case class DataSourceV2ScanRelation( lazy val runtimeFilterAttrs: AttributeSet = { checkRuntimeFilteringInterfaces() resolvedFullyPushedRuntimeFilterAttrs - val filterAttrs = scan match { - case s: SupportsRuntimeV2Filtering => s.filterAttributes - case s: SupportsRuntimeCatalystFiltering => s.filterAttributes() - case _ => Array.empty[NamedReference] - } - resolveFilterAttrs(filterAttrs, "filterAttributes()") + resolveFilterAttrs(declaredRuntimeFilterAttrs, "filterAttributes()") + } + + private lazy val declaredRuntimeFilterAttrs: Array[NamedReference] = scan match { + case s: SupportsRuntimeV2Filtering => s.filterAttributes + case s: SupportsRuntimeCatalystFiltering => s.filterAttributes() + case _ => Array.empty } private lazy val declaredFullyPushedRuntimeFilterAttrs: Array[NamedReference] = scan match { @@ -242,20 +243,8 @@ case class DataSourceV2ScanRelation( private def resolveFilterAttrs( filterAttrs: Array[NamedReference], method: String): AttributeSet = { - val resolvedAttrs = filterAttrs.map { ref => - try { - V2ExpressionUtils.resolveRef[NamedExpression](ref, this) - } catch { - case e: AnalysisException => - throw QueryCompilationErrors.cannotResolveDataSourceRuntimeFilterAttributeError( - attribute = ref.fieldNames, - method = method, - scanClass = scan.getClass.getName, - relationOutput = fromAttributes(output), - cause = e) - } - } - AttributeSet(resolvedAttrs) + DataSourceV2ScanRelation.resolveRuntimeFilterAttrs( + filterAttrs, method, scan.getClass.getName, output) } override def name: String = relation.name @@ -321,6 +310,17 @@ case class DataSourceV2ScanRelation( scanClass = scan.getClass.getName, relationOutput = fromAttributes(output)) } + declaredFullyPushedRuntimeFilterAttrs.find { fullyPushedRef => + !declaredRuntimeFilterAttrs.exists { filterRef => + fullyPushedRef.fieldNames.length == filterRef.fieldNames.length && + fullyPushedRef.fieldNames.lazyZip(filterRef.fieldNames).forall(conf.resolver) + } + }.foreach { ref => + throw QueryCompilationErrors.fullyPushedDataSourceRuntimeFilterAttributeNotFilterableError( + attribute = ref.fieldNames, + scanClass = scan.getClass.getName, + relationOutput = fromAttributes(output)) + } } override def doCanonicalize(): DataSourceV2ScanRelation = { @@ -342,6 +342,30 @@ case class DataSourceV2ScanRelation( } } +object DataSourceV2ScanRelation { + private[sql] def resolveRuntimeFilterAttrs( + filterAttrs: Array[NamedReference], + method: String, + scanClass: String, + output: Seq[AttributeReference]): AttributeSet = { + val plan = LocalRelation(output) + val resolvedAttrs = filterAttrs.map { ref => + try { + V2ExpressionUtils.resolveRef[NamedExpression](ref, plan) + } catch { + case e: AnalysisException => + throw QueryCompilationErrors.cannotResolveDataSourceRuntimeFilterAttributeError( + attribute = ref.fieldNames, + method = method, + scanClass = scanClass, + relationOutput = fromAttributes(output), + cause = e) + } + } + AttributeSet(resolvedAttrs) + } +} + /** * A specialization of [[DataSourceV2RelationBase]] that supports streaming scan. * It will be transformed to [[StreamingDataSourceV2ScanRelation]] during the planning phase of 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 ec34efed26252..bdba27f20e8b5 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 @@ -78,9 +78,10 @@ trait SupportsRuntimeCatalystFiltering extends Scan { * Implementations may use the expressions to prune initially planned * [[org.apache.spark.sql.connector.read.InputPartition]]s. * - * 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 currently 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 db4ea5344ede8..af2fd7e59690d 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 @@ -786,10 +786,9 @@ abstract class InMemoryBaseTable( } /** - * The partition-key slot that `e` reads, or None if `e` reads no identity partition - * column. The path `e` reads is compared to each partition column's name parts component-wise - * with the resolver, so a quoted top-level column `a.b` cannot collide with a nested column - * `a`.`b`. + * The partition-key slot that `e` reads, or None if `e` reads no identity partition column. + * The path `e` reads is compared to each partition column's name parts component-wise with the + * resolver, so a quoted top-level column `a.b` cannot collide with a nested column `a`.`b`. */ private def partitionAttrFor( e: CatalystExpression, @@ -838,25 +837,24 @@ abstract class InMemoryBaseTable( } override def filter(filters: Array[Filter]): Unit = { - partitioning match { - case Array(IdentityTransform(ref)) => - filters.foreach { - case In(attrName, values) if attrName == ref.toString => - val matchingKeys = values.map { value => - if (value != null) value.toString else null - }.toSet - this.data = this.data.filter(partition => { - val rows = partition.asInstanceOf[BufferedRows] - rows.key match { - // null partitions are represented as Seq(null) - case Seq(null) => matchingKeys.contains(null) - case _ => matchingKeys.contains(rows.keyString()) - } - }) - - case _ => // skip - } - case _ => + if (partitioning.length == 1 && identityPartitionReferences.length == 1) { + val ref = identityPartitionReferences.head + filters.foreach { + case In(attrName, values) if attrName == ref.toString => + val matchingKeys = values.map { value => + if (value != null) value.toString else null + }.toSet + this.data = this.data.filter(partition => { + val rows = partition.asInstanceOf[BufferedRows] + rows.key match { + // null partitions are represented as Seq(null) + case Seq(null) => matchingKeys.contains(null) + case _ => matchingKeys.contains(rows.keyString()) + } + }) + + case _ => // skip + } } } } 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 428a9215dcce0..2bbeff520f0d1 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 @@ -311,7 +311,7 @@ class InMemoryRowLevelOperationTable private ( with CatalystRuntimeFilteringScan { override def filterAttributes(): Array[NamedReference] = { - partitioning.flatMap(_.references()) + identityPartitionReferences .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 4076c63aea7d1..08230f39c4b2c 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 @@ -73,39 +73,38 @@ class InMemoryTableWithV2Filter( } override def filter(filters: Array[Predicate]): Unit = { - partitioning match { - case Array(IdentityTransform(ref)) => - filters.foreach { - case p : Predicate if p.name().equals("IN") => - if (p.children().length > 1) { - val filterRef = p.children()(0).asInstanceOf[FieldReference].references.head - if (filterRef.toString.equals(ref.toString)) { - val matchingKeys = - p.children().drop(1).map(_.asInstanceOf[LiteralValue[_]].value.toString).toSet + if (partitioning.length == 1 && identityPartitionReferences.length == 1) { + val ref = identityPartitionReferences.head + filters.foreach { + case p : Predicate if p.name().equals("IN") => + if (p.children().length > 1) { + val filterRef = p.children()(0).asInstanceOf[FieldReference].references.head + if (filterRef.toString.equals(ref.toString)) { + val matchingKeys = + p.children().drop(1).map(_.asInstanceOf[LiteralValue[_]].value.toString).toSet + data = data.filter(partition => { + val key = partition.asInstanceOf[BufferedRows].keyString() + matchingKeys.contains(key) + }) + } + } + case p : Predicate if p.name().equals("=") => + if (p.children().length == 2) { + val filterRef = p.children()(0).asInstanceOf[FieldReference].references.head + if (filterRef.toString.equals(ref.toString)) { + val matchingKey = p.children()(1).asInstanceOf[LiteralValue[_]].value + if (matchingKey != null) { data = data.filter(partition => { val key = partition.asInstanceOf[BufferedRows].keyString() - matchingKeys.contains(key) + key == matchingKey.toString }) + } else { + data = Seq.empty // NULL = anything is always false } } - case p : Predicate if p.name().equals("=") => - if (p.children().length == 2) { - val filterRef = p.children()(0).asInstanceOf[FieldReference].references.head - if (filterRef.toString.equals(ref.toString)) { - val matchingKey = p.children()(1).asInstanceOf[LiteralValue[_]].value - if (matchingKey != null) { - data = data.filter(partition => { - val key = partition.asInstanceOf[BufferedRows].keyString() - key == matchingKey.toString - }) - } else { - data = Seq.empty // NULL = anything is always false - } - } - } - case _ => // Ignore unsupported predicate types - } - case _ => + } + case _ => // Ignore unsupported predicate types + } } } } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala index 9a1d00a96a685..e252edf1013c2 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala @@ -22,7 +22,7 @@ import scala.collection.mutable import org.apache.spark.SparkException import org.apache.spark.internal.{Logging, LogKeys} import org.apache.spark.sql.AnalysisException -import org.apache.spark.sql.catalyst.expressions.{AttributeReference, AttributeSet, DynamicPruning, DynamicPruningExpression, Expression, ExpressionSet, GetStructField, Literal, NamedExpression, PythonUDF, SchemaPruning, SubqueryExpression, V2ExpressionUtils} +import org.apache.spark.sql.catalyst.expressions.{AttributeReference, AttributeSet, DynamicPruning, DynamicPruningExpression, Expression, ExpressionSet, GetStructField, Literal, NamedExpression, PythonUDF, SchemaPruning, SubqueryExpression} import org.apache.spark.sql.catalyst.plans.logical.SampleMethod import org.apache.spark.sql.catalyst.plans.physical.{KeyedPartitioning, Partitioning} import org.apache.spark.sql.catalyst.types.DataTypeUtils @@ -212,8 +212,11 @@ object PushDownUtils extends Logging { // filters whose translation was not already accepted in the first pass. (See SPARK-55596) // Only candidates whose referenced columns are declared in filterAttributes() are eligible. val partPredicatesPushed = filterableScan.supportsIterativePushdown() && { - val filterAttrs = V2ExpressionUtils.resolveAttributeRefs( - filterableScan.filterAttributes(), output) + val filterAttrs = DataSourceV2ScanRelation.resolveRuntimeFilterAttrs( + filterableScan.filterAttributes(), + "filterAttributes()", + filterableScan.getClass.getName, + output) val pushed = filterableScan.pushedPredicates().toSet val candidates = runtimeFilters.filter { f => !filtersToTranslated.get(f).exists(pushed.contains) && diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala index 1bf12a695bc4c..7a17302140563 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala @@ -80,17 +80,13 @@ object PartitionPruning extends Rule[LogicalPlan] with PredicateHelper with Join None } case (resExp, r @ ExtractV2Scan(scan: SupportsRuntimeV2Filtering)) => - val filterAttrs = V2ExpressionUtils.resolveAttributeRefs( - scan.filterAttributes, r.output) - if (resExp.references.subsetOf(filterAttrs)) { + if (resExp.references.subsetOf(r.runtimeFilterAttrs)) { Some(r) } else { None } case (resExp, r @ ExtractV2Scan(scan: SupportsRuntimeCatalystFiltering)) => - val filterAttrs = V2ExpressionUtils.resolveAttributeRefs( - scan.filterAttributes(), r.output) - if (resExp.references.subsetOf(filterAttrs)) { + if (resExp.references.subsetOf(r.runtimeFilterAttrs)) { Some(r) } else { None 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 2c3ac6710e58c..584478ade938f 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 @@ -53,32 +53,32 @@ class RowLevelOperationRuntimeGroupFiltering(optimizeSubqueries: Rule[LogicalPla override def apply(plan: LogicalPlan): LogicalPlan = plan transformDown { case GroupBasedRowLevelOperation(replaceData, _, Some(cond), - ExtractV2Scan(scan: SupportsRuntimeV2Filtering)) - if canInjectGroupFilters(cond, scan.filterAttributes) => + r @ ExtractV2Scan(scan: SupportsRuntimeV2Filtering)) + if canInjectGroupFilters(cond, r) => injectGroupFilters(replaceData, cond, scan, scan.filterAttributes) case GroupBasedRowLevelOperation(replaceData, _, Some(cond), - ExtractV2Scan(scan: SupportsRuntimeCatalystFiltering)) - if canInjectGroupFilters(cond, scan.filterAttributes()) => + r @ ExtractV2Scan(scan: SupportsRuntimeCatalystFiltering)) + if canInjectGroupFilters(cond, r) => injectGroupFilters(replaceData, cond, scan, scan.filterAttributes()) case DeltaBasedRowLevelOperation(writeDelta, _, Some(cond), - ExtractV2Scan(scan: SupportsRuntimeV2Filtering)) - if canInjectGroupFilters(cond, scan.filterAttributes) => + r @ ExtractV2Scan(scan: SupportsRuntimeV2Filtering)) + if canInjectGroupFilters(cond, r) => injectGroupFilters(writeDelta, cond, scan, scan.filterAttributes) case DeltaBasedRowLevelOperation(writeDelta, _, Some(cond), - ExtractV2Scan(scan: SupportsRuntimeCatalystFiltering)) - if canInjectGroupFilters(cond, scan.filterAttributes()) => + r @ ExtractV2Scan(scan: SupportsRuntimeCatalystFiltering)) + if canInjectGroupFilters(cond, r) => injectGroupFilters(writeDelta, cond, scan, scan.filterAttributes()) } private def canInjectGroupFilters( cond: Expression, - filterAttrs: Array[NamedReference]): Boolean = { + scanRelation: DataSourceV2ScanRelation): Boolean = { conf.runtimeRowLevelOperationGroupFilterEnabled && cond != TrueLiteral && - filterAttrs.nonEmpty + scanRelation.runtimeFilterAttrs.nonEmpty } private def injectGroupFilters( 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 57ee07515438e..9b4fd05afc5c3 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 @@ -100,7 +100,7 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession { } } - test("transformed partition source cannot be declared fully pushed") { + test("Catalyst runtime filtering fixture fully pushes only identity partition sources") { val tbl = s"$catalogName.tbl_transformed_fully_pushed" val dim = s"$catalogName.dim_transformed_fully_pushed" withTable(tbl, dim) { @@ -150,6 +150,29 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession { } } + test("fully pushed root attribute outside filterAttributes -> rejected") { + val tbl = s"$catalogName.tbl_fully_pushed_root" + withTable(tbl) { + sql(s"CREATE TABLE $tbl (id INT, s STRUCT) USING $v2Source") + val scanRelation = sql(s"SELECT * FROM $tbl").queryExecution.optimizedPlan.collectFirst { + case r: DataSourceV2ScanRelation => r + }.getOrElse(fail("Expected a DataSourceV2ScanRelation")) + val scan = new FullyPushedRootAttributeScan + val e = intercept[AnalysisException] { + scanRelation.copy(scan = scan).runtimeFilterAttrs + } + checkError( + exception = e, + condition = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.NOT_IN_FILTER_ATTRIBUTES", + parameters = Map( + "attribute" -> "`s`", + "method" -> "fullyPushedFilterAttributes()", + "scanClass" -> scan.getClass.getName, + "relationOutput" -> "\"STRUCT>\""), + sqlState = "KD000") + } + } + test("nested filter attribute under a non-struct column -> rejected during resolution") { val tbl = s"$catalogName.tbl_malformed_nested_filter_attr" withTable(tbl) { @@ -221,17 +244,13 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession { val scanClass = classOf[MissingFullyPushedFilterAttributeScan].getName checkError( exception = e, - condition = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.CANNOT_RESOLVE", + condition = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.NOT_IN_FILTER_ATTRIBUTES", parameters = Map( "attribute" -> "`missing`", "method" -> "fullyPushedFilterAttributes()", "scanClass" -> scanClass, "relationOutput" -> "\"STRUCT\""), sqlState = "KD000") - checkError( - exception = e.getCause.asInstanceOf[AnalysisException], - condition = "_LEGACY_ERROR_TEMP_1137", - parameters = Map("name" -> "missing", "outputStr" -> "id,part")) } } @@ -440,27 +459,28 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession { } } - test("filter on column outside filterAttributes -> not pushed") { + test("fully pushed attribute outside filterAttributes -> rejected") { val tbl = s"$catalogName.tbl4" - val dim = s"$catalogName.dim4" - withTable(tbl, dim) { + withTable(tbl) { sql(s"CREATE TABLE $tbl (id INT, p1 INT, p2 INT) USING $v2Source " + "PARTITIONED BY (p1, p2) " + - "TBLPROPERTIES('filter-attributes' = 'p1')") - for (i <- 0 until 5) { - sql(s"INSERT INTO $tbl VALUES ($i, $i, 10)") + "TBLPROPERTIES('filter-attributes' = 'p1', 'fully-pushed-filter-attributes' = 'p2')") + 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 } - sql(s"CREATE TABLE $dim (val INT) USING $v2Source") - sql(s"INSERT INTO $dim VALUES (10)") - - // p2 is a partition column but is not declared filterable, so no runtime filter is derived. - val df = sql(s"SELECT * FROM $tbl WHERE p2 = (SELECT max(val) FROM $dim)") - checkAnswer(df, (0 until 5).map(i => Row(i, i, 10))) - - assert(collectBatchScan(df).runtimeFilters.isEmpty, - "Expected no runtime filters for a column outside filterAttributes") - assertPushedCatalystPredicates(df, 0) - assertScalarSubqueryEvaluatedAfterScan(df, expected = true) + checkError( + exception = e, + condition = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.NOT_IN_FILTER_ATTRIBUTES", + parameters = Map( + "attribute" -> "`p2`", + "method" -> "fullyPushedFilterAttributes()", + "scanClass" -> scanClass, + "relationOutput" -> "\"STRUCT\""), + sqlState = "KD000") } } @@ -714,6 +734,21 @@ private class MissingFullyPushedFilterAttributeScan extends SupportsRuntimeCatal override def filter(expressions: Array[Expression]): Unit = {} } +/** A scan declaring a root struct fully pushed without declaring it filterable. */ +private class FullyPushedRootAttributeScan extends SupportsRuntimeCatalystFiltering { + + override def readSchema(): StructType = + StructType.fromDDL("id INT, s STRUCT") + + override def filterAttributes(): Array[NamedReference] = + Array(FieldReference(Seq("s", "part"))) + + override def fullyPushedFilterAttributes(): Array[NamedReference] = + Array(FieldReference("s")) + + 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/GroupBasedRowLevelOperationCatalystRuntimeFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedRowLevelOperationCatalystRuntimeFilterSuite.scala index becdde5aa2f4a..39138ceb0d39d 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 @@ -40,7 +40,7 @@ class GroupBasedRowLevelOperationCatalystRuntimeFilterSuite } assertCatalystGroupFilter( executedPlan, - expectedFilterAttrs = Seq("dep"), + expectedFilterPaths = Seq(Seq("dep")), expectedFilter = GroupFilter(scanSchema = "salary INT, dep STRING", groups = Seq("hr"))) checkAnswer( @@ -66,10 +66,9 @@ class GroupBasedRowLevelOperationCatalystRuntimeFilterSuite } assertCatalystGroupFilter( executedPlan, - expectedFilterAttrs = Seq("dep.name"), + expectedFilterPaths = Seq(Seq("dep", "name")), expectedFilter = GroupFilter( - scanSchema = "salary INT, dep STRUCT", groups = Seq("hr")), - expectedFilterPaths = Some(Seq(Seq("dep", "name")))) + scanSchema = "salary INT, dep STRUCT", groups = Seq("hr"))) 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 43ed7797fb5bc..01fd6591ffdaf 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,7 @@ import org.apache.spark.sql.Row 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.Expressions.bucket 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 @@ -69,7 +70,7 @@ abstract class RowLevelOperationCatalystRuntimeFilterSuiteBase } assertCatalystGroupFilter( executedPlan, - expectedFilterAttrs = Seq("dep"), + expectedFilterPaths = Seq(Seq("dep")), expectedFilter = GroupFilter(scanSchema = "id INT, dep STRING", groups = Seq("hr"))) // software was never read, so its rows must come back untouched @@ -107,7 +108,7 @@ abstract class RowLevelOperationCatalystRuntimeFilterSuiteBase } assertCatalystGroupFilter( executedPlan, - expectedFilterAttrs = Seq("dep"), + expectedFilterPaths = Seq(Seq("dep")), expectedFilter = GroupFilter(scanSchema = "pk INT, dep STRING", groups = Seq("hr"))) // software was never read, so its rows must come back untouched @@ -147,10 +148,9 @@ abstract class RowLevelOperationCatalystRuntimeFilterSuiteBase } assertCatalystGroupFilter( executedPlan, - expectedFilterAttrs = Seq("dep.name"), + expectedFilterPaths = Seq(Seq("dep", "name")), expectedFilter = GroupFilter( - scanSchema = "pk INT, dep STRUCT", groups = Seq("hr")), - expectedFilterPaths = Some(Seq(Seq("dep", "name")))) + scanSchema = "pk INT, dep STRUCT", groups = Seq("hr"))) checkAnswer( sql(s"SELECT * FROM $tableNameAsString"), @@ -160,9 +160,26 @@ abstract class RowLevelOperationCatalystRuntimeFilterSuiteBase } } + test("non-identity partition transform does not enable runtime group filtering") { + val schema = "pk INT NOT NULL, id INT, salary INT, dep STRING" + createTable(schema, Array[Transform](bucket(4, "dep"))) + append(schema, + """{ "pk": 1, "id": 1, "salary": 100, "dep": "hr" } + |{ "pk": 2, "id": 2, "salary": 200, "dep": "software" } + |""".stripMargin) + + val executedPlan = executeAndKeepPlan { + sql(s"UPDATE $tableNameAsString SET salary = -1 WHERE id = 1") + } + val batchScans = collect(executedPlan) { case s: BatchScanExec => s } + assert(batchScans.nonEmpty, "expected a batch scan for the row-level operation") + assert(batchScans.forall(_.runtimeFilters.isEmpty), + s"expected no runtime group filters, got ${batchScans.flatMap(_.runtimeFilters)}") + } + /** * Asserts the injected group filter down to its contents: the scan declares - * `expectedFilterAttrs` in `filterAttributes`, every scan node carries one dynamic pruning + * `expectedFilterPaths` in `filterAttributes`, every scan node carries one dynamic pruning * filter matching `expectedFilter`, the connector received that same filter as a Catalyst * expression, and the scan then read only `expectedFilter.groups`. * @@ -175,24 +192,22 @@ abstract class RowLevelOperationCatalystRuntimeFilterSuiteBase */ protected def assertCatalystGroupFilter( executedPlan: SparkPlan, - expectedFilterAttrs: Seq[String], - expectedFilter: GroupFilter, - expectedFilterPaths: Option[Seq[Seq[String]]] = None): Unit = { + expectedFilterPaths: Seq[Seq[String]], + expectedFilter: GroupFilter): 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) assert(batchScans.forall(_.scan eq scan), s"expected all ${batchScans.size} scan nodes to share one scan") - val filterAttrs = scan.filterAttributes().map(_.fieldNames.mkString(".")).toSeq - assert(filterAttrs === expectedFilterAttrs, - s"expected the scan to declare $expectedFilterAttrs as filter attributes, got $filterAttrs") - val filterPaths = expectedFilterPaths.getOrElse(expectedFilterAttrs.map(Seq(_))) + val filterPaths = scan.filterAttributes().map(_.fieldNames.toSeq).toSeq + assert(filterPaths === expectedFilterPaths, + s"expected the scan to declare $expectedFilterPaths as filter attributes, got $filterPaths") batchScans.foreach { batchScan => batchScan.runtimeFilters match { case Seq(DynamicPruningExpression(inSubquery: InSubqueryExec)) => - assertGroupFilter(inSubquery, filterPaths, expectedFilter) + assertGroupFilter(inSubquery, expectedFilterPaths, expectedFilter) case other => fail(s"expected a single dynamic pruning group filter, got $other") } } @@ -204,7 +219,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, filterPaths, expectedFilter) + assertGroupFilter(inSubquery, expectedFilterPaths, expectedFilter) case other => fail(s"expected the group filter pushed as an InSubqueryExec, got $other") } From 7571c91b9faef33a841003ed9922cc95e3d79e87 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Thu, 3 Sep 2026 13:06:23 -0700 Subject: [PATCH 6/8] [SPARK-59068][SQL][FOLLOWUP] Address runtime filter review feedback --- .../expressions/V2ExpressionUtils.scala | 29 +++++++++ .../datasources/v2/DataSourceV2Relation.scala | 53 ++++++---------- .../connector/catalog/InMemoryBaseTable.scala | 15 +++-- .../InMemoryCatalystRuntimeFilterTable.scala | 9 +-- .../InMemoryRowLevelOperationTable.scala | 5 +- .../catalog/InMemoryTableWithV2Filter.scala | 5 +- .../datasources/v2/PushDownUtils.scala | 8 +-- .../dynamicpruning/PartitionPruning.scala | 13 +--- ...wLevelOperationRuntimeGroupFiltering.scala | 23 ++----- ...taSourceV2CatalystRuntimeFilterSuite.scala | 60 ++++++++++++++++++- .../sql/connector/DataSourceV2SQLSuite.scala | 17 ++++++ 11 files changed, 150 insertions(+), 87 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 349a59a1aaf8f..af84df8fcc068 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 @@ -27,6 +27,7 @@ import org.apache.spark.sql.catalyst.analysis.{NoSuchFunctionException, Unresolv import org.apache.spark.sql.catalyst.encoders.EncoderUtils import org.apache.spark.sql.catalyst.expressions.objects.{Invoke, StaticInvoke} import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, LogicalPlan, SampleMethod} +import org.apache.spark.sql.catalyst.types.DataTypeUtils.fromAttributes import org.apache.spark.sql.connector.catalog.{FunctionCatalog, Identifier} import org.apache.spark.sql.connector.catalog.functions._ import org.apache.spark.sql.connector.catalog.functions.ScalarFunction.MAGIC_METHOD_NAME @@ -70,6 +71,34 @@ object V2ExpressionUtils extends SQLConfHelper with Logging { AttributeSet(resolveRefs[NamedExpression](refs.toImmutableArraySeq, plan)) } + /** + * Resolves data source runtime-filter attributes and wraps resolution failures with connector + * context. + */ + private[sql] def resolveDataSourceRuntimeFilterRefs( + refs: Array[NamedReference], + output: Seq[Attribute], + method: String, + scanClass: String): AttributeSet = { + if (refs.isEmpty) return AttributeSet.empty + + val plan = LocalRelation(output) + val resolvedAttrs = refs.map { ref => + try { + resolveRef[NamedExpression](ref, plan) + } catch { + case e: AnalysisException => + throw QueryCompilationErrors.cannotResolveDataSourceRuntimeFilterAttributeError( + attribute = ref.fieldNames, + method = method, + scanClass = scanClass, + relationOutput = fromAttributes(output), + cause = e) + } + } + AttributeSet(resolvedAttrs) + } + /** * Converts the array of input V2 [[V2SortOrder]] into their counterparts in catalyst. */ 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 27e07ca4a4e32..f651d8f1e0f1f 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,12 +20,11 @@ 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} +import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, AttributeReference, AttributeSet, Expression, SortOrder, V2ExpressionUtils} import org.apache.spark.sql.catalyst.plans.QueryPlan -import org.apache.spark.sql.catalyst.plans.logical.{ColumnStat, ExposesMetadataColumns, Histogram, HistogramBin, LeafNode, LocalRelation, LogicalPlan, Statistics} +import org.apache.spark.sql.catalyst.plans.logical.{ColumnStat, ExposesMetadataColumns, Histogram, HistogramBin, LeafNode, LogicalPlan, Statistics} import org.apache.spark.sql.catalyst.plans.logical.statsEstimation.EstimationUtils import org.apache.spark.sql.catalyst.streaming.{StreamingSourceIdentifyingName, Unassigned} import org.apache.spark.sql.catalyst.types.DataTypeUtils.{fromAttributes, toAttributes} @@ -205,10 +204,10 @@ case class DataSourceV2ScanRelation( lazy val runtimeFilterAttrs: AttributeSet = { checkRuntimeFilteringInterfaces() resolvedFullyPushedRuntimeFilterAttrs - resolveFilterAttrs(declaredRuntimeFilterAttrs, "filterAttributes()") + resolvedRuntimeFilterAttrs } - private lazy val declaredRuntimeFilterAttrs: Array[NamedReference] = scan match { + private[sql] lazy val declaredRuntimeFilterAttrs: Array[NamedReference] = scan match { case s: SupportsRuntimeV2Filtering => s.filterAttributes case s: SupportsRuntimeCatalystFiltering => s.filterAttributes() case _ => Array.empty @@ -219,10 +218,17 @@ case class DataSourceV2ScanRelation( case _ => Array.empty } + private lazy val resolvedRuntimeFilterAttrs: AttributeSet = { + resolveFilterAttrs(declaredRuntimeFilterAttrs, "filterAttributes()") + } + private lazy val resolvedFullyPushedRuntimeFilterAttrs: AttributeSet = { - checkFullyPushedFilterAttrs() - resolveFilterAttrs( + checkFullyPushedFilterAttrsAreTopLevel() + val resolvedAttrs = resolveFilterAttrs( declaredFullyPushedRuntimeFilterAttrs, "fullyPushedFilterAttributes()") + resolvedRuntimeFilterAttrs + checkFullyPushedFilterAttrsAreFilterable() + resolvedAttrs } /** @@ -243,8 +249,8 @@ case class DataSourceV2ScanRelation( private def resolveFilterAttrs( filterAttrs: Array[NamedReference], method: String): AttributeSet = { - DataSourceV2ScanRelation.resolveRuntimeFilterAttrs( - filterAttrs, method, scan.getClass.getName, output) + V2ExpressionUtils.resolveDataSourceRuntimeFilterRefs( + filterAttrs, output, method, scan.getClass.getName) } override def name: String = relation.name @@ -303,13 +309,16 @@ case class DataSourceV2ScanRelation( } } - private def checkFullyPushedFilterAttrs(): Unit = { + private def checkFullyPushedFilterAttrsAreTopLevel(): Unit = { declaredFullyPushedRuntimeFilterAttrs.find(_.fieldNames.length > 1).foreach { ref => throw QueryCompilationErrors.nestedDataSourceFullyPushedRuntimeFilterAttributeError( attribute = ref.fieldNames, scanClass = scan.getClass.getName, relationOutput = fromAttributes(output)) } + } + + private def checkFullyPushedFilterAttrsAreFilterable(): Unit = { declaredFullyPushedRuntimeFilterAttrs.find { fullyPushedRef => !declaredRuntimeFilterAttrs.exists { filterRef => fullyPushedRef.fieldNames.length == filterRef.fieldNames.length && @@ -342,30 +351,6 @@ case class DataSourceV2ScanRelation( } } -object DataSourceV2ScanRelation { - private[sql] def resolveRuntimeFilterAttrs( - filterAttrs: Array[NamedReference], - method: String, - scanClass: String, - output: Seq[AttributeReference]): AttributeSet = { - val plan = LocalRelation(output) - val resolvedAttrs = filterAttrs.map { ref => - try { - V2ExpressionUtils.resolveRef[NamedExpression](ref, plan) - } catch { - case e: AnalysisException => - throw QueryCompilationErrors.cannotResolveDataSourceRuntimeFilterAttributeError( - attribute = ref.fieldNames, - method = method, - scanClass = scanClass, - relationOutput = fromAttributes(output), - cause = e) - } - } - AttributeSet(resolvedAttrs) - } -} - /** * A specialization of [[DataSourceV2RelationBase]] that supports streaming scan. * It will be transformed to [[StreamingDataSourceV2ScanRelation]] during the planning phase of 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 af2fd7e59690d..825a276402bc5 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 @@ -515,9 +515,10 @@ abstract class InMemoryBaseTable( } private def canEvaluate(filter: Filter): Boolean = { - if (partitioning.length == 1 && partitioning.head.references.length == 1) { + val identityRefs = identityPartitionReferences + if (partitioning.length == 1 && identityRefs.length == 1) { filter match { - case In(attrName, _) if attrName == partitioning.head.references.head.toString => true + case In(attrName, _) if attrName == identityRefs.head.toString => true case _ => false } } else { @@ -635,6 +636,12 @@ abstract class InMemoryBaseTable( override def toBatch: Batch = this + protected def identityPartitionAttributes: Array[NamedReference] = { + identityPartitionReferences.distinct + .filter(ref => readSchema.findNestedField( + ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined) + } + override def estimateStatistics(): Statistics = { if (data.isEmpty) { return InMemoryStats(OptionalLong.of(0L), OptionalLong.of(0L), new util.HashMap()) @@ -831,9 +838,7 @@ abstract class InMemoryBaseTable( var pushedFilters: Array[Filter] = Array.empty override def filterAttributes(): Array[NamedReference] = { - identityPartitionReferences - .filter(ref => readSchema.findNestedField( - ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined) + identityPartitionAttributes } 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 35a94ef5a1ef2..c14c1a4eea030 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 @@ -87,13 +87,6 @@ class InMemoryCatalystRuntimeFilterTable( ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined) } - /** Identity partition columns that are present in the scan read schema. */ - private def identityPartitionAttrs: Array[NamedReference] = { - identityPartitionReferences.distinct - .filter(ref => readSchema.findNestedField( - ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined) - } - override def filterAttributes(): Array[NamedReference] = { partitionAttrs.filter { ref => restrictedFilterAttrs.forall(_.contains(ref.fieldNames.mkString("."))) @@ -103,7 +96,7 @@ class InMemoryCatalystRuntimeFilterTable( // 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] = { - identityPartitionAttrs.filter { ref => + identityPartitionAttributes.filter { ref => fullyPushedFilterAttrs.contains(ref.fieldNames.mkString(".")) } } 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 2bbeff520f0d1..7556bc96912f6 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,7 +28,6 @@ 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 @@ -311,9 +310,7 @@ class InMemoryRowLevelOperationTable private ( with CatalystRuntimeFilteringScan { override def filterAttributes(): Array[NamedReference] = { - identityPartitionReferences - .filter(ref => readSchema.findNestedField( - ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined) + identityPartitionAttributes } } } 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 08230f39c4b2c..bf4e9db9c5636 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 @@ -25,7 +25,6 @@ import org.apache.spark.sql.connector.expressions.{FieldReference, IdentityTrans 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._ @@ -67,9 +66,7 @@ class InMemoryTableWithV2Filter( extends BatchScanBaseClass(_data, readSchema, tableSchema) with SupportsRuntimeV2Filtering { override def filterAttributes(): Array[NamedReference] = { - identityPartitionReferences - .filter(ref => readSchema.findNestedField( - ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined) + identityPartitionAttributes } override def filter(filters: Array[Predicate]): Unit = { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala index e252edf1013c2..3a5796d3448fc 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala @@ -22,7 +22,7 @@ import scala.collection.mutable import org.apache.spark.SparkException import org.apache.spark.internal.{Logging, LogKeys} import org.apache.spark.sql.AnalysisException -import org.apache.spark.sql.catalyst.expressions.{AttributeReference, AttributeSet, DynamicPruning, DynamicPruningExpression, Expression, ExpressionSet, GetStructField, Literal, NamedExpression, PythonUDF, SchemaPruning, SubqueryExpression} +import org.apache.spark.sql.catalyst.expressions.{AttributeReference, AttributeSet, DynamicPruning, DynamicPruningExpression, Expression, ExpressionSet, GetStructField, Literal, NamedExpression, PythonUDF, SchemaPruning, SubqueryExpression, V2ExpressionUtils} import org.apache.spark.sql.catalyst.plans.logical.SampleMethod import org.apache.spark.sql.catalyst.plans.physical.{KeyedPartitioning, Partitioning} import org.apache.spark.sql.catalyst.types.DataTypeUtils @@ -212,11 +212,11 @@ object PushDownUtils extends Logging { // filters whose translation was not already accepted in the first pass. (See SPARK-55596) // Only candidates whose referenced columns are declared in filterAttributes() are eligible. val partPredicatesPushed = filterableScan.supportsIterativePushdown() && { - val filterAttrs = DataSourceV2ScanRelation.resolveRuntimeFilterAttrs( + val filterAttrs = V2ExpressionUtils.resolveDataSourceRuntimeFilterRefs( filterableScan.filterAttributes(), + output, "filterAttributes()", - filterableScan.getClass.getName, - output) + filterableScan.getClass.getName) val pushed = filterableScan.pushedPredicates().toSet val candidates = runtimeFilters.filter { f => !filtersToTranslated.get(f).exists(pushed.contains) && diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala index 7a17302140563..c0747cbfb1130 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala @@ -23,12 +23,11 @@ import org.apache.spark.sql.catalyst.optimizer.{JoinSelectionHelper, ReusableBro import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys import org.apache.spark.sql.catalyst.plans.logical._ import org.apache.spark.sql.catalyst.rules.Rule -import org.apache.spark.sql.connector.read.SupportsRuntimeV2Filtering import org.apache.spark.sql.execution.LogicalRDD import org.apache.spark.sql.execution.columnar.InMemoryRelation import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation} -import org.apache.spark.sql.execution.datasources.v2.ExtractV2Scan -import org.apache.spark.sql.internal.connector.SupportsRuntimeCatalystFiltering +import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation + /** * Dynamic partition pruning optimization is performed based on the type and * selectivity of the join operation. During query optimization, we insert a @@ -79,13 +78,7 @@ object PartitionPruning extends Rule[LogicalPlan] with PredicateHelper with Join } else { None } - case (resExp, r @ ExtractV2Scan(scan: SupportsRuntimeV2Filtering)) => - if (resExp.references.subsetOf(r.runtimeFilterAttrs)) { - Some(r) - } else { - None - } - case (resExp, r @ ExtractV2Scan(scan: SupportsRuntimeCatalystFiltering)) => + case (resExp, r: DataSourceV2ScanRelation) => if (resExp.references.subsetOf(r.runtimeFilterAttrs)) { Some(r) } else { 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 584478ade938f..79b4ecfc4f2f1 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 @@ -25,10 +25,9 @@ import org.apache.spark.sql.catalyst.planning.{DeltaBasedRowLevelOperation, Grou 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.connector.expressions.NamedReference -import org.apache.spark.sql.connector.read.{Scan, SupportsRuntimeV2Filtering} +import org.apache.spark.sql.connector.read.Scan import org.apache.spark.sql.connector.write.RowLevelOperation.Command.{DELETE, MERGE, UPDATE} -import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Implicits, DataSourceV2Relation, DataSourceV2ScanRelation, ExtractV2Scan} -import org.apache.spark.sql.internal.connector.SupportsRuntimeCatalystFiltering +import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Implicits, DataSourceV2Relation, DataSourceV2ScanRelation} import org.apache.spark.util.ArrayImplicits._ /** @@ -53,24 +52,14 @@ class RowLevelOperationRuntimeGroupFiltering(optimizeSubqueries: Rule[LogicalPla override def apply(plan: LogicalPlan): LogicalPlan = plan transformDown { case GroupBasedRowLevelOperation(replaceData, _, Some(cond), - r @ ExtractV2Scan(scan: SupportsRuntimeV2Filtering)) + r: DataSourceV2ScanRelation) if canInjectGroupFilters(cond, r) => - injectGroupFilters(replaceData, cond, scan, scan.filterAttributes) - - case GroupBasedRowLevelOperation(replaceData, _, Some(cond), - r @ ExtractV2Scan(scan: SupportsRuntimeCatalystFiltering)) - if canInjectGroupFilters(cond, r) => - injectGroupFilters(replaceData, cond, scan, scan.filterAttributes()) - - case DeltaBasedRowLevelOperation(writeDelta, _, Some(cond), - r @ ExtractV2Scan(scan: SupportsRuntimeV2Filtering)) - if canInjectGroupFilters(cond, r) => - injectGroupFilters(writeDelta, cond, scan, scan.filterAttributes) + injectGroupFilters(replaceData, cond, r.scan, r.declaredRuntimeFilterAttrs) case DeltaBasedRowLevelOperation(writeDelta, _, Some(cond), - r @ ExtractV2Scan(scan: SupportsRuntimeCatalystFiltering)) + r: DataSourceV2ScanRelation) if canInjectGroupFilters(cond, r) => - injectGroupFilters(writeDelta, cond, scan, scan.filterAttributes()) + injectGroupFilters(writeDelta, cond, r.scan, r.declaredRuntimeFilterAttrs) } private def canInjectGroupFilters( 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 9b4fd05afc5c3..f0c070d927ca4 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 @@ -244,13 +244,17 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession { val scanClass = classOf[MissingFullyPushedFilterAttributeScan].getName checkError( exception = e, - condition = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.NOT_IN_FILTER_ATTRIBUTES", + condition = "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.CANNOT_RESOLVE", parameters = Map( "attribute" -> "`missing`", "method" -> "fullyPushedFilterAttributes()", "scanClass" -> scanClass, "relationOutput" -> "\"STRUCT\""), sqlState = "KD000") + checkError( + exception = e.getCause.asInstanceOf[AnalysisException], + condition = "_LEGACY_ERROR_TEMP_1137", + parameters = Map("name" -> "missing", "outputStr" -> "id,part")) } } @@ -484,6 +488,60 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession { } } + test("filter on column outside filterAttributes -> not pushed") { + val tbl = s"$catalogName.tbl_restricted_filter_attrs" + val dim = s"$catalogName.dim_restricted_filter_attrs" + withTable(tbl, dim) { + sql(s"CREATE TABLE $tbl (id INT, p1 INT, p2 INT) USING $v2Source " + + "PARTITIONED BY (p1, p2) " + + "TBLPROPERTIES('filter-attributes' = 'p1')") + for (i <- 0 until 5) { + sql(s"INSERT INTO $tbl VALUES ($i, $i, $i)") + } + sql(s"CREATE TABLE $dim (val INT) USING $v2Source") + sql(s"INSERT INTO $dim VALUES (3)") + + val df = sql(s"SELECT * FROM $tbl WHERE p2 = (SELECT max(val) FROM $dim)") + checkAnswer(df, Row(3, 3, 3)) + + val scan = collectBatchScan(df) + assert(scan.runtimeFilters.isEmpty, + "Expected no runtime filters for a column outside filterAttributes") + assertPushedCatalystPredicates(df, expected = 0) + assertScalarSubqueryEvaluatedAfterScan(df, expected = true) + assert(scan.inputPartitions.size === 5) + assert(scan.filteredPartitions.flatten.size === 5) + } + } + + test("two predicates on filter attributes -> pushed together in a single filter() call") { + val tbl = s"$catalogName.tbl_two_predicates" + val dim1 = s"$catalogName.dim_two_predicates1" + val dim2 = s"$catalogName.dim_two_predicates2" + withTable(tbl, dim1, dim2) { + sql(s"CREATE TABLE $tbl (id INT, p1 INT, p2 INT) USING $v2Source PARTITIONED BY (p1, p2)") + for (i <- 0 until 5) { + sql(s"INSERT INTO $tbl VALUES ($i, $i, ${i * 10})") + } + sql(s"CREATE TABLE $dim1 (val INT) USING $v2Source") + sql(s"INSERT INTO $dim1 VALUES (3)") + sql(s"CREATE TABLE $dim2 (val INT) USING $v2Source") + sql(s"INSERT INTO $dim2 VALUES (30)") + + val df = sql(s"SELECT * FROM $tbl WHERE p1 = (SELECT max(val) FROM $dim1) " + + s"AND p2 = (SELECT max(val) FROM $dim2)") + checkAnswer(df, Row(3, 3, 30)) + + assertScalarSubqueryRuntimeFilters(df, expectedCount = 2) + val p1 = AttributeReference("p1", IntegerType, nullable = false)() + val p2 = AttributeReference("p2", IntegerType, nullable = false)() + assertPushedCatalystPredicatesEqual( + df, EqualTo(p1, Literal(3)), EqualTo(p2, Literal(30))) + assert(getCatalystScan(df).filterCallCount === 1, + "expected both predicates pushed in a single filter() call") + } + } + test("nested field of a filter attribute -> pushed with the nested access intact") { val tbl = s"$catalogName.tbl_nested" val dim = s"$catalogName.dim_nested" 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 7dd22878024bc..8b316e90ee081 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 @@ -212,6 +212,23 @@ abstract class DataSourceV2SQLSuite } } + test("filter on transformed partition source remains post-scan") { + val table = s"${catalogAndNamespace}transformed_source_filter" + withTable(table) { + sql(s"CREATE TABLE $table (id INT, part DATE) USING $v2Format " + + "PARTITIONED BY (days(part))") + sql(s"INSERT INTO $table VALUES " + + "(1, DATE '2026-08-01'), (2, DATE '2026-08-02')") + + val df = sql(s"SELECT * FROM $table WHERE part IN (DATE '2026-08-01')") + checkAnswer(df, Row(1, java.sql.Date.valueOf("2026-08-01"))) + + val scan = collect(df.queryExecution.executedPlan) { case b: BatchScanExec => b }.head + assert(scan.partitions.size === 2) + assert(scan.filteredPartitions.flatten.size === 2) + } + } + 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")) From 40a4961c43489d9c866ba09f39459276fea0bcb6 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Thu, 3 Sep 2026 15:19:28 -0700 Subject: [PATCH 7/8] [SPARK-59068][SQL] Strengthen transformed partition filter regression --- .../spark/sql/connector/DataSourceV2SQLSuite.scala | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) 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 8b316e90ee081..bd2eb571da700 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 @@ -218,14 +218,17 @@ abstract class DataSourceV2SQLSuite sql(s"CREATE TABLE $table (id INT, part DATE) USING $v2Format " + "PARTITIONED BY (days(part))") sql(s"INSERT INTO $table VALUES " + - "(1, DATE '2026-08-01'), (2, DATE '2026-08-02')") + "(1, DATE '2026-08-01'), (2, DATE '2026-08-02'), (3, DATE '2026-08-03')") - val df = sql(s"SELECT * FROM $table WHERE part IN (DATE '2026-08-01')") - checkAnswer(df, Row(1, java.sql.Date.valueOf("2026-08-01"))) + val df = sql(s"SELECT * FROM $table WHERE " + + "part IN (DATE '2026-08-01', DATE '2026-08-03')") + checkAnswer(df, Seq( + Row(1, java.sql.Date.valueOf("2026-08-01")), + Row(3, java.sql.Date.valueOf("2026-08-03")))) val scan = collect(df.queryExecution.executedPlan) { case b: BatchScanExec => b }.head - assert(scan.partitions.size === 2) - assert(scan.filteredPartitions.flatten.size === 2) + assert(scan.partitions.size === 3) + assert(scan.filteredPartitions.flatten.size === 3) } } From 9cc2d08940dc281d31bdc29109c4597c8cb08afa Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Fri, 4 Sep 2026 09:53:34 -0700 Subject: [PATCH 8/8] [SPARK-59068][SQL][FOLLOWUP] Adapt runtime filter tests for branch-4.3 --- .../sql/connector/catalog/InMemoryBaseTable.scala | 4 ++++ .../catalog/InMemoryTableWithV2Filter.scala | 2 +- .../DataSourceV2CatalystRuntimeFilterSuite.scala | 15 ++++++++++----- 3 files changed, 15 insertions(+), 6 deletions(-) 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 825a276402bc5..222776661870e 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 @@ -738,9 +738,11 @@ abstract class InMemoryBaseTable( protected def tableSchema: StructType private val catalystPredicates = ArrayBuffer.empty[CatalystExpression] + private var filterCalls = 0 override def filter(expressions: Array[CatalystExpression]): Unit = { catalystPredicates ++= expressions + filterCalls += 1 val partAttrs = partitionAttributes if (partAttrs.isEmpty) return @@ -772,6 +774,8 @@ abstract class InMemoryBaseTable( /** Predicates recorded by [[filter]], for test assertions only. */ def pushedCatalystPredicates: Seq[CatalystExpression] = catalystPredicates.toSeq + def filterCallCount: Int = filterCalls + /** * Identity partition columns paired with their bound partition-key slots. * 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 bf4e9db9c5636..102c977c9377e 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 @@ -21,7 +21,7 @@ import java.util import org.scalatest.Assertions.assert -import org.apache.spark.sql.connector.expressions.{FieldReference, IdentityTransform, LiteralValue, NamedReference, Transform} +import org.apache.spark.sql.connector.expressions.{FieldReference, LiteralValue, NamedReference, Transform} 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} 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 f0c070d927ca4..a4ac064a70557 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 @@ -709,15 +709,20 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession { }.getOrElse(fail("Expected BatchScanExec in plan")) } - private def getPushedCatalystPredicates(df: DataFrame): Seq[Expression] = { + private type CatalystScan = + InMemoryCatalystRuntimeFilterTable#InMemoryCatalystRuntimeFilterBatchScan + + private def getCatalystScan(df: DataFrame): CatalystScan = { collectBatchScan(df).scan match { - case s: InMemoryCatalystRuntimeFilterTable#InMemoryCatalystRuntimeFilterBatchScan => - s.pushedCatalystPredicates - case other => - fail(s"Expected InMemoryCatalystRuntimeFilterBatchScan, got $other") + case s: CatalystScan => s + case other => fail(s"Expected InMemoryCatalystRuntimeFilterBatchScan, got $other") } } + private def getPushedCatalystPredicates(df: DataFrame): Seq[Expression] = { + getCatalystScan(df).pushedCatalystPredicates + } + private def assertPushedCatalystPredicates(df: DataFrame, expected: Int): Unit = { val preds = getPushedCatalystPredicates(df) assert(preds.size === expected,