[SPARK-36702][SQL][FOLLOWUP] ArrayUnion handle duplicated Double.NaN and Float.NaN

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

@@ -3578,6 +3578,7 @@ case class ArrayUnion(left: Expression, right: Expression) extends ArrayBinaryLi
         val arrayBuffer = new scala.collection.mutable.ArrayBuffer[Any]
         val hs = new SQLOpenHashSet[Any]()
         val isNaN = SQLOpenHashSet.isNaN(elementType)
+        val valueNaN = SQLOpenHashSet.valueNaN(elementType)
         Seq(array1, array2).foreach { array =>
           var i = 0
           while (i < array.numElements()) {
@@ -3590,7 +3591,7 @@ case class ArrayUnion(left: Expression, right: Expression) extends ArrayBinaryLi
               val elem = array.get(i, elementType)
               if (isNaN(elem)) {
                 if (!hs.containsNaN) {
-                  arrayBuffer += elem
+                  arrayBuffer += valueNaN
                   hs.addNaN
                 }
               } else {
@@ -3688,16 +3689,18 @@ case class ArrayUnion(left: Expression, right: Expression) extends ArrayBinaryLi
 
         def withNaNCheck(body: String): String = {
           (elementType match {
-            case DoubleType => Some(s"java.lang.Double.isNaN((double)$value)")
-            case FloatType => Some(s"java.lang.Float.isNaN((float)$value)")
+            case DoubleType =>
+              Some((s"java.lang.Double.isNaN((double)$value)", "java.lang.Double.NaN"))
+            case FloatType =>
+              Some((s"java.lang.Float.isNaN((float)$value)", "java.lang.Float.NaN"))
             case _ => None
-          }).map { isNaN =>
+          }).map { case (isNaN, valueNaN) =>
             s"""
                |if ($isNaN) {
                |  if (!$hashSet.containsNaN()) {
                |     $size++;
                |     $hashSet.addNaN();
-               |     $builder.$$plus$$eq($value);
+               |     $builder.$$plus$$eq($valueNaN);
                |  }
                |} else {
                |  $body

sql/catalyst/src/main/scala/org/apache/spark/sql/util/SQLOpenHashSet.scala

@@ -69,4 +69,12 @@ object SQLOpenHashSet {
       case _ => (_: Any) => false
     }
   }
+
+  def valueNaN(dataType: DataType): Any = {
+    dataType match {
+      case DoubleType => java.lang.Double.NaN
+      case FloatType => java.lang.Float.NaN
+      case _ => null
+    }
+  }
 }