Fix Spark 4.2 collect_set float buffer conversion for mixed aggs (NVIDIA#15455)

Fixes NVIDIA#15454.

### Description

- Convert CollectSet float/double aggregation buffers between GPU
logical values and Spark 4.2 CPU normalized bit-pattern keys during
mixed hashAgg stages, so NaN and signed-zero uniqueness matches pure
CPU.
- Add `collectSetCpuBufferElementType` shim (bit-keyed on Spark 4.2,
identity earlier) and CollectSet-specific GPU↔CPU buffer converters used
by `GpuOverrides`.
- Validate with Spark 4.2.0 / Scala 2.13 and `DATAGEN_SEED=1785353212`:
`352 passed` including the previously failing
`test_hash_groupby_collect_partial_replace_fallback` /
`test_hash_groupby_collect_partial_replace_with_distinct_fallback` Float
cases; also `mvn -f scala2.13/pom.xml -Dbuildver=420
-Dcuda.version=cuda13 -DskipTests -pl sql-plugin verify`.

### Checklists

Documentation
- [ ] Updated for new or modified user-facing features or behaviors
- [x] No user-facing change

Testing
- [ ] Added or modified tests to cover new code paths
- [x] Covered by existing tests
(Please provide the names of the existing tests in the PR description.)
- [ ] Not required

Performance
- [ ] Tests ran and results are added in the PR description
- [ ] Issue filed with a link in the PR description
- [x] Not required

---------

Signed-off-by: Firestarman <firestarmanllc@gmail.com>

Author: firestarman

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala

@@ -3859,11 +3859,12 @@ object GpuOverrides extends Logging {
         }
 
         override def createCpuToGpuBufferConverter(): CpuToGpuAggregateBufferConverter =
-          new CpuToGpuCollectBufferConverter(c.child.dataType,
+          new CpuToGpuCollectSetBufferConverter(c.child.dataType,
             !TypeUtilsShims.collectSetIgnoreNulls(c))
 
         override def createGpuToCpuBufferConverter(): GpuToCpuAggregateBufferConverter =
-          new GpuToCpuCollectBufferConverter()
+          new GpuToCpuCollectSetBufferConverter(c.child.dataType,
+            !TypeUtilsShims.collectSetIgnoreNulls(c))
 
         override val supportBufferConversion: Boolean = true
 

sql-plugin/src/main/scala/org/apache/spark/sql/rapids/aggregate/aggregateFunctions.scala

@@ -2079,6 +2079,163 @@ case class GpuToCpuCollectBufferTransition(
   }
 }
 
+/**
+ * CollectSet buffer converters.
+ *
+ * Spark 4.2 changed CollectSet's CPU agg buffer for FloatType/DoubleType to store normalized
+ * bit patterns (IntegerType/LongType) so HashSet can treat NaNs and signed zeros as equal.
+ * GPU CollectSet still stores the logical float/double values, so mixed CPU/GPU aggregation
+ * stages must convert between the two buffer layouts.
+ */
+class CpuToGpuCollectSetBufferConverter(
+    elementType: DataType,
+    containsNull: Boolean = false) extends CpuToGpuAggregateBufferConverter {
+  def createExpression(child: Expression): CpuToGpuBufferTransition = {
+    CpuToGpuCollectSetBufferTransition(child, elementType, containsNull)
+  }
+}
+
+case class CpuToGpuCollectSetBufferTransition(
+    override val child: Expression,
+    private val elementType: DataType,
+    private val containsNull: Boolean) extends CpuToGpuBufferTransition {
+
+  private lazy val row = new UnsafeRow(1)
+  private lazy val cpuElementType: DataType =
+    TypeUtilsShims.collectSetCpuBufferElementType(elementType)
+
+  override def dataType: DataType = ArrayType(elementType, containsNull)
+
+  override protected def nullSafeEval(input: Any): ArrayData = {
+    val bytes = input.asInstanceOf[Array[Byte]]
+    row.pointTo(bytes, bytes.length)
+    val cpuArray = row.getArray(0)
+    if (cpuElementType == elementType) {
+      cpuArray.copy()
+    } else {
+      CollectSetBufferConversions.cpuBitsToGpuValues(cpuArray, elementType, containsNull)
+    }
+  }
+}
+
+class GpuToCpuCollectSetBufferConverter(
+    elementType: DataType,
+    containsNull: Boolean = false) extends GpuToCpuAggregateBufferConverter {
+  def createExpression(child: Expression): GpuToCpuBufferTransition = {
+    GpuToCpuCollectSetBufferTransition(child, elementType, containsNull)
+  }
+}
+
+case class GpuToCpuCollectSetBufferTransition(
+    override val child: Expression,
+    private val elementType: DataType,
+    private val containsNull: Boolean) extends GpuToCpuBufferTransition {
+
+  private lazy val cpuElementType: DataType =
+    TypeUtilsShims.collectSetCpuBufferElementType(elementType)
+  private lazy val cpuBufferType: DataType = ArrayType(cpuElementType, containsNull)
+  private lazy val projection = UnsafeProjection.create(Array[DataType](cpuBufferType))
+
+  override protected def nullSafeEval(input: Any): Array[Byte] = {
+    val arrayData = input.asInstanceOf[ArrayData]
+    val cpuArray = if (cpuElementType == elementType) {
+      arrayData
+    } else {
+      CollectSetBufferConversions.gpuValuesToCpuBits(arrayData, elementType, containsNull)
+    }
+    projection.apply(InternalRow.apply(cpuArray)).getBytes
+  }
+}
+
+object CollectSetBufferConversions {
+  // Matches Spark's NormalizeFloatingNumbers.FLOAT_NORMALIZER / DOUBLE_NORMALIZER.
+  private def normalizeFloat(f: Float): Float = {
+    if (f.isNaN) {
+      Float.NaN
+    } else if (f == -0.0f) {
+      0.0f
+    } else {
+      f
+    }
+  }
+
+  private def normalizeDouble(d: Double): Double = {
+    if (d.isNaN) {
+      Double.NaN
+    } else if (d == -0.0d) {
+      0.0d
+    } else {
+      d
+    }
+  }
+
+  def gpuValuesToCpuBits(
+      arrayData: ArrayData,
+      elementType: DataType,
+      containsNull: Boolean): ArrayData = {
+    val n = arrayData.numElements()
+    val out = new Array[Any](n)
+    var i = 0
+    elementType match {
+      case FloatType =>
+        while (i < n) {
+          if (containsNull && arrayData.isNullAt(i)) {
+            out(i) = null
+          } else {
+            out(i) = java.lang.Float.floatToIntBits(normalizeFloat(arrayData.getFloat(i)))
+          }
+          i += 1
+        }
+      case DoubleType =>
+        while (i < n) {
+          if (containsNull && arrayData.isNullAt(i)) {
+            out(i) = null
+          } else {
+            out(i) = java.lang.Double.doubleToLongBits(normalizeDouble(arrayData.getDouble(i)))
+          }
+          i += 1
+        }
+      case other =>
+        throw new IllegalStateException(
+          s"Unexpected CollectSet GPU-to-CPU buffer conversion for $other")
+    }
+    new GenericArrayData(out)
+  }
+
+  def cpuBitsToGpuValues(
+      arrayData: ArrayData,
+      elementType: DataType,
+      containsNull: Boolean): ArrayData = {
+    val n = arrayData.numElements()
+    val out = new Array[Any](n)
+    var i = 0
+    elementType match {
+      case FloatType =>
+        while (i < n) {
+          if (containsNull && arrayData.isNullAt(i)) {
+            out(i) = null
+          } else {
+            out(i) = java.lang.Float.intBitsToFloat(arrayData.getInt(i))
+          }
+          i += 1
+        }
+      case DoubleType =>
+        while (i < n) {
+          if (containsNull && arrayData.isNullAt(i)) {
+            out(i) = null
+          } else {
+            out(i) = java.lang.Double.longBitsToDouble(arrayData.getLong(i))
+          }
+          i += 1
+        }
+      case other =>
+        throw new IllegalStateException(
+          s"Unexpected CollectSet CPU-to-GPU buffer conversion for $other")
+    }
+    new GenericArrayData(out)
+  }
+}
+
 /**
  * Base class for overriding standard deviation and variance aggregations.
  * This is also a GPU-based implementation of 'CentralMomentAgg' aggregation class in Spark with

sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/TypeUtilsShims.scala

@@ -27,6 +27,7 @@ import ai.rapids.cudf.NaNEquality
 
 import org.apache.spark.sql.catalyst.expressions.aggregate.{CollectList, CollectSet}
 import org.apache.spark.sql.catalyst.util.TypeUtils
+import org.apache.spark.sql.types.DataType
 
 /**
  * Reimplement the function `checkForNumericExpr` which has been removed since
@@ -37,6 +38,9 @@ object TypeUtilsShims {
 
   val collectSetFloatNanEquality: NaNEquality = NaNEquality.UNEQUAL
 
+  // Pre-Spark 4.2 CollectSet stores child values directly in the agg buffer.
+  def collectSetCpuBufferElementType(childType: DataType): DataType = childType
+
   def collectListIgnoreNulls(_collectList: CollectList): Boolean = true
 
   def collectSetIgnoreNulls(_collectSet: CollectSet): Boolean = true

sql-plugin/src/main/spark330db/scala/com/nvidia/spark/rapids/shims/TypeUtilsShims.scala

@@ -67,6 +67,9 @@ object TypeUtilsShims {
 
   val collectSetFloatNanEquality: NaNEquality = NaNEquality.UNEQUAL
 
+  // Pre-Spark 4.2 CollectSet stores child values directly in the agg buffer.
+  def collectSetCpuBufferElementType(childType: DataType): DataType = childType
+
   def collectListIgnoreNulls(_collectList: CollectList): Boolean = true
 
   def collectSetIgnoreNulls(_collectSet: CollectSet): Boolean = true

sql-plugin/src/main/spark420/scala/com/nvidia/spark/rapids/shims/TypeUtilsShims.scala

@@ -22,7 +22,8 @@ import ai.rapids.cudf.NaNEquality
 
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
 import org.apache.spark.sql.catalyst.expressions.aggregate.{CollectList, CollectSet}
-import org.apache.spark.sql.types.{DataType, NullType, NumericType}
+import org.apache.spark.sql.types.{DataType, DoubleType, FloatType, IntegerType, LongType, NullType,
+  NumericType}
 
 /**
  * Reimplement the function `checkForNumericExpr` which has been removed since
@@ -40,6 +41,13 @@ object TypeUtilsShims {
   // Spark 4.2 stores collect_set buffers in a way that treats all NaN values as one set entry.
   val collectSetFloatNanEquality: NaNEquality = NaNEquality.ALL_EQUAL
 
+  // Spark 4.2 CollectSet keys float/double by normalized bit patterns in the agg buffer.
+  def collectSetCpuBufferElementType(childType: DataType): DataType = childType match {
+    case FloatType => IntegerType
+    case DoubleType => LongType
+    case other => other
+  }
+
   def collectListIgnoreNulls(collectList: CollectList): Boolean = collectList.ignoreNulls
 
   def collectSetIgnoreNulls(collectSet: CollectSet): Boolean = collectSet.ignoreNulls