Merge branch 'master' into eval

Conflicts:
	sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Projection.scala
	sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregates.scala
	sql/core/src/main/scala/org/apache/spark/sql/execution/aggregates.scala

Author: rxin

project/SparkBuild.scala

@@ -178,6 +178,7 @@ object SparkBuild extends Build {
     fork := true,
     javaOptions in Test += "-Dspark.home=" + sparkHome,
     javaOptions in Test += "-Dspark.testing=1",
+    javaOptions in Test += "-Dsun.io.serialization.extendedDebugInfo=true",
     javaOptions in Test ++= System.getProperties.filter(_._1 startsWith "spark").map { case (k,v) => s"-D$k=$v" }.toSeq,
     javaOptions += "-Xmx3g",
     // Show full stack trace and duration in test cases.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala

@@ -48,11 +48,17 @@ case class BoundReference(ordinal: Int, baseReference: Attribute)
   override def eval(input: Row): Any = input(ordinal)
 }
 
+/**
+ * Used to denote operators that do their own binding of attributes internally.
+ */
+trait NoBind { self: trees.TreeNode[_] => }
+
 class BindReferences[TreeNode <: QueryPlan[TreeNode]] extends Rule[TreeNode] {
   import BindReferences._
 
   def apply(plan: TreeNode): TreeNode = {
     plan.transform {
+      case n: NoBind => n.asInstanceOf[TreeNode]
       case leafNode if leafNode.children.isEmpty => leafNode
       case unaryNode if unaryNode.children.size == 1 => unaryNode.transformExpressions { case e =>
         bindReference(e, unaryNode.children.head.output)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Projection.scala

@@ -29,9 +29,9 @@ class Projection(expressions: Seq[Expression]) extends (Row => Row) {
   protected val exprArray = expressions.toArray
 
   def apply(input: Row): Row = {
-    val outputArray = new Array[Any](exprArray.size)
+    val outputArray = new Array[Any](exprArray.length)
     var i = 0
-    while (i < exprArray.size) {
+    while (i < exprArray.length) {
       outputArray(i) = exprArray(i).eval(input)
       i += 1
     }
@@ -58,7 +58,7 @@ case class MutableProjection(expressions: Seq[Expression]) extends (Row => Row)
 
   def apply(input: Row): Row = {
     var i = 0
-    while (i < exprArray.size) {
+    while (i < exprArray.length) {
       mutableRow(i) = exprArray(i).eval(input)
       i += 1
     }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregates.scala

@@ -27,7 +27,7 @@ abstract class AggregateExpression extends Expression {
    * Creates a new instance that can be used to compute this aggregate expression for a group
    * of input rows/
    */
-  def newInstance: AggregateFunction
+  def newInstance(): AggregateFunction
 }
 
 /**
@@ -75,7 +75,7 @@ abstract class AggregateFunction
   override def eval(input: Row): Any
 
   // Do we really need this?
-  override def newInstance = makeCopy(productIterator.map { case a: AnyRef => a }.toArray)
+  override def newInstance() = makeCopy(productIterator.map { case a: AnyRef => a }.toArray)
 }
 
 case class Count(child: Expression) extends PartialAggregate with trees.UnaryNode[Expression] {
@@ -89,7 +89,7 @@ case class Count(child: Expression) extends PartialAggregate with trees.UnaryNod
     SplitEvaluation(Sum(partialCount.toAttribute), partialCount :: Nil)
   }
 
-  override def newInstance = new CountFunction(child, this)
+  override def newInstance()= new CountFunction(child, this)
 }
 
 case class CountDistinct(expressions: Seq[Expression]) extends AggregateExpression {
@@ -98,7 +98,7 @@ case class CountDistinct(expressions: Seq[Expression]) extends AggregateExpressi
   override def nullable = false
   override def dataType = IntegerType
   override def toString = s"COUNT(DISTINCT ${expressions.mkString(",")}})"
-  override def newInstance = new CountDistinctFunction(expressions, this)
+  override def newInstance()= new CountDistinctFunction(expressions, this)
 }
 
 case class Average(child: Expression) extends PartialAggregate with trees.UnaryNode[Expression] {
@@ -118,7 +118,7 @@ case class Average(child: Expression) extends PartialAggregate with trees.UnaryN
       partialCount :: partialSum :: Nil)
   }
 
-  override def newInstance = new AverageFunction(child, this)
+  override def newInstance()= new AverageFunction(child, this)
 }
 
 case class Sum(child: Expression) extends PartialAggregate with trees.UnaryNode[Expression] {
@@ -134,7 +134,7 @@ case class Sum(child: Expression) extends PartialAggregate with trees.UnaryNode[
       partialSum :: Nil)
   }
 
-  override def newInstance = new SumFunction(child, this)
+  override def newInstance()= new SumFunction(child, this)
 }
 
 case class SumDistinct(child: Expression)
@@ -145,7 +145,7 @@ case class SumDistinct(child: Expression)
   override def dataType = child.dataType
   override def toString = s"SUM(DISTINCT $child)"
 
-  override def newInstance = new SumDistinctFunction(child, this)
+  override def newInstance()= new SumDistinctFunction(child, this)
 }
 
 case class First(child: Expression) extends PartialAggregate with trees.UnaryNode[Expression] {
@@ -160,7 +160,7 @@ case class First(child: Expression) extends PartialAggregate with trees.UnaryNod
       First(partialFirst.toAttribute),
       partialFirst :: Nil)
   }
-  override def newInstance = new FirstFunction(child, this)
+  override def newInstance()= new FirstFunction(child, this)
 }
 
 case class AverageFunction(expr: Expression, base: AggregateExpression)

sql/core/src/main/scala/org/apache/spark/rdd/PartitionLocalRDDFunctions.scala

@@ -76,7 +76,7 @@ case class Exchange(newPartitioning: Partitioning, child: SparkPlan) extends Una
  */
 object AddExchange extends Rule[SparkPlan] {
   // TODO: Determine the number of partitions.
-  val numPartitions = 8
+  val numPartitions = 150
 
   def apply(plan: SparkPlan): SparkPlan = plan.transformUp {
     case operator: SparkPlan =>