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 2018/12/11 06:24:36 UTC
[spark] branch branch-2.4 updated: [SPARK-26293][SQL] Cast
exception when having python udf in subquery
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch branch-2.4
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-2.4 by this push:
new 0f1b225 [SPARK-26293][SQL] Cast exception when having python udf in subquery
0f1b225 is described below
commit 0f1b225f8f30d31450a310df77db0e9d8d8e20f1
Author: Wenchen Fan <we...@databricks.com>
AuthorDate: Tue Dec 11 14:16:51 2018 +0800
[SPARK-26293][SQL] Cast exception when having python udf in subquery
This is a regression introduced by https://github.com/apache/spark/pull/22104 at Spark 2.4.0.
When we have Python UDF in subquery, we will hit an exception
```
Caused by: java.lang.ClassCastException: org.apache.spark.sql.catalyst.expressions.AttributeReference cannot be cast to org.apache.spark.sql.catalyst.expressions.PythonUDF
at scala.collection.immutable.Stream.map(Stream.scala:414)
at org.apache.spark.sql.execution.python.EvalPythonExec.$anonfun$doExecute$2(EvalPythonExec.scala:98)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2(RDD.scala:815)
...
```
https://github.com/apache/spark/pull/22104 turned `ExtractPythonUDFs` from a physical rule to optimizer rule. However, there is a difference between a physical rule and optimizer rule. A physical rule always runs once, an optimizer rule may be applied twice on a query tree even the rule is located in a batch that only runs once.
For a subquery, the `OptimizeSubqueries` rule will execute the entire optimizer on the query plan inside subquery. Later on subquery will be turned to joins, and the optimizer rules will be applied to it again.
Unfortunately, the `ExtractPythonUDFs` rule is not idempotent. When it's applied twice on a query plan inside subquery, it will produce a malformed plan. It extracts Python UDF from Python exec plans.
This PR proposes 2 changes to be double safe:
1. `ExtractPythonUDFs` should skip python exec plans, to make the rule idempotent
2. `ExtractPythonUDFs` should skip subquery
a new test.
Closes #23248 from cloud-fan/python.
Authored-by: Wenchen Fan <we...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit 7d5f6e8c493b96898ba01edede1522121fe945fc)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
python/pyspark/sql/tests.py | 9 +++++++++
.../sql/execution/python/ArrowEvalPythonExec.scala | 8 ++++++--
.../sql/execution/python/BatchEvalPythonExec.scala | 8 ++++++--
.../spark/sql/execution/python/ExtractPythonUDFs.scala | 18 +++++++++++++++---
4 files changed, 36 insertions(+), 7 deletions(-)
diff --git a/python/pyspark/sql/tests.py b/python/pyspark/sql/tests.py
index 4fc1b4f..2a0b078 100644
--- a/python/pyspark/sql/tests.py
+++ b/python/pyspark/sql/tests.py
@@ -746,6 +746,15 @@ class SQLTests(ReusedSQLTestCase):
self.assertRaisesRegexp(AnalysisException, "Can not load class non_existed_udaf",
lambda: spark.udf.registerJavaUDAF("udaf1", "non_existed_udaf"))
+ # SPARK-26293
+ def test_udf_in_subquery(self):
+ from pyspark.sql.functions import udf
+ f = udf(lambda x: x, "long")
+ self.spark.range(1).filter(f("id") >= 0).createOrReplaceTempView("v")
+ sql = self.spark.sql
+ result = sql("select i from values(0L) as data(i) where i in (select id from v)")
+ self.assertEqual(result.collect(), [Row(i=0)])
+
def test_linesep_text(self):
df = self.spark.read.text("python/test_support/sql/ages_newlines.csv", lineSep=",")
expected = [Row(value=u'Joe'), Row(value=u'20'), Row(value=u'"Hi'),
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ArrowEvalPythonExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ArrowEvalPythonExec.scala
index 2b87796..a5203da 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ArrowEvalPythonExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ArrowEvalPythonExec.scala
@@ -60,8 +60,12 @@ private class BatchIterator[T](iter: Iterator[T], batchSize: Int)
/**
* A logical plan that evaluates a [[PythonUDF]].
*/
-case class ArrowEvalPython(udfs: Seq[PythonUDF], output: Seq[Attribute], child: LogicalPlan)
- extends UnaryNode
+case class ArrowEvalPython(
+ udfs: Seq[PythonUDF],
+ output: Seq[Attribute],
+ child: LogicalPlan) extends UnaryNode {
+ override def producedAttributes: AttributeSet = AttributeSet(output.drop(child.output.length))
+}
/**
* A physical plan that evaluates a [[PythonUDF]].
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/BatchEvalPythonExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/BatchEvalPythonExec.scala
index b08b7e6..d3736d2 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/BatchEvalPythonExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/BatchEvalPythonExec.scala
@@ -32,8 +32,12 @@ import org.apache.spark.sql.types.{StructField, StructType}
/**
* A logical plan that evaluates a [[PythonUDF]]
*/
-case class BatchEvalPython(udfs: Seq[PythonUDF], output: Seq[Attribute], child: LogicalPlan)
- extends UnaryNode
+case class BatchEvalPython(
+ udfs: Seq[PythonUDF],
+ output: Seq[Attribute],
+ child: LogicalPlan) extends UnaryNode {
+ override def producedAttributes: AttributeSet = AttributeSet(output.drop(child.output.length))
+}
/**
* A physical plan that evaluates a [[PythonUDF]]
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ExtractPythonUDFs.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ExtractPythonUDFs.scala
index 90b5325..380c31b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ExtractPythonUDFs.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ExtractPythonUDFs.scala
@@ -24,7 +24,7 @@ import org.apache.spark.api.python.PythonEvalType
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
@@ -131,8 +131,20 @@ object ExtractPythonUDFs extends Rule[LogicalPlan] with PredicateHelper {
expressions.flatMap(collectEvaluableUDFs)
}
- def apply(plan: LogicalPlan): LogicalPlan = plan transformUp {
- case plan: LogicalPlan => extract(plan)
+ def apply(plan: LogicalPlan): LogicalPlan = plan match {
+ // SPARK-26293: A subquery will be rewritten into join later, and will go through this rule
+ // eventually. Here we skip subquery, as Python UDF only needs to be extracted once.
+ case _: Subquery => plan
+
+ case _ => plan transformUp {
+ // A safe guard. `ExtractPythonUDFs` only runs once, so we will not hit `BatchEvalPython` and
+ // `ArrowEvalPython` in the input plan. However if we hit them, we must skip them, as we can't
+ // extract Python UDFs from them.
+ case p: BatchEvalPython => p
+ case p: ArrowEvalPython => p
+
+ case plan: LogicalPlan => extract(plan)
+ }
}
/**
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org