[SPARK-57298][SQL] collect_set fails to dedupe float/double NaN/-0.0 by their semantics

[jiwen624]

What changes were proposed in this pull request?

CollectSet now normalizes special floating-point values (NaN and -0.0) before inserting them into its deduplication buffer, so it follows Spark's float/double equality semantics (all NaNs are equal; -0.0 equals 0.0). This covers both scalar and nested (struct/array) FLOAT/DOUBLE columns:

• Top-level DOUBLE/FLOAT: the buffer element type becomes LONG/INT and values are stored as their normalized bit pattern. convertToBufferElement reuses NormalizeFloatingNumbers.DOUBLE_NORMALIZER/FLOAT_NORMALIZER (NaN -> canonical, -0.0 -> 0.0) and then doubleToLongBits/floatToIntBits; eval converts the bits back. Keying on bits is required because the HashSet buffer compares boxed numbers with primitive equality, where NaN != NaN. Normalizing -0.0 is necessary here so that -0.0/0.0, which deduplicate today, keep deduplicating once the buffer keys on the bit pattern (otherwise doubleToLongBits(-0.0) != doubleToLongBits(0.0) would regress it).
• Complex types recursively containing FLOAT/DOUBLE (struct/array): values are normalized with NormalizeFloatingNumbers.normalize and materialized as UnsafeRow/UnsafeArrayData, so the buffer deduplicates on the canonical binary representation. MapType is unaffected (already rejected by checkInputDataTypes).

This reuses the normalization logic Spark already applies to other hash-based array set operation (NormalizeFloatingNumbers, case 5).

Why are the changes needed?

collect_set over FLOAT/DOUBLE did not follow Spark's floating-point equality semantics and returned elements that should be considered equal:

-- Top-level NaN: 
SELECT collect_set(v) FROM VALUES (double('NaN')), (double('NaN')) AS t(v);
-- before: [NaN, NaN]   after: [NaN]

-- Nested -0.0 / 0.0: 
SELECT collect_set(a) FROM VALUES (array(-0.0D)), (array(0.0D)) AS t(a);
-- before: [[-0.0], [0.0]]   after: [[0.0]]
SELECT collect_set(named_struct('a', v)) FROM VALUES (-0.0D), (0.0D) AS t(v);
-- before: [{a:-0.0}, {a:0.0}]   after: [{a:0.0}]

(Top-level -0.0/0.0 already deduplicate today; this PR preserves that while fixing the cases above.)

Does this PR introduce any user-facing change?

Yes. For collect_set over FLOAT/DOUBLE columns, including struct/array columns that contain them:

• duplicate NaN values are no longer returned, and
• -0.0 and 0.0 are deduplicated even when nested in a struct/array.

In addition, a -0.0 element is now always returned as 0.0. collect_set is already documented as non-deterministic, and -0.0/0.0 were already collapsed at the top level, so this only affects the returned representation.

How was this patch tested?

New test cases.

Was this patch authored or co-authored using generative AI tooling?

Yes. Claude Code

[jiwen624]

Hi @cloud-fan @szehon-ho when you get a chance could you take a look at this bug fix? 🙇

[cloud-fan]

0 blocking, 1 non-blocking, 0 nits.
Clean, well-scoped fix — normalization is correct on both the scalar-bits and complex-UnsafeRow paths and preserves existing -0.0/0.0 dedup.

Suggestions (1)

• collect.scala:242: scalar float/double bit-keying and the bufferElementType remap have no comment, unlike the adjacent BinaryType case — the "why bits" rationale is non-obvious. Non-blocking — see inline.

Verification

Traced the dedup correctness on both paths. Scalar: keying on doubleToLongBits/floatToIntBits collapses all NaN bit patterns to one canonical key (so NaN dedups, which the boxed-HashSet IEEE equality could not), and applying the NORMALIZER first maps -0.0 -> 0.0 so the existing -0.0/0.0 dedup is preserved rather than regressed by raw bits; eval round-trips the bits back exactly for all non-NaN values. Complex: UnsafeProjection(normalize(...)) + InternalRow.copyValue materializes a normalized, defensively-copied UnsafeRow/UnsafeArrayData, so the buffer dedups on canonical binary form. The Collect-base serialize/deserialize/merge contract holds (convertToBufferElement is only invoked in update, so no double-conversion), and the fix is complete across the Collect family (CollectList keeps dups; CollectTopK uses a NaN/-0.0-aware ordering; ListAgg is string/binary only).

[cloud-fan]

Consider adding a short comment here (and on the bufferElementType remap above) mirroring the BinaryType note. The reason scalar float/double key on the bit pattern rather than the value is non-obvious: mutable.HashSet[Any] compares boxed Double/Float with IEEE equality, where NaN != NaN, so normalizing the value alone wouldn't collapse NaNs — keying on doubleToLongBits/floatToIntBits does (and the NORMALIZER step keeps -0.0/0.0 deduped). Complex types instead dedup on a normalized UnsafeRow's binary form. A line or two would save the next reader the trace. Non-blocking.

[jiwen624]

Thanks. Comments added to explain keying on bit pattern.

[cloud-fan]

thanks, merging to master/4.x/4.2 (bug fix)!