sunchao commented on code in PR #6928:
URL: https://github.com/apache/datafusion-comet/pull/6928#discussion_r4253288053
##########
native/spark-expr/src/agg_funcs/percentile.rs:
##########
@@ -285,46 +321,228 @@ impl GroupsAccumulator for
SparkPercentileGroupsAccumulator {
.map(|values| values.capacity() * size_of::<f64>())
.sum::<usize>()
+ self.group_values.capacity() * size_of::<Vec<f64>>()
+ + self
+ .group_partial_ends
+ .iter()
+ .map(|ends| ends.capacity() * size_of::<usize>())
+ .sum::<usize>()
+ + self.group_partial_ends.capacity() * size_of::<Vec<usize>>()
}
}
-fn spark_percentile(values: &mut [f64], percentile: f64) -> Option<f64> {
+/// `partial_ends` marks where each partial state merged into `values` ends.
Values past the last
+/// one came straight from input rows, as in Spark's `Complete` mode.
+fn spark_percentile(
+ values: &mut [f64],
+ partial_ends: &[usize],
+ percentile: f64,
+ float_input: bool,
+) -> Option<f64> {
let len = values.len();
if len == 0 {
return None;
}
if len == 1 {
return Some(values[0]);
}
+ // Read before the selection below reorders the values.
+ let zeros = SignedZeros::find(values, partial_ends, float_input);
Review Comment:
[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.
##########
native/spark-expr/src/agg_funcs/percentile.rs:
##########
@@ -285,46 +321,228 @@ impl GroupsAccumulator for
SparkPercentileGroupsAccumulator {
.map(|values| values.capacity() * size_of::<f64>())
.sum::<usize>()
+ self.group_values.capacity() * size_of::<Vec<f64>>()
+ + self
+ .group_partial_ends
+ .iter()
+ .map(|ends| ends.capacity() * size_of::<usize>())
+ .sum::<usize>()
+ + self.group_partial_ends.capacity() * size_of::<Vec<usize>>()
}
}
-fn spark_percentile(values: &mut [f64], percentile: f64) -> Option<f64> {
+/// `partial_ends` marks where each partial state merged into `values` ends.
Values past the last
+/// one came straight from input rows, as in Spark's `Complete` mode.
+fn spark_percentile(
+ values: &mut [f64],
+ partial_ends: &[usize],
+ percentile: f64,
+ float_input: bool,
+) -> Option<f64> {
let len = values.len();
if len == 0 {
return None;
}
if len == 1 {
return Some(values[0]);
}
+ // Read before the selection below reorders the values.
+ let zeros = SignedZeros::find(values, partial_ends, float_input);
let position = (len - 1) as f64 * percentile;
let lower = position.floor() as usize;
let higher = position.ceil() as usize;
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)
}
+/// 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 {
+ fn find(values: &[f64], partial_ends: &[usize], float_input: bool) -> Self
{
+ if !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;
+ }
+ }
+ if positive == 0 {
+ return Self::default();
+ }
+ let first_negative = spark_iterates_negative_zero_first(values,
partial_ends, 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.
+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));
+ start = end;
+ }
+ values[start..].iter().for_each(|&v| table.add(v));
+ let first_zero = table.keys().find(|&v| v == 0.0);
+ first_zero.is_some_and(f64::is_sign_negative)
+}
+
+/// The slot layout of Spark's `OpenHashSet` with the default capacity and
load factor, holding
+/// boxed `Double` or `Float` keys.
+struct SparkKeySet {
+ slots: Vec<Option<f64>>,
+ size: usize,
+ float_input: bool,
+}
+
+impl SparkKeySet {
+ const INITIAL_CAPACITY: usize = 64;
+ const LOAD_FACTOR: f64 = 0.7;
+
+ fn new(float_input: bool) -> Self {
+ Self {
+ slots: vec![None; Self::INITIAL_CAPACITY],
+ size: 0,
+ float_input,
+ }
+ }
+
+ /// `Double.doubleToLongBits` or `Float.floatToIntBits`, which `equals`
and `hashCode` use:
+ /// the bits of the value, with every NaN turned into the same one.
+ fn key_bits(&self, v: f64) -> u64 {
+ match (self.float_input, v.is_nan()) {
+ (true, true) => f32::NAN.to_bits() as u64,
+ (true, false) => (v as f32).to_bits() as u64,
+ (false, true) => f64::NAN.to_bits(),
+ (false, false) => v.to_bits(),
+ }
+ }
+
+ /// `Double.hashCode` or `Float.hashCode`, mixed with murmur3 as
`OpenHashSet.hashcode` does.
+ fn slot(&self, v: f64, mask: usize) -> usize {
+ let bits = self.key_bits(v);
+ let hash = (bits ^ (bits >> 32)) as i32;
+ spark_compatible_murmur3_hash(hash.to_le_bytes(), 0) as usize & mask
+ }
+
+ fn add(&mut self, v: f64) {
+ let mask = self.slots.len() - 1;
+ let bits = self.key_bits(v);
+ let mut pos = self.slot(v, mask);
+ let mut delta = 1;
+ while let Some(existing) = self.slots[pos] {
+ if self.key_bits(existing) == bits {
Review Comment:
[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.
##########
native/spark-expr/src/agg_funcs/percentile.rs:
##########
@@ -238,6 +266,8 @@ impl GroupsAccumulator for SparkPercentileGroupsAccumulator
{
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);
Review Comment:
[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.
--
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]
