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

common/unsafe/src/main/java/org/apache/spark/unsafe/Platform.java

@@ -120,6 +120,9 @@ public static float getFloat(Object object, long offset) {
   }
 
   public static void putFloat(Object object, long offset, float value) {
+    if(value == -0.0f) {
+      value = 0.0f;
+    }
     _UNSAFE.putFloat(object, offset, value);
   }
 
@@ -128,6 +131,9 @@ public static double getDouble(Object object, long offset) {
   }
 
   public static void putDouble(Object object, long offset, double value) {
+    if(value == -0.0d) {
+      value = 0.0d;
+    }
     _UNSAFE.putDouble(object, offset, value);
   }
 

common/unsafe/src/test/java/org/apache/spark/unsafe/PlatformUtilSuite.java

@@ -24,6 +24,8 @@
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.nio.ByteBuffer;
+
 public class PlatformUtilSuite {
 
   @Test
@@ -157,4 +159,15 @@ public void heapMemoryReuse() {
     Assert.assertEquals(onheap4.size(), 1024 * 1024 + 7);
     Assert.assertEquals(obj3, onheap4.getBaseObject());
   }
+
+  @Test
+  // SPARK-26021
+  public void writeMinusZeroIsReplacedWithZero() {
+    byte[] doubleBytes = new byte[Double.BYTES];
+    byte[] floatBytes = new byte[Float.BYTES];
+    Platform.putDouble(doubleBytes, Platform.BYTE_ARRAY_OFFSET, -0.0d);
+    Platform.putFloat(floatBytes, Platform.BYTE_ARRAY_OFFSET, -0.0f);
+    Assert.assertEquals(0, Double.compare(0.0d, ByteBuffer.wrap(doubleBytes).getDouble()));
+    Assert.assertEquals(0, Float.compare(0.0f, ByteBuffer.wrap(floatBytes).getFloat()));
+  }
 }

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)
+  }
 }