[SPARK-59068][4.3][SQL] Correct runtime filter attribute handling - #58412
[SPARK-59068][4.3][SQL] Correct runtime filter attribute handling#58412szehon-ho wants to merge 4 commits into
Conversation
|
@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) |
| readSchema.findNestedField(path, resolver = resolver) | ||
| .orElse(tableSchema.findNestedField(path, resolver = resolver)).map { | ||
| case (_, f) => | ||
| path -> AttributeReference(ref.fieldNames.mkString("."), f.dataType, f.nullable)() |
There was a problem hiding this 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.
There was a problem hiding this 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.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Rereviewed 51ee79a with five independent agents. The original Catalyst example is fixed and its regression test passed in CI. Two remaining findings are noted inline, both limited to the in-memory test connectors and established from source. No local Spark test suites or runtime reproductions were run.
| 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) |
There was a problem hiding this 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.
There was a problem hiding this 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.
| 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("."))) |
There was a problem hiding this 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.
There was a problem hiding this 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.
szehon-ho
changed the title
|
Added the two follow-ups from cloud-fan's post-merge review of #58370:
The focused regression passes. |
|
hi @sunchao , can you take another look when you get a chance (for 4.3) ? Thanks! |
2 participants
What changes were proposed in this pull request?
This PR backports the corrected runtime-filter attribute handling from #58370 to
branch-4.3.It:
filterAttributes()and preserves nested access in the pushed filter;fullyPushedFilterAttributes()limited to top-level attributes;This is a net backport. It adapts the final behavior to the 4.3 codebase without pulling in unrelated fixture refactoring from later branches.
Why are the changes needed?
branch-4.3contains the unreleasedSupportsRuntimeCatalystFilteringAPI, so this backport aligns its runtime-filter attribute behavior with the corrected implementation.Does this PR introduce any user-facing change?
Yes. Runtime-filtering scans may report nested references from ordinary
filterAttributes(). Nested references fromfullyPushedFilterAttributes()remain unsupported, and invalid fully-pushed attributes are rejected consistently during planning.For
SupportsRuntimeCatalystFiltering, this corrects behavior on the unreleased Spark 4.3 branch. Method signatures and binary compatibility are unchanged.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'DataSourceV2SQLSuiteV1FilterandDataSourceV2SQLSuiteV2FilterGroupBasedRowLevelOperationCatalystRuntimeFilterSuiteDeltaBasedRowLevelOperationCatalystRuntimeFilterSuitecatalyst/scalastylesql/scalastylecatalyst/checkstylebuild/sbt 'sql/testOnly org.apache.spark.sql.connector.DataSourceV2CatalystRuntimeFilterSuite -- -z "missing fully pushed filter attribute"'Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex with GPT-5