[jira] [Created] (SPARK-22211) LimitPushDown optimization for FullOuterJoin generates wrong results
Benyi Wang created SPARK-22211: -- Summary: LimitPushDown optimization for FullOuterJoin generates wrong results Key: SPARK-22211 URL: https://issues.apache.org/jira/browse/SPARK-22211 Project: Spark Issue Type: Bug Components: SQL Affects Versions: 2.2.0 Environment: on community.cloude.databrick.com Runtime Version 3.2 (includes Apache Spark 2.2.0, Scala 2.11) Reporter: Benyi Wang LimitPushDown pushes LocalLimit to one side for FullOuterJoin, but this may generate a wrong result: Assume we use limit(1) and LocalLimit will be pushed to left side, and id=999 is selected, but at right side we have 100K rows including 999, the result will be - one row is (999, 999) - the rest rows are (null, xxx) Once you call show(), the row (999,999) has only 1/10th chance to be selected by CollectLimit. The actual optimization might be, - push down limit - but convert the join to Broadcast LeftOuterJoin or RightOuterJoin. Here is my notebook: https://databricks-prod-cloudfront.cloud.databricks.com/public/4027ec902e239c93eaaa8714f173bcfc/349451637617406/2750346983121008/656075277290/latest.html {code:java} import scala.util.Random._ val dl = shuffle(1 to 10).toDF("id") val dr = shuffle(1 to 10).toDF("id") println("data frame dl:") dl.explain println("data frame dr:") dr.explain val j = dl.join(dr, dl("id") === dr("id"), "outer").limit(1) j.explain j.show(false) {code} {code} data frame dl: == Physical Plan == LocalTableScan [id#10] data frame dr: == Physical Plan == LocalTableScan [id#16] == Physical Plan == CollectLimit 1 +- SortMergeJoin [id#10], [id#16], FullOuter :- *Sort [id#10 ASC NULLS FIRST], false, 0 : +- Exchange hashpartitioning(id#10, 200) : +- *LocalLimit 1 :+- LocalTableScan [id#10] +- *Sort [id#16 ASC NULLS FIRST], false, 0 +- Exchange hashpartitioning(id#16, 200) +- LocalTableScan [id#16] import scala.util.Random._ dl: org.apache.spark.sql.DataFrame = [id: int] dr: org.apache.spark.sql.DataFrame = [id: int] j: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [id: int, id: int] ++---+ |id |id | ++---+ |null|148| ++---+ {code} -- This message was sent by Atlassian JIRA (v6.4.14#64029) - To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-22211) LimitPushDown optimization for FullOuterJoin generates wrong results
[ https://issues.apache.org/jira/browse/SPARK-22211?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16202278#comment-16202278 ] Benyi Wang commented on SPARK-22211: I think my suggestion solution is correct. || Case || Left join key || Right join key || Full outer join || | 1 | Y | N | {{(left(\*), null)}} | | 2 | Y | Y | {{(left(\*), right(\*))}} | | 3 | N | Y | {{(null, right(\*))}} | | 4 | N | N | Not applied | If LimitPushDown pushes limit to the left side, whatever a limit value is and how big of left side table, you will always select some rows, in other words, the join keys are always exists, and only case 1 and 2 will happen, so it is actually a Left-join instead. It is equivalent to right-join when pushing down to the right side. The only problem of this method is: case 3 has no chance to be shown while pushing down to the left side, and case 1 for the right side. I would say this is not a big issue because we just want to see some samples of the join result, but the benefit is huge. If we want to see left-only or right-only, we might add where clause. > LimitPushDown optimization for FullOuterJoin generates wrong results > > > Key: SPARK-22211 > URL: https://issues.apache.org/jira/browse/SPARK-22211 > Project: Spark > Issue Type: Bug > Components: SQL >Affects Versions: 2.2.0 > Environment: on community.cloude.databrick.com > Runtime Version 3.2 (includes Apache Spark 2.2.0, Scala 2.11) >Reporter: Benyi Wang > > LimitPushDown pushes LocalLimit to one side for FullOuterJoin, but this may > generate a wrong result: > Assume we use limit(1) and LocalLimit will be pushed to left side, and id=999 > is selected, but at right side we have 100K rows including 999, the result > will be > - one row is (999, 999) > - the rest rows are (null, xxx) > Once you call show(), the row (999,999) has only 1/10th chance to be > selected by CollectLimit. > The actual optimization might be, > - push down limit > - but convert the join to Broadcast LeftOuterJoin or RightOuterJoin. > Here is my notebook: > https://databricks-prod-cloudfront.cloud.databricks.com/public/4027ec902e239c93eaaa8714f173bcfc/349451637617406/2750346983121008/656075277290/latest.html > {code:java} > import scala.util.Random._ > val dl = shuffle(1 to 10).toDF("id") > val dr = shuffle(1 to 10).toDF("id") > println("data frame dl:") > dl.explain > println("data frame dr:") > dr.explain > val j = dl.join(dr, dl("id") === dr("id"), "outer").limit(1) > j.explain > j.show(false) > {code} > {code} > data frame dl: > == Physical Plan == > LocalTableScan [id#10] > data frame dr: > == Physical Plan == > LocalTableScan [id#16] > == Physical Plan == > CollectLimit 1 > +- SortMergeJoin [id#10], [id#16], FullOuter >:- *Sort [id#10 ASC NULLS FIRST], false, 0 >: +- Exchange hashpartitioning(id#10, 200) >: +- *LocalLimit 1 >:+- LocalTableScan [id#10] >+- *Sort [id#16 ASC NULLS FIRST], false, 0 > +- Exchange hashpartitioning(id#16, 200) > +- LocalTableScan [id#16] > import scala.util.Random._ > dl: org.apache.spark.sql.DataFrame = [id: int] > dr: org.apache.spark.sql.DataFrame = [id: int] > j: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [id: int, id: int] > ++---+ > |id |id | > ++---+ > |null|148| > ++---+ > {code} -- This message was sent by Atlassian JIRA (v6.4.14#64029) - To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Created] (SPARK-6589) SQLUserDefinedType failed in spark-shell
Benyi Wang created SPARK-6589: - Summary: SQLUserDefinedType failed in spark-shell Key: SPARK-6589 URL: https://issues.apache.org/jira/browse/SPARK-6589 Project: Spark Issue Type: Bug Components: SQL Affects Versions: 1.2.0 Environment: CDH 5.3.2 Reporter: Benyi Wang {{DataType.fromJson}} will fail in spark-shell if the schema includes "udt". It works if running in an application. This causes that I cannot read a parquet file including a UDT field. {{DataType.fromCaseClass}} does not support UDT. I can load the class which shows that my UDT is in the classpath. {code} scala> Class.forName("com.bwang.MyTestUDT") res6: Class[_] = class com.bwang.MyTestUDT {code} But DataType fails: {code} scala> DataType.fromJson(json) java.lang.ClassNotFoundException: com.bwang.MyTestUDT at java.net.URLClassLoader$1.run(URLClassLoader.java:366) at java.net.URLClassLoader$1.run(URLClassLoader.java:355) at java.security.AccessController.doPrivileged(Native Method) at java.net.URLClassLoader.findClass(URLClassLoader.java:354) at java.lang.ClassLoader.loadClass(ClassLoader.java:425) at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308) at java.lang.ClassLoader.loadClass(ClassLoader.java:358) at java.lang.Class.forName0(Native Method) at java.lang.Class.forName(Class.java:190) at org.apache.spark.sql.catalyst.types.DataType$.parseDataType(dataTypes.scala:77) {code} The reason is DataType.fromJson tries to load {{udtClass}} using this code: {code} case JSortedObject( ("class", JString(udtClass)), ("pyClass", _), ("sqlType", _), ("type", JString("udt"))) => Class.forName(udtClass).newInstance().asInstanceOf[UserDefinedType[_]] } {code} Unfortunately, my UDT is loaded by {{SparkIMain$TranslatingClassLoader}}, but DataType is loaded by {{Launcher$AppClassLoader}}. {code} scala> DataType.getClass.getClassLoader res2: ClassLoader = sun.misc.Launcher$AppClassLoader@6876fb1b scala> this.getClass.getClassLoader res3: ClassLoader = org.apache.spark.repl.SparkIMain$TranslatingClassLoader@63d36b29 {code} -- This message was sent by Atlassian JIRA (v6.3.4#6332) - To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-6589) SQLUserDefinedType failed in spark-shell
[ https://issues.apache.org/jira/browse/SPARK-6589?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14385499#comment-14385499 ] Benyi Wang commented on SPARK-6589: --- I found a method to fix this issue. But I still think DataType should find a better way to find the correct class loader. {code} # put the UDT jar to SPARK_CLASSPATH so that Launcher$AppClassLoader can find it. export SPARK_CLASSPATH=myUDT.jar spark-shell --jars myUDT.jar ... {code} > SQLUserDefinedType failed in spark-shell > > > Key: SPARK-6589 > URL: https://issues.apache.org/jira/browse/SPARK-6589 > Project: Spark > Issue Type: Bug > Components: SQL >Affects Versions: 1.2.0 > Environment: CDH 5.3.2 >Reporter: Benyi Wang > > {{DataType.fromJson}} will fail in spark-shell if the schema includes "udt". > It works if running in an application. > This causes that I cannot read a parquet file including a UDT field. > {{DataType.fromCaseClass}} does not support UDT. > I can load the class which shows that my UDT is in the classpath. > {code} > scala> Class.forName("com.bwang.MyTestUDT") > res6: Class[_] = class com.bwang.MyTestUDT > {code} > But DataType fails: > {code} > scala> DataType.fromJson(json) > > java.lang.ClassNotFoundException: com.bwang.MyTestUDT > at java.net.URLClassLoader$1.run(URLClassLoader.java:366) > at java.net.URLClassLoader$1.run(URLClassLoader.java:355) > at java.security.AccessController.doPrivileged(Native Method) > at java.net.URLClassLoader.findClass(URLClassLoader.java:354) > at java.lang.ClassLoader.loadClass(ClassLoader.java:425) > at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308) > at java.lang.ClassLoader.loadClass(ClassLoader.java:358) > at java.lang.Class.forName0(Native Method) > at java.lang.Class.forName(Class.java:190) > at > org.apache.spark.sql.catalyst.types.DataType$.parseDataType(dataTypes.scala:77) > {code} > The reason is DataType.fromJson tries to load {{udtClass}} using this code: > {code} > case JSortedObject( > ("class", JString(udtClass)), > ("pyClass", _), > ("sqlType", _), > ("type", JString("udt"))) => > Class.forName(udtClass).newInstance().asInstanceOf[UserDefinedType[_]] > } > {code} > Unfortunately, my UDT is loaded by {{SparkIMain$TranslatingClassLoader}}, but > DataType is loaded by {{Launcher$AppClassLoader}}. > {code} > scala> DataType.getClass.getClassLoader > res2: ClassLoader = sun.misc.Launcher$AppClassLoader@6876fb1b > scala> this.getClass.getClassLoader > res3: ClassLoader = > org.apache.spark.repl.SparkIMain$TranslatingClassLoader@63d36b29 > {code} -- This message was sent by Atlassian JIRA (v6.3.4#6332) - To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org