[SPARK-59068][SQL] Restore support for nested runtime filter attributes

[szehon-ho]

What changes were proposed in this pull request?

This PR restores support for nested filterAttributes() references in:

• SupportsRuntimeFiltering
• SupportsRuntimeV2Filtering
• SupportsRuntimeCatalystFiltering

For ordinary runtime filtering, it removes the top-level-only validation added by #58296. It keeps
fullyPushedFilterAttributes() restricted to top-level attributes because Spark removes the
post-scan predicate for those attributes.

The PR also enables nested runtime group filtering for row-level operations. This is new capability,
not behavior restored from before #58296.

Invalid runtime filter references now produce structured
DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.CANNOT_RESOLVE or
DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.NOT_TOP_LEVEL errors with the offending attribute,
connector method, scan class, and scan relation output. Resolution failures preserve the original
analysis error as their cause.

Why are the changes needed?

The validation in #58296 was added following this review comment:

Accepting s.tz here widens it to s and makes filters over every field eligible.

#58296 (comment)

The concern is valid because Spark currently uses an AttributeSet for eligibility. A nested
reference such as derives.toStr contributes its root attribute, derives, so Spark cannot
distinguish it from derives.other.

However, rejecting nested references breaks an existing Iceberg runtime-filtering flow.

For example, consider an Iceberg table partitioned by:

truncate(2, derives.toStr)

The end-to-end flow before #58296 was:

• Iceberg found the partition field's source column ID.
• It converted that source column's full path into a NamedReference.
• filterAttributes() therefore returned derives.toStr.
• Spark resolved that nested reference. Its current eligibility representation reduced it to the
  root attribute derives.
• A DPP predicate on derives.toStr passed eligibility.
• Importantly, the actual predicate sent to Iceberg still contained the nested derives.toStr
  access.
• Iceberg projected that predicate through each partition spec and used the resulting evaluator
  to prune scan tasks.

Iceberg implementation:

• filterAttributes()
• Runtime-filter projection and task pruning

Therefore, this flow worked end to end even though Spark's eligibility check was too coarse.

After #58296, Spark rejects derives.toStr while constructing the scan relation, breaking the
existing Iceberg integration.

This PR restores the previous scan behavior as a compatibility fix. Separately, it makes row-level
runtime group filtering work with nested attributes. SPARK-59095 tracks making runtime-filter
eligibility path-aware so Spark can distinguish derives.toStr from sibling fields such as
derives.other. Until then, connectors cannot soundly declare a fully-pushed nested path: declaring
the root struct would also make sibling-field predicates appear fully pushed, and Spark would
remove their post-scan evaluation.

Does this PR introduce any user-facing change?

Yes. Data sources can once again advertise nested runtime filter attributes.

This restores the behavior before #58296: connectors such as Iceberg no longer fail during
planning and can receive nested runtime predicates for partition pruning. Row-level operations can
also use nested runtime filter attributes for runtime group pruning.

How was this patch tested?

Added end-to-end DPP regression tests for:

• SupportsRuntimeV2Filtering
• SupportsRuntimeCatalystFiltering
• nested row-level DELETE and MERGE group filtering
• malformed and missing ordinary and fully-pushed runtime filter attributes

The tests verify that a scan advertising derives.toStr:

• is eligible for a runtime filter on derives.toStr;
• receives a predicate retaining the nested access; and
• prunes the expected partition.

The following validation passed:

• DataSourceV2CatalystRuntimeFilterSuite,
  GroupBasedRowLevelOperationCatalystRuntimeFilterSuite, and
  DeltaBasedRowLevelOperationCatalystRuntimeFilterSuite: 30/30 tests
• focused V1/V2 nested partition source tests: 2/2 tests
• SparkThrowableSuite: 37/37 tests
• Catalyst and SQL scalastyle checks

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Codex with GPT-5

[cloud-fan]

I found one blocking correctness issue in the fully-pushed Catalyst path and one missing regression for the legacy public runtime-filtering interface.

[cloud-fan]

Nested paths resolved here are stored in an AttributeSet, which keeps only their root AttributeReference. Advertising derives.toStr in fullyPushedFilterAttributes() therefore also makes a scalar-subquery predicate on derives.other pass the subset check in DataSourceV2Strategy; Spark removes that predicate from postScanFilters even though the scan never declared or evaluated the sibling path, so non-matching rows can be returned. Please keep nested fully-pushed attributes rejected until eligibility preserves complete field paths, and add exact-path and sibling-path regressions.

[szehon-ho]

Fixed in 6019cec. Ordinary nested filterAttributes remain supported, but fullyPushedFilterAttributes now rejects nested references before conversion to AttributeSet. I also added exact-path rejection and a sibling-field regression proving the residual predicate stays above the scan. The focused Catalyst suite passes 19/19 tests.

[cloud-fan]

This regression only runs in DataSourceV2SQLSuiteV2Filter, but the restored contract also changes SupportsRuntimeFiltering. That interface has a distinct Predicate[] to Filter[] adapter and legacy scan callback, neither of which is exercised here. Please make the legacy InMemoryBatchScan retain resolvable nested partition references and run the equivalent pruning case in the V1 suite so this public compatibility path is covered end to end.

[szehon-ho]

Fixed in 6019cec. The legacy InMemoryBatchScan now retains resolvable nested references, and the nested DPP pruning test runs through the shared base suite for both SupportsRuntimeFiltering and SupportsRuntimeV2Filtering. Both focused V1 and V2 cases pass.

[sunchao]

Reviewed 4ac63be92cf191f878b6a3b9181ac272b46089cc with five independent reviewer agents. One confirmed P1 finding: a nested fully pushed Catalyst attribute can cause Spark to remove a predicate on a sibling field and return nonmatching rows. This confirms the existing concern with a runtime reproduction.

Validation:

• Using this PR's CI squash-merge build artifacts, the sibling-field query returned 3 rows instead of 1 with AQE both enabled and disabled. All four matching-field and pruning-only control cases passed. The CI checkout log confirms that the build source tree matches the reviewed head.
• A comparison with the three behavior-changing files restored from parent 2cc6ec7641002f9ae8de989f334d85a1f3e3c8da passed all six cases. This was a parent-source overlay comparison, not a complete parent rebuild.
• All 16 existing Catalyst runtime-filter tests and the V2 nested DPP regression passed locally (17 tests total).
• git diff HEAD^ HEAD --check passed. The PR CI workflow has 28 successful checks, 2 skipped checks, and none running.

[sunchao]

[P1] Preserve nested paths for fully pushed filters

This also permits nested references from fullyPushedFilterAttributes(), but AttributeSet keeps only their root AttributeReference. Declaring only s.part therefore makes the scalar-subquery predicate on s.other in SELECT * FROM fact WHERE s.other = (SELECT max(v) FROM dim) pass the fully-pushed subset check in DataSourceV2Strategy. Spark removes that equality from postScanFilters even though a scan partitioned only by s.part cannot evaluate the sibling field from its partition key.

I reproduced this with the existing in-memory Catalyst fixture, PARTITIONED BY (s.part), and TBLPROPERTIES('fully-pushed-filter-attributes'='s.part'). Given (id, s.part, s.other) values (1, 1, 10), (2, 1, 20), (3, 2, 30), and a dimension value of 10, the query should return only id 1. The PR CI build returns ids 1, 2, and 3 with AQE both on and off. Only the IS NOT NULL residual remains; the equality is no longer enforced. The matching-field and no-fully-pushed controls return the correct rows.

Please retain the nested-reference restriction for fully pushed declarations until eligibility preserves complete field paths, and cover sibling-field predicates before allowing Spark to remove their residual evaluation.

[szehon-ho]

Fixed in 6019cec. Nested fully-pushed references are rejected before they can be collapsed to the root AttributeReference in AttributeSet. The new sibling-field regression verifies that a predicate on s.other is retained when s.part is the declared nested filter attribute. Ordinary nested filterAttributes remain supported for V1, V2, and Catalyst compatibility.

[sunchao]

(oops sorry didn't intend to post "request changes" on the PR)

Removing the request-changes status. The finding and validation remain as non-blocking review feedback.

[sunchao]

Rereviewed 6019cecc278916d4f0bd7677e77c84e0f32c813a with five independent reviewer agents. No actionable findings remain. The previous P1 is fixed: nested fully pushed declarations are rejected, while ordinary nested runtime filtering remains supported.

Local validation passed 413 tests/probes (19 Catalyst, 383 V1, one V2 nested DPP regression, and ten focused probes), with one ignored test. The updated Scala sources were recompiled against the previous CI build; this was not a full Spark rebuild. Both incremental and full-PR diff checks passed.

The CI workflow is still queued.

[dongjoon-hyun]

Thanks for the fix. I agree with the direction, and with both of the judgement calls behind it:

• Restoring is right. DataSourceV2Strategy reads runtimeFilterAttrs unconditionally for every DSv2 scan relation, so the validation from #58296 does not just disable runtime filtering for a connector that advertises a nested partition source -- it fails every query over such a table with an INTERNAL_ERROR.
• The asymmetry is right. filterAttributes() only drives eligibility and the post-scan FilterExec survives, so an over-approximate (root-attribute) set is safe. fullyPushedFilterAttributes() removes the post-scan filter, so keeping the top-level check there is correct. The new sibling of a nested filter attribute remains evaluated after the scan test pins that boundary nicely.

A few comments, roughly in order of importance.

1. The fully-pushed nested check fires only for certain query shapes

fullyPushedRuntimeFilterAttrs is a lazy val, and the only access is inside the closure of scalarSubqueryFilters.filter { ... } in DataSourceV2Strategy. When scalarSubqueryFilters is empty the closure never runs, so the lazy val is never forced. A connector that declares a nested reference in fullyPushedFilterAttributes() therefore passes silently for most queries and only blows up once a scalar-subquery filter reaches that scan. That the new test has to poke the lazy val directly is a symptom of the same thing.

Since runtimeFilterAttrs is always forced, moving the nested check into checkRuntimeFilteringInterfaces() (shared by both lazy vals) would make the interface violation surface consistently and early.

2. Removing the doc paragraphs outright loses two things that are still true

Two of the statements in the deleted paragraphs still hold, and one of them matters more after this PR, not less:

• References must still resolve against the relation output. A pruned attribute still fails with _LEGACY_ERROR_TEMP_1137 -- the PR keeps that test (missing filter attribute -> rejected) but drops the doc for it.
• Eligibility is tracked per root attribute, so a scan declaring derives.toStr can receive a predicate on derives.other. This is the imprecision the PR deliberately keeps, so it is part of the connector contract now: worth stating in the filter() javadoc that a predicate may reference a sibling nested field the scan did not declare, and the scan has to match the reference itself.

Also, SupportsRuntimeCatalystFiltering.fullyPushedFilterAttributes() still carries its "must be a top-level attribute" paragraph, so the two sibling methods currently document their constraints asymmetrically.

3. resolveRefs[Attribute] does not actually return Attributes, and one caller relies on it

A nested reference resolves to Alias(GetStructField(...), "toStr"), and resolveRef's asInstanceOf[T] is erased, so nothing complains. That is fine for the two AttributeSet(...) call sites here -- AttributeSet.apply takes _.references, which is exactly the root-attribute reduction this PR relies on.

But RowLevelOperationRuntimeGroupFiltering.injectGroupFilters calls the same resolveRefs[Attribute] and uses the result as real keys: Aggregate(buildKeys, buildKeys, matchingRowsPlan) and InSubquery(pruningKeys, ...), i.e. an Alias ends up in grouping keys and inside a filter condition. This PR reopens that path (InMemoryBatchScan.filterAttributes now accepts nested references) and there is no coverage for it. A DELETE/UPDATE test over a table partitioned by a nested field would settle whether it works or breaks.

Relatedly, InMemoryCatalystRowLevelBatchScan.filterAttributes in InMemoryRowLevelOperationTable.scala still has the old top-level-only filtering, so it is the one fixture out of four that was not updated.

At minimum, consider resolveRefs[NamedExpression] in resolveFilterAttrs plus a comment that a nested reference is reduced to its root attribute -- the current type parameter reads as a guarantee that does not hold.

4. Minor

• resolveTopLevelFilterAttrs is now the fully-pushed path only; a name like resolveFullyPushedFilterAttrs would match its single caller.
• In the fixtures, readSchema.findNestedField(...) uses the default case-sensitive resolver (_ == _), and it throws invalidFieldName rather than returning None when an intermediate field is not a struct. partitionAttrFor in the same file uses SQLConf.get.resolver, so passing a resolver here would be more consistent.
• The deleted NestedFilterAttributeScan covered part.nested over an INT column. That shape now surfaces as an INVALID_EXTRACT_BASE_FIELD_TYPE AnalysisException instead of an internal error; worth pinning with a test if we care about the error shape for a malformed declaration.

5. Backport

#58296 landed on both master (d7e22ce) and branch-4.x (8e31e5e), so this fix needs to reach branch-4.x as well, otherwise 4.x ships the regression.

[szehon-ho]

Thanks for the detailed review. Addressed in 467323b126c:

1. Fully-pushed nested references are now validated eagerly through checkRuntimeFilteringInterfaces(), including queries without scalar-subquery filters.
2. Restored the resolution requirements in the interface docs and documented root-attribute eligibility and possible sibling-field predicates.
3. Runtime references now resolve as NamedExpression. Row-level group filtering materializes nested build keys before aggregation and uses the underlying nested expression for pruning. Added an end-to-end nested DELETE regression.
4. Updated the in-memory fixtures to use the configured resolver and added coverage for malformed nested references over non-struct columns.
5. Agreed that this fix should be backported to branch-4.x.

Validation:

• DataSourceV2CatalystRuntimeFilterSuite: 20 passed
• GroupBasedRowLevelOperationCatalystRuntimeFilterSuite: 4 passed
• Focused V1/V2 nested DPP regressions: 2 passed

[dongjoon-hyun]

Thanks for turning this around so quickly! Overall the direction looks right to me.

I really like the asymmetry you landed on: nested references are accepted again for filterAttributes(), where the post-scan FilterExec is retained, while fullyPushedFilterAttributes() keeps the top-level-only restriction. That is the one place DataSourceV2Strategy actually drops the post-scan filter, so widening there would have been a real correctness bug. Nice call keeping the two apart.

I ran the focused suites locally on 467323b and everything is green:

• DataSourceV2CatalystRuntimeFilterSuite + GroupBasedRowLevelOperationCatalystRuntimeFilterSuite + DeltaBasedRowLevelOperationCatalystRuntimeFilterSuite: 27/27 passed
• DataSourceV2SQLSuiteV1Filter / DataSourceV2SQLSuiteV2Filter (-z "nested partition source column"): 2/2 passed

Most of my comments are inline. Two things that do not map onto the diff:

V2ExpressionUtils.resolveAttributeRefs still uses resolveRefs[Attribute]

You fixed exactly that unsound cast in DataSourceV2ScanRelation.resolveFilterAttrs, but resolveAttributeRefs is the helper that PartitionPruning.getFilterableTableScan and PushDownUtils.pushRuntimeFilters use on the same, now-legal nested filterAttributes(). It does not fail today (T erases to NamedExpression, and AttributeSet.apply(Iterable[Expression]) only calls .references, which dispatches virtually on the Alias), but the resulting Seq[Attribute] does hold Aliases, so a future caller that touches an element as an Attribute would hit a ClassCastException. Would you mind switching it to resolveRefs[NamedExpression] while you are here, so the two paths agree?

Description nit

The row-level group filtering part reads to me as new capability rather than a restoration: before #58296, buildKeys / pruningKeys were Aliases typed as Attribute, so a nested reference would have gone into Aggregate grouping and InSubquery as an alias. Might be worth calling out, since it also justifies the separate JIRA over a [FOLLOWUP].

And since #58296 is already on branch-4.x (8e31e5e), this will want the same backport, as you noted above.

[dongjoon-hyun]

I gave this a try without the manual Project, using the shape the analyzer itself produces for GROUP BY s.a:

def unalias(e: NamedExpression): Expression = e match {
  case alias: Alias => alias.child
  case other => other
}
val buildQuery = Aggregate(buildKeys.map(unalias), buildKeys, matchingRowsPlan)
DynamicPruningExpression(
  InSubquery(pruningKeys.map(unalias), ListQuery(buildQuery, numCols = buildQuery.output.length)))

The row-level suites all still pass with that: GroupBasedRowLevelOperationCatalystRuntimeFilterSuite, DeltaBasedRowLevelOperationCatalystRuntimeFilterSuite, GroupBasedDeleteFromTableSuite, DeltaBasedDeleteFromTableSuite, GroupBasedUpdateTableSuite, GroupBasedMergeIntoTableSuite -- 236/236, including your new nested DELETE test.

My reading is that PullOutGroupingExpressions already covers this: it is in the Finish Analysis batch, OptimizeSubqueries re-runs the whole optimizer on the subquery, and it inserts the equivalent Project while preserving the alias expr IDs. If the hand-rolled version is guarding against something my run does not reach, could the comment name it? Otherwise the simpler form seems nicer.

Minor either way: if it stays, Project(buildKeys, matchingRowsPlan) is enough -- carrying all of matchingRowsPlan.output is not needed to build buildKeyAttrs.

[dongjoon-hyun]

Small readability thought: this method now does two fairly different jobs -- validating the interface combination, and validating the shape of the fully-pushed references. Since DataSourceV2Strategy always touches runtimeFilterAttrs, it also means fullyPushedFilterAttributes() is now called on every Catalyst-filtering scan at planning time, even when nothing is being filtered.

The behavior is fine (and your new test depends on it), but would a separate checkFullyPushedFilterAttrs() called from both lazy vals read better? The current name no longer quite describes what happens inside.

[dongjoon-hyun]

Not blocking, but this pins a diagnosability regression that might be worth a second look: #58296 rejected this case with an internal error naming the offending scan class, and now a connector author gets INVALID_EXTRACT_BASE_FIELD_TYPE with base = "part" / other = "INT", with no hint that runtime filtering or their filterAttributes() is involved.

Could resolveFilterAttrs catch the resolution failure and re-throw with the scan class name attached? Then this test could assert on that instead.

[dongjoon-hyun]

Tiny inconsistency: this fixture matches the table properties against ref.toString, which back-quotes any name needing quoting, while InMemoryBaseTable, InMemoryTableWithV2Filter, InMemoryRowLevelOperationTable and assertCatalystGroupFilter all use fieldNames.mkString("."). A column such as a b would silently stop matching here (same on line 108). Probably worth picking one spelling.

[dongjoon-hyun]

Nit: Seq.empty meaning "fall back to expectedFilterAttrs" took me a second -- Option[Seq[String]] = None would say that more directly.

[dongjoon-hyun]

Nice test. One coverage thought: this is the only nested row-level case, and it only exercises the group-based DELETE path. Delta-based and MERGE reach buildDynamicPruningCond with different matchingRowsPlan shapes -- MERGE goes through RewritePredicateSubquery, so the new Project would land over a join. Since that is the genuinely new code in this PR, one more nested case there would be reassuring.

[szehon-ho]

Thanks for the detailed follow-up. Addressed the review in 1983e15 and 24ac1aa:

• changed V2ExpressionUtils.resolveAttributeRefs to resolve NamedExpressions;
• simplified nested row-level aggregation and left projection insertion to PullOutGroupingExpressions;
• split fully-pushed attribute validation from interface validation;
• added DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE (SQLSTATE KD000) with the attribute, scan class, and read schema, while preserving the original resolution error as the cause;
• made the Catalyst runtime-filter fixture consistently match fieldNames.mkString(".");
• changed the optional expected filter references to Option[Seq[String]];
• added nested MERGE coverage that runs for both group- and delta-based row-level operations; and
• updated the PR description to distinguish restored scan compatibility from the new nested row-level capability.

Validation:

• DataSourceV2CatalystRuntimeFilterSuite, GroupBasedRowLevelOperationCatalystRuntimeFilterSuite, and DeltaBasedRowLevelOperationCatalystRuntimeFilterSuite: 29/29 passed
• focused V1/V2 nested partition source tests: 2/2 passed
• SparkThrowableSuite: 37/37 passed, including error-catalog formatting and SQLSTATE invariants
• Catalyst and SQL scalastyle checks passed

The error-catalog golden file was regenerated with the prescribed SparkThrowableSuite command; no SQL query golden outputs apply to these synthetic scan fixtures.

[dongjoon-hyun]

Reviewed at 24ac1aaac30 (on top of 467323b126c and 1983e1593f3). Thanks for the thorough follow-ups -- all six points from my earlier round are addressed:

Earlier comment	Status
unalias instead of a manual Project	applied as suggested
checkRuntimeFilteringInterfaces() doing two jobs	split into checkFullyPushedFilterAttrs()
Diagnosability of a failed nested resolution	DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE, original error kept as the cause
ref.toString vs fieldNames.mkString(".") in the fixture	unified on fieldNames.mkString(".")
Seq.empty as "fall back to expectedFilterAttrs"	now Option[Seq[String]] = None
Only one nested row-level case (group-based DELETE)	nested MERGE added to the shared base, so it runs for group- and delta-based

I also checked that every main-code consumer of filterAttributes() moved to NamedExpression -- DataSourceV2Relation, PushDownUtils (resolveAttributeRefs), PartitionPruning (both cases), and RowLevelOperationRuntimeGroupFiltering -- so no ClassCastException path is left for a nested reference. The asymmetry the PR settles on (root-attribute widening is acceptable for ordinary runtime filtering because the post-scan predicate stays, but not for fullyPushedFilterAttributes() where Spark drops it) reads right to me, and the new error path, the sibling regression, and the V1/V2/Catalyst coverage all look good.

Remaining comments, none of them blocking:

1. The new error message names a schema the resolution did not use

The message says the attribute "cannot be resolved against the scan read schema <readSchema>", but resolveFilterAttrs resolves against the scan relation output (this), not scan.readSchema(). The two can differ, and they already do in the new fixtures: MissingFilterAttributeScan.readSchema() is STRUCT<part: INT> while the wrapped cause reports outputStr = id,part. So the top-level message and its cause point at different things. Reporting the relation output (or both) would be more accurate.

2. Two contract violations of the same family fail in two different ways

A nested filterAttributes() that cannot resolve now raises a structured AnalysisException, while a nested fullyPushedFilterAttributes() still raises SparkException.internalError with a free-form message that the test matches with getMessage.contains("must be a top-level attribute"). Since you are already adding an error condition here, giving the fully-pushed case one too would let both tests use checkError.

3. The message does not say which method was wrong

resolveFilterAttrs is shared by filterAttributes() and fullyPushedFilterAttributes(), so a bad fully-pushed reference is reported as "The runtime filter attribute ...". A connector author cannot tell which of the two methods to fix. Passing the method name through would help.

4. Rejecting nested fully-pushed references leaves a gap worth recording

Given the AttributeSet widening, rejecting them is the right call. But it also means a connector has no sound way to declare a fully-pushed nested path: it can either drop the optimization, or declare the whole struct s, which then makes a predicate on s.other pass the fully-pushed subset check and lose its post-scan FilterExec -- exactly the failure the restriction is meant to prevent. Same root cause as the path-aware eligibility follow-up you mention in the description; worth naming explicitly in that JIRA so it does not get lost.

5. The V1 path does not document how a nested reference is spelled

SupportsRuntimeFiltering.filterAttributes() now permits nested references, but Filter has no structured nested form: PredicateUtils.toV1 uses NamedReference.toString, so quoting is load-bearing (derives.toStr for a nested path vs `derives.toStr` for a top-level column with that literal name). A connector matching on fieldNames.mkString(".") and one matching on toString will disagree. One sentence in the javadoc would save someone a debugging session.

6. The nested group-filter assertion only checks the root

expectedFilterRefs = Some(Seq("dep")) asserts the root of the pruning key, not that the key is dep.name rather than dep.region. The group values and scanned-groups assertions catch a wrong path indirectly, but asserting the path on filter.child directly would make both the intent and a failure message clearer.

7. Nit: test placement in DataSourceV2SQLSuite

"nested partition source column receives a DPP runtime filter" lands after the private checkExplain helper that belongs to the EXPLAIN tests above it, splitting that group. Moving it above checkExplain keeps them together.

8. Nit: runtimeFilterAttrs scaladoc

It does not mention that it also validates the fully-pushed references, so a scan with a nested fullyPushedFilterAttributes() fails even for a query with no runtime filter at all. The new test depends on that, so it is worth a line.

[szehon-ho]

Thanks for another careful pass. Addressed in ec4fa1afd65:

• Split DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE into structured CANNOT_RESOLVE and NOT_TOP_LEVEL conditions.
• Diagnostics now identify the connector method, scan class, and actual relation output; resolution causes remain preserved.
• Added coverage for invalid filterAttributes() and fullyPushedFilterAttributes() declarations.
• Documented V1 nested-path encoding and the fully-pushed nested-path limitation.
• Group-filter tests now assert the exact nested key path.
• Moved the shared DPP regression above the EXPLAIN helper.

Validation: 30/30 runtime-filter tests, 2/2 V1/V2 compatibility tests, 37/37 error-catalog tests, and both style checks passed. The error catalog was regenerated; no SQL golden files apply.

[szehon-ho]

Hi @dongjoon-hyun I'm going to merge as I have the other approvals and I think its a 4.3 blocker, given its a unreleased API change, if there are more follow up we can do it in another pr

[szehon-ho]

Merge Summary:

• merged into master 221f8de
• merged into branch-4.x d6a0283

Posted by merge_spark_pr.py

[cloud-fan]

Review summary

Two P2 issues remain: runtimeFilterAttrs does not fully validate a missing top-level fully-pushed declaration on its ordinary planning path, and the V1 connector Javadoc incorrectly promises unquoted nested attribute strings even though non-simple path parts are quoted. The broader nested-filter and row-level plan changes otherwise have coherent V1/V2/Catalyst coverage. No local Spark test suite was run for this review; git diff --check passed and the pinned Build check was successful.

Findings

2 total: 0 P0, 0 P1, 2 P2, 0 P3.

Non-blocking (P2)

• Describe per-part quoting in V1 runtime-filter names — sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeFiltering.java:56 — see inline.
• Resolve fully-pushed declarations through runtimeFilterAttrs — sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:210 — see inline.

[cloud-fan]

Non-blocking (P2): [P2] Describe per-part quoting in V1 filter names

PredicateUtils.toV1 uses NamedReference.toString, and FieldReference.toString applies quoteIfNeeded to each path part. Thus Seq("parent", "child.with.dot") reaches this callback as parent.child.with.dot, not as an entirely unquoted name. A connector following this text can split or bind a legal nested name incorrectly. Please say that parts are dot-separated and individually quoted as needed, with parent.child.with.dot as an example.

[cloud-fan]

Non-blocking (P2): [P2] Resolve fully-pushed attributes on this entry path

This helper only rejects nested references. A one-part value returned by fullyPushedFilterAttributes() that is absent from output therefore survives runtimeFilterAttrs: filterAttributes() is resolved, but the fully-pushed array is not resolved until fullyPushedRuntimeFilterAttrs is forced. DataSourceV2Strategy does not force that lazy value when scalarSubqueryFilters is empty, so the documented build-time validation becomes query-shape dependent. Please resolve and cache the fully-pushed declarations here, reuse them in the secondary accessor, and assert this entry point with MissingFullyPushedFilterAttributeScan.