Holden Karau created SPARK-58780:
------------------------------------
Summary: Transpilation with multiple lambdas can hide a cartesian
join
Key: SPARK-58780
URL: https://issues.apache.org/jira/browse/SPARK-58780
Project: Spark
Issue Type: Sub-task
Components: PySpark
Affects Versions: 4.3.0
Reporter: Holden Karau
When a Python UDF is the **sole join condition** and it gets transpiled to
native
Catalyst expressions, the `spark.sql.crossJoin.enabled=false` guard stops
firing --
but the query still executes as an `O(n*m)` `CartesianProduct`. A user who
deliberately disabled cross joins to catch accidental quadratic scans silently
loses that protection the moment they enable transpilation. The failure mode
changes from a loud analysis error to a silent quadratic scan.
## Reproduction
```python
from pyspark.sql import Row, SparkSession
from pyspark.sql.functions import udf
from pyspark.sql.types import BooleanType
spark = (SparkSession.builder.master("local[2]")
.config("spark.sql.ansi.enabled", "true").getOrCreate())
spark.conf.set("spark.sql.crossJoin.enabled", "false")
left = spark.createDataFrame([Row(a=1), Row(a=2)])
right = spark.createDataFrame([Row(b=1), Row(b=2)])
eq = udf(lambda a, b: a == b, BooleanType())
df = left.join(right, eq("a", "b")) # UDF is the ONLY join condition
df.collect()
```
| `transpilePyUDFs` | Result |
|---|---|
| `false` | raises `AnalysisException: Detected implicit cartesian product for
INNER join between logical plans` |
| `true` | **no error**, returns `[Row(a=1, b=1), Row(a=2, b=2)]`,
`executedPlan` contains `CartesianProduct`, no `EvalPython` |
## Why it happens
`CheckCartesianProducts.isCartesianProduct`
(`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:2652`)
asks only whether *some conjunct references both sides*:
```scala
case _ => !conditions.map(_.references).exists(refs =>
refs.exists(join.left.outputSet.contains) &&
refs.exists(join.right.outputSet.contains))
```
* **Interpreted:** the Python UDF cannot be evaluated as a join condition, so
it is
pulled out as `Filter` above a join with no condition. No conjunct references
both
sides, `isCartesianProduct` is true, and the guard raises.
* **Transpiled:** the UDF becomes a native `CASE WHEN (isnull(a) AND isnull(b))
THEN true ...`
that *does* reference both sides, so `isCartesianProduct` returns false and
the guard
is skipped. But no equi-join key is extractable from a `CASE WHEN`, so the
planner
still chooses `CartesianProduct`.
So the predicate the guard uses ("does a conjunct span both sides?") is a proxy
for
"is there a usable join key?", and lowering breaks the correspondence between
them.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
