Skip to content

fix: preserve float collect_set semantics in distinct aggregate chains - #7220

Draft
LinSimon-901101 wants to merge 6 commits into
apache:mainfrom
LinSimon-901101:fix/7203-collect-set-distinct
Draft

LinSimon-901101 wants to merge 6 commits into
apache:mainfrom
LinSimon-901101:fix/7203-collect-set-distinct

Conversation

@LinSimon-901101

Copy link
Copy Markdown
Contributor

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.

@github-actions github-actions Bot added bug Something isn't working area:aggregation Hash aggregates, aggregate expressions labels Oct 25, 2026
@LinSimon-901101
LinSimon-901101 marked this pull request as ready for review October 25, 2026 15:02

@comphead comphead left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Comment on lines +474 to +475
for {
dataType <- Seq("FLOAT", "DOUBLE")

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Copy Markdown
Contributor Author

@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
LinSimon-901101 marked this pull request as draft October 28, 2026 08:15
@LinSimon-901101

Copy link
Copy Markdown
Contributor Author

Full fork CI passed for 252bd8a

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

area:aggregation Hash aggregates, aggregate expressions bug Something isn't working

Projects

None yet

Development

Successfully merging this pull request may close these issues.

collect_set over floats beside count(DISTINCT) diverges from Spark on signed zeros and NaN

2 participants