This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 450575d41f7f [SPARK-58112][SQL] Skip pushdown for nondeterministic
Catalyst filters
450575d41f7f is described below
commit 450575d41f7fcae1dc1b4fa3731bc3224a151ae0
Author: Szehon Ho <[email protected]>
AuthorDate: Tue Jul 14 14:01:08 2026 -0700
[SPARK-58112][SQL] Skip pushdown for nondeterministic Catalyst filters
### What changes were proposed in this pull request?
This PR updates Catalyst filter pushdown for
`SupportsPushDownCatalystFilters` so Spark only passes deterministic filters to
the connector pushdown API. Nondeterministic filters are kept as post-scan
filters and evaluated by Spark after the scan.
This also adds a regression test with a test data source implementing
`SupportsPushDownCatalystFilters`; the source fails if a nondeterministic
predicate reaches `pushFilters`.
### Why are the changes needed?
Pushing nondeterministic filters into a data source can change query
semantics because the filter may be evaluated at a different point or a
different number of times than Spark expects. The existing V2 file scan path
already avoids pushing nondeterministic filters; this makes the Catalyst filter
pushdown utility apply the same guard before invoking
`SupportsPushDownCatalystFilters`.
### Does this PR introduce _any_ user-facing change?
Yes. Data sources using `SupportsPushDownCatalystFilters` will no longer
receive nondeterministic filters for pushdown. Those filters remain in Spark as
post-scan filters.
### How was this patch tested?
Added a regression test in `DataSourceV2Suite`: `catalyst filter pushdown
skips non-deterministic filters`.
Local checks run:
```bash
git diff --check
grep -rn -P "[^\x00-\x7F]"
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64
MAVEN_MIRROR_URL=https://maven-proxy.cloud.databricks.com build/sbt
'sql/testOnly *DataSourceV2Suite -- -z "catalyst filter pushdown skips
non-deterministic filters"'\n```\n\nThe focused SBT test passed locally: 1 test
run, 1 succeeded, 0 failed.\n\n### Was this patch authored or co-authored using
generative AI tooling?\n\nGenerated-by: Codex (GPT-5)\n
Closes #57235 from szehon-ho/skip-nondeterministic-catalyst-filter-pushdown.
Authored-by: Szehon Ho <[email protected]>
Signed-off-by: Szehon Ho <[email protected]>
(cherry picked from commit 233b1588b475e9c64dc10d8088baf19c1027d21c)
Signed-off-by: Szehon Ho <[email protected]>
---
.../execution/datasources/v2/PushDownUtils.scala | 3 +-
.../spark/sql/connector/DataSourceV2Suite.scala | 56 +++++++++++++++++++++-
2 files changed, 56 insertions(+), 3 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 7df6ade32c48..6685b28c7dc8 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
@@ -134,7 +134,8 @@ object PushDownUtils extends Logging {
ExpressionSet(untranslatableExprs))
(Right(r.pushedPredicates.toImmutableArraySeq), orderedPostScanFilters)
case r: SupportsPushDownCatalystFilters =>
- val postScanFilters = r.pushFilters(filters)
+ val (deterministicFilters, nonDeterministicFilters) =
filters.partition(_.deterministic)
+ val postScanFilters = r.pushFilters(deterministicFilters) ++
nonDeterministicFilters
(Right(r.pushedFilters.toImmutableArraySeq), postScanFilters)
case _ => (Left(Nil), filters)
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
index 5ae23bc3338c..c6f3c1c3886c 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
@@ -28,8 +28,10 @@ import test.org.apache.spark.sql.connector._
import org.apache.spark.SparkUnsupportedOperationException
import org.apache.spark.sql.{AnalysisException, DataFrame, Row}
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
GreaterThan => CatalystGreaterThan, Literal => CatalystLiteral, ScalarSubquery}
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Project}
+import org.apache.spark.sql.catalyst.expressions.{
+ AttributeReference, Expression => CatalystExpression, GreaterThan =>
CatalystGreaterThan,
+ Literal => CatalystLiteral, ScalarSubquery}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter =>
LogicalFilter, Project}
import org.apache.spark.sql.connector.catalog.{PartitionInternalRow,
SupportsRead, Table, TableCapability, TableProvider}
import org.apache.spark.sql.connector.catalog.TableCapability._
import org.apache.spark.sql.connector.expressions.{Expression, FieldReference,
Literal, NamedReference, NullOrdering, SortDirection, SortOrder, Transform}
@@ -46,6 +48,7 @@ import
org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.connector.SupportsPushDownCatalystFilters
import org.apache.spark.sql.sources.{Filter, GreaterThan}
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
@@ -1351,6 +1354,28 @@ class DataSourceV2Suite extends SharedSparkSession with
AdaptiveSparkPlanHelper
"pushedFilters should contain the pushed filter on column i")
}
+ test("catalyst filter pushdown skips non-deterministic filters") {
+ val df =
spark.read.format(classOf[CatalystFilterDataSourceV2].getName).load()
+ val q = df.filter($"i" > 3 && rand() > 0.5)
+
+ val scanRelation = getScanRelation(q)
+ assert(scanRelation.pushedFilters.nonEmpty,
+ "pushedFilters should contain the deterministic pushed filter")
+ assert(scanRelation.pushedFilters.forall(_.deterministic),
+ "pushedFilters should not contain non-deterministic filters")
+ val referencedCols =
scanRelation.pushedFilters.flatMap(_.references.map(_.name)).toSet
+ assert(referencedCols.contains("i"),
+ "pushedFilters should contain the pushed filter on column i")
+
+ // The non-deterministic filter is not pushed, so it must be retained as a
post-scan
+ // Filter above the scan to still be evaluated.
+ val postScanConditions = q.queryExecution.optimizedPlan.collect {
+ case f: LogicalFilter => f.condition
+ }
+ assert(postScanConditions.exists(cond => cond.exists(!_.deterministic)),
+ "non-deterministic filter should be retained as a post-scan Filter")
+ }
+
test("pushedFilters drops filters referencing pruned columns") {
// Disable constraint propagation so IsNotNull(i) is not added (it would
keep
// column i in the scan output). This simulates a connector that pushes
IsNotNull.
@@ -1503,6 +1528,33 @@ class AdvancedDataSourceV2 extends TestingV2Source {
}
}
+class CatalystFilterDataSourceV2 extends TestingV2Source {
+
+ override def getTable(options: CaseInsensitiveStringMap): Table = new
SimpleBatchTable {
+ override def newScanBuilder(options: CaseInsensitiveStringMap):
ScanBuilder = {
+ new CatalystFilterScanBuilder()
+ }
+ }
+}
+
+class CatalystFilterScanBuilder extends SimpleScanBuilder
+ with SupportsPushDownCatalystFilters {
+
+ override def pushFilters(filters: Seq[CatalystExpression]):
Seq[CatalystExpression] = {
+ if (filters.exists(!_.deterministic)) {
+ throw new IllegalArgumentException(
+ s"Non-deterministic filters should not be pushed:
${filters.mkString(", ")}")
+ }
+ Nil
+ }
+
+ override def pushedFilters: Array[Predicate] = Array.empty
+
+ override def planInputPartitions(): Array[InputPartition] = {
+ throw new IllegalArgumentException("planInputPartitions must not be
called")
+ }
+}
+
class AdvancedScanBuilder extends ScanBuilder
with Scan with SupportsPushDownFilters with SupportsPushDownRequiredColumns {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
