[SPARK-26021][SQL] replace minus zero with zero in Platform.putDouble/Float

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

@@ -20,7 +20,7 @@ package org.apache.spark.sql.catalyst.expressions
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.errors.attachTree
-import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, CodeGenerator, ExprCode, FalseLiteral, JavaCode}
+import org.apache.spark.sql.catalyst.expressions.codegen.{Block, CodegenContext, CodeGenerator, ExprCode, FalseLiteral, JavaCode}
 import org.apache.spark.sql.catalyst.expressions.codegen.Block._
 import org.apache.spark.sql.types._
 
@@ -56,17 +56,32 @@ case class BoundReference(ordinal: Int, dataType: DataType, nullable: Boolean)
       val javaType = JavaCode.javaType(dataType)
       val value = CodeGenerator.getValue(ctx.INPUT_ROW, dataType, ordinal.toString)
       if (nullable) {
-        ev.copy(code =
+        var codeBlock =
           code"""
              |boolean ${ev.isNull} = ${ctx.INPUT_ROW}.isNullAt($ordinal);
              |$javaType ${ev.value} = ${ev.isNull} ?
              |  ${CodeGenerator.defaultValue(dataType)} : ($value);
-           """.stripMargin)
+           """.stripMargin
+        codeBlock = codeBlock + genReplaceMinusZeroWithZeroCode(javaType.codeString, ev.value)
+        ev.copy(code = codeBlock)
       } else {
-        ev.copy(code = code"$javaType ${ev.value} = $value;", isNull = FalseLiteral)
+        var codeBlock = code"$javaType ${ev.value} = $value;"
+        codeBlock = codeBlock + genReplaceMinusZeroWithZeroCode(javaType.codeString, ev.value)
+        ev.copy(code = codeBlock, isNull = FalseLiteral)
       }
     }
   }
+
+  private def genReplaceMinusZeroWithZeroCode(javaType: String, value: String): Block = {
+    val code = s"\nif ($value == -0.0%c) $value = 0.0%c;"
+    var formattedCode = ""
+    javaType match {
+      case "double" | "java.lang.Double" => formattedCode = code.format('d', 'd')
+      case "float" | "java.lang.Float" => formattedCode = code.format('f', 'f')
+      case _ =>
+    }
+    code"$formattedCode"
+  }
 }
 
 object BindReferences extends Logging {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/codegen/GenerateUnsafeProjectionSuite.scala

@@ -68,6 +68,22 @@ class GenerateUnsafeProjectionSuite extends SparkFunSuite {
     assert(!result3.getStruct(0, 2).isNullAt(0))
     assert(!result3.getStruct(0, 3).isNullAt(0))
   }
+
+  test("SPARK-26021: Test replacing -0.0 with 0.0") {
+    val exprs =
+      BoundReference(0, DoubleType, nullable = false) ::
+      BoundReference(1, DoubleType, nullable = true) ::
+      BoundReference(2, FloatType, nullable = false) ::
+      BoundReference(3, FloatType, nullable = true) ::
+      Nil
+    val projection = GenerateUnsafeProjection.generate(exprs)
+    val result = projection.apply(InternalRow(-0.0d, Double.box(-0.0d), -0.0f, Float.box(-0.0f)))
+    // using compare since 0.0 == -0.0 is true
+    assert(result.getDouble(0).compareTo(0.0d) == 0)
+    assert(result.getDouble(1).compareTo(0.0d) == 0)
+    assert(result.getFloat(2).compareTo(0.0f) == 0)
+    assert(result.getFloat(3).compareTo(0.0f) == 0)
+  }
 }
 
 object AlwaysNull extends InternalRow {

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

@@ -723,4 +723,32 @@ class DataFrameAggregateSuite extends QueryTest with SharedSQLContext {
       "grouping expressions: [current_date(None)], value: [key: int, value: string], " +
         "type: GroupBy]"))
   }
+
+  test("SPARK-26021: Double and Float 0.0/-0.0 should be equal when grouping") {
+    val colName = "i"
+    def groupByCollect(df: DataFrame): Array[Row] = {
+      df.groupBy(colName).count().collect()
+    }
+    def assertResult[T](result: Array[Row], zero: T)(implicit ordering: Ordering[T]): Unit = {
+      assert(result.length == 1)
+      // using compare since 0.0 == -0.0 is true
+      assert(ordering.compare(result(0).getAs[T](0), zero) == 0)
+      assert(result(0).getLong(1) == 3)
+    }
+
+    spark.conf.set("spark.sql.codegen.wholeStage", "false")
+    val doubles =
+      groupByCollect(Seq(0.0d, 0.0d, -0.0d).toDF(colName))
+    val doublesBoxed =
+      groupByCollect(Seq(Double.box(0.0d), Double.box(0.0d), Double.box(-0.0d)).toDF(colName))
+    val floats =
+      groupByCollect(Seq(0.0f, -0.0f, 0.0f).toDF(colName))
+    val floatsBoxed =
+      groupByCollect(Seq(Float.box(0.0f), Float.box(-0.0f), Float.box(0.0f)).toDF(colName))
+
+    assertResult(doubles, 0.0d)
+    assertResult(doublesBoxed, 0.0d)
+    assertResult(floats, 0.0f)
+    assertResult(floatsBoxed, 0.0f)
+  }
 }