From 1cea766c787cc08b3de28ff794b5892cbbd126ce Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Fri, 11 Sep 2026 16:17:51 -0700 Subject: [PATCH 1/2] [MINOR][SQL] Centralize runtime filter eligibility checks --- .../datasources/v2/PushDownUtils.scala | 26 +++++++------------ 1 file changed, 10 insertions(+), 16 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala index d2b6c0bfa13c4..857a668174e1a 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala @@ -172,10 +172,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 @@ -188,15 +187,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) @@ -218,7 +219,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) } @@ -233,7 +234,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. @@ -247,17 +248,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) { From 830001fd65e2b2f7b115ca89cd9ac22919ed41b5 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Fri, 11 Sep 2026 16:55:34 -0700 Subject: [PATCH 2/2] [MINOR][SQL] Centralize static V2 filter eligibility checks --- .../sql/execution/datasources/v2/PushDownUtils.scala | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala index 857a668174e1a..f1cf9452f4f21 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala @@ -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 @@ -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 @@ -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)