This is an automated email from the ASF dual-hosted git repository.
sunchao 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 5ef5af334166 [SPARK-58310][SQL] Avoid runtime Bloom-filter subqueries
containing Python UDFs
5ef5af334166 is described below
commit 5ef5af3341665b23570a4653be690f9847cebd30
Author: Chao Sun <[email protected]>
AuthorDate: Fri Jul 24 19:57:57 2026 -0700
[SPARK-58310][SQL] Avoid runtime Bloom-filter subqueries containing Python
UDFs
### Why are the changes needed?
With adaptive execution and runtime Bloom filters enabled, a nested
broadcast join can cause `InjectRuntimeFilter` to copy a creation-side plan
containing a Python UDF into a newly constructed Bloom-filter scalar subquery.
Runtime filters are introduced after subquery optimization, so the copied
Python UDF never passes through Python-UDF extraction. The physical plan
subsequently attempts to generate JVM code for the unevaluable Python
expression while executing `BloomFilterAggregate` [...]
The problem affects ordinary Apache Spark and does not depend on an
external table provider. A self-contained PySpark setup is:
```python
from pyspark.sql.functions import pandas_udf, udf
udf("string")
def normalize_runtime_bloom_url(value):
return value
pandas_udf("string")
def normalize_runtime_bloom_pandas_url(values):
return values
spark.udf.register("normalize_runtime_bloom_url",
normalize_runtime_bloom_url)
spark.udf.register(
"normalize_runtime_bloom_pandas_url", normalize_runtime_bloom_pandas_url
)
spark.range(20000).selectExpr(
"cast(id as int) as id",
"concat('url-', cast(id % 20 as string)) as url",
).write.mode("overwrite").parquet("/tmp/spark-58310-fact")
spark.range(20).selectExpr(
"concat('url-', cast(id as string)) as url",
"case when id < 10 then 'NL' else 'US' end as country",
).write.mode("overwrite").parquet("/tmp/spark-58310-dimension")
spark.read.parquet("/tmp/spark-58310-fact").createOrReplaceTempView(
"runtime_bloom_fact"
)
spark.read.parquet("/tmp/spark-58310-dimension").createOrReplaceTempView(
"runtime_bloom_dimension"
)
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.optimizer.dynamicPartitionPruning.enabled",
"false")
spark.conf.set("spark.sql.optimizer.runtime.bloomFilter.enabled", "true")
spark.conf.set(
"spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold", "0"
)
spark.conf.set("spark.sql.optimizer.runtime.bloomFilter.creationSideThreshold",
"100MB")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")
```
The following query reproduces the failure; substituting
`normalize_runtime_bloom_pandas_url` reproduces the Arrow/Pandas variant:
```sql
WITH coverage AS (
SELECT url, normalize_runtime_bloom_url(url) AS normalized_url
FROM runtime_bloom_dimension
WHERE country = 'NL'
),
normalized_keys AS (
SELECT DISTINCT normalized_url
FROM coverage
WHERE normalized_url IS NOT NULL
),
matched_urls AS (
SELECT /*+ BROADCAST(k) */ fact.url
FROM runtime_bloom_fact fact
JOIN normalized_keys k ON fact.url = k.normalized_url
),
snapshot AS (
SELECT max(url) AS snapshot_url
FROM runtime_bloom_fact
)
SELECT /*+ BROADCAST(m), BROADCAST(snapshot) */
coverage.url,
coverage.normalized_url,
matched_urls.url AS matched_url,
snapshot.snapshot_url
FROM coverage
LEFT JOIN matched_urls ON coverage.normalized_url = matched_urls.url
CROSS JOIN snapshot
```
### What changes were proposed in this PR?
After constructing and column-pruning the runtime Bloom-filter aggregate,
return the unchanged application-side plan when the pruned aggregate still
contains the `PYTHON_UDF` tree pattern. Checking after column pruning preserves
safe Bloom filters when a Python UDF is needed only by the outer query output
and is absent from the actual Bloom-filter scalar subquery. This guard is
deliberately limited to unsafe Bloom-filter creation. It does not disable
runtime filtering, does not reject [...]
Add `SPARK-58310` coverage to `InjectRuntimeFilterSuite` that:
- Runs both standard scalar Python and Arrow/Pandas scalar UDFs when
available.
- Reproduces nested CTEs, explicit broadcast joins, scalar aggregates, and
adaptive execution with dynamic partition pruning disabled.
- Compares the complete result with the Bloom-disabled baseline and
verifies a real Parquet write/read round trip.
- Proves that safe application-side and pruned creation-side runtime Bloom
filters are still injected and produce the same results.
### Does this PR introduce _any_ user-facing change?
Yes. Queries that previously failed because a runtime Bloom-filter subquery
attempted to code-generate a Python UDF now complete successfully. Safe runtime
Bloom-filter optimizations remain enabled.
### How was this PR tested?
First, the new `SPARK-58310` regression was run against unmodified Apache
Spark `master` and failed with the expected Python-UDF
scalar-subquery/`ObjectHashAggregateExec` error. The same regression then
passed after the creation-side guard was applied, with both `pandas` and
`pyarrow` installed.
The following focused upstream regression suites and Scala-style checks
were also run on the public Apache Spark branch:
```bash
build/sbt \
'sql/testOnly org.apache.spark.sql.InjectRuntimeFilterSuite
org.apache.spark.sql.DynamicPartitionPruningV1SuiteAEOn
org.apache.spark.sql.SubquerySuite
org.apache.spark.sql.execution.adaptive.AdaptiveQueryExecSuite
org.apache.spark.sql.execution.python.ExtractPythonUDFsSuite' \
'catalyst/scalastyle' \
'sql/scalastyle' \
'sql/Test/scalastyle'
```
### How was this patch tested?
The upstream before/after reproduction, Python and Arrow regression,
adjacent SQL suites, and Scala-style checks are described above.
### Was this patch authored or co-authored using generative AI tooling?
Yes. Generated-by: OpenAI Codex (GPT-5).
Closes #57485 from sunchao/dev/chao/codex/python-udf-runtime-bloom-filter.
Authored-by: Chao Sun <[email protected]>
Signed-off-by: Chao Sun <[email protected]>
(cherry picked from commit 6f1c904686cb54e52c2282882fab898a125a7eac)
Signed-off-by: Chao Sun <[email protected]>
---
.../catalyst/optimizer/InjectRuntimeFilter.scala | 5 +
.../spark/sql/InjectRuntimeFilterSuite.scala | 125 +++++++++++++++++++++
2 files changed, 130 insertions(+)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala
index 176e927b2d21..a8f5b3ca67ed 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala
@@ -69,6 +69,11 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
val alias = Alias(bloomFilterAgg.toAggregateExpression(), "bloomFilter")()
val aggregate =
ConstantFolding(ColumnPruning(Aggregate(Nil, Seq(alias),
filterCreationSidePlan)))
+ // Runtime filters are introduced after subquery optimization, so Python
UDFs in a new
+ // creation-side subquery cannot be extracted into a Python evaluation
operator.
+ if (aggregate.containsPattern(PYTHON_UDF)) {
+ return filterApplicationSidePlan
+ }
val bloomFilterSubquery = ScalarSubquery(aggregate, Nil)
val filter = BloomFilterMightContain(bloomFilterSubquery,
new XxHash64(Seq(filterApplicationSideKey)))
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
index 53b0132b29ce..f045a92b6dc9 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
@@ -17,6 +17,8 @@
package org.apache.spark.sql
+import java.io.File
+
import org.apache.spark.sql.catalyst.expressions.{Alias,
BloomFilterMightContain, Literal}
import
org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression,
BloomFilterAggregate}
import org.apache.spark.sql.catalyst.optimizer.MergeSubplans
@@ -517,6 +519,129 @@ class InjectRuntimeFilterSuite extends SharedSparkSession
}
}
+ test("SPARK-58310: skip runtime bloom filter when pruned creation side
contains Python UDF") {
+ import IntegratedUDFTestUtils._
+
+ assume(shouldTestPythonUDFs)
+ registerTestUDF(TestPythonUDF("normalize_runtime_bloom_url"), spark)
+ val pythonUdfNames = if (shouldTestPandasUDFs) {
+
registerTestUDF(TestScalarPandasUDF("normalize_runtime_bloom_pandas_url"),
spark)
+ Seq("normalize_runtime_bloom_url", "normalize_runtime_bloom_pandas_url")
+ } else {
+ Seq("normalize_runtime_bloom_url")
+ }
+
+ withTempPath { dir =>
+ val factPath = new File(dir, "fact").getCanonicalPath
+ val dimensionPath = new File(dir, "dimension").getCanonicalPath
+
+ spark.range(20000)
+ .selectExpr("cast(id as int) as id", "concat('url-', cast(id % 20 as
string)) as url")
+ .write.parquet(factPath)
+ spark.range(20)
+ .selectExpr(
+ "concat('url-', cast(id as string)) as url",
+ "case when id < 10 then 'NL' else 'US' end as country")
+ .write.parquet(dimensionPath)
+
+ withTempView("runtime_bloom_fact", "runtime_bloom_dimension") {
+
spark.read.parquet(factPath).createOrReplaceTempView("runtime_bloom_fact")
+
spark.read.parquet(dimensionPath).createOrReplaceTempView("runtime_bloom_dimension")
+
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false",
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "0",
+ SQLConf.RUNTIME_BLOOM_FILTER_CREATION_SIDE_THRESHOLD.key -> "100MB",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ pythonUdfNames.foreach { pythonUdfName =>
+ withClue(s"$pythonUdfName: ") {
+ val query =
+ s"""
+ |WITH coverage AS (
+ | SELECT url, $pythonUdfName(url) AS normalized_url
+ | FROM runtime_bloom_dimension
+ | WHERE country = 'NL'
+ |),
+ |grader_keys AS (
+ | SELECT DISTINCT normalized_url
+ | FROM coverage
+ | WHERE normalized_url IS NOT NULL
+ |),
+ |matched_url_status AS (
+ | SELECT /*+ BROADCAST(k) */ s.url
+ | FROM runtime_bloom_fact s
+ | JOIN grader_keys k ON s.url = k.normalized_url
+ |),
+ |url_status_snapshot AS (
+ | SELECT max(url) AS snapshot_url FROM runtime_bloom_fact
+ |)
+ |SELECT /*+ BROADCAST(m), BROADCAST(snapshot) */
+ | c.url, c.normalized_url, m.url AS matched_url,
snapshot.snapshot_url
+ |FROM coverage c
+ |LEFT JOIN matched_url_status m ON c.normalized_url = m.url
+ |CROSS JOIN url_status_snapshot snapshot
+ """.stripMargin
+
+ val applicationSideQuery =
+ s"""
+ |SELECT /*+ MERGE(f, s) */ f.id, f.url
+ |FROM (
+ | SELECT id, url, $pythonUdfName(url) AS normalized_url
+ | FROM runtime_bloom_fact
+ |) f
+ |JOIN (
+ | SELECT url FROM runtime_bloom_dimension WHERE country =
'NL'
+ |) s ON f.normalized_url = s.url
+ """.stripMargin
+
+ val creationSideQuery =
+ s"""
+ |SELECT /*+ MERGE(f, s) */ f.id, s.url, s.normalized_url
+ |FROM runtime_bloom_fact f
+ |JOIN (
+ | SELECT url, $pythonUdfName(url) AS normalized_url
+ | FROM runtime_bloom_dimension
+ | WHERE country = 'NL'
+ |) s ON f.url = s.url
+ """.stripMargin
+
+ var expected: Array[Row] = null
+ var applicationSideExpected: Array[Row] = null
+ var creationSideExpected: Array[Row] = null
+ withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_ENABLED.key -> "false")
{
+ expected = sql(query).collect()
+ applicationSideExpected = sql(applicationSideQuery).collect()
+ creationSideExpected = sql(creationSideQuery).collect()
+ }
+
+ assert(expected.length == 10000)
+ assert(applicationSideExpected.length == 10000)
+ assert(creationSideExpected.length == 10000)
+
+ withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_ENABLED.key -> "true") {
+ val result = sql(query)
+ checkAnswer(result, expected.toSeq)
+
+ val outputPath = new java.io.File(dir,
pythonUdfName).getCanonicalPath
+ result.write.parquet(outputPath)
+ checkAnswer(spark.read.parquet(outputPath), expected.toSeq)
+
+ val applicationSideResult = sql(applicationSideQuery)
+
assert(getNumBloomFilters(applicationSideResult.queryExecution.optimizedPlan) >
0)
+ checkAnswer(applicationSideResult,
applicationSideExpected.toSeq)
+
+ val creationSideResult = sql(creationSideQuery)
+
assert(getNumBloomFilters(creationSideResult.queryExecution.optimizedPlan) > 0)
+ checkAnswer(creationSideResult, creationSideExpected.toSeq)
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
test("Support Left Semi join in row level runtime filters") {
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "32") {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
