Fix null case.

Author: viirya

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/functions.scala

@@ -592,31 +592,36 @@ case class Corr(
   }
 
   override def update(buffer: MutableRow, input: InternalRow): Unit = {
-    val x = left.eval(input).asInstanceOf[Double]
-    val y = right.eval(input).asInstanceOf[Double]
-
-    var xAvg = buffer.getDouble(mutableAggBufferOffset)
-    var yAvg = buffer.getDouble(mutableAggBufferOffsetPlus1)
-    var Ck = buffer.getDouble(mutableAggBufferOffsetPlus2)
-    var MkX = buffer.getDouble(mutableAggBufferOffsetPlus3)
-    var MkY = buffer.getDouble(mutableAggBufferOffsetPlus4)
-    var count = buffer.getLong(mutableAggBufferOffsetPlus5)
-
-    val deltaX = x - xAvg
-    val deltaY = y - yAvg
-    count += 1
-    xAvg += deltaX / count
-    yAvg += deltaY / count
-    Ck += deltaX * (y - yAvg)
-    MkX += deltaX * (x - xAvg)
-    MkY += deltaY * (y - yAvg)
-
-    buffer.setDouble(mutableAggBufferOffset, xAvg)
-    buffer.setDouble(mutableAggBufferOffsetPlus1, yAvg)
-    buffer.setDouble(mutableAggBufferOffsetPlus2, Ck)
-    buffer.setDouble(mutableAggBufferOffsetPlus3, MkX)
-    buffer.setDouble(mutableAggBufferOffsetPlus4, MkY)
-    buffer.setLong(mutableAggBufferOffsetPlus5, count)
+    val leftEval = left.eval(input)
+    val rightEval = right.eval(input)
+
+    if (leftEval != null && rightEval != null) {
+      val x = leftEval.asInstanceOf[Double]
+      val y = rightEval.asInstanceOf[Double]
+
+      var xAvg = buffer.getDouble(mutableAggBufferOffset)
+      var yAvg = buffer.getDouble(mutableAggBufferOffsetPlus1)
+      var Ck = buffer.getDouble(mutableAggBufferOffsetPlus2)
+      var MkX = buffer.getDouble(mutableAggBufferOffsetPlus3)
+      var MkY = buffer.getDouble(mutableAggBufferOffsetPlus4)
+      var count = buffer.getLong(mutableAggBufferOffsetPlus5)
+
+      val deltaX = x - xAvg
+      val deltaY = y - yAvg
+      count += 1
+      xAvg += deltaX / count
+      yAvg += deltaY / count
+      Ck += deltaX * (y - yAvg)
+      MkX += deltaX * (x - xAvg)
+      MkY += deltaY * (y - yAvg)
+
+      buffer.setDouble(mutableAggBufferOffset, xAvg)
+      buffer.setDouble(mutableAggBufferOffsetPlus1, yAvg)
+      buffer.setDouble(mutableAggBufferOffsetPlus2, Ck)
+      buffer.setDouble(mutableAggBufferOffsetPlus3, MkX)
+      buffer.setDouble(mutableAggBufferOffsetPlus4, MkY)
+      buffer.setLong(mutableAggBufferOffsetPlus5, count)
+    }
   }
 
   // Merge counters from other partitions. Formula can be found at:

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala

@@ -589,6 +589,13 @@ abstract class AggregationQuerySuite extends QueryTest with SQLTestUtils with Te
     val corr6 = df4.groupBy().agg(corr("a", "c")).collect()(0).getDouble(0)
     assert(math.abs(corr6 + 1.0) < 1e-12)
 
+    val df5 = Seq[(Integer, Integer)](
+      (1, null),
+      (null, -60)).toDF("a", "b")
+
+    val corr7 = df5.groupBy().agg(corr("a", "b")).collect()(0)
+    assert(corr7 == Row(null))
+
     withSQLConf(SQLConf.USE_SQL_AGGREGATE2.key -> "false") {
       val errorMessage = intercept[SparkException] {
         val df = Seq.tabulate(10)(i => (1.0 * i, 2.0 * i, i * -1.0)).toDF("a", "b", "c")