another fix

cloud-fan · cloud-fan · commit 3426310f1be9 · 2016-04-18T17:00:05.000+08:00
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/EquivalentExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/EquivalentExpressions.scala
@@ -69,8 +69,15 @@ class EquivalentExpressions {
    */
   def addExprTree(root: Expression, ignoreLeaf: Boolean = true): Unit = {
     val skip = root.isInstanceOf[LeafExpression] && ignoreLeaf
-    // the children of CodegenFallback will not be used to generate code (call eval() instead)
-    if (!skip && !addExpr(root) && !root.isInstanceOf[CodegenFallback]) {
+    // There are some special expressions that we should not recurse into children.
+    //   1. CodegenFallback: it's children will not be used to generate code (call eval() instead)
+    //   2. ReferenceToExpressions: it's kind of an explicit sub-expression elimination.
+    val shouldRecurse = root match {
+      case _: CodegenFallback => false
+      case _: ReferenceToExpressions => false
+      case _ => true
+    }
+    if (!skip && !addExpr(root) && shouldRecurse) {
       root.children.foreach(addExprTree(_, ignoreLeaf))
     }
   }
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala
@@ -110,13 +110,13 @@ class CodegenContext {
   }
 
   def declareMutableStates(): String = {
-    mutableStates.map { case (javaType, variableName, _) =>
+    mutableStates.distinct.map { case (javaType, variableName, _) =>
       s"private $javaType $variableName;"
     }.mkString("\n")
   }
 
   def initMutableStates(): String = {
-    mutableStates.map(_._3).mkString("\n")
+    mutableStates.distinct.map(_._3).mkString("\n")
   }
 
   /**
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/objects.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/objects.scala
@@ -436,7 +436,7 @@ case class MapObjects private(
 
   override def nullable: Boolean = true
 
-  override def children: Seq[Expression] = Seq(loopVar, lambdaFunction, inputData)
+  override def children: Seq[Expression] = lambdaFunction :: inputData :: Nil
 
   override def eval(input: InternalRow): Any =
     throw new UnsupportedOperationException("Only code-generated evaluation is supported")
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TypedAggregateExpression.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TypedAggregateExpression.scala
@@ -126,11 +126,9 @@ case class TypedAggregateExpression(
   override lazy val mergeExpressions: Seq[Expression] = {
     val leftBuffer = bufferDeserializer transform {
       case a: AttributeReference => a.left
-      case l: LambdaVariable => l.copy(value = l.value + "_left", isNull = l.isNull + "_left")
     }
     val rightBuffer = bufferDeserializer transform {
       case a: AttributeReference => a.right
-      case l: LambdaVariable => l.copy(value = l.value + "_right", isNull = l.isNull + "_right")
     }
     val merged = Invoke(
       aggregatorLiteral,

Original file line number	Diff line number	Diff line change
`@@ -110,13 +110,13 @@ class CodegenContext {`
`110`	`110`	`}`
`111`	`111`
`112`	`112`	`def declareMutableStates(): String = {`
`113`		`- mutableStates.map { case (javaType, variableName, _) =>`
	`113`	`+ mutableStates.distinct.map { case (javaType, variableName, _) =>`
`114`	`114`	`s"private $javaType $variableName;"`
`115`	`115`	`}.mkString("\n")`
`116`	`116`	`}`
`117`	`117`
`118`	`118`	`def initMutableStates(): String = {`
`119`		`- mutableStates.map(_._3).mkString("\n")`
	`119`	`+ mutableStates.distinct.map(_._3).mkString("\n")`
`120`	`120`	`}`
`121`	`121`
`122`	`122`	`/**`
Original file line number	Diff line number	Diff line change
`@@ -126,11 +126,9 @@ case class TypedAggregateExpression(`
`126`	`126`	`override lazy val mergeExpressions: Seq[Expression] = {`
`127`	`127`	`val leftBuffer = bufferDeserializer transform {`
`128`	`128`	`case a: AttributeReference => a.left`
`129`		`- case l: LambdaVariable => l.copy(value = l.value + "_left", isNull = l.isNull + "_left")`
`130`	`129`	`}`
`131`	`130`	`val rightBuffer = bufferDeserializer transform {`
`132`	`131`	`case a: AttributeReference => a.right`
`133`		`- case l: LambdaVariable => l.copy(value = l.value + "_right", isNull = l.isNull + "_right")`
`134`	`132`	`}`
`135`	`133`	`val merged = Invoke(`
`136`	`134`	`aggregatorLiteral,`