[SPARK-58523][SQL] Add Catalyst runtime filtering interface for DSv2 scans

[szehon-ho]

What changes were proposed in this pull request?

This PR adds an internal SupportsRuntimeCatalystFiltering mix-in for DSv2 Scans that receives runtime filters as Catalyst Expressions instead of connector Predicates.

The interface is internal because it is aimed at Spark-integrated sources that already evaluate Catalyst expressions for partition pruning (the same class of sources that bind and interpret expressions, e.g. via PartitionPredicateImpl), rather than as a general connector API that speaks V2 Predicates.

• SupportsRuntimeCatalystFiltering (in org.apache.spark.sql.internal.connector) declares filterAttributes(), filter(Array[Expression]), and fullyPushedFilterAttributes(). It is an alternative to SupportsRuntimeFiltering / SupportsRuntimeV2Filtering, not an extension of them: Spark takes exactly one of the two paths, and only one runtime filtering interface should be implemented by a data source. The name follows SupportsRuntimeV2Filtering, since in DSv2 the SupportsPushDown* prefix is used by ScanBuilder mix-ins that push down at query compilation, whereas runtime filtering interfaces are Scan mix-ins.
• PushDownUtils.pushRuntimeFilters gains a branch for the new interface. All runtime filters are pushed in a single filter call, with no translation to connector predicates, so filters that have no V2 translation still reach the source. Unavailable DPP filters (TrueLiteral / result-unavailable) are dropped explicitly. Non-deterministic filters are screened with the same pushability guard as the V2 PartitionPredicate path (isPushablePartitionFilter). The V2 path is unchanged and is matched first.
• The unwrapping logic in createRuntimePartitionPredicates (unwrap DPP, literalize scalar subqueries) is extracted into unwrapRuntimeFilterExpression and shared by both paths.
• DataSourceV2ScanRelation.runtimeFilterAttrs / fullyPushedRuntimeFilterAttrs, PartitionPruning.getFilterableTableScan, and RowLevelOperationRuntimeGroupFiltering recognize the new interface, so DPP, scalar subquery, and row-level group runtime filters are derived for these scans. Row-level group filtering covers both group-based and delta-based operations, matching the coverage the V2 interface already has.
• A scan can report fullyPushedFilterAttributes(). A runtime filter that only references those attributes is considered fully pushed and DataSourceV2Strategy drops it from the post-scan FilterExec, instead of both pushing it and re-evaluating it. Dynamic pruning filters were already excluded from the post-scan filter list. The Javadoc states that these are attributes the scan fully evaluates, and that declared attributes must be top-level fields present in readSchema, since nested references and pruned-out attributes fail to resolve when Spark builds the scan relation.
• Javadoc on SupportsRuntimeFiltering / SupportsRuntimeV2Filtering now notes that only one runtime filtering interface should be implemented, records the same top-level-attribute requirement for filterAttributes(), and clarifies that pushedPredicates() reports predicates that fully or partially help pruning rather than predicates Spark can skip evaluating.

Dropping the scalar subquery filter from the post-scan FilterExec is limited to SupportsRuntimeCatalystFiltering in this PR. Doing the same for SupportsRuntimeFiltering / SupportsRuntimeV2Filtering is deferred to a subsequent PR, because on those interfaces the filter is pushed as a translated connector Predicate and the source decides what it can fully evaluate from the semantics of that Predicate, not from the attribute alone. A source therefore cannot promise to fully evaluate every runtime filter merely because the filter references only its declared attributes: whether a given filter is fully handled depends on its translated form. Spark also has no way to know at planning time, where the post-scan FilterExec is decided, whether a filter will translate at all or whether the source will accept the translated form. Neither concern applies to the Catalyst path, where the expression reaches the scan unchanged.

Related: #57760 ([SPARK-58207][SQL][FOLLOWUP]) gates non-deterministic scalar subquery filters out of runtimeFilters in DataSourceV2Strategy. This PR also screens in the Catalyst pushRuntimeFilters branch; once #57760 lands and this branch is rebased, that screen is redundant but harmless.

Why are the changes needed?

The existing runtime filtering interfaces receive connector predicates, so a runtime filter is only pushed if it can be translated to a V2 Predicate. Filters that cannot be translated - a complex expression such as part > (subquery) + 1, RLIKE, a UDF over a partition column - are silently dropped and never reach the data source, even when the source could use them to prune input partitions. Sources that already work with Catalyst expressions have no way to receive them.

The interface also gives sources a way to state that they fully evaluate a runtime filter, so Spark can skip the redundant post-scan evaluation of that filter.

Does this PR introduce any user-facing change?

No. SupportsRuntimeCatalystFiltering lives in org.apache.spark.sql.internal.connector and is not public API. The changes to the public SupportsRuntimeFiltering and SupportsRuntimeV2Filtering interfaces are documentation only.

How was this patch tested?

Added DataSourceV2CatalystRuntimeFilterSuite, backed by InMemoryCatalystRuntimeFilterTable / InMemoryTableCatalystRuntimeFilterCatalog, covering:

• a scalar subquery on a partition column pushed as a Catalyst expression, and still evaluated after the scan because the attribute is not declared fully pushed;
• the same predicate no longer evaluated after the scan once the attribute is declared in fullyPushedFilterAttributes, with matching and nonmatching partitions so the fixture must prune itself;
• a predicate that references both a fully pushed attribute and one that is not, so it stays in the post-scan FilterExec;
• an untranslatable filter (part > (subquery) + 1) pushed instead of dropped, with the scalar subquery literalized and the surrounding expression preserved;
• a DPP filter pushed as the raw InSubqueryExec expression;
• a filter on a column outside filterAttributes not being pushed;
• no runtime filter meaning filter() is never called.

The fixture scan binds and interprets any pushed expression whose references are all partition columns and prunes its partitions accordingly, the same way PartitionPredicateImpl does, so a fully pushed predicate is enforced by the source alone once Spark drops the post-scan FilterExec.

Added RowLevelOperationCatalystRuntimeFilterSuiteBase, with GroupBasedRowLevelOperationCatalystRuntimeFilterSuite and DeltaBasedRowLevelOperationCatalystRuntimeFilterSuite, covering RowLevelOperationRuntimeGroupFiltering for Catalyst runtime-filtering scans:

• UPDATE and MERGE for both group-based and delta-based operations. Each asserts that the scan declares the group key in filterAttributes, that every scan node carries a dynamic pruning filter keyed on it, that the filter subquery projects only the columns the row-level condition needs and resolves to the expected groups, that the same Catalyst expression reaches the connector once per scan node, and that the scan then reads only those groups.
• group-based DELETE, which additionally checks the replaced partition and the delete metrics;
• delta-based DELETE, which asserts that no filter is injected: that plan scans the row ID, the condition columns and the metadata columns, so the group key is not available to filter on.

build/sbt 'sql/testOnly *DataSourceV2CatalystRuntimeFilterSuite *RowLevelOperationCatalystRuntimeFilterSuite *DataSourceV2EnhancedRuntimePartitionFilterSuite'

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

Generated-by: Cursor

[cloud-fan]

0 blocking, 1 non-blocking, 1 nit.
The production design is coherent, but the fully-pushed test fixture does not actually enforce the contract it is meant to validate; there is also one prose nit.

Correctness (1)

• sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalystRuntimeFilterTable.scala:105: The fully-pushed test fixture records predicates without applying them, so the suite does not validate source-side full evaluation against nonmatching partitions. -- see inline

Nits: 1 minor item (see inline comments).

Verification

I traced scalar-subquery filters from DataSourceV2Strategy into BatchScanExec and PushDownUtils, including the non-empty reference guard and the subset check for fully pushed attributes. I also compared the new dispatch with the existing V2 path and verified that unavailable DPP filters are omitted while scalar subqueries are replaced by their evaluated literals.

[cloud-fan]

A scan that declares a filter fully pushed must ensure its returned partitions satisfy that predicate. This implementation only records the expression, while the fully-pushed test uses rows that all happen to match, so checkAnswer cannot catch an incorrect post-scan-filter removal. Please make this fixture filter its partitions (or use a dedicated fully-evaluating fixture) and test with both matching and nonmatching partitions.

[szehon-ho]

Good catch, thanks. The fixture now prunes rather than just recording. filter() remaps the expression's references onto the partition attributes, and when every reference is a partition column it binds the expression and evaluates it against the partition key, dropping partitions that don't match -- the same bind-and-interpret approach as PartitionPredicateImpl, rather than pattern matching operators. An incorrect post-scan filter removal now shows up as a wrong answer.

The fully-pushed test inserts ($i, $i) for i in 0..4, so four of the five partitions don't match, and it expects a single row back.

[cloud-fan]

The plural subject needs agreement here: The exceptions are filters.

[szehon-ho]

Fixed.

[peter-toth]

Thanks for the PR, @szehon-ho!

The shape reads well: a separate Scan mix-in that receives runtime filters as raw Catalyst expressions, matched after the V2 path so the two are mutually exclusive, with the unwrap logic shared between them. One thing I think has to change before merge — the new branch pushes runtime filters with no determinism guard, and with fullyPushedFilterAttributes declared, a non-deterministic predicate can end up with no evaluator at all: I reproduced a plan where part = (subquery) OR rand() < 0.5 is pushed to the source and no FilterExec is left above the scan. The closest sibling, FileScanBuilder.pushFilters, screens for exactly this. The rest is docs and coverage. Since nothing in the tree implements the trait yet and the description doesn't name the intended consumer, the Javadoc is all an adopter will have to go on, which is what finding 2 is about.

I hit the fully-pushed fixture problem @cloud-fan raised at InMemoryCatalystRuntimeFilterTable.scala:105 independently, so I haven't opened a second thread — as a data point for it: with fully-pushed-filter-attributes='part' and rows part = 0..4, SELECT * FROM t WHERE part = (SELECT max(val) FROM dim) returns all five rows on this head. The current test only passes because every row satisfies the predicate.

Blocking

• 1. Non-deterministic runtime filters are pushed, and can be evaluated nowhere: the branch pushes every unwrapped filter, unlike the V2 second pass (isPushablePartitionFilter) and both pushFilters branches (SPARK-58112, SPARK-58207). With a fully-pushed attribute the post-scan FilterExec disappears too, so rand() is evaluated only by the connector, once per partition. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:228]

Non-blocking

• 2. Spell out what fullyPushedFilterAttributes promises: the attribute-level shape matches v2 file sources, but that works because partition pruning is exact per row. Nothing restricts filterAttributes to partition columns, and the source can't refuse an individual predicate, so the Javadoc should require exact per-row evaluation of an arbitrary deterministic predicate. [inline: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala:52]
• 3. Third dispatch site not updated: RowLevelOperationRuntimeGroupFiltering.scala:54,58 matches SupportsRuntimeV2Filtering, so a source that adopts the new interface silently loses runtime group filtering for MERGE/UPDATE/DELETE. Add the cases or document the gap. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala:90]
• 4. Missing partitioning-preservation contract: replanWithRuntimeFilters enforces it on this path with hard SparkExceptions, but only the SupportsRuntimeV2Filtering Javadoc states it. [inline: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala:63]
• 5. pushedPredicates() is read by nothing in Spark: the V2 one drives the iterative-pass dedup; this one has no consumer outside the new suite. [inline: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala:76]
• 6. Undocumented resolution requirement: resolveRefs throws (or ClassCastExceptions on a nested reference) unless the declared attributes are top-level and present in the scan's readSchema. [inline: sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:221]

Minor

• 7. Vacuous assertion: case _ => Seq.empty lets the two assertPushedCatalystPredicates(df, 0) checks pass when the scan isn't the expected type. [inline: sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala:268]
• 8. Description is stale: the test list omits the seventh test (predicate on partly fully pushed filter attributes), and "All 21 tests pass" is now 25 (7 new + 18 in DataSourceV2EnhancedRuntimePartitionFilterSuite, both green on this head for me).

[peter-toth]

Finding 1. This branch pushes every unwrapped runtime filter with no determinism guard. Every sibling screens for it: the V2 second pass runs its runtime filters through isPushablePartitionFilter (PushDownUtils.scala:471 — deterministic && !hasSubquery && no PythonUDF), both pushFilters branches partition out non-deterministic filters (SPARK-58112, SPARK-58207), and the Catalyst pushdown path this interface is modelled on does the same — FileScanBuilder.pushFilters (FileScanBuilder.scala:73-79) splits on _.deterministic and drops subquery/PythonUDF filters from the partition filters it keeps. Here nothing does.

Run on this head against your own fixture:

SELECT * FROM tbl WHERE part = (SELECT max(val) FROM dim) OR rand() < 0.5

pushes ((part#270 = 3) OR (rand(6694523947714441432) < 0.5)) — deterministic = false — while the same predicate stays in the post-scan FilterExec. Any partition the source prunes on its own roll of rand() is gone for good, and Spark re-rolls for the rows that survive, so rows that should have passed are dropped.

Add TBLPROPERTIES('fully-pushed-filter-attributes' = 'part') and it gets worse — the post-scan filter disappears entirely and the connector is the only evaluator of rand(), once per partition rather than once per row:

*(1) Project [id#37, part#38]
+- BatchScan ...tbl_f[id#37, part#38] ... RuntimeFilters: [((part#38 = Subquery subquery#36, [id=#100]) OR (rand(-656537516222552506) < 0.5))]

The V2 first pass shares the missing guard (translateScalarSubqueryFilterV2 translates Rand fine — V2ExpressionBuilder.scala:155), so that half is pre-existing and deserves its own ticket. But the V2 path can never drop a post-scan filter, so the "evaluated nowhere" case is new here.

Screening with the guard the sibling pass already applies keeps the two consistent:

Suggested change

	.filterNot(_ == Literal.TrueLiteral)
	.filterNot(_ == Literal.TrueLiteral)
	.filter(isPushablePartitionFilter)

DPP filters still go through: isPushablePartitionFilter's subquery check is on catalyst SubqueryExpression, and InSubqueryExec/ScalarSubquery are ExecSubqueryExpression.

[peter-toth]

Follow-up on this: the non-determinism gap isn't specific to your branch, so I've fixed it at the root under the same ticket rather than asking you to carry it here — #57760 ([SPARK-58207][SQL][FOLLOWUP]).

The gate goes in DataSourceV2Strategy rather than in pushRuntimeFilters: scalarSubqueryFilters now also requires f.deterministic, so a non-deterministic filter never enters runtimeFilters in the first place — which also keeps it out of EXPLAIN's RuntimeFilters: [...], out of BatchScanExec.equals/doCanonicalize, and skips a pointless filter() + planInputPartitions() round. On master that stops (part = 3) OR (RAND() < 0.5) reaching a SupportsRuntimeV2Filtering source; on your branch the same predicate then never reaches the new Catalyst branch either, and — since it stays in postScanFilters — fullyPushedRuntimeFilters can no longer drop the only evaluator of a non-deterministic predicate. So once that lands and you rebase, I'd consider this finding resolved and the .filter(isPushablePartitionFilter) I suggested above optional.

For completeness on the other half of runtimeFilters: DPP filters bypass that gate, but they can't carry non-determinism here either. A DynamicPruningSubquery whose filtering plan is non-deterministic makes the whole expression non-deterministic (PlanExpression.deterministic folds in plan.deterministic), so the Filter above the scan fails NodeWithOnlyDeterministicProjectAndFilter and CleanupDynamicPruningFilters rewrites the DynamicPruning to TrueLiteral before planning. Adding the screen on your side would be harmless, just redundant.

[szehon-ho]

Great catch, and the most valuable one in the review -- a filter that is pushed, evaluated by the source on its own roll of rand(), and then re-rolled by Spark is exactly the kind of bug that never shows up in a test with a deterministic fixture. Thank you as well for not just reporting it but fixing it at the root in #57760, and for working out that the fullyPushedFilterAttributes case makes it a correctness issue rather than a cosmetic one.

I applied your suggestion, so the Catalyst branch now screens with isPushablePartitionFilter. I kept it even though #57760 is still open: until that lands this branch needs the guard on its own, and afterwards it is redundant but harmless and keeps the two passes symmetric. Happy to drop it once #57760 is in if you would rather have the single gate in DataSourceV2Strategy.

Agreed on the DPP half. PlanExpression.deterministic folds in the filtering plan, so a non-deterministic DynamicPruningSubquery fails NodeWithOnlyDeterministicProjectAndFilter and CleanupDynamicPruningFilters rewrites it to TrueLiteral before planning -- it cannot reach the source either way.

[peter-toth]

Finding 2. No objection to the attribute-level shape — v2 file sources already work this way. FileScanBuilder.pushFilters (FileScanBuilder.scala:72-95) keeps every deterministic partition filter for itself and returns only dataFilters ++ nonDeterministicFilters as post-scan filters, so "any predicate over these attributes is fully evaluated by the source" is established practice. What I'd like is for the Javadoc to say what makes it sound, since nothing in the tree implements this trait yet and two things are easy to get wrong, both silently:

1. Exactness, not just reachability. The file-source precedent holds because partition pruning is exact — every row of a surviving file carries that partition value. Nothing restricts filterAttributes to partition columns: SupportsRuntimeV2Filtering documents it as "attributes this scan can be filtered by at runtime", and a scan may prune files or row groups by min/max statistics on a data column. Statistics-based pruning is not exact, so declaring such an attribute here returns extra rows with no error.

2. Any shape, not the shapes you recognize. The source can't refuse an individual predicate — by the time filter() runs, DataSourceV2Strategy has already removed the FilterExec. On this head, with fully-pushed-filter-attributes='part':

   SELECT * FROM t WHERE part > (SELECT max(val) FROM dim) + 1 AND CAST(part AS STRING) RLIKE '4'
   

   leaves only Filter (isnotnull(part#341) AND RLIKE(cast(part#341 as string), 4)) above the scan; part > (2 + 1) is gone, pushed as (part#341 > (2 + 1)). A source that hand-matches operators and ignores the rest — InMemoryTableWithV2Filter.filter handles only = and IN — drops it on the floor. InMemoryEnhancedRuntimePartitionFilterTable gets it right by delegating to PartitionPredicate.eval, i.e. bind and interpret (PartitionPredicateImpl.boundPredicate), which is what the file index does too.

Something along these lines:

  /**
   * Returns attributes for which this scan fully evaluates runtime predicates.
   *
   * Any runtime predicate that references only attributes in this set is considered fully pushed
   * and will not be evaluated again after the scan. These attributes must also be returned by
   * [[filterAttributes]].
   *
   * Only declare an attribute here if this scan evaluates an arbitrary deterministic Catalyst
   * predicate over it exactly, for every row it returns -- e.g. an identity partition column,
   * whose value is known for every row of a surviving partition. Do not declare an attribute
   * whose predicates only guide approximate pruning, such as file or row-group statistics.
   * Spark may push any expression that references only these attributes, so do not assume a
   * fixed set of operators: bind and evaluate the expression (see
   * [[PartitionPredicateImpl]]) instead of pattern matching it.
   */

While you're here, it would help to name the intended implementor in the PR description — it makes the contract judgeable and tells a reader why the interface is internal.

[peter-toth]

One correction here in light of #57760: that PR adds f.deterministic to the scalarSubqueryFilters routing in DataSourceV2Strategy, so a non-deterministic filter never reaches runtimeFilters and can never be a fully-pushed candidate. My "including rand() (finding 1)" above is wrong once you rebase — the set of expressions this promise has to cover is still unbounded in shape (> with arithmetic, RLIKE, a cast chain, ...) but bounded to deterministic ones, which is what the suggested wording already says. Nothing else in this finding changes.

[szehon-ho]

Thanks -- both failure modes are real, and your RLIKE example is a good demonstration that the source cannot refuse an individual predicate once FilterExec is gone.

On the wording, I landed on keeping the doc shorter. "Returns attributes for which this scan fully evaluates runtime predicates", together with "will not be evaluated again after the scan", already carries the obligation: fully evaluating a predicate means evaluating it exactly, for every row returned, whatever the predicate looks like. That rules out approximate statistics pruning and hand-matching a fixed set of operators without spelling either out. The longer text reads as implementation guidance for one particular way of satisfying the contract, and it is fairly technical for a trait Javadoc, so I would rather leave it out.

I did take the top-level attribute requirement from your finding 6, since nothing in the existing wording implies it and getting it wrong fails at planning time.

On naming the implementor, the description explains the class of source this targets -- Spark-integrated sources that already bind and interpret Catalyst expressions for partition pruning, the same ones that go through PartitionPredicateImpl -- which is also why the interface is internal.

Noted on your follow-up: after rebasing onto #57760 the set of expressions is bounded to deterministic ones, so the remaining concern is shape, which is what the current wording covers.

[peter-toth]

Finding 3. This is one of three places that dispatch on SupportsRuntimeV2Filtering; the third isn't updated. RowLevelOperationRuntimeGroupFiltering.scala:54,58 matches ExtractV2Scan(scan: SupportsRuntimeV2Filtering) for group-based and delta-based row-level operations, and canInjectGroupFilters / injectGroupFilters are typed on that interface. Since the Javadoc you added says only one runtime filtering interface should be implemented, a source that adopts SupportsRuntimeCatalystFiltering silently loses runtime group filtering for MERGE/UPDATE/DELETE — no error, just a rule that stops firing and whole unmodified groups getting read.

The rule needs nothing but filterAttributes and injects a plain DynamicPruningExpression(InSubquery(...)), which the new branch in pushRuntimeFilters already handles, so adding the two cases looks mechanical. If you'd rather keep it out of scope, please say so in the trait's Javadoc so an adopter isn't surprised.

[szehon-ho]

Good catch -- silently losing group filtering for MERGE/UPDATE/DELETE is much worse than an error, and I had missed that this rule dispatches on the interface too.

Added, and it was mechanical as you expected. canInjectGroupFilters and injectGroupFilters are now typed on Array[NamedReference] and Scan rather than on SupportsRuntimeV2Filtering, and there are four cases: group-based and delta-based, each for both interfaces.

Also added tests, since the delta side of this had no Catalyst coverage at all: RowLevelOperationCatalystRuntimeFilterSuiteBase with a group-based and a delta-based subclass. They assert the injected filter is keyed on the group key, that its subquery projects only the columns the row-level condition needs and resolves to the expected groups, that the same Catalyst expression reaches the connector once per scan node, and that the scan then reads only those groups. A delta-based DELETE scans the row ID, the condition columns and the metadata columns, so the group key is not in the read schema and no filter is injected -- that case is asserted explicitly.

[peter-toth]

Finding 4. SupportsRuntimeV2Filtering.filter documents the partitioning-preservation contract — "If the scan also implements SupportsReportPartitioning, it must preserve the originally reported partitioning ... The scan must not report new partition values that were not present in the original partitioning" — and PushDownUtils.replanWithRuntimeFilters enforces it for whatever scan it was handed, this interface included: it calls pushRuntimeFilters, then scan.toBatch.planInputPartitions(), then the KeyedPartitioning checks that throw SparkException on a missing HasPartitionKey, a new partition key, or a grown per-key partition count. An SPJ-active adopter reading only this Javadoc finds out from "Data source must have preserved the original partitioning during runtime filtering". Please carry that paragraph over.

[szehon-ho]

Done, carried the paragraph over. An SPJ-active adopter reading only this Javadoc should not have to learn the contract from a SparkException thrown by replanWithRuntimeFilters.

[peter-toth]

Finding 5. Nothing in Spark reads this. SupportsRuntimeV2Filtering.pushedPredicates() earns its place — PushDownUtils.scala:133,205 uses it to avoid pushing the same predicate twice across the two iterative passes — but this path pushes once and never consults the result; grepping sql/core/src/main and sql/catalyst/src/main for pushedPredicates finds no call on this trait, only the new suite's assertions.

Either drop it and let the fixture expose its own accessor (InMemoryEnhancedRuntimePartitionFilterTable.pushedPartitionPredicates is the precedent), or keep it and say in the Javadoc that it exists for inspection/testing and Spark does not consult it — as written the doc reads like part of a contract Spark relies on.

[szehon-ho]

You're right that nothing reads it -- dropped from the trait. The fixture exposes its own pushedCatalystPredicates accessor instead, following the InMemoryEnhancedRuntimePartitionFilterTable.pushedPartitionPredicates precedent, so the assertions still work without implying a contract Spark relies on.

[peter-toth]

Finding 6. resolveRefs → V2ExpressionUtils.resolveRef throws cannotResolveAttributeError when a reference doesn't resolve against the plan's output, and casts the result to Attribute — so a nested reference, which LogicalPlan.resolve hands back as an Alias(GetStructField(...)), throws a ClassCastException. fullyPushedFilterAttributes() therefore has an unwritten requirement: top-level attributes only, and only ones that survived column pruning into the scan's readSchema. Break it and the query fails at planning time.

filterAttributes carries the same requirement and is equally undocumented, but it's forced for every scan relation, so an adopter trips it on the first query. This one is only forced when a scalar-subquery runtime filter is present (scalarSubqueryFilters.filter doesn't evaluate its closure on an empty Seq), which makes it a query-shape-dependent failure. Worth a line on the trait alongside finding 2; the new fixture quietly depends on it via the scanFields.contains(name) guard at InMemoryCatalystRuntimeFilterTable.scala:267.

[szehon-ho]

Documented on both filterAttributes() and fullyPushedFilterAttributes(): each reference must be a top-level attribute present in readSchema, since nested references and attributes pruned out of the read schema fail to resolve when Spark builds the scan relation.

Since you noted the requirement is equally undocumented on the existing interfaces, I added the same note to SupportsRuntimeFiltering.filterAttributes() and SupportsRuntimeV2Filtering.filterAttributes(). Documentation only, no behaviour change there.

[peter-toth]

Finding 7. This swallows a wrong-scan-type case, and the two assertPushedCatalystPredicates(df, 0) assertions (:191, :207) are exactly the ones that then pass for the wrong reason — "nothing was pushed" and "we didn't find the scan we expected" become indistinguishable.

Suggested change

	case _ => Seq.empty
	case other => fail(s"Expected InMemoryCatalystRuntimeFilterBatchScan, got $other")

[szehon-ho]

Applied -- a test that passes because it couldn't find the scan is worse than no test.

[uros-b]

@szehon-ho Please make sure to coordinate properly with #57760 (cc @peter-toth)

Edit: I see that this has already been noted above, alongside more suggestions; otherwise, I don't have any additional concerns

[szehon-ho]

Thanks @cloud-fan and @peter-toth for the reviews. @peter-toth, these were excellent catches, and the non-determinism one in particular -- a predicate that the source evaluates on its own roll of rand() and Spark then re-rolls is a silent wrong-answer bug, and it only becomes a correctness issue once fullyPushedFilterAttributes removes the post-scan FilterExec, which is new in this PR. Thank you for chasing it to the root and fixing it upstream in #57760 instead of handing it back.

Pushed, addressing everything:

• The test fixture now binds and interprets pushed predicates over partition columns and prunes accordingly, instead of only recording them, and the fully-pushed test uses matching and nonmatching partitions so an incorrect FilterExec removal fails the test.
• The Catalyst branch screens runtime filters with isPushablePartitionFilter, matching the V2 PartitionPredicate pass. Redundant once [SPARK-58207][SQL][FOLLOWUP] Skip runtime filter pushdown for nondeterministic filters #57760 lands and this rebases, but harmless.
• RowLevelOperationRuntimeGroupFiltering now recognizes the new interface for both group-based and delta-based operations, so adopters keep runtime group filtering for DELETE, UPDATE and MERGE. Added RowLevelOperationCatalystRuntimeFilterSuiteBase with a group-based and a delta-based subclass to cover it.
• Javadoc: carried over the partitioning-preservation contract, and documented that filter attributes must be top-level fields present in readSchema -- on the new trait and on the existing SupportsRuntimeFiltering / SupportsRuntimeV2Filtering as well, since they share the requirement.
• Dropped pushedPredicates() from the trait; the fixture exposes its own accessor.
• Test helper fails on an unexpected scan type rather than returning empty.
• Grammar fix.

The one thing I did not take is the longer fullyPushedFilterAttributes wording, discussed inline -- I think "fully evaluates" already carries the obligation and the expanded version is quite technical for a trait Javadoc. Happy to revisit if you disagree.

[cloud-fan]

2 addressed, 0 remaining, 1 new. (1 newly introduced, 0 late catches, 0 previously raised.)
1 blocking, 0 non-blocking, 0 nits.
The earlier fixture and integration gaps are addressed, but the new pushability guard can now remove the only evaluator for a fully-pushed non-deterministic filter.

Correctness (1)

• sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:232: The new pushability guard can reject a fully-pushed scalar filter after planning has removed its post-scan evaluator. -- see inline

Verification

I traced scalar-subquery filters from DataSourceV2Strategy into BatchScanExec and PushDownUtils. DataSourceV2Strategy classifies fully-pushed filters solely by referenced attributes and removes them from postScanFilters, while PushDownUtils independently rejects non-deterministic expressions; those decisions can disagree and leave no evaluator. I also verified that the V2 branch remains first, unavailable DPP filters are dropped, and both row-level operation forms converge on the shared group-filter injection path.

[cloud-fan]

This guard can now remove the predicate's only evaluator. DataSourceV2Strategy puts a non-deterministic scalar-subquery filter in fullyPushedRuntimeFilters based only on its references and removes it from postScanFilters; this line then rejects it before scan.filter. For example, part = (SELECT max(val) FROM dim) OR rand() < 0.5 is evaluated nowhere when part is fully pushed. Please apply the same pushability gate before removing the post-scan filter, or keep that filter until delivery is guaranteed.

[peter-toth]

@szehon-ho, I merged #57760, which might have caused the conflicts. Please rebase or merge latest master into this PR.