andygrove commented on code in PR #6928:
URL: https://github.com/apache/datafusion-comet/pull/6928#discussion_r4284151724
##########
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
Review Comment:
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()`.
--
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]
