You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2019/02/13 13:14:43 UTC
[spark] branch master updated: [SPARK-26864][SQL] Query may return
incorrect result when python udf is used as a join condition and the udf
uses attributes from both legs of left semi join.
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 7f44c9a [SPARK-26864][SQL] Query may return incorrect result when python udf is used as a join condition and the udf uses attributes from both legs of left semi join.
7f44c9a is described below
commit 7f44c9a2522f17e72733f6bf8c3f9614d3001250
Author: Dilip Biswal <db...@us.ibm.com>
AuthorDate: Wed Feb 13 21:14:19 2019 +0800
[SPARK-26864][SQL] Query may return incorrect result when python udf is used as a join condition and the udf uses attributes from both legs of left semi join.
## What changes were proposed in this pull request?
In SPARK-25314, we supported the scenario of having a python UDF that refers to attributes from both legs of a join condition by rewriting the plan to convert an inner join or left semi join to a filter over a cross join. In case of left semi join, this transformation may cause incorrect results when the right leg of join condition produces duplicate rows based on the join condition. This fix disallows the rewrite for left semi join and raises an error in the case like we do for other [...]
**Python**
```SQL
>>> from pyspark import SparkContext
>>> from pyspark.sql import SparkSession, Column, Row
>>> from pyspark.sql.functions import UserDefinedFunction, udf
>>> from pyspark.sql.types import *
>>> from pyspark.sql.utils import AnalysisException
>>>
>>> spark.conf.set("spark.sql.crossJoin.enabled", "True")
>>> left = spark.createDataFrame([Row(lc1=1, lc2=1), Row(lc1=2, lc2=2)])
>>> right = spark.createDataFrame([Row(rc1=1, rc2=1), Row(rc1=1, rc2=1)])
>>> func = udf(lambda a, b: a == b, BooleanType())
>>> df = left.join(right, func("lc1", "rc1"), "leftsemi").show()
19/02/12 16:07:10 WARN PullOutPythonUDFInJoinCondition: The join condition:<lambda>(lc1#0L, rc1#4L) of the join plan contains PythonUDF only, it will be moved out and the join plan will be turned to cross join.
+---+---+
|lc1|lc2|
+---+---+
| 1| 1|
| 1| 1|
+---+---+
```
**Scala**
```SQL
scala> val left = Seq((1, 1), (2, 2)).toDF("lc1", "lc2")
left: org.apache.spark.sql.DataFrame = [lc1: int, lc2: int]
scala> val right = Seq((1, 1), (1, 1)).toDF("rc1", "rc2")
right: org.apache.spark.sql.DataFrame = [rc1: int, rc2: int]
scala> val equal = udf((p1: Integer, p2: Integer) => {
| p1 == p2
| })
equal: org.apache.spark.sql.expressions.UserDefinedFunction = SparkUserDefinedFunction($Lambda$2141/11016292394666f1b5,BooleanType,List(Some(Schema(IntegerType,true)), Some(Schema(IntegerType,true))),None,false,true)
scala> val df = left.join(right, equal(col("lc1"), col("rc1")), "leftsemi")
df: org.apache.spark.sql.DataFrame = [lc1: int, lc2: int]
scala> df.show()
+---+---+
|lc1|lc2|
+---+---+
| 1| 1|
+---+---+
```
## How was this patch tested?
Modified existing tests.
Closes #23769 from dilipbiswal/dkb_python_udf_in_join.
Authored-by: Dilip Biswal <db...@us.ibm.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
python/pyspark/sql/tests/test_udf.py | 24 ++--------------------
.../spark/sql/catalyst/optimizer/joins.scala | 8 ++------
.../PullOutPythonUDFInJoinConditionSuite.scala | 23 ++-------------------
3 files changed, 6 insertions(+), 49 deletions(-)
diff --git a/python/pyspark/sql/tests/test_udf.py b/python/pyspark/sql/tests/test_udf.py
index 12cf8c7d..0a56ba8 100644
--- a/python/pyspark/sql/tests/test_udf.py
+++ b/python/pyspark/sql/tests/test_udf.py
@@ -215,17 +215,6 @@ class UDFTests(ReusedSQLTestCase):
with self.sql_conf({"spark.sql.crossJoin.enabled": True}):
self.assertEqual(df.collect(), [Row(a=1, b=1)])
- def test_udf_in_left_semi_join_condition(self):
- # regression test for SPARK-25314
- left = self.spark.createDataFrame([Row(a=1, a1=1, a2=1), Row(a=2, a1=2, a2=2)])
- right = self.spark.createDataFrame([Row(b=1, b1=1, b2=1)])
- f = udf(lambda a, b: a == b, BooleanType())
- df = left.join(right, f("a", "b"), "leftsemi")
- with self.assertRaisesRegexp(AnalysisException, 'Detected implicit cartesian product'):
- df.collect()
- with self.sql_conf({"spark.sql.crossJoin.enabled": True}):
- self.assertEqual(df.collect(), [Row(a=1, a1=1, a2=1)])
-
def test_udf_and_common_filter_in_join_condition(self):
# regression test for SPARK-25314
# test the complex scenario with both udf and common filter
@@ -236,19 +225,9 @@ class UDFTests(ReusedSQLTestCase):
# do not need spark.sql.crossJoin.enabled=true for udf is not the only join condition.
self.assertEqual(df.collect(), [Row(a=1, a1=1, a2=1, b=1, b1=1, b2=1)])
- def test_udf_and_common_filter_in_left_semi_join_condition(self):
- # regression test for SPARK-25314
- # test the complex scenario with both udf and common filter
- left = self.spark.createDataFrame([Row(a=1, a1=1, a2=1), Row(a=2, a1=2, a2=2)])
- right = self.spark.createDataFrame([Row(b=1, b1=1, b2=1), Row(b=1, b1=3, b2=1)])
- f = udf(lambda a, b: a == b, BooleanType())
- df = left.join(right, [f("a", "b"), left.a1 == right.b1], "left_semi")
- # do not need spark.sql.crossJoin.enabled=true for udf is not the only join condition.
- self.assertEqual(df.collect(), [Row(a=1, a1=1, a2=1)])
-
def test_udf_not_supported_in_join_condition(self):
# regression test for SPARK-25314
- # test python udf is not supported in join type besides left_semi and inner join.
+ # test python udf is not supported in join type except inner join.
left = self.spark.createDataFrame([Row(a=1, a1=1, a2=1), Row(a=2, a1=2, a2=2)])
right = self.spark.createDataFrame([Row(b=1, b1=1, b2=1), Row(b=1, b1=3, b2=1)])
f = udf(lambda a, b: a == b, BooleanType())
@@ -262,6 +241,7 @@ class UDFTests(ReusedSQLTestCase):
runWithJoinType("left", "LeftOuter")
runWithJoinType("right", "RightOuter")
runWithJoinType("leftanti", "LeftAnti")
+ runWithJoinType("leftsemi", "LeftSemi")
def test_udf_without_arguments(self):
self.spark.catalog.registerFunction("foo", lambda: "bar")
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala
index 251ece3..22704b2 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala
@@ -180,8 +180,8 @@ object PullOutPythonUDFInJoinCondition extends Rule[LogicalPlan] with PredicateH
override def apply(plan: LogicalPlan): LogicalPlan = plan transformUp {
case j @ Join(_, _, joinType, Some(cond), _) if hasUnevaluablePythonUDF(cond, j) =>
- if (!joinType.isInstanceOf[InnerLike] && joinType != LeftSemi) {
- // The current strategy only support InnerLike and LeftSemi join because for other type,
+ if (!joinType.isInstanceOf[InnerLike]) {
+ // The current strategy supports only InnerLike join because for other types,
// it breaks SQL semantic if we run the join condition as a filter after join. If we pass
// the plan here, it'll still get a an invalid PythonUDF RuntimeException with message
// `requires attributes from more than one child`, we throw firstly here for better
@@ -202,10 +202,6 @@ object PullOutPythonUDFInJoinCondition extends Rule[LogicalPlan] with PredicateH
val newJoin = j.copy(condition = newCondition)
joinType match {
case _: InnerLike => Filter(udf.reduceLeft(And), newJoin)
- case LeftSemi =>
- Project(
- j.left.output.map(_.toAttribute),
- Filter(udf.reduceLeft(And), newJoin.copy(joinType = Inner)))
case _ =>
throw new AnalysisException("Using PythonUDF in join condition of join type" +
s" $joinType is not supported.")
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PullOutPythonUDFInJoinConditionSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PullOutPythonUDFInJoinConditionSuite.scala
index 3f1c91d..4a25ddf 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PullOutPythonUDFInJoinConditionSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PullOutPythonUDFInJoinConditionSuite.scala
@@ -64,7 +64,7 @@ class PullOutPythonUDFInJoinConditionSuite extends PlanTest {
PythonEvalType.SQL_BATCHED_UDF,
udfDeterministic = true)
- val unsupportedJoinTypes = Seq(LeftOuter, RightOuter, FullOuter, LeftAnti)
+ val unsupportedJoinTypes = Seq(LeftOuter, RightOuter, FullOuter, LeftAnti, LeftSemi)
private def comparePlanWithCrossJoinEnable(query: LogicalPlan, expected: LogicalPlan): Unit = {
// AnalysisException thrown by CheckCartesianProducts while spark.sql.crossJoin.enabled=false
@@ -99,25 +99,6 @@ class PullOutPythonUDFInJoinConditionSuite extends PlanTest {
comparePlans(Optimize.execute(query2), query2)
}
- test("left semi join condition with python udf") {
- val query1 = testRelationLeft.join(
- testRelationRight,
- joinType = LeftSemi,
- condition = Some(unevaluableJoinCond))
- val expected1 = testRelationLeft.join(
- testRelationRight,
- joinType = Inner,
- condition = None).where(unevaluableJoinCond).select('a, 'b).analyze
- comparePlanWithCrossJoinEnable(query1, expected1)
-
- // evaluable PythonUDF will not be touched
- val query2 = testRelationLeft.join(
- testRelationRight,
- joinType = LeftSemi,
- condition = Some(evaluableJoinCond))
- comparePlans(Optimize.execute(query2), query2)
- }
-
test("unevaluable python udf and common condition") {
val query = testRelationLeft.join(
testRelationRight,
@@ -195,7 +176,7 @@ class PullOutPythonUDFInJoinConditionSuite extends PlanTest {
comparePlans(optimized, expected)
}
- test("throw an exception for not support join type") {
+ test("throw an exception for not supported join types") {
for (joinType <- unsupportedJoinTypes) {
val e = intercept[AnalysisException] {
val query = testRelationLeft.join(
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org