[SPARK-58523][SQL] Add Catalyst runtime filtering interface for DSv2 scans - #57727
[SPARK-58523][SQL] Add Catalyst runtime filtering interface for DSv2 scans#57727szehon-ho wants to merge 5 commits into
Conversation
… DSv2 scans The existing runtime filtering interfaces receive connector predicates, so a runtime filter that has no data source V2 translation (a complex expression, a UDF, RLIKE) is dropped and never reaches the source. Add an internal SupportsPushDownCatalystRuntimeFiltering that hands scans the Catalyst expressions directly, and let a scan declare which attributes it fully evaluates so Spark can stop re-evaluating those filters after the scan.
…alystFiltering In DSv2 the SupportsPushDown* prefix is used by ScanBuilder mix-ins, whose pushdown happens at query compilation. Runtime filtering interfaces are Scan mix-ins named SupportsRuntime<variant>Filtering, so align the new trait with SupportsRuntimeV2Filtering instead. Also drop the empty parameter lists on the trait's accessors so all of them agree with filterAttributes.
Matches the Java runtime filtering interfaces, where filterAttributes() and pushedPredicates() carry empty parameter lists, and keeps the trait's own members consistent with each other.
The catalog only needs to be registered once for the suite's session, and every test already scopes its tables with withTable, so the per-test registration and catalogManager.reset() are not needed.
Pins down that a runtime predicate is only dropped from the post-scan filter when every attribute it references is declared fully pushed.
a3a9570 to
6563ae2
Compare
cloud-fan
left a comment
There was a problem hiding this comment.
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.
| } | ||
|
|
||
| override def filter(expressions: Array[Expression]): Unit = | ||
| _catalystPredicates ++= expressions |
There was a problem hiding this comment.
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.
| // These filters stay in postScanFilters for correctness (FilterExec above scan), | ||
| // but are also routed into runtimeFilters so BatchScanExec can use them for | ||
| // partition pruning via SupportsRuntimeV2Filtering.filter(). | ||
| // partition pruning via SupportsRuntimeV2Filtering.filter(). The exception is filters |
There was a problem hiding this comment.
The plural subject needs agreement here: The exceptions are filters.
peter-toth
left a comment
There was a problem hiding this comment.
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 bothpushFiltersbranches (SPARK-58112, SPARK-58207). With a fully-pushed attribute the post-scanFilterExecdisappears too, sorand()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
fullyPushedFilterAttributespromises: the attribute-level shape matches v2 file sources, but that works because partition pruning is exact per row. Nothing restrictsfilterAttributesto 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,58matchesSupportsRuntimeV2Filtering, 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:
replanWithRuntimeFiltersenforces it on this path with hardSparkExceptions, but only theSupportsRuntimeV2FilteringJavadoc 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:
resolveRefsthrows (orClassCastExceptions on a nested reference) unless the declared attributes are top-level and present in the scan'sreadSchema. [inline:sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:221]
Minor
- 7. Vacuous assertion:
case _ => Seq.emptylets the twoassertPushedCatalystPredicates(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 inDataSourceV2EnhancedRuntimePartitionFilterSuite, both green on this head for me).
| // so filter them out explicitly. | ||
| val catalystFilters = runtimeFilters | ||
| .flatMap(unwrapRuntimeFilterExpression) | ||
| .filterNot(_ == Literal.TrueLiteral) |
There was a problem hiding this comment.
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:
| .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.
| * and will not be evaluated again after the scan. These attributes must also be returned by | ||
| * [[filterAttributes]]. | ||
| */ | ||
| def fullyPushedFilterAttributes(): Array[NamedReference] = Array.empty |
There was a problem hiding this comment.
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:
-
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
filterAttributesto partition columns:SupportsRuntimeV2Filteringdocuments 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. -
Any shape, not the shapes you recognize. The source can't refuse an individual predicate — by the time
filter()runs,DataSourceV2Strategyhas already removed theFilterExec. On this head, withfully-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.filterhandles only=andIN— drops it on the floor.InMemoryEnhancedRuntimePartitionFilterTablegets it right by delegating toPartitionPredicate.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.
| } else { | ||
| None | ||
| } | ||
| case (resExp, r @ ExtractV2Scan(scan: SupportsRuntimeCatalystFiltering)) => |
There was a problem hiding this comment.
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.
| * | ||
| * Note that Spark will call [[Scan.toBatch]] again after filtering the scan at runtime. | ||
| */ | ||
| def filter(expressions: Array[Expression]): Unit |
There was a problem hiding this comment.
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.
| * It's possible that there are no runtime predicates and [[filter]] is never called; | ||
| * an empty array should be returned for this case. | ||
| */ | ||
| def pushedPredicates(): Array[Expression] = Array.empty |
There was a problem hiding this comment.
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.
| case s: SupportsRuntimeCatalystFiltering => s.fullyPushedFilterAttributes() | ||
| case _ => Array.empty[NamedReference] | ||
| } | ||
| AttributeSet(V2ExpressionUtils.resolveRefs[Attribute]( |
There was a problem hiding this comment.
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.
| collectBatchScan(df).scan match { | ||
| case s: InMemoryCatalystRuntimeFilterTable#InMemoryCatalystRuntimeFilterBatchScan => | ||
| s.pushedPredicates().toSeq | ||
| case _ => Seq.empty |
There was a problem hiding this comment.
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.
| case _ => Seq.empty | |
| case other => fail(s"Expected InMemoryCatalystRuntimeFilterBatchScan, got $other") |
What changes were proposed in this pull request?
This PR adds an internal
SupportsRuntimeCatalystFilteringmix-in for DSv2Scans that receives runtime filters as CatalystExpressions instead of connectorPredicates.SupportsRuntimeCatalystFiltering(inorg.apache.spark.sql.internal.connector) declaresfilterAttributes(),filter(Array[Expression]),pushedPredicates()andfullyPushedFilterAttributes(). It is an alternative toSupportsRuntimeFiltering/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 followsSupportsRuntimeV2Filtering, since in DSv2 theSupportsPushDown*prefix is used byScanBuildermix-ins that push down at query compilation, whereas runtime filtering interfaces areScanmix-ins.PushDownUtils.pushRuntimeFiltersgains a branch for the new interface. All runtime filters are pushed in a singlefiltercall, with no translation to connector predicates, so filters that have no V2 translation still reach the source. DPP filters whose subquery was pruned away degrade toTrueLiteraland are dropped explicitly, since there is no translation step to drop them implicitly. The V2 path is unchanged and is matched first.createRuntimePartitionPredicates(unwrap DPP, literalize scalar subqueries) is extracted intounwrapRuntimeFilterExpressionand shared by both paths.DataSourceV2ScanRelation.runtimeFilterAttrsandPartitionPruning.getFilterableTableScanrecognize the new interface, so DPP and scalar subquery runtime filters are derived for these scans.fullyPushedFilterAttributes(). A runtime filter that only references those attributes is considered fully pushed andDataSourceV2Strategydrops it from the post-scanFilterExec, instead of both pushing it and re-evaluating it. Dynamic pruning filters were already excluded from the post-scan filter list.SupportsRuntimeFiltering/SupportsRuntimeV2Filteringnow notes that only one runtime filtering interface should be implemented, and clarifies thatpushedPredicates()reports predicates that fully or partially help pruning rather than predicates Spark can skip evaluating.Dropping the scalar subquery filter from the post-scan
FilterExecis limited toSupportsRuntimeCatalystFilteringin this PR. Doing the same forSupportsRuntimeFiltering/SupportsRuntimeV2Filteringis deferred to a subsequent PR, because on those interfaces the filter is pushed as a translated connectorPredicateand the source decides what it can fully evaluate from the semantics of thatPredicate, 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-scanFilterExecis 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.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 aspart > (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.
SupportsRuntimeCatalystFilteringlives inorg.apache.spark.sql.internal.connectorand is not public API. The changes to the publicSupportsRuntimeFilteringandSupportsRuntimeV2Filteringinterfaces are documentation only.How was this patch tested?
Added
DataSourceV2CatalystRuntimeFilterSuite, backed by a newInMemoryCatalystRuntimeFilterTableandInMemoryTableCatalystRuntimeFilterCatalog, covering:fullyPushedFilterAttributes;part > (subquery) + 1) pushed instead of dropped, with the scalar subquery literalized and the surrounding expression preserved;InSubqueryExecexpression;filterAttributesnot being pushed;filter()is never called.All 21 tests pass, including the existing
DataSourceV2EnhancedRuntimePartitionFilterSuitefor the unchanged V2 path.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor