[SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and test fixtures by szehon-ho · Pull Request #58503 · apache/spark

szehon-ho · 2026-09-03T18:17:38Z

What changes were proposed in this pull request?

This follow-up addresses the upstream review findings from #58412 before its Spark 4.3 backport proceeds.

It:

restricts the in-memory V1, V2, Catalyst, and row-level runtime-filtering fixtures to identity partition transforms while preserving the real partition-key ordinal;
requires every fullyPushedFilterAttributes() entry to be an exact entry in filterAttributes();
reports invalid runtime-filter declarations consistently from DPP, row-level group filtering, and iterative pushdown paths;
clarifies that Spark currently tracks runtime-filter eligibility by root attribute; and
adds regressions for transformed partition sources and invalid fully-pushed declarations.

Why are the changes needed?

The test fixtures could advertise a transformed source column and then compare its source value directly with a transformed partition key. That could incorrectly remove a matching partition before the residual predicate was evaluated.

Also, a scan could declare a top-level struct as fully pushed while only declaring one nested field as filterable. Because Spark tracks eligibility by root attribute, this could remove a required post-scan predicate. Invalid connector declarations could additionally surface different errors depending on the planning path.

Does this PR introduce any user-facing change?

Yes, on the unreleased master branch. Invalid Catalyst runtime-filter scan declarations are now rejected consistently during planning. Correct connector declarations and released Spark behavior are unchanged.

How was this patch tested?

Tests were added for Catalyst runtime filtering, V1/V2 DPP, and both group-based and delta-based row-level operations.

build/sbt catalyst/Test/compile sql/Test/compile
SPARK_GENERATE_GOLDEN_FILES=1 build/sbt 'core/testOnly *SparkThrowableSuite -- -t "Error conditions are correctly formatted"'
SPARK_LOCAL_IP=localhost build/sbt 'sql/testOnly *DataSourceV2CatalystRuntimeFilterSuite'
Focused transformed-partition DPP tests in DataSourceV2SQLSuiteV1Filter and DataSourceV2SQLSuiteV2Filter
Focused non-identity partition transform tests in GroupBasedRowLevelOperationCatalystRuntimeFilterSuite and DeltaBasedRowLevelOperationCatalystRuntimeFilterSuite
build/sbt catalyst/scalastyle sql/scalastyle catalyst/checkstyle
git diff --check

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

Generated-by: Codex with GPT-5

…st fixtures

szehon-ho · 2026-09-03T18:26:09Z

This is the upstream follow-up requested during the review of #58412.

The changes map to that feedback as follows:

Transform safety: all affected in-memory runtime-filtering fixtures now advertise and bind only identity-transform source attributes, while retaining their real partition-key ordinals. This includes InMemoryRowLevelOperationTable, which was called out in the follow-up review.
Fully-pushed validation: fullyPushedFilterAttributes() must be an exact subset of filterAttributes(). This rejects a top-level struct root when only a nested path is filterable and exercises the previously unused invalid-declaration test hook.
Consistent diagnostics: DPP, row-level group filtering, and iterative pushdown now use the same structured runtime-filter attribute resolution instead of surfacing raw resolution errors. This is why RowLevelOperationRuntimeGroupFiltering.scala is part of the patch.
Tests and cleanup: transformed-source regressions cover Catalyst, V1/V2 DPP, and both row-level implementations; expected nested paths are represented once without lossy dot splitting; the V1/V2 evaluator guards remain minimal; and the transformed-partition test name describes fixture behavior.
API wording: the runtime-filter interfaces now say Spark currently tracks eligibility by root attribute.

The focused suites pass with the documented local-network setting (SPARK_LOCAL_IP=localhost), along with compilation, error-condition validation, scalastyle, and checkstyle.

szehon-ho · 2026-09-03T18:28:59Z

FYI @dongjoon-hyun

dongjoon-hyun · 2026-09-03T18:29:52Z

Thank you, @szehon-ho .

dongjoon-hyun

Thank you for the follow-up, @szehon-ho. I went through the diff and the affected call sites. One fixture inconsistency looks like a real bug, one diagnostic ordering issue, one coverage gap, and a few cleanup items. Details inline.

dongjoon-hyun · 2026-09-03T18:55:34Z

    override def filter(filters: Array[Filter]): Unit = {
-      if (partitioning.length == 1 && partitioning.head.references().length == 1) {
-        val ref = partitioning.head.references().head
+      if (partitioning.length == 1 && identityPartitionReferences.length == 1) {


InMemoryScanBuilder.canEvaluate (line 519) still uses partitioning.length == 1 && partitioning.head.references.length == 1, so an In on a single non-identity transform (e.g. PARTITIONED BY (days(part))) is still classified as fully evaluable and removed from postScanFilters by pushFilters. With this guard now identity-only, build() hands that In to filter() and it is skipped, so the static filter is evaluated nowhere.

Example: CREATE TABLE t (id INT, part DATE) PARTITIONED BY (days(part)), two dates inserted, SELECT * FROM t WHERE part IN (DATE '2026-08-01') now returns both rows.

Could we make canEvaluate use the same identity-only guard (identityPartitionReferences.length == 1) so the two sides agree?

Fixed in 5cd4730. canEvaluate now uses the same identity-transform condition as the scan evaluator. I also added a shared V1/V2 regression for days(part) that verifies the IN filter remains residual, returns only the matching row, and retains both source partitions.

dongjoon-hyun · 2026-09-03T18:55:34Z

        scanClass = scan.getClass.getName,
        relationOutput = fromAttributes(output))
    }
+    declaredFullyPushedRuntimeFilterAttrs.find { fullyPushedRef =>


This name-only membership check runs before fullyPushedFilterAttributes() is resolved against the output. A fully pushed reference that does not exist at all is therefore reported as NOT_IN_FILTER_ATTRIBUTES ("must also be returned by filterAttributes()") rather than CANNOT_RESOLVE. That is exactly the MissingFullyPushedFilterAttributeScan case, whose expectation was flipped in this PR and whose getCause assertion was dropped.

A connector author following that message would add missing to filterAttributes() and only then get CANNOT_RESOLVE on the next run. Checking resolvability of the fully pushed refs before the membership check (or resolving both lists first) would surface the root cause in one round and let the original test expectation stand.

Fixed in 5cd4730. Fully-pushed references are now checked for the top-level constraint, resolved against the output, and only then checked for exact membership after ordinary filter attributes are also resolved. The missing-reference test again expects CANNOT_RESOLVE and checks its underlying resolution cause.

dongjoon-hyun · 2026-09-03T18:55:34Z

  }

-  test("filter on column outside filterAttributes -> not pushed, even if declared fully pushed") {
+  test("fully pushed attribute outside filterAttributes -> rejected") {


Turning the old tbl4 test into a pure rejection test drops the suite's only coverage of the valid case: a scalar-subquery filter on a partition column outside a restricted filterAttributes() must not be routed as a runtime filter, filter() must not be called, and the post-scan FilterExec must be kept. 'filter-attributes' now only appears in this invalid-declaration test.

Could we keep a sibling test with 'filter-attributes' = 'p1' and no fully-pushed property, asserting runtimeFilters.isEmpty, assertPushedCatalystPredicates(df, 0), assertScalarSubqueryEvaluatedAfterScan(df, expected = true) and all 5 partitions retained?

Restored in 5cd4730 as a sibling valid-case test. It restricts filterAttributes() to p1 without a fully-pushed declaration and verifies that the scalar-subquery predicate on p2 creates no runtime filter, calls no Catalyst pushdown, remains post-scan, and retains all five partitions.

dongjoon-hyun · 2026-09-03T18:55:34Z

        val partPredicatesPushed = filterableScan.supportsIterativePushdown() && {
-          val filterAttrs = V2ExpressionUtils.resolveAttributeRefs(
-            filterableScan.filterAttributes(), output)
+          val filterAttrs = DataSourceV2ScanRelation.resolveRuntimeFilterAttrs(


With this and the two PartitionPruning call sites rewired, V2ExpressionUtils.resolveAttributeRefs has no remaining callers, and the new companion helper re-implements its body plus the error wrapping. Leaving both around means a future call site can pick the public V2ExpressionUtils one and surface _LEGACY_ERROR_TEMP_1137 for the same misdeclaration every other path now reports as DATA_SOURCE_INVALID_RUNTIME_FILTER_ATTRIBUTE.CANNOT_RESOLVE.

Either remove it (it shipped in 4.2.0 on a public object, so worth a note in the PR description) or make it the home of the error-wrapping version and drop the new object DataSourceV2ScanRelation.

Addressed in 5cd4730. The contextual error-wrapping resolver now lives in V2ExpressionUtils, and the new DataSourceV2ScanRelation companion was removed. I retained the existing public resolveAttributeRefs method because it shipped in 4.2; the runtime-filter-specific helper is private[sql] and is used by both relation and pushdown paths.

dongjoon-hyun · 2026-09-03T18:55:34Z

    case GroupBasedRowLevelOperation(replaceData, _, Some(cond),
-        ExtractV2Scan(scan: SupportsRuntimeV2Filtering))
-        if canInjectGroupFilters(cond, scan.filterAttributes) =>
+        r @ ExtractV2Scan(scan: SupportsRuntimeV2Filtering))


Nit: now that eligibility comes from r.runtimeFilterAttrs, the four arms here differ only by the interface type test and filterAttributes vs filterAttributes(), and the two V2 arms in PartitionPruning.getFilterableTableScan are identical with scan unused. declaredRuntimeFilterAttrs already folds both interfaces (and the neither-interface case) into one array.

PartitionPruning could collapse to a single case (resExp, r: DataSourceV2ScanRelation) if resExp.references.subsetOf(r.runtimeFilterAttrs), and this rule to two arms (Group/Delta) taking r and using r.scan plus a private[sql] declaredRuntimeFilterAttrs for the raw NamedReference array that injectGroupFilters still needs for nested paths. A scan implementing neither interface yields an empty set without throwing, so this is behavior-preserving.

Simplified in 5cd4730. PartitionPruning now has one DataSourceV2ScanRelation arm, and row-level group filtering has one arm each for Group and Delta operations. The exact declared references are exposed private[sql] for building nested pruning keys.

dongjoon-hyun · 2026-09-03T18:55:34Z


    override def filterAttributes(): Array[NamedReference] = {
-      partitioning.flatMap(_.references)
+      identityPartitionReferences


Nit: this identityPartitionReferences.filter(readSchema.findNestedField(...).isDefined) expression is now copied in four scans (here, InMemoryTableWithV2Filter, InMemoryRowLevelOperationTable, and InMemoryCatalystRuntimeFilterTable.identityPartitionAttrs), and only the last one carries .distinct. A single protected helper on BatchScanBaseClass next to identityPartitionReferences would remove the drift.

Fixed in 5cd4730. BatchScanBaseClass now provides one protected identityPartitionAttributes helper, including deduplication and read-schema resolution, and the V1, V2, Catalyst fully-pushed, and row-level scans all use it.

dongjoon-hyun · 2026-09-03T18:55:34Z

+      method: String,
+      scanClass: String,
+      output: Seq[AttributeReference]): AttributeSet = {
+    val plan = LocalRelation(output)


Nit: LocalRelation(output) is built even when filterAttrs is empty, and the instance path now forwards here instead of resolving against this as before. DataSourceV2Strategy reads runtimeFilterAttrs for every V2 scan relation, including scans with no runtime-filter interface, so each of those now allocates two LocalRelations for nothing. Small cost, but a one-line if (filterAttrs.isEmpty) return AttributeSet.empty short-circuit (or resolving against this in the instance path) avoids it.

Fixed in 5cd4730. The centralized runtime-filter resolver returns AttributeSet.empty before constructing a LocalRelation (or deriving the relation output type), so scans with no declared attributes avoid those allocations.

dongjoon-hyun

Thanks for the quick turnaround, @szehon-ho. The fixes look good. One remaining note on the new regression test, inline.

dongjoon-hyun · 2026-09-03T21:50:22Z

+      sql(s"INSERT INTO $table VALUES " +
+        "(1, DATE '2026-08-01'), (2, DATE '2026-08-02')")
+
+      val df = sql(s"SELECT * FROM $table WHERE part IN (DATE '2026-08-01')")


This single-element IN does not reach canEvaluate as an In: OptimizeIn (Operator Optimization batch) rewrites part IN (x) to part = x before the Early Filter and Projection Push-Down batch runs V2ScanRelationPushDown, so pushFilters receives an EqualTo, canEvaluate returns false for it regardless of the transform, and the test passes with or without the canEvaluate fix (I checked by reverting it mentally: the filter stays post-scan either way).

Could we use at least two distinct values, e.g. part IN (DATE '2026-08-01', DATE '2026-08-03')? Then the filter arrives as an In and the identity-only guard in canEvaluate is what keeps it post-scan.

Addressed in a07290f. The test now inserts three distinct transformed partitions and filters with IN over two distinct dates, so OptimizeIn leaves it as an In and the regression exercises the identity-only canEvaluate guard. The focused cases pass in both DataSourceV2SQLSuiteV1Filter and DataSourceV2SQLSuiteV2Filter.

dongjoon-hyun

+1, LGTM (Pending CIs). Thank you, @szehon-ho .

[SPARK-59068][SQL][FOLLOWUP] Correct runtime filter validation and te…

87b02e2

…st fixtures

szehon-ho mentioned this pull request Sep 3, 2026

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

Open

dongjoon-hyun reviewed Sep 3, 2026

View reviewed changes

[SPARK-59068][SQL][FOLLOWUP] Address runtime filter review feedback

5cd4730

dongjoon-hyun reviewed Sep 3, 2026

View reviewed changes

[SPARK-59068][SQL] Strengthen transformed partition filter regression

a07290f

dongjoon-hyun approved these changes Sep 3, 2026

View reviewed changes

Uh oh!

Conversation

szehon-ho commented Sep 3, 2026

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

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

Uh oh!

szehon-ho commented Sep 3, 2026

Uh oh!

szehon-ho commented Sep 3, 2026

Uh oh!

dongjoon-hyun commented Sep 3, 2026

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

szehon-ho Sep 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

szehon-ho Sep 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

szehon-ho Sep 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

szehon-ho Sep 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

szehon-ho Sep 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

szehon-ho Sep 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

szehon-ho Sep 3, 2026 •

edited

Loading

szehon-ho Sep 3, 2026 •

edited

Loading

szehon-ho Sep 3, 2026 •

edited

Loading

szehon-ho Sep 3, 2026 •

edited

Loading

szehon-ho Sep 3, 2026 •

edited

Loading

szehon-ho Sep 3, 2026 •

edited

Loading

dongjoon-hyun left a comment •

edited

Loading