Fix

maropu · maropu · commit 1bf4f32924df · 2020-09-17T13:37:50.000+09:00
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingNumbers.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingNumbers.scala
@@ -129,10 +129,10 @@ object NormalizeFloatingNumbers extends Rule[LogicalPlan] {
       Coalesce(children.map(normalize))
 
     case _ if expr.dataType.isInstanceOf[StructType] =>
-      val fields = expr.dataType.asInstanceOf[StructType].fields.indices.map { i =>
-        normalize(GetStructField(expr, i))
+      val fields = expr.dataType.asInstanceOf[StructType].fieldNames.zipWithIndex.map {
+        case (name, i) => Seq(Literal(name), normalize(GetStructField(expr, i)))
       }
-      val struct = CreateStruct(fields)
+      val struct = CreateNamedStruct(fields.flatten.toSeq)
       KnownFloatingPointNormalized(If(IsNull(expr), Literal(null, struct.dataType), struct))
 
     case _ if expr.dataType.isInstanceOf[ArrayType] =>
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
@@ -1043,6 +1043,14 @@ class DataFrameAggregateSuite extends QueryTest
     checkAnswer(sql(queryTemplate("FIRST")), Row(1))
     checkAnswer(sql(queryTemplate("LAST")), Row(3))
   }
+
+  test("SPARK-32906: struct field names should not change after normalizing floats") {
+    val df = Seq(Tuple1(Tuple2(-0.0d, Double.NaN)), Tuple1(Tuple2(0.0d, Double.NaN))).toDF("k")
+    val aggs = df.distinct().queryExecution.sparkPlan.collect { case a: HashAggregateExec => a }
+    assert(aggs.length == 2)
+    assert(aggs.head.output.map(_.dataType.simpleString).head ===
+      aggs.last.output.map(_.dataType.simpleString).head)
+  }
 }
 
 case class B(c: Option[Double])

Original file line number	Diff line number	Diff line change
`@@ -1043,6 +1043,14 @@ class DataFrameAggregateSuite extends QueryTest`
`1043`	`1043`	`checkAnswer(sql(queryTemplate("FIRST")), Row(1))`
`1044`	`1044`	`checkAnswer(sql(queryTemplate("LAST")), Row(3))`
`1045`	`1045`	`}`
	`1046`	`+`
	`1047`	`+ test("SPARK-32906: struct field names should not change after normalizing floats") {`
	`1048`	`+ val df = Seq(Tuple1(Tuple2(-0.0d, Double.NaN)), Tuple1(Tuple2(0.0d, Double.NaN))).toDF("k")`
	`1049`	`+ val aggs = df.distinct().queryExecution.sparkPlan.collect { case a: HashAggregateExec => a }`
	`1050`	`+ assert(aggs.length == 2)`
	`1051`	`+ assert(aggs.head.output.map(_.dataType.simpleString).head ===`
	`1052`	`+ aggs.last.output.map(_.dataType.simpleString).head)`
	`1053`	`+ }`
`1046`	`1054`	`}`
`1047`	`1055`
`1048`	`1056`	`case class B(c: Option[Double])`