[SPARK-59068][4.3][SQL] Correct runtime filter attribute handling

[szehon-ho]

What changes were proposed in this pull request?

This PR backports the corrected runtime-filter attribute handling from #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;
• resolves and caches fully-pushed attributes during normal scan planning;
• reports invalid runtime-filter attributes with structured errors;
• enables nested runtime group filtering for row-level operations;
• documents how nested attribute paths are represented in V1 filters; and
• adds Catalyst, V1, V2, DPP, invalid-attribute, and row-level operation coverage.

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.3 contains the unreleased SupportsRuntimeCatalystFiltering API, 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 from fullyPushedFilterAttributes() 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'
• Focused nested DPP tests in DataSourceV2SQLSuiteV1Filter and DataSourceV2SQLSuiteV2Filter
• GroupBasedRowLevelOperationCatalystRuntimeFilterSuite
• DeltaBasedRowLevelOperationCatalystRuntimeFilterSuite
• catalyst/scalastyle
• sql/scalastyle
• catalyst/checkstyle
• build/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

[szehon-ho]

@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)

[sunchao]

Reviewed 7d6278d with five independent agents. One source-established issue in the in-memory test connector is noted inline; no additional production-code findings. Spark test suites were not run locally.

[sunchao]

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.

[szehon-ho]

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).
• Resolved and cached fullyPushedFilterAttributes() from the primary runtimeFilterAttrs planning path, and updated the regression to exercise that entry point (review comment).

The focused regression passes.

[szehon-ho]

hi @sunchao , can you take another look when you get a chance (for 4.3) ? Thanks!

[dongjoon-hyun]

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)
c845783 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 (51ee79a / 070a430) 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

[dongjoon-hyun]

Hi, @szehon-ho . It seems that we need to revisit master and branch-4.x first before considering this backporting PR.

[szehon-ho]

Thanks for the detailed review. The merge-order concern is now resolved: #58439 landed in both master (6864fbb) and branch-4.x (91c3e7d). Both commits have the same patch ID as c845783 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.

[dongjoon-hyun]

Follow-up review of c845783 against the latest master (bcea2b2). 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.

[dongjoon-hyun]

To prevent any accidental merging, I'm simply blocking this PR because we need to fix master branch first before this PR as @szehon-ho agreed here #58412 (comment) .

[szehon-ho]

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.

[dongjoon-hyun]

Thank you, @szehon-ho . Please address the review comments on that PR.

• [SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures #58503

[dongjoon-hyun]

Thank you, @szehon-ho . I approved the followup. Please make this PR up-to-date to make it consistent across branches.

• [SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures #58503