add regression test

apache · Jan 29, 2016 · ffa8e6b · ffa8e6b
1 parent f90b38d
commit ffa8e6b
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 4 deletions.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregate.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregate.scala
@@ -215,17 +215,19 @@ case class TungstenAggregate(
 
     ctx.currentVars = bufVars ++ input
     // TODO: support subexpression elimination
-    val updates = updateExpr.zipWithIndex.map { case (e, i) =>
-      val ev = BindReferences.bindReference[Expression](e, inputAttrs).gen(ctx)
+    val aggVals = updateExpr.map(BindReferences.bindReference(_, inputAttrs).gen(ctx))
+    // aggregate buffer should be updated atomic
+    val updates = aggVals.zipWithIndex.map { case (ev, i) =>
       s"""
-         | ${ev.code}
          | ${bufVars(i).isNull} = ${ev.isNull};
          | ${bufVars(i).value} = ${ev.value};
        """.stripMargin
     }
 
     s"""
-       | // do aggregate and update aggregation buffer
+       | // do aggregate
+       | ${aggVals.map(_.code).mkString("\n")}
+       | // update aggregation buffer
        | ${updates.mkString("")}
      """.stripMargin
   }

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
@@ -153,6 +153,13 @@ class DataFrameAggregateSuite extends QueryTest with SharedSQLContext {
     )
   }
 
+  test("agg without groups and functions") {
+    checkAnswer(
+      testData2.agg(lit(1)),
+      Row(1)
+    )
+  }
+
   test("average") {
     checkAnswer(
       testData2.agg(avg('a), mean('a)),