[SPARK-32906][SQL] Struct field names should not change after normalizing floats

[maropu]

What changes were proposed in this pull request?

This PR intends to fix a minor bug when normalizing floats for struct types;

scala> import org.apache.spark.sql.execution.aggregate.HashAggregateExec
scala> val df = Seq(Tuple1(Tuple1(-0.0d)), Tuple1(Tuple1(0.0d))).toDF("k")
scala> val agg = df.distinct()
scala> agg.explain()
== Physical Plan ==
*(2) HashAggregate(keys=[k#40], functions=[])
+- Exchange hashpartitioning(k#40, 200), true, [id=#62]
   +- *(1) HashAggregate(keys=[knownfloatingpointnormalized(if (isnull(k#40)) null else named_struct(col1, knownfloatingpointnormalized(normalizenanandzero(k#40._1)))) AS k#40], functions=[])
      +- *(1) LocalTableScan [k#40]

scala> val aggOutput = agg.queryExecution.sparkPlan.collect { case a: HashAggregateExec => a.output.head }
scala> aggOutput.foreach { attr => println(attr.prettyJson) }
### Final Aggregate ###
[ {
  "class" : "org.apache.spark.sql.catalyst.expressions.AttributeReference",
  "num-children" : 0,
  "name" : "k",
  "dataType" : {
    "type" : "struct",
    "fields" : [ {
      "name" : "_1",
                ^^^
      "type" : "double",
      "nullable" : false,
      "metadata" : { }
    } ]
  },
  "nullable" : true,
  "metadata" : { },
  "exprId" : {
    "product-class" : "org.apache.spark.sql.catalyst.expressions.ExprId",
    "id" : 40,
    "jvmId" : "a824e83f-933e-4b85-a1ff-577b5a0e2366"
  },
  "qualifier" : [ ]
} ]

### Partial Aggregate ###
[ {
  "class" : "org.apache.spark.sql.catalyst.expressions.AttributeReference",
  "num-children" : 0,
  "name" : "k",
  "dataType" : {
    "type" : "struct",
    "fields" : [ {
      "name" : "col1",
                ^^^^
      "type" : "double",
      "nullable" : true,
      "metadata" : { }
    } ]
  },
  "nullable" : true,
  "metadata" : { },
  "exprId" : {
    "product-class" : "org.apache.spark.sql.catalyst.expressions.ExprId",
    "id" : 40,
    "jvmId" : "a824e83f-933e-4b85-a1ff-577b5a0e2366"
  },
  "qualifier" : [ ]
} ]

Why are the changes needed?

bugfix.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Added tests.

[SparkQA]

Test build #128795 has finished for PR 29780 at commit 1bf4f32.

• This patch fails due to an unknown error code, -9.
• This patch merges cleanly.
• This patch adds no public classes.

[maropu]

GA passed. cc: @cloud-fan @viirya

[viirya]

thanks for catching and fixing this.

[HyukjinKwon]

Looks logically fine but just to confirm that I understood correctly, it's just a cleanup fix to match the column names, right? and doesn't affect the final output?

[cloud-fan]

Yea I think so. CreateNamedStruct allows you to specify the name for each field.

[maropu]

Yeah, you're right and this is not an user-facing bug. When writing code to check paln integrity for SparkPlans, I found this issue; canonicalized attributes between final/partial HaashAggregates were not equal incorrectly because of this issue (schema mismatch).

[viirya]

Thanks all! Merging to master/3.0.

[dongjoon-hyun]

Thank you, @maropu and @viirya and all!