[
https://issues.apache.org/jira/browse/SPARK-58780?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-58780:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Priority: Major
> Labels: pull-request-available
>
> 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
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
