[SPARK-59012][SQL] Improve SupportsRuntimeCatalystFiltering test coverage and reject nested filter attributes by szehon-ho · Pull Request #58296 · apache/spark

szehon-ho · 2026-08-25T23:49:36Z

What changes were proposed in this pull request?

SupportsRuntimeCatalystFiltering was added by SPARK-58523, which covered row-level operations (RowLevelOperationCatalystRuntimeFilterSuiteBase and its group- and delta-based subclasses) and a core set of pushdown tests. It was not exercised in the two other places the predicate-based runtime filtering interfaces are: dynamic partition pruning and storage-partitioned joins.

This PR closes that gap by reusing the existing suites rather than cloning them, and hardens the filterAttributes() contract.

Behavior change

DataSourceV2ScanRelation.runtimeFilterAttrs (and fullyPushedRuntimeFilterAttrs) now reject a nested (multi-part) runtime-filter reference, including one over a struct column. Both runtime filtering interfaces require each filterAttributes() reference to be a top-level read-schema attribute; a nested reference such as s.tz previously resolved and widened to the enclosing struct column s, making runtime filters over every field of s eligible.

Test fixtures

InMemoryBaseTable: CatalystRuntimeFilteringScan now prunes on nested partition keys. It previously looked up a partition attribute by joining fieldNames into a single top-level name, so a GetStructField chain such as s.part never matched and no pruning happened. Partition attributes now keep their fieldNames as name-part sequences, and partitionAttrFor matches the expression's path component-wise, so a quoted top-level column `a.b` stays distinct from a nested a.b.
InMemoryCatalystRuntimeFilterTable: threads the full table-creation metadata (constraints, distribution, ordering, partition counts, advisory size, strictness, and numRowsPerSplit) through to InMemoryBaseTable, and derives filterAttributes() / fullyPushedFilterAttributes() from a shared helper.
InMemoryTableCatalystRuntimeFilterCatalog: adds InMemoryCatalystRuntimeFilterCatalog, the InMemoryCatalog counterpart, so the Catalyst fixture can be used where functions and procedures are needed.
InMemoryTableWithV2Filter: threads the same table-creation metadata through to InMemoryBaseTable.

Reused suites

DynamicPartitionPruningSuite: adds DynamicPartitionPruningV2CatalystFilterSuiteAEOff / AEOn, mirroring the existing DynamicPartitionPruningV2FilterSuite pair.
KeyGroupedPartitioningSuite: the shared SPJ fixtures move to a new KeyGroupedPartitioningSuiteBase, and the three runtime-filtering tests move to a KeyGroupedPartitioningRuntimeFilterTests trait. Two small suites then run those three tests once per interface: KeyGroupedPartitioningRuntimeFilterSuite (predicate-based, the default InMemoryCatalog) and KeyGroupedPartitioningCatalystRuntimeFilterSuite.
DistributionAndOrderingSuiteBase: catalogClassName becomes overridable so a subclass can vary the catalog behind testcat.

Note for reviewers: SPARK-42038: partially clustered: with dynamic partition filtering and SPARK-45652: SPJ should handle empty partition after dynamic filtering are unchanged, but now report under KeyGroupedPartitioningRuntimeFilterSuite instead of KeyGroupedPartitioningSuite. Most of the diff in that file is this movement.

New tests in DataSourceV2CatalystRuntimeFilterSuite

Covering behavior specific to pushing Catalyst expressions: a DPP filter on a nested partition field arriving with the nested access intact, multiple predicates pushed in a single filter() call, a filter with no V2 translation being pushed instead of dropped, a scan implementing both runtime filtering interfaces being rejected, two partition columns whose dotted names collide (a quoted top-level `x.y` and a nested x.y) binding to the correct partition slot, and the filterAttributes() contract when a reported attribute is not a top-level scan attribute -- a missing attribute and a nested reference (over both an int and a struct column) are all rejected.

One test calls PushDownUtils.replanWithRuntimeFilters directly, to reach the SPJ partitioning-preservation checks that a well-behaved source cannot trigger: dropping HasPartitionKey, reporting a partition key that was not in the original partitioning, or growing a key's split count.

Why are the changes needed?

SupportsRuntimeCatalystFiltering is the path a scan takes when runtime filters are pushed as Catalyst expressions instead of connector predicates. Dynamic partition pruning and storage-partitioned joins are the two features that produce those filters, and neither was tested against this interface, so regressions in the Catalyst path would not have been caught by the suites that cover the equivalent predicate-based path.

Does this PR introduce any user-facing change?

No end-user-facing behavior change. For connector authors, a scan that declares a nested runtime-filter attribute -- a violation of the filterAttributes() contract -- is now rejected with an internal error instead of silently widening to the enclosing top-level column.

How was this patch tested?

Existing and new unit tests. Locally, on the rebased branch:

DataSourceV2CatalystRuntimeFilterSuite: 16 tests passed, including the new dotted/nested collision test and the nested-reference rejection.
DynamicPartitionPruningV2CatalystFilterSuiteAEOff / AEOn and DynamicPartitionPruningV2FilterSuiteAEOff / AEOn: 154 tests passed.
KeyGroupedPartitioningSuite: 96 tests passed.

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

Generated-by: Cursor with Claude Opus 5

cloud-fan

0 blocking, 3 non-blocking, 2 nits.
The coverage expansion is well targeted, but the new fixture behavior needs contract and fidelity corrections before the review is clean.

Design / architecture (1)

Non-blocking: sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalystRuntimeFilterCatalog.scala:84: Forward the complete table-creation metadata through the new Catalyst test catalog. -- see inline

Correctness (2)

Non-blocking: sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:772: Preserve partition paths as name-part sequences so dotted and nested names cannot collide. -- see inline
Non-blocking: sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala:412: Reject nested filterAttributes references consistently instead of pinning their accidental widening. -- see inline

Nits: 2 minor items (see inline comments).

Verification

Traced runtime-filter attributes through DataSourceV2ScanRelation and PushDownUtils into the Catalyst test scan, compared the new catalog factory with InMemoryTableCatalog, and verified that Spark supports quoted partition-column names containing dots. No test suite was run as part of this review.

cloud-fan · 2026-08-26T03:41:06Z

+
+    val tableName = s"$name.${ident.quoted}"
+    val table = new InMemoryCatalystRuntimeFilterTable(
+      tableName, columns, partitions, properties, numRowsPerSplit)


Non-blocking:

This override accepts distribution, ordering, partition counts, advisory size, constraints, and strictness, but this constructor call drops all of them. Please thread the full metadata through the Catalyst table constructors, as InMemoryTableCatalog does, so shared suites do not silently exercise defaults.

Done -- InMemoryTableWithV2Filter and InMemoryCatalystRuntimeFilterTable now take the full table-creation metadata (constraints, distribution, ordering, partition counts, advisory size, strictness) like InMemoryTable, and InMemoryCatalystRuntimeFilterCatalog.createTable forwards all of it.

cloud-fan · 2026-08-26T03:41:06Z

+        val path = ref.fieldNames.toImmutableArraySeq
+        readSchema.findNestedField(path).orElse(tableSchema.findNestedField(path)).map {
+          case (_, f) =>
+            AttributeReference(ref.fieldNames.mkString("."), f.dataType, f.nullable)()


Non-blocking:

Please preserve fieldNames as parts here. Flattening them makes a quoted top-level a.b indistinguishable from nested a.b; if both are partition columns, partitionAttrFor can bind a predicate to the first, wrong partition-key slot. Compare path parts component-wise with the resolver and add that collision case.

Done -- partition columns now keep their fieldNames as an unflattened Seq[String], and partitionAttrFor compares path parts component-wise with the resolver, so a quoted top-level `a.b` no longer collides with nested a.b. Added a test (dotted top-level and nested partition columns -> bound to the correct partition slot) covering the collision.

cloud-fan · 2026-08-26T03:41:06Z

+      // is shared with the two predicate-based interfaces, which resolve the same way.
+      val widened = scanRelation.copy(scan = new StructNestedFilterAttributeScan)
+        .runtimeFilterAttrs
+      assert(widened.map(_.name).toSeq === Seq("s"),


Non-blocking:

filterAttributes() requires top-level read-schema attributes. Please make runtimeFilterAttrs reject nested references even when the parent is a struct and change this assertion to expect that error; accepting s.tz here widens it to s and makes filters over every field eligible.

Done -- runtimeFilterAttrs and fullyPushedRuntimeFilterAttrs now reject any multi-part reference, including one over a struct, instead of widening. The assertion now expects the rejection for both the int-parent and struct-parent cases, and the SupportsRuntimeCatalystFiltering doc is updated to match.

cloud-fan · 2026-08-26T03:41:06Z

+
+      // The scan declares the top-level struct column `s` as its filter attribute, so the
+      // predicate qualifies for pushdown even though it reaches into `s.tz`. Matching the nested
+      // access against the partition layout is left to the scan, which the fixture does not do.


Nit:

The current sentence describes the pre-change fixture behavior.

Suggested change

// access against the partition layout is left to the scan, which the fixture does not do.

// access against the partition layout is left to the scan, which this fixture now does.

Done. I removed the temporal word so the comment states the current invariant rather than a before/after transition.

cloud-fan · 2026-08-26T03:41:06Z

+  }.toSet.toSeq
+
+  protected def collectGroupPartitions(plan: SparkPlan): Seq[GroupPartitionsExec] = {
+    // here we skip collecting shuffle operators that are not associated with SMJ


Nit:

This helper returns GroupPartitionsExec nodes, not shuffle operators.

Suggested change

// here we skip collecting shuffle operators that are not associated with SMJ

// here we skip collecting group-partition operators that are not associated with SMJ

uros-b · 2026-08-26T17:41:39Z

+  override def filter(expressions: Array[Expression]): Unit = {
+    filtered = true
+  }
+}


Suggested change

}

}

(file should end with a newline)

Done -- added the trailing newline.

…iltering SupportsRuntimeCatalystFiltering was added by SPARK-58523 with coverage for row-level operations and a core set of pushdown tests, but it was not exercised in the two places where the predicate-based runtime filtering interfaces are: dynamic partition pruning and storage-partitioned joins. Extend the in-memory Catalyst fixtures and reuse the existing DPP and SPJ suites against them, so the Catalyst interface is covered by the same tests as SupportsRuntimeFiltering rather than by a parallel set of its own. This is test-only; there is no production code change.

- Reject nested (multi-part) references in runtimeFilterAttrs / fullyPushedRuntimeFilterAttrs instead of widening to the enclosing column - Thread full table-creation metadata through InMemoryTableWithV2Filter and InMemoryCatalystRuntimeFilterTable - Compare partition name parts component-wise in InMemoryBaseTable so a quoted top-level `a.b` cannot collide with nested `a`.`b`; add a collision test - Comment/nit fixes and trailing newline

cloud-fan

5 addressed, 0 remaining, 2 new to this AI review. (0 newly introduced, 2 late catches, 0 previously raised, 0 unattributed findings.)
0 blocking, 1 non-blocking, 1 nit.
The production contract change and its expanded coverage look sound. Two non-blocking test-maintenance issues remain: catalog reconstruction can switch the Catalyst test fixture to the wrong runtime-filter interface, and one SPJ comment overstates behavior when partition values are not pushed down.

Design / architecture (1)

Non-blocking: sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalystRuntimeFilterCatalog.scala:59: Preserve the Catalyst runtime-filter table type across catalog reconstruction. -- see inline

Nits: 1 minor item (see inline comments).

Verification

Inspected the changed runtime-filter contracts, planner dispatch, in-memory table construction and reconstruction paths, and the DPP and storage-partitioned join assertions. I also compared both new findings against the prior AI-reviewed commit to classify them. No tests were run.

cloud-fan · 2026-08-27T03:33:08Z

+ * out tables whose scans take runtime filters as Catalyst expressions, and honors
+ * `numRowsPerSplit` so that a partition key can have several splits.
+ */
+class InMemoryCatalystRuntimeFilterCatalog extends InMemoryCatalog {


Non-blocking:

Preserve the Catalyst-filtering table type through ALTER TABLE as well as creation. The inherited alterTable path matches this subclass as InMemoryTableWithV2Filter and rebuilds the predicate-filtering fixture, so later DPP or SPJ tests exercise the wrong interface. Please route both create and alter through a shared overridable table factory that forwards the full metadata.

Done. createTable and alterTable now both go through a shared, overridable newInMemoryTable factory on BasicInMemoryTableCatalog. The Catalyst catalogs mix in InMemoryCatalystRuntimeFilterTableFactory, so ALTER TABLE reconstructs InMemoryCatalystRuntimeFilterTable instead of matching the subclass as InMemoryTableWithV2Filter and rebuilding the predicate-filtering fixture. InMemoryTableWithV2FilterCatalog overrides the same factory so its ALTER path stays on the V2-filter table. Added a test (ALTER TABLE keeps the Catalyst runtime-filter table type) that asserts the reconstructed table type after ADD COLUMNS.

cloud-fan · 2026-08-27T03:33:08Z

+              SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> pushDownValues.toString,
+              SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> enable) {
+
+            // storage-partitioned join should kick in and fill the missing partitions & splits


Nit:

Scope this comment to pushDownValues=true. In the false branch immediately below, the test requires a shuffle and no GroupPartitionsExec, so SPJ does not kick in for every iteration described here.

Done. Scoped the comment to the pushDownValues=true case, since the false branch below requires a shuffle and no GroupPartitionsExec.

… ALTER TABLE Route CREATE and ALTER through a shared overridable newInMemoryTable factory so the Catalyst runtime-filter catalogs reconstruct InMemoryCatalystRuntimeFilterTable instead of falling back to InMemoryTableWithV2Filter on ALTER. Scope the SPJ comment to the pushDownValues=true case.

cloud-fan reviewed Aug 26, 2026

View reviewed changes

uros-b reviewed Aug 26, 2026

View reviewed changes

szehon-ho changed the title ~~[SPARK-59012][SQL] Improve test coverage for SupportsRuntimeCatalystFiltering~~ [SPARK-59012][SQL] Improve SupportsRuntimeCatalystFiltering test coverage and reject nested filter attributes Aug 27, 2026

szehon-ho added 2 commits August 26, 2026 17:19

szehon-ho force-pushed the catalyst-runtime-filter-coverage branch from c587c07 to 8b7d138 Compare August 27, 2026 00:34

cloud-fan approved these changes Aug 27, 2026

View reviewed changes

	// access against the partition layout is left to the scan, which the fixture does not do.
	// access against the partition layout is left to the scan, which this fixture now does.

	// here we skip collecting shuffle operators that are not associated with SMJ
	// here we skip collecting group-partition operators that are not associated with SMJ

Uh oh!

Conversation

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

Uh oh!

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!

cloud-fan left a comment

Choose a reason for hiding this comment

Design / architecture (1)

Correctness (2)

Verification

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!

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!

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!

Choose a reason for hiding this comment

Uh oh!

cloud-fan left a comment

Choose a reason for hiding this comment

Design / architecture (1)

Verification

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

szehon-ho commented Aug 25, 2026 •

edited

Loading