fix: pick the signed zero Spark ranks first in native percentile by zhangfengcdt · Pull Request #6928 · apache/datafusion-comet

zhangfengcdt · 2026-10-12T22:39:25Z

Which issue does this PR close?

Closes #6799.

Rationale for this change

Spark keeps 0.0 and -0.0 as separate keys in the percentile buffer (SPARK-45599), then stable-sorts them under SQL ordering, where they are equal. So the zero its OpenHashMap iterates first takes the lower ranks, and which one that is depends on the table's capacity and collisions. Comet's native percentile picked either zero, so the result could have the wrong sign.

What changes are included in this PR?

When a requested rank lands on a zero and the group holds both zeros, the native percentile replays Spark's hash tables (murmur3 slots, growth, quadratic probing) to find which zero Spark ranks first. Spark builds one table per task partial buffer and a fresh one per merge stage, so the replay follows the same structure: each partial is replayed into its own table, and the tables are merged in Spark's order. Other ranks and other inputs pay nothing beyond one scan for a -0.0.
The accumulator state becomes {values, partial_ends, partial_depths} so the partial boundaries and merge-stage nesting survive state(), spills and shuffles. The two lists stay empty unless the serde sets signed_zero_keys.
Three fields on the Percentile proto: float_input (Spark hashes FLOAT input as a float, before Comet's cast to double), signed_zero_keys (set only for FLOAT/DOUBLE input on Spark 3.5.2+, where both zeros stay separate keys), and merge_stage (set for a PartialMerge aggregate so the replay nests that stage's partials as Spark does).
The serde reports Compatible with a note on FLOAT/DOUBLE input: before 3.5.2, that Spark can merge the two zeros and drop a count; from 3.5.2, that the replay does not model Spark's sort-based fallback (spark.sql.objectHashAggregate.sortBased.fallbackThreshold) or its aggregates emitting or spilling under memory pressure. Both are runtime conditions the serde cannot see at plan time, and in both Spark's own answer moves as well.

How are these changes tested?

Rust unit tests in percentile.rs: replays_each_partial_separately, keeps_the_partials_through_a_spill, nests_the_partials_of_each_merge_stage (22 rows that give -0.0 with no merge stage and 0.0 after two), tracks_no_partials_unless_spark_keeps_both_zeros, and replays_only_when_a_rank_lands_on_a_zero.
percentile.sql: both signs of zero in separate groups, for FLOAT and DOUBLE.
percentile_zero_tie.sql (Spark 3.5+): both zero insertion orders, a grouped case, a 502-row table large enough to make Spark's table grow, and the 22-row table with count(DISTINCT k) beside the percentile so the plan has PartialMerge stages. The large-table case fails without the replay (Spark 0.0, Comet -0.0), and the merge-stage case fails without the nesting (Spark 0.0, Comet -0.0).
percentile_partials.sql (Spark 3.5+): two files in separate partitions, merged in either order, which fails if the partials are replayed as one buffer.
CometSqlFileTestSuite percentile passes on Spark 3.4 (the 3.5+ fixtures skip, and the flag is off) and Spark 4.1.

Spark keeps 0.0 and -0.0 as separate keys in the percentile buffer and ranks first the one its OpenHashMap iterates first, which depends on the table's capacity and collisions. When a group holds both zeros, replay Spark's partial, deserialized and final tables to pick the same sign. Pass whether the input was FLOAT so the replay hashes values as Spark does. Closes apache#6799.

Spark builds, serializes and deserializes one map per partition before merging them into the final buffer, so replaying all merged values as one partial can rank the other signed zero first. Keep the boundaries of the merged partial states and replay each one in turn. Add a two-partition regression test, and write the zero-tie fixtures as one file per insert.

…ve-zero

sunchao

Reviewed 71d4834b67d6a94bedfaa406efe574f0d34986ec against 21e57333a5f7c60386998207de442700802ee9e8. Found three introduced P2 issues.

Reviewed all 12 base-to-head changed paths and both PR commits. The CASE/IN safeguard differences come from base-only commit #6763, rather than a revert introduced by this PR. Existing reviews, conversation comments, inline comments and the supplied threads were empty. The PR is not a draft.

Routed skills: review-comet-pr, review-comet-expression-pr, and review-comet-memory-pr.

Summary

Prior state and problem: Native percentile selected an arbitrary signed zero among equal values. Spark's percentile buffer can retain both signs, with hash-table iteration determining their ranks. This PR addresses #6799.
Design approach: Replay Spark's partial, deserialized and final key tables when both zero signs occur, then correct the selected zero's sign.
Correctness: The normal partial/final examples agree with Spark 4.1.3, including FLOAT input and table growth. However, serialized aggregate state discards the new partial boundaries. A reproducible spill/restore transformation changes a previously correct 0.0 result to -0.0.
Compatibility analysis: Compared Spark sources for 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. Spark 3.4.3 uses cooperative key equality rather than the post-SPARK-45599 object equality reproduced here. Applying the new behavior unconditionally introduces a signed-zero mismatch on that supported profile.
Key design decisions: float_input correctly preserves the input type needed for hashing after the serde casts values to DOUBLE. Keeping partial boundaries in memory fixes the earlier flattened replay, but those boundaries also need a durable representation.
Implementation sketch: The serde sets protobuf field 4, the planner passes it to SparkPercentile, both accumulator implementations record partial ends, and SignedZeros uses SparkKeySet to resolve zero ranks.
Performance: Replay runs even when neither requested rank contains zero. In an optimized standalone benchmark of the exact kernels, a one-million-value median with two additional signed zeros took about 281 ms at head versus 3.2 ms at base, with identical results. This unnecessary replay also allocates multiple key tables.
Design: Reproducing Spark's table layout explains the zero ordering, but the design must distinguish original partial buffers from subsequently serialized merged state. The current state contract loses information required by evaluation.
Abstraction & complexity: SignedZeros and SparkKeySet keep the compatibility machinery localized. Restricting replay to percentile brackets that intersect zero would avoid most of its cost without changing the result.
Behavioral changes worth calling out: Compared with branch-1.1 at dc4295ffe4492de42e1b863a3bb5fe96c8551990, correcting signed-zero selection is intended. The older-Spark mismatch, spill-dependent sign regression and nonzero-percentile slowdown are unintended. Supported expression forms and return types remain unchanged.
Suggested improvements: Gate the key-equality behavior in Scala or fall back on affected Spark versions, preserve replay information across serialized states, and skip table replay when the requested ranks cannot select zero. Add regression coverage for the concrete cases below.

Exact-head CI: run 38320108804 reports this exact SHA. Native build/tests, lint, Spark 4.1 scans/shuffle/exec and TPC-H/TPC-DS checks passed. The expression job was still running at the final check. Spark SQL suites for 3.4–4.2, macOS and benchmark checks were skipped. No failures were reported.

Validation limits: Ran optimized standalone Rust probes extracted from the exact base/head kernels, compiled upstream Spark 3.4.3 hash-map sources with Scala 2.12, and exercised Spark 4.1.3's actual Percentile aggregate. The focused Cargo test could not resolve DataFusion 55.2.0 from the configured registry, both offline and online. Full Comet/JVM tests and a forced-spill integration query were not run. The spill finding is supported by the exact kernel/state transformation and verified DataFusion 55.2 spill call paths.

sunchao · 2026-10-12T23:40:04Z

+        let mut pos = self.slot(v, mask);
+        let mut delta = 1;
+        while let Some(existing) = self.slots[pos] {
+            if self.key_bits(existing) == bits {


[P2] Preserve the supported Spark version's key-equality behavior. This bitwise equality reproduces Spark after SPARK-45599, but Spark 3.4.3 still uses cooperative equality in OpenHashSet. For one ordered DOUBLE partial containing 1.0 through 43.0, followed by 0.0 and -0.0, percentile(v, 0.0) returns 0.0 in Spark 3.4.3 and at the PR base. This head returns -0.0. The new replay therefore introduces a result mismatch on a supported profile. Select the behavior through the Scala serde, or fall back on Spark versions whose buffer semantics are not reproduced.

Evidence: Compiled the upstream v3.4.3 OpenHashMap and OpenHashSet sources with Scala 2.12.18, changing only their package and the BitSet import. Inserting the 45 rows, rebuilding in serialization order, then merging produces one zero entry (0.0, 2), so percentile 0 is positive zero. The exact extracted base kernel returns bits 0x0000000000000000; the head kernel with partial_ends=[45], percentile 0.0 and float_input=false returns 0x8000000000000000. Spark 3.5.0/3.5.1 sources also retain the older equality implementation.

Confirmed on 3.4.3 and 3.5.0. The serde now sets signed_zero_keys only on spark >=3.5.2.

sunchao · 2026-10-12T23:40:04Z

    fn state(&mut self, emit_to: EmitTo) -> Result<Vec<ArrayRef>> {
        let emit_group_values = emit_to.take_needed(&mut self.group_values);
+        // A state holds no boundaries, so a later merge sees these values as one partial.
+        emit_to.take_needed(&mut self.group_partial_ends);


[P2] Preserve replay information when emitting merged aggregate state. DataFusion calls state() when a final grouped aggregate spills and calls merge_batch() when reading it back. Discarding group_partial_ends turns several original partials into one synthetic partial, changing the hash-table replay and potentially the percentile's sign. The concrete two-partial example below returns 0.0 in Spark and at the base, but returns -0.0 after this head's state/restore transformation. Serialize sufficient partial/layout metadata, or an equivalent representation that preserves evaluation across the round trip.

Evidence: Use one group with DOUBLE partial A=[503707.0, 0.0, 320108.0] and B=[816344.0, 982714.0, 824673.0, 754010.0, 941794.0, 961567.0, 167263.0, 680472.0, -0.0]. Spark 4.1.3's actual update/serialize/deserialize/merge/eval pipeline returns positive zero for percentile 0 in either partial arrival order. The exact head kernel on A+B with ends [3,12] returns positive zero. state() emits only the concatenated values, so restoring it supplies ends [12], which returns negative zero. The base kernel returns positive zero. Verified DataFusion 55.2 grouped_hash_stream.rs calls acc.state(emit_to) for spilling and acc.merge_batch(...) during spill replay.

Yes, that is fair observation. I changed the state to include values and partial_ends so that it should keep the partitial boundaries.

sunchao · 2026-10-12T23:40:04Z

        return Some(values[0]);
    }
+    // Read before the selection below reorders the values.
+    let zeros = SignedZeros::find(values, partial_ends, float_input);


[P2] Skip hash-table replay when the requested percentile bracket cannot select zero. SignedZeros::find currently reconstructs every table whenever both signs occur, even for a nonzero median or maximum. This adds substantial time and allocations to previously correct results. Compute the requested ranks before replay and use the existing below-zero/zero counts to bypass table construction when neither rank intersects the zero interval.

Evidence: Compiled standalone copies of the exact base/head kernels with rustc +1.99.0 -O. Input was shuffled DOUBLE values 1..=1_000_000 followed by [-0.0, 0.0], represented as one partial. Across eight measured runs after warmup, median evaluation time for percentile 0.5 was 3,243 microseconds at base versus 280,669 microseconds at head, about 86 times slower. Both returned 499999.5. Percentile 1.0 similarly took 985 versus 277,688 microseconds and returned 1000000.0 in both. These are kernel measurements, not end-to-end query timings.

Done! Now, the ranks are computed first and the replay only runs upon failures on a zero. A median or maximum away from zero can skip it.

… on Spark version - Store where each partial ends in the percentile state so a spilled and restored final aggregate replays the same partials. - Replay Spark's tables only when a requested rank lands on a zero. - Replay only on Spark 3.5.2+, which keeps 0.0 and -0.0 as separate keys (SPARK-45599); earlier releases keep the previous behavior, documented as a compatibility note.

…ve-zero

comphead

Thanks for the quick turnaround. I rechecked the three earlier points at 3e0a220b0a. The serde gate (keepsSignedZeroKeys, 3.5.2+) matches where OpenHashSet.keyExistsAtPos switched to equals (v3.5.2, v4.0.x, v4.1.x). partial_ends now survives state()/merge_batch, and the replay only runs when a rank lands on a zero. I also compared SparkKeySet with OpenHashSet and TypedAggregateWithHashMapAsBuffer at v3.5.2 and v4.1.3 (murmur3 hashInt seed 0, 64 slots, 0.7 load factor, quadratic probing, rehash in slot order, doubleToLongBits/floatToIntBits keys) and found no difference. The comments below cover one Spark path the replay does not model, the cost of partial_ends where it can never be used, replay memory, and the tests.

comphead · 2026-10-14T00:17:40Z

+/// Each partition inserts its rows into a partial buffer and serializes it in iteration order.
+/// The final aggregation deserializes each one into a new map and merges that into its buffer,
+/// again in iteration order, one partial after another. This replays those tables, taking the
+/// partials in the order Comet merged them. Spark merges them in the order its shuffle delivers
+/// them, so when that changes the zero Spark reaches first, Spark's own answer varies too. A
+/// distinct aggregate's PartialMerge hands on the partials it merged where Spark hands on one
+/// merged buffer, so there the order can also differ from Spark's.


This models a partial buffer as every row of the partition inserted in order, and the final buffer as each partial merged once. That only holds while Spark's ObjectHashAggregateExec stays hash based. Once a task holds spark.sql.objectHashAggregate.sortBased.fallbackThreshold groups (default 128), ObjectAggregationIterator falls back. Each group then gets a fresh buffer, takes the rows that came after the fallback first, and then merges the pre-fallback buffer back in after a serialize/deserialize round trip (SortBasedAggregator.findNextSortedGroup). The same happens in the final aggregate for the partials. In a Python model of these tables, I split one group's rows at a fallback point, and its answer differed from this replay for about 2% of random splits (model only, not run on Spark). So for a GROUP BY with more than 128 groups per task, Comet can still pick the other zero, deterministically. Could you either model the fallback or state the limit here and in signedZeroKeysNote?

Documented rather than modeled. Spark's sort-based fallback changes Spark's own answer, and modeling it would need every row's position relative to the 128th group on both the partial and merge sides. The doc comment on spark_iterates_negative_zero_first and the new compat note call out the limit.

comphead · 2026-10-14T00:17:40Z

+    for (group, group_ends) in group_values.iter().zip(group_partial_ends) {
+        ends.extend(group_ends.iter().map(|&end| end as i64));
+        if group_ends.last().copied().unwrap_or(0) < group.len() {
+            ends.push(group.len() as i64);
+        }


partial_ends is recorded, shuffled and kept per group even when SignedZeros::find can never read it. That covers Spark before 3.5.2 (signed_zero_keys false) and every non-floating child, since an integral or decimal value cast to double is never -0.0. Each state row now carries an extra list offset plus one i64, and each final group keeps one usize per merged partial. For a high-cardinality GROUP BY with about one value per group per partition, I expect this to roughly double the state shuffle bytes and the final accumulator's memory (estimate, not measured). Could the serde set the flag only for FloatType | DoubleType children, and the accumulators skip the ends (emitting an empty partial_ends) when it is false?

Done. The lists are populated only when signed_zero_keys is set, i.e. FLOAT/DOUBLE input on Spark 3.5.2+; otherwise they stay empty end to end. Covered by tracks_no_partials_unless_spark_keeps_both_zeros.

comphead · 2026-10-14T00:17:40Z

+struct SparkKeySet {
+    slots: Vec<Option<f64>>,
+    size: usize,
+    float_input: bool,
+}


The replay is allocated outside the memory pool and is large next to the tracked values. Option<f64> takes 16 bytes per slot, a table holds 1.43 to 2.86 slots per distinct key, and partial, deserialized and table are all alive inside the loop at lines 527-531. That is roughly 70 to 140 bytes per distinct value against the 8 that size() reports (my arithmetic, not measured). Storing the key bits in a Vec<u64> with u64::MAX as the empty marker would halve the slot size and drop the repeated key_bits(existing) per probe. u64::MAX is never a key, because NaNs canonicalize to 0x7ff8... and float keys fit in 32 bits. Dropping partial once deserialized is built would also remove one table from the peak.

Done. The key set is packed bits (Vec<u64>, u64::MAX as the empty marker), and the recursive replay drops each partial's table as soon as it is merged into its parent, so only one table per nesting level is live.

comphead · 2026-10-14T00:17:40Z

+    #[test]
+    fn picks_the_zero_spark_iterates_first() {
+        for float_input in [false, true] {
+            let mut values = spark_45599_rows();
+            let ends = [values.len()];
+            let result =
+                spark_percentile(&mut values, &ends, 0.1, zero_order(float_input)).unwrap();
+            assert_eq!(result.to_bits(), (-0.0_f64).to_bits());
+        }
+    }
+
+    #[test]
+    fn follows_the_table_as_it_grows() {
+        // -0.0 comes first in Spark's 64-slot table, and 0.0 once it has grown to 1024 slots,
+        // which it does past 358 keys.
+        for (keys, first) in [(10, -0.0_f64), (500, 0.0_f64)] {
+            let mut values: Vec<f64> = (1..=keys).map(f64::from).collect();
+            values.extend([-0.0, 0.0]);
+            let ends = [values.len()];
+            let result = spark_percentile(&mut values, &ends, 0.0, zero_order(false)).unwrap();
+            assert_eq!(result.to_bits(), first.to_bits());
+        }
+    }
+
+    #[test]
+    fn keeps_the_sign_of_a_lone_zero() {
+        let mut values = vec![-1.0, -0.0, 1.0];
+        let result = spark_percentile(&mut values, &[], 0.5, zero_order(false)).unwrap();
+        assert_eq!(result.to_bits(), (-0.0_f64).to_bits());
+    }


These three tests repeat cases the SQL files already check against Spark. picks_the_zero_spark_iterates_first uses the same rows and percentile as the first query in percentile_zero_tie.sql. follows_the_table_as_it_grows is the 500-key test_percentile_zero_tie_large case plus the small grouped table. keeps_the_sign_of_a_lone_zero is test_percentile_signed_zero in percentile.sql. Could they be removed, keeping the spill, partial and replay-gating tests that SQL cannot pin?

Removed the three tests and the matching fixture; the SQL files cover those cases against Spark.

comphead · 2026-10-14T00:17:40Z

+-- Spark ranks -0.0 first for these rows in either arrival order, while replaying them as one
+-- partial ranks 0.0 first.


Per-partial replay ranks -0.0 first in both arrival orders. Replaying the rows as one partial ranks 0.0 first only when the 15-row file is merged first. With the 5-row file first, it also ranks -0.0 first. So this case only catches a regression to the flattened replay for one shuffle order. Could the comment say so? The Rust comment at percentile.rs line 699 says the reverse ("when the second is merged first"), while the test there flattens the first partial first.

Right, the comment had it backwards. Fixed in both places: the flattened replay ranks 0.0 first only when the 15-row file is merged first.

sunchao

Reviewed commit 3e0a220b0a17e301f5d0a1d5b5e373a30911a078. New P1/P2 findings are detailed inline.

sunchao · 2026-10-14T02:19:03Z

+    for &end in partial_ends {
+        let partial = build(&mut values[start..end].iter().copied());
+        let deserialized = build(&mut partial.keys());
+        deserialized.keys().for_each(|v| table.add(v));


[P2] Preserve logical PartialMerge stages in the replay. Adding COUNT(DISTINCT d) beside percentile introduces intermediate Spark merge stages, each of which serializes and reconstructs the merged table. Comet’s MergeAsPartial instead forwards the original partial_ends, so this loop still replays only the original partials. For the input below, Spark and the PR base return +0.0, but this head returns -0.0. This is deterministic with one input partition and does not depend on shuffle arrival order. Represent logical merge stages separately from spill restoration and replay their table rebuilds, or fall back for this plan shape.

Evidence: Spark 4.1.3, local[1], AQE disabled, one shuffle partition: create one ordered partition with DOUBLE v=[572803,866096,866879,433001,33996,75441,418281,919347,710172,746536,68317,830168,127901,662299,132114,986230,624172,784519,199287,200265,0.0,-0.0] and INT d=1 on every row. Executed SELECT percentile(v,0.0), count(DISTINCT d) FROM t: Spark returns bits 0x0000000000000000. Its plan has two merge_percentile stages before the final aggregate. Compiled exact-head state helpers with Arrow 59.2.0: after each of three merge/state rounds, ends remain [22] and evaluation returns 0x8000000000000000. Exact base and latest-release kernels return 0x0000000000000000. The unchanged MergeAsPartial wrappers call inner.merge_batch and inner.state, confirming this state transformation.

Fixed by modeling the merge stages. The state now carries partial_depths next to partial_ends, and the serde sets a merge_stage flag for PartialMerge, so each stage's partials are nested under one node and replayed as Spark's OpenHashMap would merge them. Your 22-row example gives +0.0 and is checked in nests_the_partials_of_each_merge_stage and the new percentile_zero_tie.sql query with count(DISTINCT k). One remaining gap: a spill inside a merge stage nests one level deeper than Spark; documented in the code.

sunchao · 2026-10-14T02:19:03Z

+    for (group, group_ends) in group_values.iter().zip(group_partial_ends) {
+        ends.extend(group_ends.iter().map(|&end| end as i64));
+        if group_ends.last().copied().unwrap_or(0) < group.len() {
+            ends.push(group.len() as i64);


[P2] Keep early-emitted chunks from one task in the same replay buffer. DataFusion’s grouped Partial aggregate can emit state under memory pressure and continue consuming that task’s rows. This code marks every emitted chunk as a completed partial, so the final replay treats chunks from one Spark partition as separate Spark buffers. Splitting the input below after row 15 changes the head’s result from +0.0 to -0.0, although Spark and the base return +0.0. Thus memory pressure or batch boundaries can change a previously correct result. Preserve original task/chunk identity so replay can reconstruct the task’s actual partial buffer.

Evidence: Use DOUBLE values [369513,998311,8278,156933,286454,349279,125841,796806,556996,0.0,667190,624668,335166,220654,315604,223815,201661,68860,928422,-0.0] in one ordered partition/group. Spark 4.1.3’s executed grouped percentile(v,0.0) returns positive-zero bits. Exact-head build_state/merge_state_row helpers compiled with Arrow 59.2.0 produce ends=[20] and positive zero when emitted together, but ends=[15,20] and negative zero when emitted as two chunks. The base kernel returns positive zero. DataFusion 55.2.0 grouped_hash_stream.rs selects EmitEarly for Partial mode, calls acc.state when a reservation fails with more than one group, and continues reading input. Another group enables that path without reaching Spark’s 128-group fallback threshold. The state transformation and upstream call path were verified; a complete Comet query forcing the reservation failure was not run.

Documented as a limit rather than modeled. Early emit only happens on a reservation failure, and at that point Spark's own sort-based fallback changes its answer too, so matching it would need task identity per state row. The doc comment lists this with the sort-based fallback.

andygrove

I reproduced the COUNT(DISTINCT) case sunchao reported with a real query, and it is still open. On this head, a one-partition parquet table with 22 doubles ending in 0.0 then -0.0 (the values from sunchao's repro, plus a constant int d) and SELECT cast(percentile(v, 0.0) as string), count(DISTINCT d) returns -0.0 from Comet and 0.0 from Spark. The same query passes when I force signed_zero_keys to false in the serde, so the mismatch comes from the new replay and not from the existing selection. The doc comment on spark_iterates_negative_zero_first already says the order can differ here, but CometPercentile still reports Compatible() for that plan shape. Could the serde skip the replay (set signed_zero_keys to false) or fall back when the aggregate is in a PartialMerge stage, so a query that matched Spark does not start returning the other zero?

The early-emit chunking case and the sort-based fallback case that sunchao and comphead described have the same shape. Each is a path where the replay model does not match what Spark builds, and the replay moves a previously matching answer. Could the PR either model those paths or limit the replay to the shapes it is known to reproduce, and say in signedZeroKeysNote which shapes remain unmatched? I would also like replies on comphead's questions about partial_ends being shuffled for non-float inputs and for Spark before 3.5.2, since that extra state ships on every query that uses percentile today.

For the tests, I ran percentile_zero_tie.sql and percentile_partials.sql through CometSqlFileTestSuite on this head and both pass. With the serde forced to setSignedZeroKeys(false) both fail, so they do pin the new behavior. The datafusion-comet-spark-expr percentile unit tests also pass. None of them cover a PartialMerge plan, so a case with percentile next to count(DISTINCT ...) would be worth adding once the behavior is decided.

andygrove · 2026-10-15T14:19:11Z

+/// again in iteration order, one partial after another. This replays those tables, taking the
+/// partials in the order Comet merged them. Spark merges them in the order its shuffle delivers
+/// them, so when that changes the zero Spark reaches first, Spark's own answer varies too. A
+/// distinct aggregate's PartialMerge hands on the partials it merged where Spark hands on one


Should this caveat be a fallback instead? With percentile(v, 0.0) next to count(DISTINCT d) on one partition, Spark returns 0.0 and this head returns -0.0, while the same query with the replay switched off matches Spark. A comment does not stop a user from hitting it while the expression is reported as Compatible().

I changed the code so that in stead of fall back, this query now returns 0.0 like Spark (covered in percentile_zero_tie.sql).

… signed zeros Spark's PartialMerge stages merge each incoming state into a fresh OpenHashMap, so the iteration order of signed-zero keys depends on how many merge stages a row passed through. Record the merge stage in the state (partial_depths) so the replay can nest partials the same way. Track partial boundaries only when the input is FLOAT/DOUBLE on a Spark that keeps both signed zeros as keys; otherwise the lists stay empty. Store the key set as packed bits and drop each partial's table once it has been merged. Document the two unmodeled paths (Spark's sort-based fallback and DataFusion early emit) in the code and the compat note. Remove the redundant unit tests and fix the stale fixture comment.

…ve-zero # Conflicts: # spark/src/main/scala/org/apache/comet/serde/aggregates.scala

…k 3.5.2+ The serde returned a plain Compatible() for FLOAT and DOUBLE input on Spark versions that keep both signed zeros as keys, even though the replay does not model Spark's sort-based fallback or its aggregates emitting under memory pressure. Return the note with the support level, as the serde already does with signedZeroKeysNote on older Spark, so the limit is visible where the plan shape is accepted.

zhangfengcdt · 2026-10-15T16:15:46Z

I reproduced the COUNT(DISTINCT) case sunchao reported with a real query, and it is still open. On this head, a one-partition parquet table with 22 doubles ending in 0.0 then -0.0 (the values from sunchao's repro, plus a constant int d) and SELECT cast(percentile(v, 0.0) as string), count(DISTINCT d) returns -0.0 from Comet and 0.0 from Spark. The same query passes when I force signed_zero_keys to false in the serde, so the mismatch comes from the new replay and not from the existing selection. The doc comment on spark_iterates_negative_zero_first already says the order can differ here, but CometPercentile still reports Compatible() for that plan shape. Could the serde skip the replay (set signed_zero_keys to false) or fall back when the aggregate is in a PartialMerge stage, so a query that matched Spark does not start returning the other zero?

The early-emit chunking case and the sort-based fallback case that sunchao and comphead described have the same shape. Each is a path where the replay model does not match what Spark builds, and the replay moves a previously matching answer. Could the PR either model those paths or limit the replay to the shapes it is known to reproduce, and say in signedZeroKeysNote which shapes remain unmatched? I would also like replies on comphead's questions about partial_ends being shuffled for non-float inputs and for Spark before 3.5.2, since that extra state ships on every query that uses percentile today.

For the tests, I ran percentile_zero_tie.sql and percentile_partials.sql through CometSqlFileTestSuite on this head and both pass. With the serde forced to setSignedZeroKeys(false) both fail, so they do pin the new behavior. The datafusion-comet-spark-expr percentile unit tests also pass. None of them cover a PartialMerge plan, so a case with percentile next to count(DISTINCT ...) would be worth adding once the behavior is decided.

Thank for confirming the two issues, and I think they both should have been fixed now with the few new commits.

The replay now maps to PartialMerge stages, so the count(DISTINCT) case matches Spark and is covered by a new SQL fixture and unit test.

The partial boundaries are only tracked for FLOAT/DOUBLE on Spark 3.5.2+; and the sort-based fallback and early-emit paths, which cannot be detected at plan time, are named in the note attached to the Compatible result. comphead's threads are answered inline, and the branch is merged with main.

…ve-zero # Conflicts: # native/spark-expr/benches/aggregate_stats.rs # spark/src/main/scala/org/apache/comet/serde/aggregates.scala

github-actions Bot added bug Something isn't working area:aggregation Hash aggregates, aggregate expressions labels Oct 12, 2026

zhangfengcdt added 2 commits October 12, 2026 16:03

Merge remote-tracking branch 'origin/main' into fix/percentile-negati…

aaa4600

…ve-zero

sunchao reviewed Oct 12, 2026

View reviewed changes

zhangfengcdt added 2 commits October 13, 2026 07:22

Merge remote-tracking branch 'origin/main' into fix/percentile-negati…

3e0a220

…ve-zero

comphead reviewed Oct 14, 2026

View reviewed changes

sunchao reviewed Oct 14, 2026

View reviewed changes

andygrove requested changes Oct 15, 2026

View reviewed changes

zhangfengcdt added 3 commits October 15, 2026 08:38

Merge remote-tracking branch 'origin/main' into fix/percentile-negati…

e9586d3

…ve-zero # Conflicts: # spark/src/main/scala/org/apache/comet/serde/aggregates.scala

Merge remote-tracking branch 'origin/main' into fix/percentile-negati…

c476649

…ve-zero # Conflicts: # native/spark-expr/benches/aggregate_stats.rs # spark/src/main/scala/org/apache/comet/serde/aggregates.scala

		-- Spark ranks -0.0 first for these rows in either arrival order, while replaying them as one
		-- partial ranks 0.0 first.

Conversation

zhangfengcdt commented Oct 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

How are these changes tested?

Uh oh!

sunchao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

comphead left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sunchao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

andygrove left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zhangfengcdt commented Oct 15, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

zhangfengcdt commented Oct 12, 2026 •

edited

Loading