zhangfengcdt commented on code in PR #6928:
URL: https://github.com/apache/datafusion-comet/pull/6928#discussion_r4259952329
##########
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:
Confirmed on 3.4.3 and 3.5.0. The serde now sets signed_zero_keys only on
spark >=3.5.2.
--
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]
