szehon-ho opened a new pull request, #57727: URL: https://github.com/apache/spark/pull/57727
### What changes were proposed in this pull request? This PR adds an internal `SupportsPushDownCatalystRuntimeFiltering` mix-in for DSv2 `Scan`s that receives runtime filters as Catalyst `Expression`s instead of connector `Predicate`s. - `SupportsPushDownCatalystRuntimeFiltering` (in `org.apache.spark.sql.internal.connector`) declares `filterAttributes`, `filter(Array[Expression])`, `pushedPredicates()` 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. - `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. DPP filters whose subquery was pruned away degrade to `TrueLiteral` and are dropped explicitly, since there is no translation step to drop them implicitly. 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` and `PartitionPruning.getFilterableTableScan` recognize the new interface, so DPP and scalar subquery runtime filters are derived for these scans. - 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. - Javadoc on `SupportsRuntimeFiltering` / `SupportsRuntimeV2Filtering` now notes that only one runtime filtering interface should be implemented, and clarifies that `pushedPredicates()` reports predicates that fully or partially help pruning rather than predicates Spark can skip evaluating. ### 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. `SupportsPushDownCatalystRuntimeFiltering` 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 a new `InMemoryCatalystRuntimeFilterTable` and `InMemoryTableCatalystRuntimeFilterCatalog`, covering: - a scalar subquery on a partition column pushed as a Catalyst expression; - 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; - a filter on `fullyPushedFilterAttributes` not being evaluated after the scan, and the same filter still being evaluated after the scan when the attribute is not declared fully pushed. ``` build/sbt 'sql/testOnly *DataSourceV2CatalystRuntimeFilterSuite *DataSourceV2EnhancedRuntimePartitionFilterSuite' ``` All 21 tests pass, including the existing `DataSourceV2EnhancedRuntimePartitionFilterSuite` for the unchanged V2 path. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Cursor -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
