Repository navigation
Conversation
Before SPARK-45599, Spark's OpenHashMap hashes a float by its bits but matches keys with ==, so no two NaNs match and each NaN counts as a key of its own. Native mode counts all NaNs as one value, so with allowIncompatible set it could return NaN where Spark 3.4 returns the value that is actually most frequent. Report FLOAT and DOUBLE input as Unsupported on Spark 3.4, 3.5.0 and 3.5.1, so allowIncompatible does not reach it, and correct the comments that claimed every version collapses NaN. Closes apache#6738.
sunchao
left a comment
There was a problem hiding this comment.
Reviewed head 6747cef65f2658ab5247b65d31d0e675638a92f3 against base 498956132ced6f4e45453841aeae4e56e8e98e61. The PR is non-draft. Reviewed all eight files in the full PR diff. The release-script differences in a direct tree comparison belong solely to the newer base commit, not this PR.
Routed skills: review-comet-pr and review-comet-expression-pr. Read AGENTS.md, the discussion snapshot, live comments and reviews, and issue #6738. No existing review concerns remain to track.
Summary
- Prior state and problem: With
Mode.allowIncompatible=true, nativemodecombined NaNs on older Spark versions that counted them separately, producing different results even without ties. - Design approach: Add a Scala version predicate and return
UnsupportedforFLOATandDOUBLEbefore Spark 3.5.2. - Correctness: Spark’s
Mode,OpenHashMap, andOpenHashSetsources confirm the equality change at 3.5.2. Standalone probes reproduced1.0before that change and NaN afterward for three NaNs plus two1.0s, for both floating-point types. The planner handlesUnsupportedbefore consulting the incompatibility opt-in. - Compatibility analysis: Checked Spark 3.4.3, 3.5.0/1/2/9, 4.0.4, 4.1.3, and 4.2.0 sources. The predicate matches the relevant boundary. Compared with
branch-1.1at89c729a9697467e8f8d89e3eadf1797a75e2328c: the older-version fallback is an intended, documented behavior change. - Key design decisions:
Unsupportedprevents the tie-breaking opt-in from enabling a separate counting divergence. Existing deterministic-ordering, collation, and unsupported-type guards retain precedence. - Implementation sketch: The change adds
keysNaNsTogether, a floating-point support guard, documentation reasons, version tests, and separate native/fallback SQL fixtures. Rust changes are comments only. - Performance: The additional work is version parsing during planning, with no new per-row native work. Older Spark floating-point aggregates lose native execution intentionally to preserve correctness. No benchmark-based performance claim is warranted.
- Design: A conservative type-level fallback addresses the mismatch without reproducing Spark’s historical hash-table behavior inside the accumulator.
- Abstraction & complexity: One behavior-named predicate reuses the existing version parser and keeps version policy in Scala. No new protocol fields or native abstractions are introduced.
- Behavioral changes worth calling out: On Spark 3.4 and 3.5.0/1, floating-point
modenow falls back even withallowIncompatible=trueand even when the data contains no NaNs. Later versions retain existing opt-in and signed-zero behavior. - Suggested improvements: No additional issue meeting the P1/P2 reporting bar was identified.
Exact-head CI: CodeQL passed. Preflight, check-pr-title, label, and Analyze Actions remain queued. There is no completed build/test verdict and no run-* label coverage.
Validation: Standalone Scala probes using Spark collection sources passed, and the exact PR version predicate passed 17 boundary/suffix checks. git diff --check passed. The checkout has no native/JVM build outputs, and I did not run the Comet suites, Spark SQL matrix, or a full build. The SQL fixtures use minor-version gates, so they do not validate the 3.5.0/1 fallback correctly; those patch boundaries received predicate-level validation here.
No introduced P1/P2 issues found within this review. No project code or GitHub state was changed.
comphead
left a comment
There was a problem hiding this comment.
Gating on the exact patch boundary where SPARK-45599 landed (3.5.2 and 4.0.0) is right, and the fixture split by Spark version is clear. One comment on the unit tests.
| } | ||
| } | ||
|
|
||
| test("mode support level for float input follows the Spark version") { |
There was a problem hiding this comment.
This test computes its expectation from CometMode.keysNaNsTogether(SPARK_VERSION), the same predicate getSupportLevel uses, so the float branch is only checked against itself. mode float keys by Spark patch version pins the predicate at each boundary, mode_floating_point_spark_3_4.sql asserts the fallback end to end on 3.4, mode_floating_point.sql asserts the native path from 3.5, and mode.sql already covers the IntegerType Incompatible case. Could this test be dropped, keeping only the boundary test?
Which issue does this PR close?
Closes #6738.
Part of #6385.
Rationale for this change
Before SPARK-45599 (fixed in Spark 3.5.2 and 4.0.0),
modecounts values in anOpenHashMapthat hashes a float by its bits but matches keys with==. No two NaNs match, so every NaN row is a key of its own with a count of 1. Nativemodecanonicalizes NaN and counts them together. Withspark.comet.expression.Mode.allowIncompatible=true,modeoverNaN, NaN, NaN, 1.0, 1.0returnsNaNon Spark 3.4.3 where Spark returns1.0. There is no tie, so the opt-in's documented reason (tie-breaking, #3970) does not cover it.What changes are included in this PR?
CometMode.getSupportLevelreturnsUnsupportedforFLOATandDOUBLEinput on Spark 3.4, 3.5.0 and 3.5.1, with the reason in the fallback message and the generated compatibility guide.Unsupportedrather thanIncompatible, becauseallowIncompatiblewould otherwise run it and return a wrong answer with no tie. From 3.5.2, nothing changes.CometMode.keysNaNsTogether, which reads the patch version asArraySetSupport.normalizesFloatsdoes.Modeinmode.rsand the comment inCometMode.convertsaid that every version collapses NaN and that versions before 4.2 key byDouble.equals. Both hold only from 3.5.2 and 4.0.0, and now say so.mode.sqlloses itsDOUBLEsections, which move tomode_floating_point.sql(MinSparkVersion: 3.5) unchanged except for the corrected comments. Newmode_floating_point_spark_3_4.sql(MaxSparkVersion: 3.4) holds the reproduction and theDOUBLEandFLOATqueries asexpect_fallback.modeand the contributor guide'smodebullet mention the fallback.An alternative is to count each NaN separately in the native accumulator, which would keep
modenative on 3.4. It is opt-in and incompatible for ties already, and the epic keeps version differences in one Scala policy (#6385), so this takes the fallback. The issue allows either.How are these changes tested?
mode_floating_point_spark_3_4.sqlfails on 3.4 without the new branch (checked by removing it), and passes with it.mode_floating_point.sqlpasses on 3.5, 4.1 and 4.2;mode.sqlpasses on 3.4, 3.5, 4.1 and 4.2.CometAggregateSuitegetsmode float keys by Spark patch version(3.4.3, 3.5.0, 3.5.1 and 3.5.1-SNAPSHOT on one side, 3.5.2 through 4.2 on the other) andmode support level for float input follows the Spark version, which pass on 3.4 and 4.1.MinSparkVersionreads only the minor version, somode_floating_point.sqlalso runs on a local 3.5.0 or 3.5.1, where it would fail. CI pins 3.5.8.