fix: normalize floating-point values in native collect_set

[peterxcli]

Which issue does this PR close?

Closes #4966.

Rationale for this change

Spark 4.2 changed collect_set to normalize NaN values and signed zero before deduplication. Comet passed the raw input to the native partial aggregate, whose set distinguishes floating-point bit patterns, so its result could differ from Spark for NaN and -0.0 values.

What changes are included in this PR?

• Reuse Spark's recursive floating-point normalizer for Spark 4.2+ collect_set partial aggregation.
• Keep final aggregation state merging unchanged and preserve the strict floating-point fallback on older Spark versions.
• Add Spark-version-specific SQL coverage for scalar, array, and struct values across partial aggregation buffers.

How are these changes tested?

• make core
• ./mvnw test -Pspark-4.2 -Dtest=none -Dsuites='org.apache.comet.CometSqlFileTestSuite collect_set' -Dscalastyle.skip=true (4 tests passed)
• ./mvnw test -Pspark-4.1 -Dtest=none -Dsuites='org.apache.comet.CometSqlFileTestSuite collect_set_floating_fallback' -Dscalastyle.skip=true (1 test passed)

[andygrove]

Thanks for taking this on. Reusing Spark's own NormalizeFloatingNumbers.normalize rather than reimplementing the normalization in the native set is the right call, and gating on Partial mirrors Spark, where normalization only happens in update() via convertToBufferElement.

I checked a couple of things that could have gone wrong and they look fine. The nested paths are already reachable, since CometKnownFloatingPointNormalized in contraintExpressions.scala handles both the scalar NormalizeNaNAndZero case and the nested ArrayTransform / CreateNamedStruct cases. The mixed Comet-partial / JVM-final buffer hazard is already guarded by hasNativeArrayBufferAgg plus the COMET_UNSAFE_PARTIAL tagging in CometExecRule, so there is no new interop concern here. CI is fully green including all four Spark 4.2 shards.

A few things I would like to see addressed.

1. aggregates.scala:855 - version-gate the incompatible reason.

On Spark 4.2+ getSupportLevel now returns Compatible(), but getIncompatibleReasons() still returns a bullet unconditionally. GenerateDocs writes those bullets into the per-Spark-version compatibility page, so the 4.2 aggregate page would show collect_set carrying an incompatibility that this PR just removed. Could this mirror the support level?

override def getIncompatibleReasons(): Seq[String] =
  if (isSpark42Plus) Nil else Seq(...)

That would also let the wording drop the "Before Spark 4.2" qualifier, which reads a little oddly on a page that is already version-specific.

2. aggregates.scala:889 - should Complete normalize too?

Spark normalizes inside convertToBufferElement, which is reached from update(), and that runs in both Partial and Complete mode. Comet rejects Complete today in operators.scala:1737, so this is not reachable, but I think it is worth writing case Partial | Complete now with a one-line comment. Spark 4.3 adds spark.sql.execution.bypassPartialAggregation, which plans exactly a single Complete-mode aggregate in AggUtils.planAggregateWithoutDistinct, and CometWindowExec already asserts Complete for window aggregates. If either path gains collect_set support later, the current gate silently returns wrong results rather than falling back.

3. collect_set_normalization.sql - assert values, not just sizes.

Most queries here check size(collect_set(...)). That confirms the dedup count but not which representative survived, which is the actual thing SPARK-57298 changed. Spark's own test asserts values (Row(Seq(0.0d, 1.0d, Double.NaN))). A size of 3 for the mixed group would pass even if Comet kept a non-canonical NaN payload. Could these use sort_array(collect_set(...)) so the framework compares the values against Spark?

4. collect_set.sql - float and double coverage is now missing on Spark 4.2+.

The removed cs_src_float / cs_src_double blocks covered NULL, Infinity, -Infinity and ordinary values, and they ran under that file's ConfigMatrix: parquet.enable.dictionary=false,true sweep. They now only live in collect_set_floating_fallback.sql, which is MaxSparkVersion: 4.1 and expect_fallback. So on 4.2+, where these cases finally execute natively, none of them are exercised, and the dictionary sweep is gone. Would it work to bring those two tables into collect_set_normalization.sql as native query cases with sort_array, so 4.2+ gets strictly more coverage than before rather than less?

5. collect_set_normalization.sql - nulls and deeper nesting on the nested path.

NormalizeFloatingNumbers.normalize takes visibly different branches depending on shape. A nullable struct becomes KnownFloatingPointNormalized(If(IsNull(expr), Literal(null, structType), CreateNamedStruct(...))), and an array becomes ArrayTransform with a lambda. cs_norm_nested has no NULL struct row, no NULL array row, no array containing a NULL element, and no double nesting such as array<struct<v:double>> or struct<a:array<double>>. Since a serde gap in any of those produces a silent fallback rather than a wrong answer, query cases are what actually prove the native path stays native. Would you mind adding a few rows?

6. collect_set_normalization.sql - add the global aggregate case.

Both repartitioned queries have a GROUP BY. Spark's test uses the no-GROUP BY form, Seq(Double.NaN, Double.NaN, 0.0d, -0.0d, 1.0d).toDF("v").repartition(3) then sort_array(collect_set(v)), which goes through a different aggregate shape. The grp = 'zero' query here has no GROUP BY but also no repartition, so it likely stays in one partition and never merges across partial buffers. Adding the repartitioned global variant would close that.

7. Confirm the Spark SQL test is unblocked.

The issue says DataFrameAggregateSuite "SPARK-57298: collect_set normalizes NaN and -0.0 for floating-point types" is marked IgnoreComet in dev/diffs/4.2.0.diff. That file does not exist on this branch (only 3.4.3, 3.5.8, 4.0.2, 4.1.2), so there may be nothing to remove. Could you confirm? The fix is only proven end to end once that test runs against Comet, and there is a second one to account for, "SPARK-57298: collect_set normalizes NaN and -0.0 nested in complex types". If the 4.2 diff is landing separately, a tracking issue linked from here would keep it from getting lost.

[andygrove]

A follow-up pass on top of my review above. I went through the Spark 4.2 source and the native side, and most of what I chased came back clean, so let me record that first and then raise one thing and answer my own point 7.

Things I checked that turned out correct, so nobody needs to re-verify them:

• Struct field names survive normalization. The struct branch of normalize builds its replacement with fieldNames.zipWithIndex and an explicit Literal(name), so struct<v:double> stays struct<v:double>. The serializeDataType(expr.dataType) computed from the original child still matches what native produces.
• Spark 4.2 changed CollectSet.bufferElementType to LongType / IntegerType for double / float, since convertToBufferElement now keys on doubleToLongBits. That does not leak into Comet, because adjustOutputForNativeState in operators.scala already rewrites the CollectSet buffer attribute to ArrayType(cs.children.head.dataType). Together with the hasNativeArrayBufferAgg guard forbidding mixed execution, there is no cross-engine buffer hazard.
• normalize_float in native/spark-expr/src/math_funcs/internal/normalize_nan.rs maps any NaN to T::nan() and -0.0 to 0.0, matching Spark's DOUBLE_NORMALIZER and FLOAT_NORMALIZER exactly. A non-canonical NaN payload will not slip through.
• The nullable-struct branch of normalize emits Literal(null, structType), and CometLiteral.getSupportLevel sets allowComplex = expr.value == null, so that literal serializes and the nested path does not silently fall back.
• normalize does have a TransformValues branch for maps, but it is unreachable here, because CollectSet.checkInputDataTypes rejects any type containing MapType.
• There is no cost for non-floating-point children, since needNormalize short-circuits and normalize returns the expression untouched.

8. aggregates.scala:855 - the pre-4.2 reason documents one divergence when there are two.

This builds on point 1. Since that string is being rewritten anyway, it is worth making it complete. On 3.4 through 4.1 there are two differences and they point in opposite directions.

Comet's native set is HashSet<ScalarValue> by way of DistinctArrayAggAccumulator, and ScalarValue compares and hashes Float32 and Float64 by to_bits(), so -0.0 and 0.0 stay as two entries. Spark before 4.2 buffers into mutable.HashSet[Any] of boxed doubles where Scala's == is IEEE, so -0.0 == 0.0 collapses to one entry while NaN == NaN is false and the NaNs are kept apart. That gives:

case	Spark pre-4.2	Comet pre-4.2
two NaNs	2 entries	1 entry
-0.0 and 0.0	1 entry	2 entries

Only the NaN row is documented today. spark.comet.exec.strictFloatingPoint defaults to false, so both are live by default on those versions, and this bullet is the only warning a user gets. The config's own doc already uses "comparing or sorting -0.0 and 0.0" as its headline example, which makes the omission read as accidental. Could signed zero be added here, and to the what string passed to strictFloatingPointReason on line 872, which carries the same wording?

Related: the 'd' group in the cs_src_float and cs_src_double tables this PR moves into collect_set_floating_fallback.sql is 0.0, -0.0, 1.0, NULL, which is exactly the signed-zero case, and in that file it only asserts fallback. So the reason string is the only record of that behavior, which is a further argument for getting its wording right.

9. Answering my own point 7 so you do not have to.

There is no dev/diffs/4.2*.diff in the repo. dev/diffs/ holds 3.4.3, 3.5.8, 4.0.2 and 4.1.2, and ci.yml wires spark_sql_test_reusable.yml for those same four versions, so there is no Spark SQL Tests job for 4.2 at all. Nothing to un-ignore.

The consequence matters. The SQL file tests in this PR are the only thing verifying this fix, so points 3 through 6 are carrying all of the weight. Could you file a tracking issue for enabling the two DataFrameAggregateSuite tests, "SPARK-57298: collect_set normalizes NaN and -0.0 for floating-point types" and "SPARK-57298: collect_set normalizes NaN and -0.0 nested in complex types", once a 4.2 diff lands, and link it here?

Concrete vectors for points 3, 5 and 6.

Spark's own two tests give you most of what those points ask for:

-- value-asserting scalar cases
SELECT collect_set(v) FROM VALUES (double('NaN')), (double('NaN')) AS t(v)     -- Seq(NaN)
SELECT collect_set(v) FROM VALUES (-0.0D), (0.0D) AS t(v)                      -- Seq(0.0)
SELECT collect_set(v) FROM VALUES (float(-0.0)), (float(0.0)) AS t(v)          -- Seq(0.0f)
-- Seq(NaN, NaN, 0.0, -0.0, 1.0) repartitioned to 3, then sort_array           -- Seq(0.0, 1.0, NaN)

-- value-asserting nested cases
SELECT collect_set(named_struct('a', v)) FROM VALUES (-0.0D), (0.0D) AS t(v)   -- Seq(Row(0.0))
SELECT collect_set(a) FROM VALUES (array(-0.0D)), (array(0.0D)) AS t(a)        -- Seq(Seq(0.0))
SELECT collect_set(a) FROM VALUES (array(float(-0.0))), (array(float(0.0))) AS t(a)
SELECT collect_set(named_struct('a', v)) FROM VALUES (double('NaN')), (double('NaN')) AS t(v)
SELECT collect_set(a) FROM VALUES (array(double('NaN'))), (array(double('NaN'))) AS t(a)
SELECT collect_set(a) FROM VALUES (array(float('NaN'))), (array(float('NaN'))) AS t(a)

Note that array<float> NaN and the struct NaN guardrail are in Spark's set but not in cs_norm_nested. The NULL rows and the double nesting from point 5 are missing from Spark's coverage too, so those still need writing from scratch.