You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by rx...@apache.org on 2016/12/15 00:40:30 UTC
[1/2] spark git commit: Revert "Revert "[SPARK-18854][SQL]
numberedTreeString and apply(i) inconsistent for subqueries""
Repository: spark
Updated Branches:
refs/heads/branch-2.0 a5c178bc0 -> a32317845
Revert "Revert "[SPARK-18854][SQL] numberedTreeString and apply(i) inconsistent for subqueries""
This reverts commit a5c178bc07092b698ee17894a439deb47699db0f.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/2085a103
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/2085a103
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/2085a103
Branch: refs/heads/branch-2.0
Commit: 2085a1033a23b4344963703f73fa4fdfd26f1e95
Parents: a5c178b
Author: Reynold Xin <rx...@databricks.com>
Authored: Wed Dec 14 16:35:25 2016 -0800
Committer: Reynold Xin <rx...@databricks.com>
Committed: Wed Dec 14 16:35:25 2016 -0800
----------------------------------------------------------------------
.../spark/sql/catalyst/plans/QueryPlan.scala | 9 ++++
.../spark/sql/catalyst/trees/TreeNode.scala | 43 ++++++++++++--------
.../execution/columnar/InMemoryRelation.scala | 3 +-
.../org/apache/spark/sql/SubquerySuite.scala | 18 ++++++++
4 files changed, 54 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/2085a103/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
----------------------------------------------------------------------
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
index 41c4e00..bcc7521 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
@@ -24,6 +24,15 @@ import org.apache.spark.sql.types.{DataType, StructType}
abstract class QueryPlan[PlanType <: QueryPlan[PlanType]] extends TreeNode[PlanType] {
self: PlanType =>
+ /**
+ * Override [[TreeNode.apply]] to so we can return a more narrow type.
+ *
+ * Note that this cannot return BaseType because logical plan's plan node might return
+ * physical plan for innerChildren, e.g. in-memory relation logical plan node has a reference
+ * to the physical plan node it is referencing.
+ */
+ override def apply(number: Int): QueryPlan[_] = super.apply(number).asInstanceOf[QueryPlan[_]]
+
def output: Seq[Attribute]
/**
http://git-wip-us.apache.org/repos/asf/spark/blob/2085a103/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala
----------------------------------------------------------------------
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala
index 931d14d..877a16d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala
@@ -20,7 +20,6 @@ package org.apache.spark.sql.catalyst.trees
import java.util.UUID
import scala.collection.Map
-import scala.collection.mutable.Stack
import scala.reflect.ClassTag
import org.apache.commons.lang3.ClassUtils
@@ -483,7 +482,10 @@ abstract class TreeNode[BaseType <: TreeNode[BaseType]] extends Product {
/**
* Returns a string representation of the nodes in this tree, where each operator is numbered.
- * The numbers can be used with [[trees.TreeNode.apply apply]] to easily access specific subtrees.
+ * The numbers can be used with [[TreeNode.apply]] to easily access specific subtrees.
+ *
+ * The numbers are based on depth-first traversal of the tree (with innerChildren traversed first
+ * before children).
*/
def numberedTreeString: String =
treeString.split("\n").zipWithIndex.map { case (line, i) => f"$i%02d $line" }.mkString("\n")
@@ -491,17 +493,24 @@ abstract class TreeNode[BaseType <: TreeNode[BaseType]] extends Product {
/**
* Returns the tree node at the specified number.
* Numbers for each node can be found in the [[numberedTreeString]].
+ *
+ * Note that this cannot return BaseType because logical plan's plan node might return
+ * physical plan for innerChildren, e.g. in-memory relation logical plan node has a reference
+ * to the physical plan node it is referencing.
*/
- def apply(number: Int): BaseType = getNodeNumbered(new MutableInt(number))
+ def apply(number: Int): TreeNode[_] = getNodeNumbered(new MutableInt(number)).orNull
- protected def getNodeNumbered(number: MutableInt): BaseType = {
+ private def getNodeNumbered(number: MutableInt): Option[TreeNode[_]] = {
if (number.i < 0) {
- null.asInstanceOf[BaseType]
+ None
} else if (number.i == 0) {
- this
+ Some(this)
} else {
number.i -= 1
- children.map(_.getNodeNumbered(number)).find(_ != null).getOrElse(null.asInstanceOf[BaseType])
+ // Note that this traversal order must be the same as numberedTreeString.
+ innerChildren.map(_.getNodeNumbered(number)).find(_ != None).getOrElse {
+ children.map(_.getNodeNumbered(number)).find(_ != None).flatten
+ }
}
}
@@ -517,6 +526,8 @@ abstract class TreeNode[BaseType <: TreeNode[BaseType]] extends Product {
* The `i`-th element in `lastChildren` indicates whether the ancestor of the current node at
* depth `i + 1` is the last child of its own parent node. The depth of the root node is 0, and
* `lastChildren` for the root node should be empty.
+ *
+ * Note that this traversal (numbering) order must be the same as [[getNodeNumbered]].
*/
def generateTreeString(
depth: Int,
@@ -524,19 +535,16 @@ abstract class TreeNode[BaseType <: TreeNode[BaseType]] extends Product {
builder: StringBuilder,
verbose: Boolean,
prefix: String = ""): StringBuilder = {
+
if (depth > 0) {
lastChildren.init.foreach { isLast =>
- val prefixFragment = if (isLast) " " else ": "
- builder.append(prefixFragment)
+ builder.append(if (isLast) " " else ": ")
}
-
- val branch = if (lastChildren.last) "+- " else ":- "
- builder.append(branch)
+ builder.append(if (lastChildren.last) "+- " else ":- ")
}
builder.append(prefix)
- val headline = if (verbose) verboseString else simpleString
- builder.append(headline)
+ builder.append(if (verbose) verboseString else simpleString)
builder.append("\n")
if (innerChildren.nonEmpty) {
@@ -547,9 +555,10 @@ abstract class TreeNode[BaseType <: TreeNode[BaseType]] extends Product {
}
if (children.nonEmpty) {
- children.init.foreach(
- _.generateTreeString(depth + 1, lastChildren :+ false, builder, verbose, prefix))
- children.last.generateTreeString(depth + 1, lastChildren :+ true, builder, verbose, prefix)
+ children.init.foreach(_.generateTreeString(
+ depth + 1, lastChildren :+ false, builder, verbose, prefix))
+ children.last.generateTreeString(
+ depth + 1, lastChildren :+ true, builder, verbose, prefix)
}
builder
http://git-wip-us.apache.org/repos/asf/spark/blob/2085a103/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala
index 56bd5c1..03cc046 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala
@@ -24,7 +24,6 @@ import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.MultiInstanceRelation
import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.catalyst.plans.QueryPlan
import org.apache.spark.sql.catalyst.plans.logical
import org.apache.spark.sql.catalyst.plans.logical.Statistics
import org.apache.spark.sql.execution.SparkPlan
@@ -64,7 +63,7 @@ case class InMemoryRelation(
val batchStats: LongAccumulator = child.sqlContext.sparkContext.longAccumulator)
extends logical.LeafNode with MultiInstanceRelation {
- override protected def innerChildren: Seq[QueryPlan[_]] = Seq(child)
+ override protected def innerChildren: Seq[SparkPlan] = Seq(child)
override def producedAttributes: AttributeSet = outputSet
http://git-wip-us.apache.org/repos/asf/spark/blob/2085a103/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
index b145c69..eff8894 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
@@ -54,6 +54,24 @@ class SubquerySuite extends QueryTest with SharedSQLContext {
t.createOrReplaceTempView("t")
}
+ test("SPARK-18854 numberedTreeString for subquery") {
+ val df = sql("select * from range(10) where id not in " +
+ "(select id from range(2) union all select id from range(2))")
+
+ // The depth first traversal of the plan tree
+ val dfs = Seq("Project", "Filter", "Union", "Project", "Range", "Project", "Range", "Range")
+ val numbered = df.queryExecution.analyzed.numberedTreeString.split("\n")
+
+ // There should be 8 plan nodes in total
+ assert(numbered.size == dfs.size)
+
+ for (i <- dfs.indices) {
+ val node = df.queryExecution.analyzed(i)
+ assert(node.nodeName == dfs(i))
+ assert(numbered(i).contains(node.nodeName))
+ }
+ }
+
test("rdd deserialization does not crash [SPARK-15791]") {
sql("select (select 1 as b) as b").rdd.count()
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org
[2/2] spark git commit: Fix compilation error
Posted by rx...@apache.org.
Fix compilation error
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/a3231784
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/a3231784
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/a3231784
Branch: refs/heads/branch-2.0
Commit: a32317845277f327664734f7c3af6c80a65dc3e6
Parents: 2085a10
Author: Reynold Xin <rx...@databricks.com>
Authored: Wed Dec 14 16:40:17 2016 -0800
Committer: Reynold Xin <rx...@databricks.com>
Committed: Wed Dec 14 16:40:17 2016 -0800
----------------------------------------------------------------------
.../main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/a3231784/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala
----------------------------------------------------------------------
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala
index 877a16d..78cc0fa 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala
@@ -20,6 +20,7 @@ package org.apache.spark.sql.catalyst.trees
import java.util.UUID
import scala.collection.Map
+import scala.collection.mutable.Stack
import scala.reflect.ClassTag
import org.apache.commons.lang3.ClassUtils
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org