Merge pull request #79 from markhamstra/csd-1.4_empty_sum_null

Sum of NULL values should return NULL

Author: markhamstra

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

@@ -619,7 +619,7 @@ case class SumFunction(expr: Expression, base: AggregateExpression) extends Aggr
   private val sum = MutableLiteral(null, calcType)
 
   private val addFunction =
-    Coalesce(Seq(Add(Coalesce(Seq(sum, zero)), Cast(expr, calcType)), sum, zero))
+    Coalesce(Seq(Add(Coalesce(Seq(sum, zero)), Cast(expr, calcType)), sum))
 
   override def update(input: Row): Unit = {
     sum.update(addFunction, input)

sql/core/src/main/scala/org/apache/spark/sql/execution/GeneratedAggregate.scala

@@ -107,7 +107,7 @@ case class GeneratedAggregate(
           Add(
             Coalesce(currentSum :: zero :: Nil),
             Cast(expr, calcType)
-          ) :: currentSum :: zero :: Nil)
+          ) :: currentSum :: Nil)
         val result =
           expr.dataType match {
             case DecimalType.Fixed(_, _) =>

sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

@@ -26,7 +26,7 @@ import org.apache.spark.sql.catalyst.errors.DialectException
 import org.apache.spark.sql.execution.GeneratedAggregate
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.TestData._
-import org.apache.spark.sql.test.TestSQLContext
+import org.apache.spark.sql.test.{SQLTestUtils, TestSQLContext}
 import org.apache.spark.sql.test.TestSQLContext.{udf => _, _}
 
 import org.apache.spark.sql.types._
@@ -253,7 +253,7 @@ class SQLQuerySuite extends QueryTest with BeforeAndAfterAll {
     // Aggregate with Code generation handling all null values
     testCodeGen(
       "SELECT  sum('a'), avg('a'), count(null) FROM testData",
-      Row(0, null, 0) :: Nil)
+      Row(null, null, 0) :: Nil)
 
     dropTempTable("testData3x")
     setConf(SQLConf.CODEGEN_ENABLED, originalValue.toString)

sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala

@@ -814,7 +814,6 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "udaf_covar_pop",
     "udaf_covar_samp",
     "udaf_histogram_numeric",
-    "udaf_number_format",
     "udf2",
     "udf5",
     "udf6",

sql/hive/src/test/resources/golden/udaf_number_format-0-eff4ef3c207d14d5121368f294697964

@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql.hive.execution
 
+import org.apache.spark.sql.TestData.NullInts
 import org.apache.spark.sql.catalyst.DefaultParserDialect
 import org.apache.spark.sql.catalyst.analysis.EliminateSubQueries
 import org.apache.spark.sql.catalyst.errors.DialectException
@@ -27,6 +28,7 @@ import org.apache.spark.sql.hive.test.TestHive.implicits._
 import org.apache.spark.sql.hive.{HiveQLDialect, HiveShim, MetastoreRelation}
 import org.apache.spark.sql.parquet.ParquetRelation2
 import org.apache.spark.sql.sources.LogicalRelation
+import org.apache.spark.sql.test.SQLTestUtils
 import org.apache.spark.sql.types._
 
 case class Nested1(f1: Nested2)
@@ -59,7 +61,9 @@ class MyDialect extends DefaultParserDialect
  * Hive to generate them (in contrast to HiveQuerySuite).  Often this is because the query is
  * valid, but Hive currently cannot execute it.
  */
-class SQLQuerySuite extends QueryTest {
+class SQLQuerySuite extends QueryTest with SQLTestUtils {
+  override val sqlContext: SQLContext = TestHive
+
   test("SPARK-6835: udtf in lateral view") {
     val df = Seq((1, 1)).toDF("c1", "c2")
     df.registerTempTable("table1")
@@ -946,4 +950,28 @@ class SQLQuerySuite extends QueryTest {
 
     checkAnswer(sql("SELECT a.`c.b`, `b.$q`[0].`a@!.q`, `q.w`.`w.i&`[0] FROM t"), Row(1, 1, 1))
   }
+
+  test("SPARK-8828 sum should return null if all input values are null") {
+    val allNulls =
+      TestHive.sparkContext.parallelize(
+        NullInts(null) ::
+        NullInts(null) ::
+        NullInts(null) ::
+        NullInts(null) :: Nil).toDF()
+    allNulls.registerTempTable("allNulls")
+
+    withSQLConf(SQLConf.CODEGEN_ENABLED -> "true") {
+      checkAnswer(
+        sql("select sum(a), avg(a) from allNulls"),
+        Seq(Row(null, null))
+      )
+    }
+    withSQLConf(SQLConf.CODEGEN_ENABLED -> "false") {
+      checkAnswer(
+        sql("select sum(a), avg(a) from allNulls"),
+        Seq(Row(null, null))
+      )
+    }
+  }
+
 }