sunchao commented on code in PR #6928:
URL: https://github.com/apache/datafusion-comet/pull/6928#discussion_r4266096222
##########
native/spark-expr/src/agg_funcs/percentile.rs:
##########
@@ -301,30 +403,231 @@ fn spark_percentile(values: &mut [f64], percentile: f64)
-> Option<f64> {
let lower = position.floor() as usize;
let higher = position.ceil() as usize;
+ // Read before the selection below reorders the values.
+ let zeros = SignedZeros::find(values, partial_ends, [lower, higher],
zero_order);
+
let (_, lower_value, _) = values.select_nth_unstable_by(lower, |x, y|
compare_floats(*x, *y));
- let lower_value = *lower_value;
+ let lower_value = zeros.resolve(lower, *lower_value);
if lower == higher {
return Some(lower_value);
}
let (_, higher_value, _) = values.select_nth_unstable_by(higher, |x, y|
compare_floats(*x, *y));
- let higher_value = *higher_value;
+ let higher_value = zeros.resolve(higher, *higher_value);
if compare_floats(lower_value, higher_value).is_eq() {
return Some(lower_value);
}
Some((higher as f64 - position) * lower_value + (position - lower as f64)
* higher_value)
}
+/// What decides where Spark ranks `0.0` against `-0.0`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+struct ZeroOrder {
+ /// Whether Spark's input was `FLOAT` before the serde cast it to
`DOUBLE`. Spark hashes the
+ /// values as floats then, which decides which signed zero it reaches
first.
+ float_input: bool,
+ /// Whether Spark keeps `0.0` and `-0.0` as separate keys, which it does
from 3.5.2
+ /// (SPARK-45599). Earlier releases can merge the two and drop one's
count, which this does
+ /// not reproduce.
+ signed_zero_keys: bool,
+}
+
+/// Where `0.0` and `-0.0` sit in Spark's sorted order, when the values hold
both.
+///
+/// Spark counts the values in an `OpenHashMap`, which keeps the two zeros as
separate keys
+/// (SPARK-45599), then sorts the keys with a stable sort under SQL ordering,
where they are
+/// equal. So the zero that Spark's hash table iterates first takes the lower
ranks, and which
+/// one that is depends on the table's capacity and collisions.
+#[derive(Debug, Default)]
+struct SignedZeros {
+ /// The rank of the first zero, `None` unless both signs are present.
+ start: Option<usize>,
+ /// How many values have the sign Spark reaches first.
+ first_count: usize,
+ first_negative: bool,
+}
+
+impl SignedZeros {
+ /// Only works out the order when one of `ranks` lands on a zero, since
replaying Spark's
+ /// tables is far slower than the selection itself.
+ fn find(
+ values: &[f64],
+ partial_ends: &[usize],
+ ranks: [usize; 2],
+ zero_order: ZeroOrder,
+ ) -> Self {
+ if !zero_order.signed_zero_keys || !values.iter().any(|v| v.to_bits()
== NEGATIVE_ZERO_BITS)
+ {
+ return Self::default();
+ }
+ let (mut below, mut negative, mut positive) = (0, 0, 0);
+ for &v in values {
+ if v == 0.0 {
+ if v.is_sign_negative() {
+ negative += 1;
+ } else {
+ positive += 1;
+ }
+ } else if v < 0.0 {
+ below += 1;
+ }
+ }
+ let zero_ranks = below..below + negative + positive;
+ if positive == 0 || !ranks.iter().any(|rank|
zero_ranks.contains(rank)) {
+ return Self::default();
+ }
+ let first_negative =
+ spark_iterates_negative_zero_first(values, partial_ends,
zero_order.float_input);
+ Self {
+ start: Some(below),
+ first_count: if first_negative { negative } else { positive },
+ first_negative,
+ }
+ }
+
+ /// The zero Spark has at `rank`, or `value` itself when it is not a zero.
+ fn resolve(&self, rank: usize, value: f64) -> f64 {
+ match self.start {
+ Some(start) if value == 0.0 => {
+ if (rank < start + self.first_count) == self.first_negative {
+ -0.0
+ } else {
+ 0.0
+ }
+ }
+ _ => value,
+ }
+ }
+}
+
+const NEGATIVE_ZERO_BITS: u64 = 0x8000_0000_0000_0000;
+
+/// Whether `-0.0` comes before `0.0` in Spark's final percentile buffer.
+///
+/// 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.
+fn spark_iterates_negative_zero_first(
+ values: &[f64],
+ partial_ends: &[usize],
+ float_input: bool,
+) -> bool {
+ let build = |values: &mut dyn Iterator<Item = f64>| {
+ let mut table = SparkKeySet::new(float_input);
+ values.for_each(|v| table.add(v));
+ table
+ };
+ let mut table = SparkKeySet::new(float_input);
+ let mut start = 0;
+ 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));
Review Comment:
[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.
##########
native/spark-expr/src/agg_funcs/percentile.rs:
##########
@@ -121,32 +132,108 @@ fn validate_percentile(percentile: f64) -> Result<()> {
Ok(())
}
+/// Fields of the single struct state column `{values: list<f64>,
partial_ends: list<i64>}`.
+/// `partial_ends` keeps the boundaries `spark_percentile` replays, so a final
aggregate that
+/// spills its state and merges it back still sees the partials it merged.
+fn state_struct_fields() -> Fields {
+ let list_of = |data_type|
DataType::List(Arc::new(Field::new_list_field(data_type, true)));
+ Fields::from(vec![
+ Field::new("values", list_of(DataType::Float64), false),
+ Field::new("partial_ends", list_of(DataType::Int64), false),
+ ])
+}
+
+/// One state row per group. Values past a group's last partial end came from
input rows, and
+/// the state hands them on as one more partial, as Spark's partial aggregate
does.
+fn build_state(group_values: &[Vec<f64>], group_partial_ends: &[Vec<usize>])
-> StructArray {
+ let num_values = group_values.iter().map(Vec::len).sum();
+ let num_ends = group_partial_ends.iter().map(|ends| ends.len() + 1).sum();
+ let mut values = Vec::with_capacity(num_values);
+ let mut ends = Vec::with_capacity(num_ends);
+ let mut value_offsets = Vec::with_capacity(group_values.len() + 1);
+ let mut end_offsets = Vec::with_capacity(group_values.len() + 1);
+ value_offsets.push(0);
+ end_offsets.push(0);
+ 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);
Review Comment:
[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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
