From e6927c55eae2736af670ed2a8433d540d5c2e391 Mon Sep 17 00:00:00 2001
From: Dongjoon Hyun <dongjoon@apache.org>
Date: Wed, 17 Aug 2016 13:41:04 -0700
Subject: [PATCH 1/2] [SPARK-17098][SQL] Fix `NullPropagation` optimizer to
 handle `COUNT(NULL)` correctly

---
 .../org/apache/spark/sql/catalyst/optimizer/Optimizer.scala  | 2 ++
 .../src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala  | 5 +++++
 2 files changed, 7 insertions(+)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
index ce57f05868fe1..9a0ff8a9b3211 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
@@ -619,6 +619,8 @@ object NullPropagation extends Rule[LogicalPlan] {
 
   def apply(plan: LogicalPlan): LogicalPlan = plan transform {
     case q: LogicalPlan => q transformExpressionsUp {
+      case e @ WindowExpression(Cast(Literal(0L, _), _), _) =>
+        Cast(Literal(0L), e.dataType)
       case e @ AggregateExpression(Count(exprs), _, _, _) if !exprs.exists(nonNullLiteral) =>
         Cast(Literal(0L), e.dataType)
       case e @ IsNull(c) if !c.nullable => Literal.create(false, BooleanType)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
index eac266cba55b8..93629acc0df30 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
@@ -2647,6 +2647,11 @@ class SQLQuerySuite extends QueryTest with SharedSQLContext {
     }
   }
 
+  test("SPARK-17098 COUNT(NULL) on window should be 0") {
+    checkAnswer(sql("SELECT COUNT(NULL) OVER ()"), Row(0) :: Nil)
+    checkAnswer(sql("SELECT COUNT(1 + NULL) OVER ()"), Row(0) :: Nil)
+  }
+
   test("SPARK-16674: field names containing dots for both fields and partitioned fields") {
     withTempPath { path =>
       val data = (1 to 10).map(i => (i, s"data-$i", i % 2, if ((i % 2) == 0) "a" else "b"))

From 25aa91de8d2d41914b1d6dab85e0f89631c71dfe Mon Sep 17 00:00:00 2001
From: Dongjoon Hyun <dongjoon@apache.org>
Date: Thu, 18 Aug 2016 07:47:48 -0700
Subject: [PATCH 2/2] Move testcase to `null-propagation.sql`.

---
 .../sql-tests/inputs/null-propagation.sql     |  9 +++++
 .../results/null-propagation.sql.out          | 38 +++++++++++++++++++
 .../org/apache/spark/sql/SQLQuerySuite.scala  |  5 ---
 3 files changed, 47 insertions(+), 5 deletions(-)
 create mode 100644 sql/core/src/test/resources/sql-tests/inputs/null-propagation.sql
 create mode 100644 sql/core/src/test/resources/sql-tests/results/null-propagation.sql.out

diff --git a/sql/core/src/test/resources/sql-tests/inputs/null-propagation.sql b/sql/core/src/test/resources/sql-tests/inputs/null-propagation.sql
new file mode 100644
index 0000000000000..66549da7971d3
--- /dev/null
+++ b/sql/core/src/test/resources/sql-tests/inputs/null-propagation.sql
@@ -0,0 +1,9 @@
+
+-- count(null) should be 0
+SELECT COUNT(NULL) FROM VALUES 1, 2, 3;
+SELECT COUNT(1 + NULL) FROM VALUES 1, 2, 3;
+
+-- count(null) on window should be 0
+SELECT COUNT(NULL) OVER () FROM VALUES 1, 2, 3;
+SELECT COUNT(1 + NULL) OVER () FROM VALUES 1, 2, 3;
+
diff --git a/sql/core/src/test/resources/sql-tests/results/null-propagation.sql.out b/sql/core/src/test/resources/sql-tests/results/null-propagation.sql.out
new file mode 100644
index 0000000000000..ed3a651aa6614
--- /dev/null
+++ b/sql/core/src/test/resources/sql-tests/results/null-propagation.sql.out
@@ -0,0 +1,38 @@
+-- Automatically generated by SQLQueryTestSuite
+-- Number of queries: 4
+
+
+-- !query 0
+SELECT COUNT(NULL) FROM VALUES 1, 2, 3
+-- !query 0 schema
+struct<count(NULL):bigint>
+-- !query 0 output
+0
+
+
+-- !query 1
+SELECT COUNT(1 + NULL) FROM VALUES 1, 2, 3
+-- !query 1 schema
+struct<count((1 + CAST(NULL AS INT))):bigint>
+-- !query 1 output
+0
+
+
+-- !query 2
+SELECT COUNT(NULL) OVER () FROM VALUES 1, 2, 3
+-- !query 2 schema
+struct<count(NULL) OVER (ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING):bigint>
+-- !query 2 output
+0
+0
+0
+
+
+-- !query 3
+SELECT COUNT(1 + NULL) OVER () FROM VALUES 1, 2, 3
+-- !query 3 schema
+struct<count((1 + CAST(NULL AS INT))) OVER (ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING):bigint>
+-- !query 3 output
+0
+0
+0
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
index 93629acc0df30..eac266cba55b8 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
@@ -2647,11 +2647,6 @@ class SQLQuerySuite extends QueryTest with SharedSQLContext {
     }
   }
 
-  test("SPARK-17098 COUNT(NULL) on window should be 0") {
-    checkAnswer(sql("SELECT COUNT(NULL) OVER ()"), Row(0) :: Nil)
-    checkAnswer(sql("SELECT COUNT(1 + NULL) OVER ()"), Row(0) :: Nil)
-  }
-
   test("SPARK-16674: field names containing dots for both fields and partitioned fields") {
     withTempPath { path =>
       val data = (1 to 10).map(i => (i, s"data-$i", i % 2, if ((i % 2) == 0) "a" else "b"))