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 8879d98836d0 [SPARK-60486][SQL] Preserve Python UDF partition filters
after V2 pushdown
8879d98836d0 is described below
commit 8879d98836d01f721c37ae60503d5ac9ee3e5684
Author: 小哥不太逍遥 <[email protected]>
AuthorDate: Wed Oct 28 12:05:19 2026 -0700
[SPARK-60486][SQL] Preserve Python UDF partition filters after V2 pushdown
### What changes were proposed in this pull request?
Preserve partition predicates rejected by `FileScanBuilder` as post-scan
filters. Python UDFs remain excluded from partition pruning, while their
predicates remain available for subsequent Python UDF extraction and evaluation.
Extend the existing shared PySpark DSv2 UDF regression with a partitioned
input and exact-row checks for a partition UDF, `AND`, and `OR`. Add a SQL
migration note for the corrected results.
### Why are the changes needed?
[SPARK-60486](https://issues.apache.org/jira/browse/SPARK-60486) reports
incorrect results when filtering partitioned files through Data Source V2 with
Python UDFs. The pruning guard added by
[SPARK-48666](https://github.com/apache/spark/pull/47313) removes such
predicates from the pushed partition filters, but the returned post-scan
filters currently omit them as well. The scan can therefore return rows that do
not satisfy the original filter.
For example, after writing six rows with `i = 0..5` and partition column `p
= i % 3`, filtering a V2 Parquet read with `udf(lambda p: p == 1,
"boolean")("p")` returns all six rows on the unmodified master. It should
return only `(i=1, p=1)` and `(i=4, p=1)`.
### Does this PR introduce _any_ user-facing change?
Yes. Affected Data Source V2 file queries now apply the Python UDF filter
and exclude rows that were incorrectly retained. This was independently
reproduced on released Spark 4.2.0 and current master; this PR targets master.
Built-in Parquet/ORC file reads use the V1 path by default through
`spark.sql.sources.useV1SourceList`; the affected queries are those routed to
Data Source V2. The V1 path already evaluates these predicates after the scan.
### How was this patch tested?
- Built the unmodified master with Hive support, then rebuilt the candidate
SQL reactor with Maven. `PruneFileSourcePartitionsSuite`: **9/9 passed**.
- The two expanded regressions: **2 failures before the production change,
2/2 passed afterward**. An additional **20/20** existing filter, join,
subquery, and nondeterministic UDF tests passed across pickle and Arrow UDF
classes.
- An independent **44-case** matrix covering Parquet/ORC, V1/V2,
pickle/Arrow, NULL partitions, compound predicates, and controls: **12
incorrect results before the fix, 44/44 correct afterward**, with identical
input file bytes and expected results. The 32 control cases remained correct.
- Ruff check/format, Scala formatter validation, and `git diff --check`
passed. The only difference between the locally built Scala source and the
final source is two whitespace-only scalafmt changes (indentation and argument
wrapping).
- [Fork CI](https://github.com/xgbtxy/spark/actions/runs/39878240762)
completed successfully for implementation commit `f886f78b`: 30 checks passed
and 2 integration checks were skipped. The subsequent documentation-only
revision moves the migration note to the Spark 4.4 section per review; its CI
is tracked separately.
Local baseline: master `7e9a966c`, Java 17, Python 3.12.
Local PySpark tests use Python 3.12 on Windows through `unittest`. A
process-local fixture adapter closes only `ReusedSQLTestCase.setUpClass`'s
temporary file before its existing unlink call, because Windows rejects
unlinking the open handle. Product code, test assertions, and SQL/Arrow
settings are unchanged by that adapter. The stock `python/run-tests` driver
requires POSIX `pty` support.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex (GPT-6)
This contribution is original work submitted under the project's Apache
License 2.0.
Closes #59732 from xgbtxy/fix/spark-60486-python-udf-partition.
Authored-by: 小哥不太逍遥 <[email protected]>
Signed-off-by: Szehon Ho <[email protected]>
(cherry picked from commit 925d91b19cb5c01dbb57be85915efd60cf06c71a)
Signed-off-by: Szehon Ho <[email protected]>
---
docs/sql-migration-guide.md | 2 ++
python/pyspark/sql/tests/test_udf.py | 21 +++++++++++++++++++--
.../execution/datasources/v2/FileScanBuilder.scala | 10 ++++++----
3 files changed, 27 insertions(+), 6 deletions(-)
diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index ee590cfb4f10..2b0c576517e3 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -24,7 +24,9 @@ license: |
## Upgrading from Spark SQL 4.3 to 4.4
+- Since Spark 4.4, Data Source V2 file scans correctly evaluate deterministic
Python UDF filter predicates that reference only partition columns. Previously,
these predicates could be dropped, returning rows that did not satisfy the
filter (SPARK-60486). Data derived from affected query results may need to be
recomputed.
- Since Spark 4.4, quoted Variant path keys decode `\\`, `\/`, `\b`, `\f`,
`\n`, `\r`, `\t`, `\uXXXX` (including valid surrogate pairs), and escapes of
the enclosing quote (`\'` or `\"`). Previously, backslashes were literal, so
existing paths can match different keys or become invalid. For example, use
`$['a\\']` instead of `$['a\']`. Unknown escapes and unpaired surrogate escapes
stay literal. Dot-notation keys are unchanged. To keep the old behavior, set
`spark.sql.legacy.variantPathB [...]
+
- Since Spark 4.4, `spark.sql.execution.arrow.pythonUDF.columnarInput.enabled`
is `false` by default: Arrow-based Python UDFs read rows from an Arrow-backed
upstream operator, such as a Data Source V2 scan or the Arrow cache, as before
Spark 4.2. With columnar input, they kept the columns that they pass through
until the results of their batch came back from Python, and failed or returned
wrong results when the operator reused or released the vectors of the batch in
the meantime. To enab [...]
- Since Spark 4.4, the conversion of columnar batches to rows releases each
batch with `ColumnVector.closeIfFreeable` once its rows are consumed also when
it is not whole-stage code-generated, e.g. when the query has more than
`spark.sql.codegen.maxFields` fields or `spark.sql.codegen.wholeStage` is
`false`, as the generated code has done since Spark 3.5.4 and 4.0. A
`ColumnVector` that a Data Source V2 source reuses across batches, such as an
`ArrowColumnVector` over a `VectorSchemaRoot [...]
- Since Spark 4.4, `spark.sql.charVarchar.standardSemantics.enabled` (default
`false`) is a preview feature that may change in future releases. When true, it
enables first-class `CHAR(n)` and `VARCHAR(n)` semantics throughout schemas and
query plans: `CHAR` -> `VARCHAR` -> `STRING` form a type-precedence family,
explicit casts can return constrained types, least-common-type resolution for
`COALESCE`/`CASE`/`UNION` can return `CHAR` or `VARCHAR`, and string
expressions that transform thei [...]
diff --git a/python/pyspark/sql/tests/test_udf.py
b/python/pyspark/sql/tests/test_udf.py
index 9fd365f131e1..6477dd6e2fdc 100644
--- a/python/pyspark/sql/tests/test_udf.py
+++ b/python/pyspark/sql/tests/test_udf.py
@@ -898,12 +898,29 @@ class BaseUDFTestsMixin:
try:
with self.sql_conf({"spark.sql.sources.useV1SourceList": ""}):
-
self.spark.range(1).write.mode("overwrite").format("parquet").save(path)
- df = self.spark.read.parquet(path).toDF("i")
+ (
+ self.spark.range(6)
+ .selectExpr("id AS i", "id % 3 AS p")
+ .write.mode("overwrite")
+ .partitionBy("p")
+ .parquet(path)
+ )
+ df = self.spark.read.parquet(path)
f = udf(lambda x: False, "boolean")(lit(1))
result = df.filter(f)
self.assertEqual(0, result.count())
+ # SPARK-60486: partition predicates that cannot be pushed must
still be evaluated.
+ matches_partition = udf(lambda x: x == 1, "boolean")(col("p"))
+ expected = [Row(i=1, p=1), Row(i=4, p=1)]
+ assertDataFrameEqual(df.filter(col("p") == 1), expected)
+ assertDataFrameEqual(df.filter(matches_partition), expected)
+ assertDataFrameEqual(df.filter(matches_partition & (col("p") >
0)), expected)
+ assertDataFrameEqual(
+ df.filter(matches_partition | (col("p") == 0)),
+ [Row(i=0, p=0), Row(i=1, p=1), Row(i=3, p=0), Row(i=4,
p=1)],
+ )
+
finally:
shutil.rmtree(path)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScanBuilder.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScanBuilder.scala
index 46f6702003ff..1e0b5ae05c90 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScanBuilder.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScanBuilder.scala
@@ -31,7 +31,7 @@ abstract class FileScanBuilder(
sparkSession: SparkSession,
fileIndex: PartitioningAwareFileIndex,
dataSchema: StructType)
- extends ScanBuilder
+ extends ScanBuilder
with SupportsPushDownRequiredColumns
with SupportsPushDownCatalystFilters {
private val partitionSchema = fileIndex.partitionSchema
@@ -73,17 +73,19 @@ abstract class FileScanBuilder(
val (deterministicFilters, nonDeterminsticFilters) =
filters.partition(_.deterministic)
val (partitionFilters, dataFilters) =
DataSourceUtils.getPartitionFiltersAndDataFilters(partitionSchema,
deterministicFilters)
- this.partitionFilters = partitionFilters.filter { f =>
+ val (partitionFiltersToPush, postScanPartitionFilters) =
partitionFilters.partition { f =>
// Python UDFs might exist because this rule is applied before
``ExtractPythonUDFs``.
!SubqueryExpression.hasSubquery(f) &&
!f.exists(_.isInstanceOf[PythonUDF])
}
+ this.partitionFilters = partitionFiltersToPush
this.dataFilters = dataFilters
// Expand struct equality predicates into field-level predicates for
pushdown only.
// The original struct predicates stay in `dataFilters` (returned below as
post-scan
// filters), so the expanded field predicates only need to be a sound
over-approximation.
val expandedDataFilters =
DataSourceStrategy.expandStructPredicatesForPushdown(
- dataFilters, sparkSession.sessionState.conf)
+ dataFilters,
+ sparkSession.sessionState.conf)
val translatedFilters = mutable.ArrayBuffer.empty[sources.Filter]
for (filterExpr <- expandedDataFilters) {
val translated = DataSourceStrategy.translateFilter(filterExpr, true)
@@ -92,7 +94,7 @@ abstract class FileScanBuilder(
}
}
pushedDataFilters = pushDataFilters(translatedFilters.toArray)
- dataFilters ++ nonDeterminsticFilters
+ dataFilters ++ postScanPartitionFilters ++ nonDeterminsticFilters
}
override def pushedFilters: Array[Predicate] = pushedDataFilters.map(_.toV2)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
