zhangfengcdt commented on code in PR #6928:
URL: https://github.com/apache/datafusion-comet/pull/6928#discussion_r4285080374
##########
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.
Review Comment:
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.
##########
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:
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`.
##########
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));
+ 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,
+}
Review Comment:
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.
##########
native/spark-expr/src/agg_funcs/percentile.rs:
##########
@@ -333,7 +636,137 @@ mod tests {
fn orders_nan_above_every_other_value() {
// A NaN with the sign bit set, which IEEE 754 total order would put
first.
let mut values = vec![f64::from_bits(0xfff8_0000_0000_0000), 2.0, 1.0];
- assert_eq!(spark_percentile(&mut values, 0.0), Some(1.0));
- assert!(spark_percentile(&mut values, 1.0).unwrap().is_nan());
+ assert_eq!(
+ spark_percentile(&mut values, &[], 0.0, zero_order(false)),
+ Some(1.0)
+ );
+ assert!(spark_percentile(&mut values, &[], 1.0, zero_order(false))
+ .unwrap()
+ .is_nan());
+ }
+
+ // The rows of Spark's SPARK-45599 test, for which Spark returns -0.0.
+ fn spark_45599_rows() -> Vec<f64> {
+ vec![
+ 0.0,
+ 2.0,
+ 153.0,
+ 168.0,
+ 3252411229536261.0,
+ 7.205759403792794e16,
+ f64::MAX,
+ 0.25,
+ f64::NAN,
+ f64::NAN,
+ -0.0,
+ -128.0,
+ f64::NAN,
+ f64::NAN,
+ ]
+ }
+
+ #[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());
+ }
Review Comment:
Removed the three tests and the matching fixture; the SQL files cover those
cases against Spark.
##########
spark/src/test/resources/sql-tests/expressions/aggregate/percentile_partials.sql:
##########
@@ -0,0 +1,45 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- MinSparkVersion: 3.5
+
+-- When a group holds both 0.0 and -0.0, Spark ranks first the zero its final
hash table reaches
+-- first, which depends on each partition's partial buffer, not only on the
rows. Each INSERT
+-- coalesces to one task, so it writes one file in row order, and an open cost
as large as a
+-- split keeps every file in its own partition.
+-- Spark ranks -0.0 first for these rows in either arrival order, while
replaying them as one
+-- partial ranks 0.0 first.
Review Comment:
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.
--
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]
