This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new d6a02836bca1 [SPARK-59068][SQL] Restore support for nested runtime
filter attributes
d6a02836bca1 is described below
commit d6a02836bca1ad460c900328ce41c4a22b0c45e0
Author: Szehon Ho <[email protected]>
AuthorDate: Sat Aug 29 08:27:27 2026 -0700
[SPARK-59068][SQL] Restore support for nested runtime filter attributes
### What changes were proposed in this pull request?
This PR restores support for nested `filterAttributes()` references in:
- `SupportsRuntimeFiltering`
- `SupportsRuntimeV2Filtering`
- `SupportsRuntimeCatalystFiltering`
For ordinary runtime filtering, it removes the top-level-only validation
added by #58296. It keeps
`fullyPushedFilterAttributes()` restricted to top-level attributes because
Spark removes the
post-scan predicate for those attributes.
The PR also enables nested runtime group filtering for row-level
operations. This is new capability,
not behavior restored from before #58296.
Invalid runtime filter references now produce structured
`DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.CANNOT_RESOLVE` or
`DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.NOT_TOP_LEVEL` errors with
the offending attribute,
connector method, scan class, and scan relation output. Resolution failures
preserve the original
analysis error as their cause.
### Why are the changes needed?
The validation in #58296 was added following this review comment:
> Accepting `s.tz` here widens it to `s` and makes filters over every field
eligible.
https://github.com/apache/spark/pull/58296#discussion_r3859456719
The concern is valid because Spark currently uses an `AttributeSet` for
eligibility. A nested
reference such as `derives.toStr` contributes its root attribute,
`derives`, so Spark cannot
distinguish it from `derives.other`.
However, rejecting nested references breaks an existing Iceberg
runtime-filtering flow.
For example, consider an Iceberg table partitioned by:
```
truncate(2, derives.toStr)
```
The end-to-end flow before #58296 was:
- Iceberg found the partition field's source column ID.
- It converted that source column's full path into a `NamedReference`.
- `filterAttributes()` therefore returned `derives.toStr`.
- Spark resolved that nested reference. Its current eligibility
representation reduced it to the
root attribute `derives`.
- A DPP predicate on `derives.toStr` passed eligibility.
- Importantly, the actual predicate sent to Iceberg still contained the
nested `derives.toStr`
access.
- Iceberg projected that predicate through each partition spec and used the
resulting evaluator
to prune scan tasks.
Iceberg implementation:
-
[`filterAttributes()`](https://github.com/apache/iceberg/blob/ef8a69dc3dab5858d612770062bbf5a6043170f0/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java#L105-L123)
- [Runtime-filter projection and task
pruning](https://github.com/apache/iceberg/blob/ef8a69dc3dab5858d612770062bbf5a6043170f0/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java#L127-L146)
Therefore, this flow worked end to end even though Spark's eligibility
check was too coarse.
After #58296, Spark rejects `derives.toStr` while constructing the scan
relation, breaking the
existing Iceberg integration.
This PR restores the previous scan behavior as a compatibility fix.
Separately, it makes row-level
runtime group filtering work with nested attributes.
[SPARK-59095](https://issues.apache.org/jira/browse/SPARK-59095) tracks making
runtime-filter
eligibility path-aware so Spark can distinguish `derives.toStr` from
sibling fields such as
`derives.other`. Until then, connectors cannot soundly declare a
fully-pushed nested path: declaring
the root struct would also make sibling-field predicates appear fully
pushed, and Spark would
remove their post-scan evaluation.
### Does this PR introduce _any_ user-facing change?
Yes. Data sources can once again advertise nested runtime filter attributes.
This restores the behavior before #58296: connectors such as Iceberg no
longer fail during
planning and can receive nested runtime predicates for partition pruning.
Row-level operations can
also use nested runtime filter attributes for runtime group pruning.
### How was this patch tested?
Added end-to-end DPP regression tests for:
- `SupportsRuntimeV2Filtering`
- `SupportsRuntimeCatalystFiltering`
- nested row-level DELETE and MERGE group filtering
- malformed and missing ordinary and fully-pushed runtime filter attributes
The tests verify that a scan advertising `derives.toStr`:
- is eligible for a runtime filter on `derives.toStr`;
- receives a predicate retaining the nested access; and
- prunes the expected partition.
The following validation passed:
- `DataSourceV2CatalystRuntimeFilterSuite`,
`GroupBasedRowLevelOperationCatalystRuntimeFilterSuite`, and
`DeltaBasedRowLevelOperationCatalystRuntimeFilterSuite`: 30/30 tests
- focused V1/V2 nested partition source tests: 2/2 tests
- `SparkThrowableSuite`: 37/37 tests
- Catalyst and SQL scalastyle checks
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex with GPT-5
Closes #58370 from szehon-ho/codex/runtime-filter-nested-compat.
Authored-by: Szehon Ho <[email protected]>
Signed-off-by: Szehon Ho <[email protected]>
(cherry picked from commit 221f8de0d2e793646dcbbda4095ef9dcca8b9022)
Signed-off-by: Szehon Ho <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 18 ++
.../connector/read/SupportsRuntimeFiltering.java | 12 +-
.../connector/read/SupportsRuntimeV2Filtering.java | 10 +-
.../catalyst/expressions/V2ExpressionUtils.scala | 2 +-
.../spark/sql/errors/QueryCompilationErrors.scala | 40 ++++
.../datasources/v2/DataSourceV2Relation.scala | 80 ++++---
.../SupportsRuntimeCatalystFiltering.scala | 16 +-
.../sql/connector/catalog/InMemoryBaseTable.scala | 8 +-
.../InMemoryCatalystRuntimeFilterTable.scala | 27 +--
.../catalog/InMemoryRowLevelOperationTable.scala | 5 +-
.../catalog/InMemoryTableWithV2Filter.scala | 5 +-
.../RowLevelOperationRuntimeGroupFiltering.scala | 21 +-
.../DataSourceV2CatalystRuntimeFilterSuite.scala | 235 ++++++++++++++-------
.../spark/sql/connector/DataSourceV2SQLSuite.scala | 44 +++-
...wLevelOperationCatalystRuntimeFilterSuite.scala | 31 +++
...elOperationCatalystRuntimeFilterSuiteBase.scala | 69 +++++-
.../sql/connector/RowLevelOperationSuiteBase.scala | 9 +
17 files changed, 478 insertions(+), 154 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index 31ec0418d1b4..2b73c1759733 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -2265,6 +2265,24 @@
],
"sqlState" : "KD010"
},
+ "DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE" : {
+ "message" : [
+ "The runtime filter attribute <attribute> reported by `<method>` in data
source scan <scanClass> is invalid for the scan relation output
<relationOutput>."
+ ],
+ "subClass" : {
+ "CANNOT_RESOLVE" : {
+ "message" : [
+ "The attribute cannot be resolved."
+ ]
+ },
+ "NOT_TOP_LEVEL" : {
+ "message" : [
+ "The attribute must be top-level, but it is a nested reference."
+ ]
+ }
+ },
+ "sqlState" : "KD000"
+ },
"DATA_SOURCE_METADATA_SCHEMA_NOT_IMPLEMENTED" : {
"message" : [
"<class> does not implement metadataSchema."
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 067202a36270..c752c8c95d54 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.
* <p>
- * Each reference must be a top-level attribute present in {@link
Scan#readSchema()}.
- * Nested references and attributes pruned out of the read schema fail to
resolve when
- * Spark builds the scan relation.
+ * Each reference must resolve against the scan relation output when Spark
builds it. Attributes
+ * pruned out of {@link Scan#readSchema()} fail to resolve.
*/
NamedReference[] filterAttributes();
@@ -51,6 +50,13 @@ public interface SupportsRuntimeFiltering extends
SupportsRuntimeV2Filtering {
* The provided expressions must be interpreted as a set of filters that are
ANDed together.
* Implementations may use the filters to prune initially planned {@link
InputPartition}s.
* <p>
+ * Spark tracks runtime-filter eligibility by root attribute. If {@link
#filterAttributes()}
+ * returns a nested reference, this method may receive a filter on another
nested field under
+ * the same root. Implementations must inspect each filter and use only
filters they can apply.
+ * Nested paths are encoded in a V1 {@link Filter} as unquoted dot-separated
names such as
+ * {@code parent.child}. A top-level column whose name contains a dot
remains quoted, such as
+ * {@code `parent.child`}.
+ * <p>
* If the scan also implements {@link SupportsReportPartitioning}, it must
preserve
* the originally reported partitioning during runtime filtering. While
applying runtime filters,
* the scan may detect that some {@link InputPartition}s have no matching
data, in which case
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 6b286f041b01..671a822096bb 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.
* <p>
- * Each reference must be a top-level attribute present in {@link
Scan#readSchema()}.
- * Nested references and attributes pruned out of the read schema fail to
resolve when
- * Spark builds the scan relation.
+ * Each reference must resolve against the scan relation output when Spark
builds it. Attributes
+ * pruned out of {@link Scan#readSchema()} fail to resolve.
*/
NamedReference[] filterAttributes();
@@ -64,6 +63,11 @@ public interface SupportsRuntimeV2Filtering extends Scan {
* The provided expressions must be interpreted as a set of predicates that
are ANDed together.
* Implementations may use the predicates to prune initially planned {@link
InputPartition}s.
* <p>
+ * Spark tracks runtime-filter eligibility by root attribute. If {@link
#filterAttributes()}
+ * returns a nested reference, this method may receive a predicate on
another nested field under
+ * the same root. Implementations must inspect each predicate and use only
predicates they can
+ * apply.
+ * <p>
* If the scan also implements {@link SupportsReportPartitioning}, it must
preserve
* the originally reported partitioning during runtime filtering. While
applying runtime
* predicates, the scan may detect that some {@link InputPartition}s have no
matching data, in
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 702255e07574..349a59a1aaf8 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 ada71af43d3b..a662eee4ea41 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -4654,6 +4654,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 7b2cfacb652f..c13b44ab5fc6 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,15 +20,16 @@ package org.apache.spark.sql.execution.datasources.v2
import java.util.{Collections, Optional, OptionalLong}
import org.apache.spark.SparkException
+import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation,
NamedRelation, TimeTravelSpec}
import org.apache.spark.sql.catalyst.catalog.{CatalogColumnStat,
CatalogStatistics}
-import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap,
AttributeReference, AttributeSet, Expression, SortOrder, V2ExpressionUtils}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap,
AttributeReference, AttributeSet, Expression, NamedExpression, SortOrder,
V2ExpressionUtils}
import org.apache.spark.sql.catalyst.plans.QueryPlan
import org.apache.spark.sql.catalyst.plans.logical.{ColumnStat,
ExposesMetadataColumns, Histogram, HistogramBin, LeafNode, LogicalPlan,
Statistics}
import
org.apache.spark.sql.catalyst.plans.logical.statsEstimation.EstimationUtils
import
org.apache.spark.sql.catalyst.streaming.{StreamingSourceIdentifyingName,
Unassigned}
import
org.apache.spark.sql.catalyst.trees.TreePattern.{DATA_SOURCE_V2_RELATION,
DATA_SOURCE_V2_SCAN_RELATION, TreePattern}
-import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes
+import org.apache.spark.sql.catalyst.types.DataTypeUtils.{fromAttributes,
toAttributes}
import org.apache.spark.sql.catalyst.util.{removeInternalMetadata,
truncatedString, CharVarcharUtils}
import org.apache.spark.sql.connector.catalog.{CatalogPlugin, FunctionCatalog,
Identifier, SupportsMetadataColumns, Table, TableCapability, TableCatalog,
V2TableUtil}
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.CatalogHelper
@@ -36,10 +37,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
/**
@@ -201,16 +202,23 @@ case class DataSourceV2ScanRelation(
* Resolved attributes that the scan declares for runtime filtering via
* [[SupportsRuntimeV2Filtering.filterAttributes]] or
* [[SupportsRuntimeCatalystFiltering.filterAttributes]]. Empty when the scan
- * implements neither interface or exposes no attributes.
+ * implements neither interface or exposes no attributes. Accessing this
value also validates
+ * attributes returned by
[[SupportsRuntimeCatalystFiltering.fullyPushedFilterAttributes]].
*/
lazy val runtimeFilterAttrs: AttributeSet = {
checkRuntimeFilteringInterfaces()
+ checkFullyPushedFilterAttrs()
val filterAttrs = scan match {
case s: SupportsRuntimeV2Filtering => s.filterAttributes
case s: SupportsRuntimeCatalystFiltering => s.filterAttributes()
case _ => Array.empty[NamedReference]
}
- resolveTopLevelFilterAttrs(filterAttrs)
+ resolveFilterAttrs(filterAttrs, "filterAttributes()")
+ }
+
+ private lazy val declaredFullyPushedRuntimeFilterAttrs:
Array[NamedReference] = scan match {
+ case s: SupportsRuntimeCatalystFiltering => s.fullyPushedFilterAttributes()
+ case _ => Array.empty
}
/**
@@ -219,27 +227,34 @@ case class DataSourceV2ScanRelation(
*/
lazy val fullyPushedRuntimeFilterAttrs: AttributeSet = {
checkRuntimeFilteringInterfaces()
- val filterAttrs = scan match {
- case s: SupportsRuntimeCatalystFiltering =>
s.fullyPushedFilterAttributes()
- case _ => Array.empty[NamedReference]
- }
- resolveTopLevelFilterAttrs(filterAttrs)
+ checkFullyPushedFilterAttrs()
+ resolveFilterAttrs(
+ declaredFullyPushedRuntimeFilterAttrs, "fullyPushedFilterAttributes()")
}
/**
- * Resolves the given runtime-filter references against this relation's
output. Both runtime
- * filtering interfaces require each reference to be a top-level attribute
of the read schema,
- * so a nested reference is rejected.
+ * Resolves runtime-filter references against this relation's output.
+ *
+ * [[AttributeSet]] reduces nested references to their root attributes. This
is sufficient for
+ * ordinary runtime-filter eligibility because Spark retains the post-scan
predicate.
*/
- private def resolveTopLevelFilterAttrs(filterAttrs: Array[NamedReference]):
AttributeSet = {
- filterAttrs.find(_.fieldNames.length > 1).foreach { ref =>
- throw SparkException.internalError(
- s"Runtime filter attribute '${ref.fieldNames.mkString(".")}' declared
by " +
- s"${scan.getClass.getName} must be a top-level attribute of the scan
read schema, " +
- "but it is a nested reference.")
+ private def resolveFilterAttrs(
+ filterAttrs: Array[NamedReference],
+ method: String): AttributeSet = {
+ val resolvedAttrs = filterAttrs.map { ref =>
+ try {
+ V2ExpressionUtils.resolveRef[NamedExpression](ref, this)
+ } catch {
+ case e: AnalysisException =>
+ throw
QueryCompilationErrors.cannotResolveDataSourceRuntimeFilterAttributeError(
+ attribute = ref.fieldNames,
+ method = method,
+ scanClass = scan.getClass.getName,
+ relationOutput = fromAttributes(output),
+ cause = e)
+ }
}
- AttributeSet(V2ExpressionUtils.resolveRefs[Attribute](
- filterAttrs.toImmutableArraySeq, this))
+ AttributeSet(resolvedAttrs)
}
override val nodePatterns: Seq[TreePattern] =
Seq(DATA_SOURCE_V2_SCAN_RELATION)
@@ -290,12 +305,23 @@ case class DataSourceV2ScanRelation(
Statistics(sizeInBytes = conf.defaultSizeInBytes)
}
- private def checkRuntimeFilteringInterfaces(): Unit = scan match {
- case _: SupportsRuntimeV2Filtering with SupportsRuntimeCatalystFiltering =>
- throw SparkException.internalError(
- "A scan must not implement both SupportsRuntimeV2Filtering and " +
- s"SupportsRuntimeCatalystFiltering, but ${scan.getClass.getName}
implements both.")
- case _ =>
+ private def checkRuntimeFilteringInterfaces(): Unit = {
+ scan match {
+ case _: SupportsRuntimeV2Filtering with SupportsRuntimeCatalystFiltering
=>
+ throw SparkException.internalError(
+ "A scan must not implement both SupportsRuntimeV2Filtering and " +
+ s"SupportsRuntimeCatalystFiltering, but ${scan.getClass.getName}
implements both.")
+ case _ =>
+ }
+ }
+
+ private def checkFullyPushedFilterAttrs(): Unit = {
+ declaredFullyPushedRuntimeFilterAttrs.find(_.fieldNames.length >
1).foreach { ref =>
+ throw
QueryCompilationErrors.nestedDataSourceFullyPushedRuntimeFilterAttributeError(
+ attribute = ref.fieldNames,
+ scanClass = scan.getClass.getName,
+ relationOutput = fromAttributes(output))
+ }
}
override def doCanonicalize(): DataSourceV2ScanRelation = {
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 eaa4857f1abe..ec34efed2625 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 are rejected, and attributes pruned out of the read schema
fail to resolve, when
- * Spark builds the scan relation.
+ * Each reference must resolve against the scan relation output when Spark
builds it. Attributes
+ * pruned out of [[Scan.readSchema]] fail to resolve.
*/
def filterAttributes(): Array[NamedReference]
@@ -65,7 +64,10 @@ trait SupportsRuntimeCatalystFiltering extends Scan {
*
* Each reference must be a top-level attribute present in
[[Scan.readSchema]]. Nested
* references are rejected, and attributes pruned out of the read schema
fail to resolve, when
- * Spark builds the scan relation.
+ * Spark builds the scan relation. Spark cannot currently represent an
individual fully pushed
+ * nested path. A scan must not return the root struct as a substitute
unless it can fully
+ * evaluate predicates over every nested field, since Spark would remove
their post-scan
+ * evaluation as well.
*/
def fullyPushedFilterAttributes(): Array[NamedReference] = Array.empty
@@ -76,9 +78,9 @@ trait SupportsRuntimeCatalystFiltering extends Scan {
* Implementations may use the expressions to prune initially planned
* [[org.apache.spark.sql.connector.read.InputPartition]]s.
*
- * An expression may access nested fields of an attribute returned by
[[filterAttributes]], as
- * that attribute is required to be top-level. The scan is responsible for
matching such
- * accesses against its own partition layout.
+ * Spark tracks runtime-filter eligibility by root attribute. If
[[filterAttributes]] returns a
+ * nested reference, an expression may access another nested field under the
same root. The scan
+ * must match each access against its own partition layout and use only
expressions it can apply.
*
* Spark may call this method more than once for the same scan instance: a
plan can hold several
* scan nodes sharing one scan (e.g. the two branches of a group-based
UPDATE), and each pushes
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 06e4158edbe5..ed7df753bc0a 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
@@ -780,7 +780,9 @@ abstract class InMemoryBaseTable(
private def partitionAttributes: Seq[(Seq[String], AttributeReference)] = {
partitioning.flatMap(_.references()).flatMap { ref =>
val path = ref.fieldNames.toImmutableArraySeq
-
readSchema.findNestedField(path).orElse(tableSchema.findNestedField(path)).map {
+ val resolver = SQLConf.get.resolver
+ readSchema.findNestedField(path, resolver = resolver)
+ .orElse(tableSchema.findNestedField(path, resolver = resolver)).map {
case (_, f) =>
path -> AttributeReference(ref.fieldNames.mkString("."),
f.dataType, f.nullable)()
}
@@ -837,9 +839,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 369e8123f1d7..58f07307dcbe 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,8 +23,9 @@ import InMemoryCatalystRuntimeFilterTable._
import org.apache.spark.sql.connector.catalog.constraints.Constraint
import org.apache.spark.sql.connector.distributions.{Distribution,
Distributions}
-import org.apache.spark.sql.connector.expressions.{FieldReference,
NamedReference, SortOrder, Transform}
+import org.apache.spark.sql.connector.expressions.{NamedReference, SortOrder,
Transform}
import org.apache.spark.sql.connector.read.{InputPartition, Scan, ScanBuilder}
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.util.ArrayImplicits._
@@ -90,29 +91,23 @@ class InMemoryCatalystRuntimeFilterTable(
.map(_.split(",").map(_.trim).toSet)
.getOrElse(Set.empty)
- /**
- * The partition columns, each named by the top level read schema column
it lives under, the
- * form both interface methods require. Columns pruned out of the read
schema are dropped,
- * since neither method may name one. Examples:
- * - `PARTITIONED BY (part)` -> `"part"`
- * - `PARTITIONED BY (s.nested)` -> `"s"`, the struct column holding the
partition field
- */
- private def partitionAttrNames: Array[String] = {
- val scanFields = readSchema.fields.map(_.name).toSet
- partitioning.flatMap(_.references()).map(_.fieldNames.head).distinct
- .filter(scanFields.contains)
+ /** Partition source columns that are present in the scan read schema. */
+ private def partitionAttrs: Array[NamedReference] = {
+ partitioning.flatMap(_.references()).distinct
+ .filter(ref => readSchema.findNestedField(
+ ref.fieldNames.toImmutableArraySeq, resolver =
SQLConf.get.resolver).isDefined)
}
override def filterAttributes(): Array[NamedReference] = {
- partitionAttrNames
- .filter(name => restrictedFilterAttrs.forall(_.contains(name)))
- .map(FieldReference.column)
+ partitionAttrs.filter { ref =>
+ restrictedFilterAttrs.forall(_.contains(ref.fieldNames.mkString(".")))
+ }
}
// Not intersected with `filterAttributes()`, so a table can declare a
fully pushed attribute
// that is not a filter attribute, a combination the interface forbids.
override def fullyPushedFilterAttributes(): Array[NamedReference] = {
-
partitionAttrNames.filter(fullyPushedFilterAttrs.contains).map(FieldReference.column)
+ partitionAttrs.filter(ref =>
fullyPushedFilterAttrs.contains(ref.fieldNames.mkString(".")))
}
}
}
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 6cb23d505f78..428a9215dcce 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 b59a81cc2909..fa50da6732bf 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala
@@ -27,6 +27,7 @@ import
org.apache.spark.sql.connector.expressions.{FieldReference, LiteralValue,
import org.apache.spark.sql.connector.expressions.filter.{And, Predicate}
import org.apache.spark.sql.connector.read.{InputPartition, Scan, ScanBuilder,
SupportsRuntimeV2Filtering}
import org.apache.spark.sql.connector.write.{LogicalWriteInfo,
SupportsOverwriteV2, WriteBuilder, WriterCommitMessage}
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.util.ArrayImplicits._
@@ -77,9 +78,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 ae7b432d72f5..b17e224c4a5a 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}
@@ -97,8 +97,9 @@ class
RowLevelOperationRuntimeGroupFiltering(optimizeSubqueries: Rule[LogicalPla
val relation = r.relation.copy(table = originalTable)
val matchingRowsPlan = buildMatchingRowsPlan(write, relation, cond)
val filterAttrsSeq = filterAttrs.toImmutableArraySeq
- val buildKeys =
V2ExpressionUtils.resolveRefs[Attribute](filterAttrsSeq, matchingRowsPlan)
- val pruningKeys =
V2ExpressionUtils.resolveRefs[Attribute](filterAttrsSeq, r)
+ val buildKeys =
+ V2ExpressionUtils.resolveRefs[NamedExpression](filterAttrsSeq,
matchingRowsPlan)
+ val pruningKeys =
V2ExpressionUtils.resolveRefs[NamedExpression](filterAttrsSeq, r)
Filter(buildDynamicPruningCond(matchingRowsPlan, buildKeys,
pruningKeys), r)
}
// optimize subqueries to rewrite them as joins and trigger job planning
@@ -143,13 +144,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 c0a1faa03263..11e7461bc7f1 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala
@@ -116,6 +116,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<part: INT, other: INT>) 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<id: INT, s: STRUCT<part: INT, other:
INT>>\""),
+ 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<id: INT, part: INT>\""),
+ 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<id: INT, part: INT>\""),
+ sqlState = "KD000")
+ checkError(
+ exception = e.getCause.asInstanceOf[AnalysisException],
+ condition = "_LEGACY_ERROR_TEMP_1137",
+ parameters = Map("name" -> "missing", "outputStr" -> "id,part"))
+ }
+ }
+
+ test("missing fully pushed filter attribute -> identifies the declaring
method") {
+ val tbl = s"$catalogName.tbl_missing_fully_pushed_filter_attr"
+ withTable(tbl) {
+ sql(s"CREATE TABLE $tbl (id INT, part INT) USING $v2Source PARTITIONED
BY (part)")
+
+ val scanRelation = sql(s"SELECT * FROM
$tbl").queryExecution.optimizedPlan.collectFirst {
+ case r: DataSourceV2ScanRelation => r
+ }.getOrElse(fail("Expected a DataSourceV2ScanRelation"))
+ val e = intercept[AnalysisException] {
+ scanRelation.copy(scan = new MissingFullyPushedFilterAttributeScan)
+ .fullyPushedRuntimeFilterAttrs
+ }
+ val scanClass = classOf[MissingFullyPushedFilterAttributeScan].getName
+ checkError(
+ exception = e,
+ condition =
"DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.CANNOT_RESOLVE",
+ parameters = Map(
+ "attribute" -> "`missing`",
+ "method" -> "fullyPushedFilterAttributes()",
+ "scanClass" -> scanClass,
+ "relationOutput" -> "\"STRUCT<id: INT, part: INT>\""),
+ sqlState = "KD000")
+ checkError(
+ exception = e.getCause.asInstanceOf[AnalysisException],
+ condition = "_LEGACY_ERROR_TEMP_1137",
+ parameters = Map("name" -> "missing", "outputStr" -> "id,part"))
+ }
+ }
+
test("non-deterministic predicate on fully pushed attributes -> evaluated
after the scan") {
val tbl = s"$catalogName.tbl_nondeterministic"
val dim = s"$catalogName.dim_nondeterministic"
@@ -244,40 +355,64 @@ class DataSourceV2CatalystRuntimeFilterSuite extends
SharedSparkSession {
}
}
- test("DPP filter on a nested partition field -> pushed with the nested
access intact") {
+ test("DPP filter on a nested partition source -> pushed with the nested
access intact") {
val fact = s"$catalogName.fact_nested_dpp"
val dim = s"$catalogName.dim_nested_dpp"
withTable(fact, dim) {
- sql(s"CREATE TABLE $fact (id INT, s STRUCT<part: INT>) USING $v2Source "
+
- "PARTITIONED BY (s.part)")
- for (i <- 0 until 5) {
- sql(s"INSERT INTO $fact VALUES ($i, named_struct('part', $i))")
- }
- sql(s"CREATE TABLE $dim (dim_id INT, dim_val STRING) USING $v2Source")
- sql(s"INSERT INTO $dim VALUES (2, 'two')")
+ sql(s"CREATE TABLE $fact " +
+ "(id INT, derives STRUCT<toStr: STRING, other: STRING>) USING " +
+ s"$v2Source PARTITIONED BY (derives.toStr)")
+ sql(s"INSERT INTO $fact VALUES " +
+ "(1, named_struct('toStr', 'AA', 'other', 'a')), " +
+ "(2, named_struct('toStr', 'BB', 'other', 'b')), " +
+ "(3, named_struct('toStr', 'CC', 'other', 'c'))")
+ sql(s"CREATE TABLE $dim (value STRING, selected INT) USING $v2Source")
+ sql(s"INSERT INTO $dim VALUES ('AA', 0), ('BB', 1)")
withDPPConf {
val df = sql(
s"""SELECT f.id FROM $fact f JOIN $dim d
- |ON f.s.part = d.dim_id WHERE d.dim_val = 'two'""".stripMargin)
+ |ON f.derives.toStr = d.value WHERE d.selected = 1""".stripMargin)
checkAnswer(df, Row(2))
assertDPPRuntimeFilters(df)
- // The scan only reports `s`, the struct holding the partition field,
but the predicate it
- // receives keeps the nested access, so it can still tell which
partition to keep.
+ // The scan reports `derives.toStr`, and the predicate it receives
keeps that nested access.
val pushed = getPushedCatalystPredicates(df)
assert(pushed.size === 1, s"expected a single pushed predicate, got
$pushed")
assert(pushed.head.exists(_.isInstanceOf[GetStructField]),
s"expected the pushed predicate to keep the nested access, got
${pushed.head}")
val batchScan = collectBatchScan(df)
- assert(batchScan.inputPartitions.size === 5)
+ assert(batchScan.inputPartitions.size === 3)
assert(batchScan.filteredPartitions.flatten.size === 1,
s"expected 1 partition after pruning, got
${batchScan.filteredPartitions.flatten.size}")
}
}
}
+ test("sibling of a nested filter attribute remains evaluated after the
scan") {
+ val fact = s"$catalogName.fact_nested_sibling"
+ val dim = s"$catalogName.dim_nested_sibling"
+ withTable(fact, dim) {
+ sql(s"CREATE TABLE $fact (id INT, s STRUCT<part: INT, other: INT>) USING
$v2Source " +
+ "PARTITIONED BY (s.part)")
+ sql(s"INSERT INTO $fact VALUES " +
+ "(1, named_struct('part', 1, 'other', 10)), " +
+ "(2, named_struct('part', 1, 'other', 20)), " +
+ "(3, named_struct('part', 2, 'other', 30))")
+ sql(s"CREATE TABLE $dim (value INT) USING $v2Source")
+ sql(s"INSERT INTO $dim VALUES (10)")
+
+ val df = sql(s"SELECT id FROM $fact WHERE s.other = (SELECT max(value)
FROM $dim)")
+ checkAnswer(df, Row(1))
+
+ // A nested reference currently contributes its root attribute to
eligibility, so `s.other`
+ // may be routed to a scan that advertises `s.part`. It must remain
above the scan unless
+ // eligibility becomes path-aware.
+ assertScalarSubqueryEvaluatedAfterScan(df, expected = true)
+ }
+ }
+
test("scan implementing both runtime filtering interfaces -> rejected") {
val tbl = s"$catalogName.tbl_both_interfaces"
withTable(tbl) {
@@ -364,9 +499,9 @@ class DataSourceV2CatalystRuntimeFilterSuite extends
SharedSparkSession {
sql(s"CREATE TABLE $dim (val STRING) USING $v2Source")
sql(s"INSERT INTO $dim VALUES ('tz1')")
- // The scan declares the top-level struct column `s` as its filter
attribute, so the
- // predicate qualifies for pushdown even though it reaches into `s.tz`.
Matching the nested
- // access against the partition layout is left to the scan, which this
fixture does.
+ // The scan declares the nested partition source `s.tz` as its filter
attribute. The
+ // predicate arrives with the nested access intact, and matching it
against the partition
+ // layout is left to the scan, which this fixture does.
val df = sql(s"SELECT * FROM $tbl WHERE s.tz = (SELECT max(val) FROM
$dim)")
checkAnswer(df, Row(1, Row("tz1")))
@@ -414,46 +549,6 @@ class DataSourceV2CatalystRuntimeFilterSuite extends
SharedSparkSession {
}
}
- test("filterAttributes that is not a top-level scan attribute") {
- val tbl = s"$catalogName.tbl_unresolvable_attr"
- withTable(tbl) {
- sql(s"CREATE TABLE $tbl (id INT, part INT, s STRUCT<tz: STRING>) USING
$v2Source " +
- "PARTITIONED BY (part)")
- sql(s"INSERT INTO $tbl VALUES (1, 1, named_struct('tz', 'a'))")
-
- val scanRelation = sql(s"SELECT * FROM
$tbl").queryExecution.optimizedPlan.collectFirst {
- case r: DataSourceV2ScanRelation => r
- }.getOrElse(fail("Expected a DataSourceV2ScanRelation"))
-
- // An attribute the read schema does not carry, such as one pruned out
of the projection.
- val missing = intercept[AnalysisException] {
- scanRelation.copy(scan = new
MissingFilterAttributeScan).runtimeFilterAttrs
- }
- checkError(
- exception = missing,
- condition = "_LEGACY_ERROR_TEMP_1137",
- parameters = Map("name" -> "missing", "outputStr" -> "id,part,s"))
-
- // A nested reference is rejected up front, since `filterAttributes()`
must return
- // top-level read-schema attributes. This holds even over an int column
that could never
- // carry a nested field.
- val nested = intercept[SparkException] {
- scanRelation.copy(scan = new
NestedFilterAttributeScan).runtimeFilterAttrs
- }
- assert(nested.getMessage.contains("must be a top-level attribute"),
- s"expected the nested reference to be rejected, got
${nested.getMessage}")
-
- // Over a struct it is rejected the same way, rather than widening to
the struct column:
- // accepting `s.tz` would make runtime filters over every field of `s`
eligible, not just
- // `s.tz`.
- val struct = intercept[SparkException] {
- scanRelation.copy(scan = new
StructNestedFilterAttributeScan).runtimeFilterAttrs
- }
- assert(struct.getMessage.contains("must be a top-level attribute"),
- s"expected the nested struct reference to be rejected, got
${struct.getMessage}")
- }
- }
-
test("no runtime filter -> filter() is never called") {
val tbl = s"$catalogName.tbl5"
withTable(tbl) {
@@ -637,8 +732,8 @@ private class BothRuntimeFilteringInterfacesScan
override def filter(expressions: Array[Expression]): Unit = {}
}
-/** A scan declaring a filter attribute the read schema does not carry. */
-private class MissingFilterAttributeScan extends Scan with
SupportsRuntimeCatalystFiltering {
+/** 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)
@@ -647,30 +742,26 @@ private class MissingFilterAttributeScan extends Scan
with SupportsRuntimeCataly
override def filter(expressions: Array[Expression]): Unit = {}
}
-/**
- * A scan breaking the rule that a filter attribute must be a top level read
schema column: it
- * reports `part.nested` over the int column `part`, so it is rejected as a
nested reference.
- */
-private class NestedFilterAttributeScan extends Scan with
SupportsRuntimeCatalystFiltering {
+/** 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(Seq("part", "nested")))
+ 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 breaking the same rule over a struct column: it reports `s.tz` where
`s` is a struct.
- * The nested reference is rejected rather than widening to the struct column
`s`.
- */
-private class StructNestedFilterAttributeScan extends Scan with
SupportsRuntimeCatalystFiltering {
+/** A scan declaring a nested runtime-filter attribute beneath an integer
column. */
+private class NestedFilterAttributeScan extends
SupportsRuntimeCatalystFiltering {
- override def readSchema(): StructType =
- new StructType().add("s", new StructType().add("tz", StringType))
+ override def readSchema(): StructType = new StructType().add("part",
IntegerType)
- override def filterAttributes(): Array[NamedReference] =
Array(FieldReference(Seq("s", "tz")))
+ 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 76c7902188ed..cc003d440eb5 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<toStr: STRING, other: STRING>) USING " +
+ s"$v2Format PARTITIONED BY (derives.toStr)")
+ sql(s"INSERT INTO $fact VALUES " +
+ "(1, named_struct('toStr', 'AA', 'other', 'a')), " +
+ "(2, named_struct('toStr', 'BB', 'other', 'b')), " +
+ "(3, named_struct('toStr', 'CC', 'other', 'c'))")
+ sql(s"CREATE TABLE $dim (value STRING, selected INT) USING $v2Format")
+ sql(s"INSERT INTO $dim VALUES ('AA', 0), ('BB', 1)")
+
+ withSQLConf(
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "10") {
+ val df = sql(
+ s"""SELECT f.id FROM $fact f JOIN $dim d
+ |ON f.derives.toStr = d.value WHERE d.selected = 1""".stripMargin)
+ checkAnswer(df, Row(2))
+
+ val batchScans = collect(df.queryExecution.executedPlan) {
+ case b: BatchScanExec if b.runtimeFilters.nonEmpty => b
+ }
+ assert(batchScans.nonEmpty,
+ s"expected a scan with runtime filters, got ${df.queryExecution}")
+ val batchScan = batchScans.head
+
assert(batchScan.runtimeFilters.exists(_.exists(_.isInstanceOf[GetStructField])),
+ s"expected a runtime filter on derives.toStr, got
${batchScan.runtimeFilters}")
+ assert(batchScan.partitions.size === 3)
+ assert(batchScan.filteredPartitions.flatten.size === 1,
+ s"expected 1 partition after pruning, got
${batchScan.filteredPartitions.flatten.size}")
+ }
+ }
+ }
+
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 34b43e2354ab..becdde5aa2f4 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedRowLevelOperationCatalystRuntimeFilterSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedRowLevelOperationCatalystRuntimeFilterSuite.scala
@@ -19,6 +19,8 @@ package org.apache.spark.sql.connector
import org.apache.spark.sql.Row
import org.apache.spark.sql.connector.catalog.InMemoryTable
+import
org.apache.spark.sql.connector.expressions.LogicalExpressions.{identity,
reference}
+import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.connector.write.DeleteSummary
class GroupBasedRowLevelOperationCatalystRuntimeFilterSuite
@@ -49,6 +51,35 @@ class GroupBasedRowLevelOperationCatalystRuntimeFilterSuite
checkDeleteMetrics(numDeletedRows = 1, numCopiedRows = 1)
}
+ test("delete runtime group filtering by a nested attribute") {
+ val schema = "pk INT NOT NULL, id INT, salary INT, " +
+ "dep STRUCT<name: STRING, region: STRING>"
+ createTable(schema, Array[Transform](identity(reference(Seq("dep",
"name")))))
+ append(schema,
+ """{"pk":1,"id":1,"salary":300,"dep":{"name":"hr","region":"west"}}
+ |{"pk":2,"id":2,"salary":150,"dep":{"name":"software","region":"west"}}
+ |{"pk":3,"id":3,"salary":120,"dep":{"name":"hr","region":"east"}}
+ |""".stripMargin)
+
+ val executedPlan = executeAndKeepPlan {
+ sql(s"DELETE FROM $tableNameAsString WHERE salary IN (300, 400, 500)")
+ }
+ assertCatalystGroupFilter(
+ executedPlan,
+ expectedFilterAttrs = Seq("dep.name"),
+ expectedFilter = GroupFilter(
+ scanSchema = "salary INT, dep STRUCT<name: STRING>", groups =
Seq("hr")),
+ 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 810131eb54ce..43ed7797fb5b 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<name: STRING, region: STRING>"
+ createTable(schema, Array[Transform](identity(reference(Seq("dep",
"name")))))
+ append(schema,
+ """{"pk":1,"id":1,"salary":100,"dep":{"name":"hr","region":"west"}}
+ |{"pk":2,"id":2,"salary":200,"dep":{"name":"hr","region":"east"}}
+
|{"pk":3,"id":3,"salary":300,"dep":{"name":"software","region":"west"}}
+ |""".stripMargin)
+
+ Seq(1, 2).toDF("pk").createOrReplaceTempView("source")
+
+ val executedPlan = executeAndKeepPlan {
+ sql(
+ s"""MERGE INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk
+ |WHEN MATCHED THEN
+ | UPDATE SET t.salary = t.salary + 1
+ |""".stripMargin)
+ }
+ assertCatalystGroupFilter(
+ executedPlan,
+ expectedFilterAttrs = Seq("dep.name"),
+ expectedFilter = GroupFilter(
+ scanSchema = "pk INT, dep STRUCT<name: STRING>", groups = Seq("hr")),
+ 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 21e27e9b6c1c..c62cc48cffa1 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)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
