Skip to content

[MINOR][SQL] Centralize static and runtime filter eligibility checks #58752

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
szehon-ho wants to merge 2 commits into
base: master
Choose a base branch
from
+14 −24
Open

[MINOR][SQL] Centralize static and runtime filter eligibility checks #58752

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1cea766
[MINOR][SQL] Centralize runtime filter eligibility checks
szehon-ho Sep 11, 2026
830001f
[MINOR][SQL] Centralize static V2 filter eligibility checks
szehon-ho Sep 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 14 additions & 24 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,10 @@ object PushDownUtils extends Logging {
filters: Seq[Expression],
partitionFields: Option[Seq[PartitionPredicateField]])
: (Either[Seq[sources.Filter], Seq[Predicate]], Seq[Expression]) = {
// Both V2 filter APIs keep non-deterministic filters for post-scan evaluation. A source may
// evaluate a pushed predicate a different number of times or at a different point than Spark,
// so a partial push could evaluate a non-deterministic predicate twice with different results.
lazy val (deterministicFilters, nonDeterministicFilters) = filters.partition(_.deterministic)
scanBuilder match {
case r: SupportsPushDownFilters =>
// A map from translated data source leaf node filters to original catalyst filter
Expand Down Expand Up @@ -93,13 +97,6 @@ object PushDownUtils extends Logging {
(postScanFilters ++ untranslatableExprs).toImmutableArraySeq)

case r: SupportsPushDownV2Filters =>
// Non-deterministic filters should not be pushed down: a data source may evaluate a pushed
// predicate a different number of times or at a different point than Spark, and a partial
// push (a predicate used for pruning yet also returned for post-scan re-evaluation, e.g. a
// parquet row group filter) would evaluate a non-deterministic predicate twice with
// different results. Keep them as post-scan filters, matching the
// SupportsPushDownCatalystFilters branch below.
val (deterministicFilters, nonDeterministicFilters) = filters.partition(_.deterministic)
// Divide the filters into those translatable and untranslatable to data source filters.
// For the translated filters, we will try to push them down to the data source,
// and the data source will return the filters that it cannot guarantee to be true
Expand Down Expand Up @@ -141,7 +138,6 @@ object PushDownUtils extends Logging {
ExpressionSet(untranslatableExprs)) ++ nonDeterministicFilters
(Right(r.pushedPredicates.toImmutableArraySeq), orderedPostScanFilters)
case r: SupportsPushDownCatalystFilters =>
val (deterministicFilters, nonDeterministicFilters) = filters.partition(_.deterministic)
val postScanFilters = r.pushFilters(deterministicFilters) ++ nonDeterministicFilters
(Right(r.pushedFilters.toImmutableArraySeq), postScanFilters)
case _ => (Left(Nil), filters)
Expand Down Expand Up @@ -172,10 +168,9 @@ object PushDownUtils extends Logging {
* group-based UPDATE), each pushing its own copy of the runtime filters. Successive calls are
* additive: a scan ANDs the newly pushed predicates with those it already holds.
*
* Note: `runtimeFilters` must not contain non-deterministic filters. A runtime filter is also
* evaluated by the `FilterExec` above the scan, so pushing a non-deterministic one would
* evaluate it twice with different results. `DataSourceV2Strategy` enforces this where
* `runtimeFilters` is built (SPARK-58207).
* Non-deterministic filters are not pushed. A runtime filter is also evaluated by the
* `FilterExec` above the scan, so pushing a non-deterministic one would evaluate it twice with
* different results.
*
* A scan implementing [[SupportsRuntimeCatalystFiltering]] takes a separate path: all
* runtime filters are pushed as Catalyst expressions in a single call, with no translation to
Expand All @@ -188,15 +183,17 @@ object PushDownUtils extends Logging {
runtimeFilters: Seq[Expression],
table: Table,
output: Seq[AttributeReference]): Boolean = {
val pushableFilters = runtimeFilters.filter(
isPushablePartitionFilter(_, includeSubquery = true))
scan match {
case _: SupportsRuntimeV2Filtering with SupportsRuntimeCatalystFiltering =>
throw SparkException.internalError(
"A scan must not implement both SupportsRuntimeV2Filtering and " +
s"SupportsRuntimeCatalystFiltering, but ${scan.getClass.getName} implements both.")

case filterableScan: SupportsRuntimeV2Filtering if runtimeFilters.nonEmpty =>
case filterableScan: SupportsRuntimeV2Filtering if pushableFilters.nonEmpty =>
// Push down translatable runtime filters.
val filtersToTranslated = runtimeFilters.flatMap { f =>
val filtersToTranslated = pushableFilters.flatMap { f =>
(f match {
case DynamicPruningExpression(e) => DataSourceV2Strategy.translateRuntimeFilterV2(e)
case o => DataSourceV2Strategy.translateScalarSubqueryFilterV2(o)
Expand All @@ -218,7 +215,7 @@ object PushDownUtils extends Logging {
"filterAttributes()",
filterableScan.getClass.getName)
val pushed = filterableScan.pushedPredicates().toSet
val candidates = runtimeFilters.filter { f =>
val candidates = pushableFilters.filter { f =>
!filtersToTranslated.get(f).exists(pushed.contains) &&
f.references.subsetOf(filterAttrs)
}
Expand All @@ -233,7 +230,7 @@ object PushDownUtils extends Logging {

translatedFiltersPushed || partPredicatesPushed

case catalystScan: SupportsRuntimeCatalystFiltering if runtimeFilters.nonEmpty =>
case catalystScan: SupportsRuntimeCatalystFiltering if pushableFilters.nonEmpty =>
// A runtime filter is normally evaluated twice: the source prunes with it, and the
// FilterExec above the scan applies it again. The two have to agree, so this screen
// pushes only predicates the source can be trusted to evaluate in Spark's place.
Expand All @@ -247,17 +244,10 @@ object PushDownUtils extends Logging {
// below, and we use the same method (isPushablePartitionFilter) to determine if we
// should push it.
//
// Note: today every filter reaching either site passes this check
// (deleted by DataSourceV2Strategy, and pushed by this method), since runtimeFilters
// holds only deterministic filters (SPARK-58207) and ExtractPythonUDFs has already
// lifted any Python UDF out of the post-scan filters. But sharing the check keeps the two
// decisions consistent if non-deterministic filters reach the site.
//
// A DPP filter degrades to TrueLiteral once its subquery is pruned away, so it matches
// every row. The V2 path above drops these implicitly, since translateRuntimeFilterV2
// returns None; here we push Catalyst expressions directly, so we remove them explicitly.
val catalystFilters = runtimeFilters
.filter(isPushablePartitionFilter(_, includeSubquery = true))
val catalystFilters = pushableFilters
.flatMap(unwrapRuntimeFilterExpression)
.filterNot(_ == Literal.TrueLiteral)
if (catalystFilters.nonEmpty) {
Expand Down