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/08/29 16:00:32 UTC
[PR] [SPARK-59068][4.3][SQL] Restore support for nested runtime filter attributes [spark]
szehon-ho opened a new pull request, #58412:
URL: https://github.com/apache/spark/pull/58412
### What changes were proposed in this pull request?
This is a targeted backport of #58370 to `branch-4.3`.
It:
- allows nested references from ordinary `filterAttributes()` and preserves nested access in the pushed filter;
- keeps `fullyPushedFilterAttributes()` limited to top-level attributes, where Spark can safely remove post-scan evaluation;
- replaces the prior internal failure with the structured `DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE` error condition;
- fixes nested runtime group filtering for row-level operations; and
- adds coverage for Catalyst, V1, V2, DPP, dotted-name collisions, invalid attributes, and group- and delta-based row-level operations.
This is a net backport. The 4.4 commit was applied onto code that already contained the intermediate #59012 fixture refactoring. This PR adapts the final behavior to the 4.3 codebase without pulling in those unrelated refactors.
### Why are the changes needed?
`branch-4.3` already contains the unreleased `SupportsRuntimeCatalystFiltering` API introduced by `87064f614d8`, but it does not contain the final correction from #58370. Without this backport, connectors cannot report nested runtime-filter attributes correctly, and nested row-level runtime filtering can resolve the wrong expression shape.
Because Spark 4.3 is unreleased, its API contract should match the corrected behavior before release. Method signatures and binary compatibility are unchanged.
Spark continues to track eligibility by root attribute. Making that tracking fully path-aware is separate follow-up work in [SPARK-59095](https://issues.apache.org/jira/browse/SPARK-59095).
### Does this PR introduce _any_ user-facing change?
Yes. Runtime-filtering scans may report nested references from ordinary `filterAttributes()`. Nested references from `fullyPushedFilterAttributes()` remain unsupported and now produce a structured analysis error.
For `SupportsRuntimeCatalystFiltering`, this corrects behavior on the unreleased Spark 4.3 branch. There is no API signature change.
### How was this patch tested?
- `SPARK_GENERATE_GOLDEN_FILES=1 build/sbt 'core/testOnly *SparkThrowableSuite -- -t "Error conditions are correctly formatted"'`
- `build/sbt 'core/testOnly org.apache.spark.SparkThrowableSuite'`
- `build/sbt 'sql/testOnly org.apache.spark.sql.connector.DataSourceV2CatalystRuntimeFilterSuite'`
- Focused nested DPP tests in `DataSourceV2SQLSuiteV1Filter` and `DataSourceV2SQLSuiteV2Filter`
- `GroupBasedRowLevelOperationCatalystRuntimeFilterSuite`
- `DeltaBasedRowLevelOperationCatalystRuntimeFilterSuite`
- `catalyst/scalastyle`
- `sql/scalastyle`
- `catalyst/checkstyle`
### 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][4.3][SQL] Correct runtime filter attribute handling [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #58412:
URL: https://github.com/apache/spark/pull/58412#issuecomment-5488160246
hi @sunchao , can you take another look when you get a chance? Thanks!
--
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][4.3][SQL] Correct runtime filter attribute handling [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #58412:
URL: https://github.com/apache/spark/pull/58412#issuecomment-5497280754
Hi, @szehon-ho . It seems that we need to revisit `master` and `branch-4.x` first before considering this backporting PR.
--
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][4.3][SQL] Correct runtime filter attribute handling [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho closed pull request #58412: [SPARK-59068][4.3][SQL] Correct runtime filter attribute handling
URL: https://github.com/apache/spark/pull/58412
--
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][4.3][SQL] Correct runtime filter attribute handling [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #58412:
URL: https://github.com/apache/spark/pull/58412#issuecomment-5497263304
I reviewed this backport with a multi-agent review (8 finder angles, then a verification pass), cross-checking against `master` and the existing review threads. Findings already raised in earlier comments are excluded. All findings are established from source; no runtime reproduction was attempted.
### Merge order / cross-branch sync
**1. [P2] The last commit pre-includes the still-open master follow-up #58439** (`DataSourceV2Relation.scala`)
c845783a906 is the same change as #58439 (eager resolution of `fullyPushedFilterAttributes()` from `runtimeFilterAttrs`). If this merges first, branch-4.3 becomes stricter than master: a connector whose `fullyPushedFilterAttributes()` names a column pruned out of the read schema plans fine on master today but fails every `SELECT` on 4.3 with `DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.CANNOT_RESOLVE`. Suggest landing #58439 first, or noting the intended order here.
**2. [P2] The fixture transform-semantics fixes have no master counterpart** (`InMemoryBaseTable.scala`, `InMemoryTableWithV2Filter.scala`, `InMemoryCatalystRuntimeFilterTable.scala`)
The identity-only binding fixes from this PR's review (51ee79a5692 / 070a430c90b) and their regression tests exist only on this branch. master's `CatalystRuntimeFilteringScan` still binds transformed-source predicates (e.g. `days()`, `truncate()`) against transformed partition-key slots -- the exact mis-pruning fixed here -- and only the validation commit is covered by #58439. Suggest sending the fixture fixes to master as well so future backports of these files do not conflict.
### Production code (same code exists on master)
**3. [P2] A top-level struct root in `fullyPushedFilterAttributes()` is unguarded** (`DataSourceV2Relation.scala`, `checkFullyPushedFilterAttrs`)
The check rejects only nested refs (`fieldNames.length > 1`). A scan partitioned by `s.part` declaring `filterAttributes=[s.part]` and `fullyPushedFilterAttributes=[s]` lets a scalar-subquery filter on the sibling `s.other` (references `{s}`) pass the subset check in `DataSourceV2Strategy`, so `postScanFilters.diff(fullyPushedRuntimeFilters)` removes its `FilterExec` while the scan can only prune by `s.part` -- rows violating the `s.other` predicate are returned. The only guard is the new scaladoc sentence ("must not return the root struct as a substitute..."); consider a code check or at least a test.
**4. [P3] DPP and row-level paths bypass the new structured error** (`PartitionPruning.scala`, `RowLevelOperationRuntimeGroupFiltering.scala`, `PushDownUtils.scala`)
These sites still resolve `filterAttributes()` directly via `V2ExpressionUtils`, so an invalid attribute surfaces there as raw `_LEGACY_ERROR_TEMP_1137` / `INVALID_EXTRACT_BASE_FIELD_TYPE` from an optimizer rule, while a plain `SELECT` over the same scan produces the new `DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE` error -- inconsistent diagnostics for the same connector mistake, and the DPP-shaped path is untested.
### Cleanup (same code exists on master; best fixed upstream, then re-backported)
**5. [P3]** `unalias` in `RowLevelOperationRuntimeGroupFiltering.buildDynamicPruningCond` re-implements `AliasHelper.trimAliases`, which is already in scope via `PredicateHelper`.
**6. [P3]** The `readSchema.findNestedField(...).isDefined` filter block is copy-pasted five times across four in-memory fixture files; a shared helper next to `identityPartitionReferences` would collapse all call sites.
**7. [P3]** `partitionKeyPath` in `InMemoryBaseTable` matches `GetStructField` ordinals manually, while this PR's own `fieldPath` (`RowLevelOperationCatalystRuntimeFilterSuiteBase`) uses the existing `GetStructFieldObject` extractor for the identical recursion.
**8. [P3]** `SQLConf.get.resolver` is re-fetched per expression-tree node in `partitionAttrFor` (and per partitioning entry in `partitionAttributes`); the pre-PR code hoisted it once per `filter()` call.
Generated-by: Claude Fable 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][4.3][SQL] Correct runtime filter attribute handling [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #58412:
URL: https://github.com/apache/spark/pull/58412#issuecomment-5593429602
As the test failure is explained, Ill merge this to unblock. Initially was thinking to wait till the fix is in to re-run, but maybe not necessary. @HeartSaVioR is it sound ok?
--
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][4.3][SQL] Correct runtime filter attribute handling [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #58412:
URL: https://github.com/apache/spark/pull/58412#issuecomment-5623130226
Thanks all for reviews!
--
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][4.3][SQL] Correct runtime filter attribute handling [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #58412:
URL: https://github.com/apache/spark/pull/58412#issuecomment-5530257020
I opened #58503 as the `master` follow-up requested by the review here.
It addresses the upstream items before this 4.3 backport proceeds:
- restricts the V1, V2, Catalyst, and row-level in-memory fixtures to identity partition transforms while preserving the actual partition-key ordinal;
- validates that every `fullyPushedFilterAttributes()` entry is also an exact `filterAttributes()` entry, including the top-level struct-root case;
- routes DPP, row-level group filtering, and iterative pushdown through the structured runtime-filter attribute validation;
- changes the API wording to say Spark *currently* tracks eligibility by root attribute; and
- adds regressions for transformed partition sources and invalid fully-pushed declarations.
The focused Catalyst, V1/V2 DPP, and both row-level implementations pass, along with compilation, error-condition validation, scalastyle, and checkstyle.
Once #58503 lands in `master` and `branch-4.x`, this backport can be updated and reconsidered.
--
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][4.3][SQL] Restore support for nested runtime filter attributes [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #58412:
URL: https://github.com/apache/spark/pull/58412#discussion_r3897440150
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalystRuntimeFilterTable.scala:
##########
@@ -74,23 +75,28 @@ class InMemoryCatalystRuntimeFilterTable(
Option(InMemoryCatalystRuntimeFilterTable.this.properties.get(FilterAttributesKey))
.map(_.split(",").map(_.trim).toSet)
+ private val fullyPushedFilterAttrs: Set[String] = Option(
+ InMemoryCatalystRuntimeFilterTable.this.properties.get(FullyPushedFilterAttributesKey))
+ .map(_.split(",").map(_.trim).toSet)
+ .getOrElse(Set.empty)
+
+ /** Partition source columns that are present in the scan read schema. */
+ private def partitionAttrs: Array[NamedReference] = {
+ partitioning.flatMap(_.references()).distinct
+ .filter(ref => readSchema.findNestedField(
+ ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined)
+ }
+
override def filterAttributes(): Array[NamedReference] = {
- val scanFields = readSchema.fields.map(_.name).toSet
- partitioning.flatMap(_.references()).filter { ref =>
- val name = ref.fieldNames.mkString(".")
- scanFields.contains(name) &&
- restrictedFilterAttrs.forall(_.contains(name))
+ partitionAttrs.filter { ref =>
+ restrictedFilterAttrs.forall(_.contains(ref.fieldNames.mkString(".")))
}
}
+ // Not intersected with `filterAttributes()`, so a table can declare a fully pushed attribute
+ // that is not a filter attribute, a combination the interface forbids.
override def fullyPushedFilterAttributes(): Array[NamedReference] = {
- val fullyPushedFilterAttrs = Option(
- InMemoryCatalystRuntimeFilterTable.this.properties.get(FullyPushedFilterAttributesKey))
- .map(_.split(",").map(_.trim).toSet)
- .getOrElse(Set.empty)
- filterAttributes().filter { ref =>
- fullyPushedFilterAttrs.contains(ref.fieldNames.mkString("."))
- }
+ partitionAttrs.filter(ref => fullyPushedFilterAttrs.contains(ref.fieldNames.mkString(".")))
Review Comment:
Thanks. `fullyPushedFilterAttributes()` is now restricted to identity transforms, matching what the fixture evaluator can enforce directly. Non-identity sources may still receive ordinary Catalyst runtime filters, but Spark retains the post-scan residual filter. I added a `days(part)` regression verifying the runtime filter is pushed, both partitions remain at the source, and the residual returns only the matching date.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala:
##########
@@ -66,9 +67,9 @@ class InMemoryTableWithV2Filter(
extends BatchScanBaseClass(_data, readSchema, tableSchema) with SupportsRuntimeV2Filtering {
override def filterAttributes(): Array[NamedReference] = {
- val scanFields = readSchema.fields.map(_.name).toSet
partitioning.flatMap(_.references)
- .filter(ref => scanFields.contains(ref.fieldNames.mkString(".")))
+ .filter(ref => readSchema.findNestedField(
+ ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined)
Review Comment:
Thanks, good catch. I restricted both public V1/V2 fixtures to advertise only identity-transform source attributes, and added the same identity guard in their evaluators. The shared regression uses `truncate(derives.toStr, 1)` and runs against both fixtures; it verifies no DPP filter is pushed on the transformed source and that all partitions remain available for the residual join.
--
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][4.3][SQL] Restore support for nested runtime filter attributes [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #58412:
URL: https://github.com/apache/spark/pull/58412#discussion_r3897438092
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala:
##########
@@ -66,9 +67,9 @@ class InMemoryTableWithV2Filter(
extends BatchScanBaseClass(_data, readSchema, tableSchema) with SupportsRuntimeV2Filtering {
override def filterAttributes(): Array[NamedReference] = {
- val scanFields = readSchema.fields.map(_.name).toSet
partitioning.flatMap(_.references)
- .filter(ref => scanFields.contains(ref.fieldNames.mkString(".")))
+ .filter(ref => readSchema.findNestedField(
+ ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined)
Review Comment:
Thanks, good catch. I restricted both public V1/V2 fixtures to advertise only identity-transform source attributes, and added the same identity guard in their evaluators. The shared regression uses and runs against both fixtures; it verifies no DPP filter is pushed on the transformed source and that all partitions remain available for the residual join.
--
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][4.3][SQL] Correct runtime filter attribute handling [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #58412:
URL: https://github.com/apache/spark/pull/58412#issuecomment-5501803520
Thanks for the detailed review. The merge-order concern is now resolved: #58439 landed in both master (6864fbbbc3f) and branch-4.x (91c3e7d5ea1). Both commits have the same patch ID as c845783a906 in this PR.
The transform-semantics changes are limited to test fixtures. I propose handling their upstream synchronization separately rather than expanding this 4.3 backport. Items 3–8 likewise describe existing upstream behavior or cleanup and should be addressed upstream first if pursued.
--
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][4.3][SQL] Correct runtime filter attribute handling [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #58412:
URL: https://github.com/apache/spark/pull/58412#issuecomment-5546315788
looks like known failure: https://github.com/apache/spark/pull/58533
--
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][4.3][SQL] Restore support for nested runtime filter attributes [spark]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #58412:
URL: https://github.com/apache/spark/pull/58412#discussion_r3889865198
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala:
##########
@@ -66,9 +67,9 @@ class InMemoryTableWithV2Filter(
extends BatchScanBaseClass(_data, readSchema, tableSchema) with SupportsRuntimeV2Filtering {
override def filterAttributes(): Array[NamedReference] = {
- val scanFields = readSchema.fields.map(_.name).toSet
partitioning.flatMap(_.references)
- .filter(ref => scanFields.contains(ref.fieldNames.mkString(".")))
+ .filter(ref => readSchema.findNestedField(
+ ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined)
Review Comment:
[P2] Apply transform safety to the public V1/V2 fixtures
Both public fixtures still advertise nested source references from every partition transform, but their evaluators compare source predicate values directly against transformed partition keys. With `PARTITIONED BY (truncate(s.part, 1))`, a row containing `s.part = 'AB'` is stored under key `'A'`. A V2 predicate such as `s.part = (SELECT max(value) FROM dim)`, with `dim.value = 'AB'`, removes that matching partition. V1/V2 DPP has the same problem when it pushes `IN ('AB')`. The equivalent V1 declaration is in `InMemoryBaseTable.filterAttributes()`.
The base did not advertise nested `s.part`, so this pruning path was not enabled. The new identity-transform guard applies only to `CatalystRuntimeFilteringScan`; it does not protect these public-filter evaluators. Please restrict them to supported identity transforms or translate source predicates through the partition transform. Retained residual filters cannot recover a partition already removed.
This finding is limited to the in-memory test connectors and is established from source, without runtime reproduction.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalystRuntimeFilterTable.scala:
##########
@@ -74,23 +75,28 @@ class InMemoryCatalystRuntimeFilterTable(
Option(InMemoryCatalystRuntimeFilterTable.this.properties.get(FilterAttributesKey))
.map(_.split(",").map(_.trim).toSet)
+ private val fullyPushedFilterAttrs: Set[String] = Option(
+ InMemoryCatalystRuntimeFilterTable.this.properties.get(FullyPushedFilterAttributesKey))
+ .map(_.split(",").map(_.trim).toSet)
+ .getOrElse(Set.empty)
+
+ /** Partition source columns that are present in the scan read schema. */
+ private def partitionAttrs: Array[NamedReference] = {
+ partitioning.flatMap(_.references()).distinct
+ .filter(ref => readSchema.findNestedField(
+ ref.fieldNames.toImmutableArraySeq, resolver = SQLConf.get.resolver).isDefined)
+ }
+
override def filterAttributes(): Array[NamedReference] = {
- val scanFields = readSchema.fields.map(_.name).toSet
- partitioning.flatMap(_.references()).filter { ref =>
- val name = ref.fieldNames.mkString(".")
- scanFields.contains(name) &&
- restrictedFilterAttrs.forall(_.contains(name))
+ partitionAttrs.filter { ref =>
+ restrictedFilterAttrs.forall(_.contains(ref.fieldNames.mkString(".")))
}
}
+ // Not intersected with `filterAttributes()`, so a table can declare a fully pushed attribute
+ // that is not a filter attribute, a combination the interface forbids.
override def fullyPushedFilterAttributes(): Array[NamedReference] = {
- val fullyPushedFilterAttrs = Option(
- InMemoryCatalystRuntimeFilterTable.this.properties.get(FullyPushedFilterAttributesKey))
- .map(_.split(",").map(_.trim).toSet)
- .getOrElse(Set.empty)
- filterAttributes().filter { ref =>
- fullyPushedFilterAttrs.contains(ref.fieldNames.mkString("."))
- }
+ partitionAttrs.filter(ref => fullyPushedFilterAttrs.contains(ref.fieldNames.mkString(".")))
Review Comment:
[P2] Align fully pushed declarations with the new evaluator
`fullyPushedFilterAttributes()` still advertises source columns of non-identity transforms, but the new evaluator skips those transforms. For a `DATE part` column partitioned by `days(part)` with `TBLPROPERTIES('fully-pushed-filter-attributes' = 'part')`, a predicate such as `part = (SELECT max(value) FROM dim)` selecting one of two dates has its residual filter removed by `DataSourceV2Strategy`. The fixture then returns without filtering because no identity slots exist, so both dates are returned.
This worked on the base: `getKey` preserves the original DATE day value for `DaysTransform`, so the old evaluator correctly evaluated the source predicate. Please restrict fully pushed declarations to sources the evaluator actually enforces, or retain their correct evaluation.
This is a regression in the test connector introduced by the follow-up commit and is established from source, without runtime reproduction.
--
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][4.3][SQL] Correct runtime filter attribute handling [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #58412:
URL: https://github.com/apache/spark/pull/58412#issuecomment-5483315916
Added the two follow-ups from cloud-fan's post-merge review of #58370:
- Corrected the V1 runtime-filter Javadoc to describe dot-separated paths with each path part quoted as needed ([review comment](https://github.com/apache/spark/pull/58370#discussion_r3892275058)).
- Resolved and cached `fullyPushedFilterAttributes()` from the primary `runtimeFilterAttrs` planning path, and updated the regression to exercise that entry point ([review comment](https://github.com/apache/spark/pull/58370#discussion_r3892275062)).
The focused regression passes.
--
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][4.3][SQL] Restore support for nested runtime filter attributes [spark]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #58412:
URL: https://github.com/apache/spark/pull/58412#discussion_r3887066268
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -765,15 +765,60 @@ abstract class InMemoryBaseTable(
/** Predicates recorded by [[filter]], for test assertions only. */
def pushedCatalystPredicates: Seq[CatalystExpression] = catalystPredicates.toSeq
- /** AttributeReferences matching the partition-key InternalRow field order. */
- private def partitionAttributes: Seq[AttributeReference] = {
+ /**
+ * The `AttributeReference`s standing for the partition key InternalRow fields, in its field
+ * order, each paired with the name-part sequence of its partition column. The parts are kept
+ * unflattened so a quoted top-level column `a.b` (parts `Seq("a.b")`) stays distinct from a
+ * nested column `a`.`b` (parts `Seq("a", "b")`). Example:
+ * - `PARTITIONED BY (part, s.nested)` -> `(Seq("part"), AttributeReference(part))`, then
+ * `(Seq("s", "nested"), AttributeReference(s.nested))`
+ */
+ private def partitionAttributes: Seq[(Seq[String], AttributeReference)] = {
partitioning.flatMap(_.references()).flatMap { ref =>
- val name = ref.fieldNames.mkString(".")
- readSchema.find(_.name == name).orElse(tableSchema.find(_.name == name)).map { f =>
- AttributeReference(f.name, f.dataType, f.nullable)()
+ val path = ref.fieldNames.toImmutableArraySeq
+ val resolver = SQLConf.get.resolver
+ readSchema.findNestedField(path, resolver = resolver)
+ .orElse(tableSchema.findNestedField(path, resolver = resolver)).map {
+ case (_, f) =>
+ path -> AttributeReference(ref.fieldNames.mkString("."), f.dataType, f.nullable)()
Review Comment:
[P2] Preserve transform semantics for nested partition filters
`partitioning.flatMap(_.references())` includes non-identity transforms, but this nested lookup binds each source field as if its original value occupied the partition-key slot. For the fixture's supported `PARTITIONED BY (truncate(s.part, 1))`, a row with `s.part = 'AB'` is stored under key `'A'` by `getKey`. A scalar-subquery runtime predicate such as `s.part = (SELECT max(val) FROM dim)`, where `max(val)` is `'AB'`, is then remapped to `'A' = 'AB'`, and `filter()` drops the matching partition before residual evaluation can recover it. Before this change the fixture did not advertise `s.part`, so the row survived for the residual filter.
Please restrict direct key-slot binding to identity transforms while preserving actual partition-key ordinals, or project predicates through the transform. This finding is limited to the in-memory test connector and was independently checked by two reviewers from source; it was not reproduced at runtime.
--
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][4.3][SQL] Correct runtime filter attribute handling [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #58412:
URL: https://github.com/apache/spark/pull/58412#issuecomment-5533002321
Thank you, @szehon-ho . I approved the followup. Please make this PR up-to-date to make it consistent across branches.
- 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][4.3][SQL] Correct runtime filter attribute handling [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #58412:
URL: https://github.com/apache/spark/pull/58412#issuecomment-5543791383
Updated this 4.3 backport to reflect all approved changes from #58503.
What changed:
- Rebased onto the current `branch-4.3`.
- Ported the fully-pushed runtime-filter attribute validation and the related planning-path coverage from the master PR.
- Kept the 4.3-specific nested-attribute coverage.
- Updated the test fixture to evaluate only identity transforms, added shared scan/filter-call tracking needed by the ported tests, and strengthened the transformed-source regression coverage for both V1 and V2 filtering.
Validation completed:
- `DataSourceV2CatalystRuntimeFilterSuite`: 23 tests passed.
- V1 transformed-source tests: 2 tests passed.
- V2 transformed-source tests: 2 tests passed.
- Group row-level runtime filtering suite: 6 tests passed.
- Delta row-level runtime filtering suite: 5 tests passed.
- Catalyst and SQL main/test Scalastyle checks passed with no warnings or errors.
--
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][4.3][SQL] Correct runtime filter attribute handling [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #58412:
URL: https://github.com/apache/spark/pull/58412#issuecomment-5623119375
**Merge Summary:**
- merged into branch-4.3 https://github.com/apache/spark/commit/6940f9d3dff342d239f713f39394961c2b327ea4
*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][4.3][SQL] Restore support for nested runtime filter attributes [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #58412:
URL: https://github.com/apache/spark/pull/58412#issuecomment-5463425584
@cloud-fan @dongjoon-hyun @sunchao @HeartSaVioR can you help review (this is minimal DSV2 API-facing backport for Spark 4.3 to prevent breaking change)
--
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][4.3][SQL] Restore support for nested runtime filter attributes [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #58412:
URL: https://github.com/apache/spark/pull/58412#discussion_r3888736341
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -765,15 +765,60 @@ abstract class InMemoryBaseTable(
/** Predicates recorded by [[filter]], for test assertions only. */
def pushedCatalystPredicates: Seq[CatalystExpression] = catalystPredicates.toSeq
- /** AttributeReferences matching the partition-key InternalRow field order. */
- private def partitionAttributes: Seq[AttributeReference] = {
+ /**
+ * The `AttributeReference`s standing for the partition key InternalRow fields, in its field
+ * order, each paired with the name-part sequence of its partition column. The parts are kept
+ * unflattened so a quoted top-level column `a.b` (parts `Seq("a.b")`) stays distinct from a
+ * nested column `a`.`b` (parts `Seq("a", "b")`). Example:
+ * - `PARTITIONED BY (part, s.nested)` -> `(Seq("part"), AttributeReference(part))`, then
+ * `(Seq("s", "nested"), AttributeReference(s.nested))`
+ */
+ private def partitionAttributes: Seq[(Seq[String], AttributeReference)] = {
partitioning.flatMap(_.references()).flatMap { ref =>
- val name = ref.fieldNames.mkString(".")
- readSchema.find(_.name == name).orElse(tableSchema.find(_.name == name)).map { f =>
- AttributeReference(f.name, f.dataType, f.nullable)()
+ val path = ref.fieldNames.toImmutableArraySeq
+ val resolver = SQLConf.get.resolver
+ readSchema.findNestedField(path, resolver = resolver)
+ .orElse(tableSchema.findNestedField(path, resolver = resolver)).map {
+ case (_, f) =>
+ path -> AttributeReference(ref.fieldNames.mkString("."), f.dataType, f.nullable)()
Review Comment:
Thanks, good catch. The fixture now binds only identity transforms and uses each transform’s actual partition-key ordinal. Predicates on transformed source columns remain for residual evaluation. I added a regression covering a nested truncate transform in slot 0 and a nested identity transform in slot 1.
--
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][4.3][SQL] Correct runtime filter attribute handling [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #58412:
URL: https://github.com/apache/spark/pull/58412#issuecomment-5528646817
Follow-up review of c845783a906 against the latest `master` (bcea2b23caa). Items already covered by the existing threads and my earlier comment are excluded. All findings are established from source; no runtime reproduction was attempted.
### Test fixtures (this PR)
**1. [P2] `InMemoryRowLevelOperationTable` did not receive the identity-transform restriction** (`InMemoryRowLevelOperationTable.scala`, `filterAttributes`)
After the review fixes, the V1/V2/Catalyst fixtures advertise only identity-transform sources, but the row-level fixture still returns `partitioning.flatMap(_.references())` while sharing the same `CatalystRuntimeFilteringScan` evaluator that binds only identity slots. With the new `createTable(columns, transforms)` overload, a `bucket(4, dep)` or `days(...)` row-level table gets a group filter injected and recorded, but `filter()` returns early on `partAttrs.isEmpty` and prunes nothing, so a future `checkReplacedPartitions` assertion would rewrite every partition and fail without pointing at the fixture asymmetry. Suggest using `identityPartitionReferences` here as well, or documenting why row-level differs.
**2. [P3] The test name "transformed partition source cannot be declared fully pushed" implies a Spark-side guard that does not exist** (`DataSourceV2CatalystRuntimeFilterSuite.scala`)
The test exercises no rejection: the fixture's `fullyPushedFilterAttributes()` silently drops `part` via `identityPartitionAttrs`, and `checkFullyPushedFilterAttrs` only rejects nested references. A real connector partitioned by `days(part)` that lists `part` in `fullyPushedFilterAttributes()` still passes validation, `DataSourceV2Strategy` drops the post-scan `FilterExec`, and the scan compares a date against a day ordinal. Suggest renaming the test to describe the fixture behavior, or adding a Spark-side guard and a test for it.
**3. [P3] `InMemoryCatalystRuntimeFilterTable.fullyPushedFilterAttributes()` advertises an unused test hook** (`InMemoryCatalystRuntimeFilterTable.scala`)
It switched from `filterAttributes().filter(...)` to `identityPartitionAttrs.filter(...)`, and the new comment says a table can now declare a fully pushed attribute outside `filterAttributes()`, "a combination the interface forbids". No test sets both `filter-attributes` and `fully-pushed-filter-attributes`, so the hook is unused and `partitionAttrs` / `identityPartitionAttrs` are near-duplicates to keep in sync. `filterAttributes().filter(ref => identityRefs.contains(ref) && fullyPushedFilterAttrs.contains(...))` covers the actual need without the extra helper or the comment.
**4. [P3] `expectedFilterPaths` duplicates `expectedFilterAttrs`** (`RowLevelOperationCatalystRuntimeFilterSuiteBase.scala`, `assertCatalystGroupFilter`)
`expectedFilterAttrs` is already asserted equal to `scan.filterAttributes().map(_.fieldNames.mkString("."))`, so both new callers pass the same information twice (`Seq("dep.name")` and `Some(Seq(Seq("dep", "name")))`). `expectedFilterAttrs.map(_.split('.').toSeq)` works for every existing call site and lets the two nested tests drop the extra argument.
**5. [P3] ~40 unchanged lines re-indented in `InMemoryBatchScan.filter` and `InMemoryTableWithV2Filter.filter`**
Both bodies were wrapped in `partitioning match { case Array(IdentityTransform(ref)) => ... case _ => }`, adding a nesting level to every existing line although the only semantic change is that the single transform must be an identity. `git blame` on both bodies now points at this backport instead of the SPARK-56467 history, and later cherry-picks across branches will conflict on every line. A two-line change to the original guard keeps the body intact: `if (partitioning.length == 1 && identityPartitionReferences.length == 1) { val ref = identityPartitionReferences.head; ... }`.
### Interface docs (same text exists on master; best fixed upstream)
**6. [P3] Root-attribute eligibility is documented as a contract rather than current behavior** (`SupportsRuntimeFiltering.java`, `SupportsRuntimeV2Filtering.java`, `SupportsRuntimeCatalystFiltering.scala`)
Eligibility is `references.subsetOf(AttributeSet)` in `PartitionPruning`, `DataSourceV2Strategy` and `PushDownUtils`, so a scan advertising `derives.toStr` also receives predicates on `derives.other`. Beyond the connector-side burden the new docs describe, this injects a `DynamicPruningSubquery` for a join on the sibling field, runs (and may broadcast) the build side, and the source then ignores the pushed filter. Results stay correct, but no test observes the DPP path (the "sibling" test covers only scalar subqueries). Since path-aware eligibility is deferred to SPARK-59095, the Javadoc could say Spark "currently" tracks eligibility by root attribute rather than stating it as the interface contract.
Generated-by: Claude Fable 5.1
--
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][4.3][SQL] Correct runtime filter attribute handling [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #58412:
URL: https://github.com/apache/spark/pull/58412#issuecomment-5531422871
Thank you, @szehon-ho . Please address the review comments on that PR.
- 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