[LinSimon-901101]

Which issue does this PR close?

Closes #7203.

Rationale for this change

Before Spark 4.2, collect_set over FLOAT or DOUBLE collapses signed zeros and retains repeated NaNs in Spark, whereas native execution distinguishes the zeros and deduplicates the NaNs. Enabling native PartialMerge chains exposed this difference when collect_set is combined with a DISTINCT aggregate. Comet 1.1.0 kept that shape in Spark.

What changes are included in this PR?

Decline FLOAT/DOUBLE collect_set in PartialMerge mode before Spark 4.2. The existing unsafe-partial pass and buffer compatibility checks keep the aggregate chain in Spark, restoring the previous results. Document this restriction in the CollectSet compatibility notes.

The restriction is specific to scalar FLOAT/DOUBLE PartialMerge on older Spark versions. Ordinary collect_set, collect_list, other element types, and Spark 4.2 retain their existing native support.

How are these changes tested?

• Add 16 regressions covering FLOAT/DOUBLE, Spark partial versus Parquet input, AQE on/off, and object-hash versus sort aggregation. Inputs include repeated NaNs, signed zeros, nulls, and an all-null group. Assertions check both results and the same executed plan: the aggregate chain stays in Spark before 4.2, while merge/final remain native on 4.2.
• Add a control that checks native ordinary float collect_set, float collect_list with DISTINCT, and struct-valued collect_set with DISTINCT. All 16 regressions fail with wrong results on the unmodified baseline; the control passes.
• All 17 cases pass in the full fork CI run at commit 6029ad2, on five Linux Spark profiles (3.4 through 4.2) and macOS. All 20 Linux Comet matrix rows, all 45 Spark SQL matrix rows, and Required Checks pass. The related SPARK-17616 integration cases also pass on each Spark version.

[comphead]

Declining the merge in convert is the first of the two options in #7203 and restores the 1.1.0 result for the distinct shape. One point on where the tests live.

[comphead]

The decision this matrix pins depends only on the element type, the mode and the Spark version, and the SQL file harness covers all of it. collect_set_floating_fallback.sql (-- MaxSparkVersion: 4.1) already has FLOAT and DOUBLE tables with NaN, signed zeros, infinities and NULLs, and a ConfigMatrix over useObjectHashAggregateExec. A sibling file with -- Config: spark.comet.exec.strictFloatingPoint=false, ConfigMatrix lines over useObjectHashAggregateExec and spark.sql.adaptive.enabled, and query expect_fallback(collect_set on FLOAT/DOUBLE in PartialMerge requires Spark 4.2) over count(DISTINCT ...) beside collect_set would cover the native-input half. A VALUES source with spark.comet.exec.localTableScan.enabled=false would cover the Spark-partial half. On 4.2, the same query added to collect_set_normalization.sql (-- MinSparkVersion: 4.2) checks the native answer. That replaces the 16 generated Scala tests with a few queries compared against Spark, and the control test fits as plain query blocks in the same files.

[LinSimon-901101]

@comphead Thanks for the suggestions. I'll move the result/fallback matrix and control cases to SQL fixtures using the version gates and ConfigMatrix settings you outlined.
I'll also check the Spark 4.2 VALUES case with LocalTableScan disabled. If the SQL harness cannot assert its mixed Spark-partial/native-merge plan, I'll keep a focused Scala test for that path.

[LinSimon-901101]

Full fork CI passed for 252bd8a

[rich7420]

LGTM. Relevant regressions passed fork CI. No local Comet tests.

[LinSimon-901101]

Fork Comet CI passed on 5eb84ee.

[andygrove]

Thanks for the fix. Declining the PartialMerge in convert for FLOAT and DOUBLE before Spark 4.2 matches the first option in #7203 and restores the 1.1.0 result for the distinct shape. I compared against branch-1.2 and the only behavior change I found in the collect_set path is that regression fix. I read the diff and the existing threads and did not build or run anything.

The last commit looks like a separate change. 5eb84ee ("refresh materialized AQE inputs and repair CI compatibility") adds about 65 lines to CometExecRule.refreshedNativeOp, a macos-14 to macos-15 bump, an rm -rf native/target/ci step in spark_sql_test_reusable.yml, and edits to the Iceberg, CollectMetrics and Parquet writer tests. None of it is needed for #7203, and fork CI passed at 252bd8a without it. Both existing reviews also predate it.

The CometExecRule hunk rewrites serialized native plans under AQE for every query that reuses a node, not only collect_set. It matches operators by object identity and planId, and it is not gated on Spark version even though the doc text says Spark 3.4. This PR carries backport-1.1 and backport-1.2, and a backport must not carry an unrelated change.

Could you move these to their own PR? For the planner change, an issue that shows the failure on main, names the Spark version, and has a test that fails without it would help a lot. Which failures on main motivated the workflow and test edits?

Nested floats. The guard only checks expr.child.dataType == FloatType || DoubleType. The struct control in both SQL files uses named_struct('v', f) restricted to grp = 3, which only holds finite non-zero values, so it can't catch a signed-zero or NaN divergence. Could you add a query with -0.0, 0.0 and NaN inside array<double> and named_struct('v', d) elements next to count(DISTINCT id) on Spark before 4.2? If Spark and Comet agree, the control becomes meaningful. If they differ, the guard needs to cover types that contain floats.

[LinSimon-901101]

@andygrove, thanks for pointing these out. I've removed the unrelated changes introduced in 5eb84ee and added tests for nested arrays and structs containing signed zeros, NaNs, and nulls.
These cases match Spark, so the guard remains limited to scalar FLOAT/DOUBLE in PartialMerge chains before Spark 4.2.

[LinSimon-901101]

[Full fork CI] passed on 239daf1.