You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "szehon-ho (via GitHub)" <gi...@apache.org> on 2026/09/03 18:17:41 UTC
[PR] [SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures [spark]
szehon-ho opened a new pull request, #58503:
URL: https://github.com/apache/spark/pull/58503
### What changes were proposed in this pull request?
This follow-up addresses the upstream review findings from #58412 before its Spark 4.3 backport proceeds.
It:
- restricts the in-memory V1, V2, Catalyst, and row-level runtime-filtering fixtures to identity partition transforms while preserving the real partition-key ordinal;
- requires every `fullyPushedFilterAttributes()` entry to be an exact entry in `filterAttributes()`;
- reports invalid runtime-filter declarations consistently from DPP, row-level group filtering, and iterative pushdown paths;
- clarifies that Spark currently tracks runtime-filter eligibility by root attribute; and
- adds regressions for transformed partition sources and invalid fully-pushed declarations.
### Why are the changes needed?
The test fixtures could advertise a transformed source column and then compare its source value directly with a transformed partition key. That could incorrectly remove a matching partition before the residual predicate was evaluated.
Also, a scan could declare a top-level struct as fully pushed while only declaring one nested field as filterable. Because Spark tracks eligibility by root attribute, this could remove a required post-scan predicate. Invalid connector declarations could additionally surface different errors depending on the planning path.
### Does this PR introduce _any_ user-facing change?
Yes, on the unreleased `master` branch. Invalid Catalyst runtime-filter scan declarations are now rejected consistently during planning. Correct connector declarations and released Spark behavior are unchanged.
### How was this patch tested?
Tests were added for Catalyst runtime filtering, V1/V2 DPP, and both group-based and delta-based row-level operations.
- `build/sbt catalyst/Test/compile sql/Test/compile`
- `SPARK_GENERATE_GOLDEN_FILES=1 build/sbt 'core/testOnly *SparkThrowableSuite -- -t "Error conditions are correctly formatted"'`
- `SPARK_LOCAL_IP=localhost build/sbt 'sql/testOnly *DataSourceV2CatalystRuntimeFilterSuite'`
- Focused transformed-partition DPP tests in `DataSourceV2SQLSuiteV1Filter` and `DataSourceV2SQLSuiteV2Filter`
- Focused non-identity partition transform tests in `GroupBasedRowLevelOperationCatalystRuntimeFilterSuite` and `DeltaBasedRowLevelOperationCatalystRuntimeFilterSuite`
- `build/sbt catalyst/scalastyle sql/scalastyle catalyst/checkstyle`
- `git diff --check`
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex with GPT-5
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #58503:
URL: https://github.com/apache/spark/pull/58503#issuecomment-5530302231
Thank you, @szehon-ho .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho closed pull request #58503: [SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures
URL: https://github.com/apache/spark/pull/58503
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #58503:
URL: https://github.com/apache/spark/pull/58503#discussion_r3928268897
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -841,14 +837,14 @@ 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
+ if (partitioning.length == 1 && identityPartitionReferences.length == 1) {
Review Comment:
Fixed in 5cd473027a7. `canEvaluate` now uses the same identity-transform condition as the scan evaluator. I also added a shared V1/V2 regression for `days(part)` that verifies the `IN` filter remains residual, returns only the matching row, and retains both source partitions.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -326,6 +315,17 @@ case class DataSourceV2ScanRelation(
scanClass = scan.getClass.getName,
relationOutput = fromAttributes(output))
}
+ declaredFullyPushedRuntimeFilterAttrs.find { fullyPushedRef =>
Review Comment:
Fixed in 5cd473027a7. Fully-pushed references are now checked for the top-level constraint, resolved against the output, and only then checked for exact membership after ordinary filter attributes are also resolved. The missing-reference test again expects `CANNOT_RESOLVE` and checks its underlying resolution cause.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala:
##########
@@ -432,30 +475,28 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession {
}
}
- test("filter on column outside filterAttributes -> not pushed, even if declared fully pushed") {
+ test("fully pushed attribute outside filterAttributes -> rejected") {
Review Comment:
Restored in 5cd473027a7 as a sibling valid-case test. It restricts `filterAttributes()` to `p1` without a fully-pushed declaration and verifies that the scalar-subquery predicate on `p2` creates no runtime filter, calls no Catalyst pushdown, remains post-scan, and retains all five partitions.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -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(
Review Comment:
Addressed in 5cd473027a7. The contextual error-wrapping resolver now lives in `V2ExpressionUtils`, and the new `DataSourceV2ScanRelation` companion was removed. I retained the existing public `resolveAttributeRefs` method because it shipped in 4.2; the runtime-filter-specific helper is `private[sql]` and is used by both relation and pushdown paths.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala:
##########
@@ -55,32 +55,32 @@ class RowLevelOperationRuntimeGroupFiltering(optimizeSubqueries: Rule[LogicalPla
override def apply(plan: LogicalPlan): LogicalPlan = plan.transformDownWithPruning(
_.containsAnyPattern(REPLACE_DATA, WRITE_DELTA)) {
case GroupBasedRowLevelOperation(replaceData, _, Some(cond),
- ExtractV2Scan(scan: SupportsRuntimeV2Filtering))
- if canInjectGroupFilters(cond, scan.filterAttributes) =>
+ r @ ExtractV2Scan(scan: SupportsRuntimeV2Filtering))
Review Comment:
Simplified in 5cd473027a7. `PartitionPruning` now has one `DataSourceV2ScanRelation` arm, and row-level group filtering has one arm each for Group and Delta operations. The exact declared references are exposed `private[sql]` for building nested pruning keys.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -841,14 +837,14 @@ abstract class InMemoryBaseTable(
var pushedFilters: Array[Filter] = Array.empty
override def filterAttributes(): Array[NamedReference] = {
- partitioning.flatMap(_.references)
+ identityPartitionReferences
Review Comment:
Fixed in 5cd473027a7. `BatchScanBaseClass` now provides one protected `identityPartitionAttributes` helper, including deduplication and read-schema resolution, and the V1, V2, Catalyst fully-pushed, and row-level scans all use it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -347,6 +347,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)
Review Comment:
Fixed in 5cd473027a7. The centralized runtime-filter resolver returns `AttributeSet.empty` before constructing a `LocalRelation` (or deriving the relation output type), so scans with no declared attributes avoid those allocations.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on PR #58503:
URL: https://github.com/apache/spark/pull/58503#issuecomment-5537117172
Thank you @szehon-ho and @dongjoon-hyun! CI failure seems unrelated, can we merge to unblock https://github.com/apache/spark/pull/58412? or try to re-run
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #58503:
URL: https://github.com/apache/spark/pull/58503#issuecomment-5544857805
Thanks all, will work on https://github.com/apache/spark/pull/58412 next
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #58503:
URL: https://github.com/apache/spark/pull/58503#issuecomment-5544854102
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/bce5ed1a072329d0cc3fa4e394dcb7872414d060
- merged into branch-4.x https://github.com/apache/spark/commit/5d40df1c63d6e7d3023c37c5869ca712f25068d1
*Posted by `merge_spark_pr.py`*
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #58503:
URL: https://github.com/apache/spark/pull/58503#discussion_r3929204986
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala:
##########
@@ -178,6 +178,57 @@ 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<toStr: STRING, other: STRING>) 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)
+ }
+ }
+ }
+
+ 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')")
Review Comment:
Addressed in a07290f6d5a. The test now inserts three distinct transformed partitions and filters with `IN` over two distinct dates, so `OptimizeIn` leaves it as an `In` and the regression exercises the identity-only `canEvaluate` guard. The focused cases pass in both `DataSourceV2SQLSuiteV1Filter` and `DataSourceV2SQLSuiteV2Filter`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #58503:
URL: https://github.com/apache/spark/pull/58503#discussion_r3927706529
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -841,14 +837,14 @@ abstract class InMemoryBaseTable(
var pushedFilters: Array[Filter] = Array.empty
override def filterAttributes(): Array[NamedReference] = {
- partitioning.flatMap(_.references)
+ identityPartitionReferences
Review Comment:
Nit: this `identityPartitionReferences.filter(readSchema.findNestedField(...).isDefined)` expression is now copied in four scans (here, `InMemoryTableWithV2Filter`, `InMemoryRowLevelOperationTable`, and `InMemoryCatalystRuntimeFilterTable.identityPartitionAttrs`), and only the last one carries `.distinct`. A single protected helper on `BatchScanBaseClass` next to `identityPartitionReferences` would remove the drift.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -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(
Review Comment:
With this and the two `PartitionPruning` call sites rewired, `V2ExpressionUtils.resolveAttributeRefs` has no remaining callers, and the new companion helper re-implements its body plus the error wrapping. Leaving both around means a future call site can pick the public `V2ExpressionUtils` one and surface `_LEGACY_ERROR_TEMP_1137` for the same misdeclaration every other path now reports as `DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.CANNOT_RESOLVE`.
Either remove it (it shipped in 4.2.0 on a public object, so worth a note in the PR description) or make it the home of the error-wrapping version and drop the new `object DataSourceV2ScanRelation`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -326,6 +315,17 @@ case class DataSourceV2ScanRelation(
scanClass = scan.getClass.getName,
relationOutput = fromAttributes(output))
}
+ declaredFullyPushedRuntimeFilterAttrs.find { fullyPushedRef =>
Review Comment:
This name-only membership check runs before `fullyPushedFilterAttributes()` is resolved against the output. A fully pushed reference that does not exist at all is therefore reported as `NOT_IN_FILTER_ATTRIBUTES` ("must also be returned by `filterAttributes()`") rather than `CANNOT_RESOLVE`. That is exactly the `MissingFullyPushedFilterAttributeScan` case, whose expectation was flipped in this PR and whose `getCause` assertion was dropped.
A connector author following that message would add `missing` to `filterAttributes()` and only then get `CANNOT_RESOLVE` on the next run. Checking resolvability of the fully pushed refs before the membership check (or resolving both lists first) would surface the root cause in one round and let the original test expectation stand.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala:
##########
@@ -55,32 +55,32 @@ class RowLevelOperationRuntimeGroupFiltering(optimizeSubqueries: Rule[LogicalPla
override def apply(plan: LogicalPlan): LogicalPlan = plan.transformDownWithPruning(
_.containsAnyPattern(REPLACE_DATA, WRITE_DELTA)) {
case GroupBasedRowLevelOperation(replaceData, _, Some(cond),
- ExtractV2Scan(scan: SupportsRuntimeV2Filtering))
- if canInjectGroupFilters(cond, scan.filterAttributes) =>
+ r @ ExtractV2Scan(scan: SupportsRuntimeV2Filtering))
Review Comment:
Nit: now that eligibility comes from `r.runtimeFilterAttrs`, the four arms here differ only by the interface type test and `filterAttributes` vs `filterAttributes()`, and the two V2 arms in `PartitionPruning.getFilterableTableScan` are identical with `scan` unused. `declaredRuntimeFilterAttrs` already folds both interfaces (and the neither-interface case) into one array.
`PartitionPruning` could collapse to a single `case (resExp, r: DataSourceV2ScanRelation) if resExp.references.subsetOf(r.runtimeFilterAttrs)`, and this rule to two arms (Group/Delta) taking `r` and using `r.scan` plus a `private[sql]` `declaredRuntimeFilterAttrs` for the raw `NamedReference` array that `injectGroupFilters` still needs for nested paths. A scan implementing neither interface yields an empty set without throwing, so this is behavior-preserving.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -841,14 +837,14 @@ 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
+ if (partitioning.length == 1 && identityPartitionReferences.length == 1) {
Review Comment:
`InMemoryScanBuilder.canEvaluate` (line 519) still uses `partitioning.length == 1 && partitioning.head.references.length == 1`, so an `In` on a single non-identity transform (e.g. `PARTITIONED BY (days(part))`) is still classified as fully evaluable and removed from `postScanFilters` by `pushFilters`. With this guard now identity-only, `build()` hands that `In` to `filter()` and it is skipped, so the static filter is evaluated nowhere.
Example: `CREATE TABLE t (id INT, part DATE) PARTITIONED BY (days(part))`, two dates inserted, `SELECT * FROM t WHERE part IN (DATE '2026-08-01')` now returns both rows.
Could we make `canEvaluate` use the same identity-only guard (`identityPartitionReferences.length == 1`) so the two sides agree?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -347,6 +347,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)
Review Comment:
Nit: `LocalRelation(output)` is built even when `filterAttrs` is empty, and the instance path now forwards here instead of resolving against `this` as before. `DataSourceV2Strategy` reads `runtimeFilterAttrs` for every V2 scan relation, including scans with no runtime-filter interface, so each of those now allocates two `LocalRelation`s for nothing. Small cost, but a one-line `if (filterAttrs.isEmpty) return AttributeSet.empty` short-circuit (or resolving against `this` in the instance path) avoids it.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala:
##########
@@ -432,30 +475,28 @@ class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession {
}
}
- test("filter on column outside filterAttributes -> not pushed, even if declared fully pushed") {
+ test("fully pushed attribute outside filterAttributes -> rejected") {
Review Comment:
Turning the old tbl4 test into a pure rejection test drops the suite's only coverage of the valid case: a scalar-subquery filter on a partition column outside a restricted `filterAttributes()` must not be routed as a runtime filter, `filter()` must not be called, and the post-scan `FilterExec` must be kept. `'filter-attributes'` now only appears in this invalid-declaration test.
Could we keep a sibling test with `'filter-attributes' = 'p1'` and no fully-pushed property, asserting `runtimeFilters.isEmpty`, `assertPushedCatalystPredicates(df, 0)`, `assertScalarSubqueryEvaluatedAfterScan(df, expected = true)` and all 5 partitions retained?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #58503:
URL: https://github.com/apache/spark/pull/58503#issuecomment-5530291796
FYI @dongjoon-hyun
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #58503:
URL: https://github.com/apache/spark/pull/58503#issuecomment-5530257338
This is the upstream follow-up requested during the review of #58412.
The changes map to that feedback as follows:
- **Transform safety:** all affected in-memory runtime-filtering fixtures now advertise and bind only identity-transform source attributes, while retaining their real partition-key ordinals. This includes `InMemoryRowLevelOperationTable`, which was called out in the follow-up review.
- **Fully-pushed validation:** `fullyPushedFilterAttributes()` must be an exact subset of `filterAttributes()`. This rejects a top-level struct root when only a nested path is filterable and exercises the previously unused invalid-declaration test hook.
- **Consistent diagnostics:** DPP, row-level group filtering, and iterative pushdown now use the same structured runtime-filter attribute resolution instead of surfacing raw resolution errors. This is why `RowLevelOperationRuntimeGroupFiltering.scala` is part of the patch.
- **Tests and cleanup:** transformed-source regressions cover Catalyst, V1/V2 DPP, and both row-level implementations; expected nested paths are represented once without lossy dot splitting; the V1/V2 evaluator guards remain minimal; and the transformed-partition test name describes fixture behavior.
- **API wording:** the runtime-filter interfaces now say Spark *currently* tracks eligibility by root attribute.
The focused suites pass with the documented local-network setting (`SPARK_LOCAL_IP=localhost`), along with compilation, error-condition validation, scalastyle, and checkstyle.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #58503:
URL: https://github.com/apache/spark/pull/58503#issuecomment-5543275122
yea it doesnt look related, re-trigger just that job
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #58503:
URL: https://github.com/apache/spark/pull/58503#discussion_r3928268897
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -841,14 +837,14 @@ 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
+ if (partitioning.length == 1 && identityPartitionReferences.length == 1) {
Review Comment:
Fixed in 5cd473027a7. now uses the same identity-transform condition as the scan evaluator. I also added a shared V1/V2 regression for that verifies the filter remains residual, returns only the matching row, and retains both source partitions.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -326,6 +315,17 @@ case class DataSourceV2ScanRelation(
scanClass = scan.getClass.getName,
relationOutput = fromAttributes(output))
}
+ declaredFullyPushedRuntimeFilterAttrs.find { fullyPushedRef =>
Review Comment:
Fixed in 5cd473027a7. Fully-pushed references are now checked for the top-level constraint, resolved against the output, and only then checked for exact membership after ordinary filter attributes are also resolved. The missing-reference test again expects and checks its underlying resolution cause.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -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(
Review Comment:
Addressed in 5cd473027a7. The contextual error-wrapping resolver now lives in , and the new companion was removed. I retained the existing public method because it shipped in 4.2; the runtime-filter-specific helper is and is used by both relation and pushdown paths.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala:
##########
@@ -55,32 +55,32 @@ class RowLevelOperationRuntimeGroupFiltering(optimizeSubqueries: Rule[LogicalPla
override def apply(plan: LogicalPlan): LogicalPlan = plan.transformDownWithPruning(
_.containsAnyPattern(REPLACE_DATA, WRITE_DELTA)) {
case GroupBasedRowLevelOperation(replaceData, _, Some(cond),
- ExtractV2Scan(scan: SupportsRuntimeV2Filtering))
- if canInjectGroupFilters(cond, scan.filterAttributes) =>
+ r @ ExtractV2Scan(scan: SupportsRuntimeV2Filtering))
Review Comment:
Simplified in 5cd473027a7. now has one arm, and row-level group filtering has one arm each for Group and Delta operations. The exact declared references are exposed for building nested pruning keys.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -841,14 +837,14 @@ abstract class InMemoryBaseTable(
var pushedFilters: Array[Filter] = Array.empty
override def filterAttributes(): Array[NamedReference] = {
- partitioning.flatMap(_.references)
+ identityPartitionReferences
Review Comment:
Fixed in 5cd473027a7. now provides one protected helper, including deduplication and read-schema resolution, and the V1, V2, Catalyst fully-pushed, and row-level scans all use it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -347,6 +347,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)
Review Comment:
Fixed in 5cd473027a7. The centralized runtime-filter resolver returns before constructing a (or deriving the relation output type), so scans with no declared attributes avoid those allocations.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #58503:
URL: https://github.com/apache/spark/pull/58503#discussion_r3929018945
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala:
##########
@@ -178,6 +178,57 @@ 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<toStr: STRING, other: STRING>) 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)
+ }
+ }
+ }
+
+ 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')")
Review Comment:
This single-element `IN` does not reach `canEvaluate` as an `In`: `OptimizeIn` (Operator Optimization batch) rewrites `part IN (x)` to `part = x` before the Early Filter and Projection Push-Down batch runs `V2ScanRelationPushDown`, so `pushFilters` receives an `EqualTo`, `canEvaluate` returns false for it regardless of the transform, and the test passes with or without the `canEvaluate` fix (I checked by reverting it mentally: the filter stays post-scan either way).
Could we use at least two distinct values, e.g. `part IN (DATE '2026-08-01', DATE '2026-08-03')`? Then the filter arrives as an `In` and the identity-only guard in `canEvaluate` is what keeps it post-scan.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org