comphead commented on code in PR #6928:
URL: https://github.com/apache/datafusion-comet/pull/6928#discussion_r4265451326
##########
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:
This models a partial buffer as every row of the partition inserted in
order, and the final buffer as each partial merged once. That only holds while
Spark's `ObjectHashAggregateExec` stays hash based. Once a task holds
`spark.sql.objectHashAggregate.sortBased.fallbackThreshold` groups (default
128), `ObjectAggregationIterator` falls back. Each group then gets a fresh
buffer, takes the rows that came after the fallback first, and then merges the
pre-fallback buffer back in after a serialize/deserialize round trip
(`SortBasedAggregator.findNextSortedGroup`). The same happens in the final
aggregate for the partials. In a Python model of these tables, I split one
group's rows at a fallback point, and its answer differed from this replay for
about 2% of random splits (model only, not run on Spark). So for a `GROUP BY`
with more than 128 groups per task, Comet can still pick the other zero,
deterministically. Could you either model the fallback or state the limit here
and in `signedZer
oKeysNote`?
##########
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:
`partial_ends` is recorded, shuffled and kept per group even when
`SignedZeros::find` can never read it. That covers Spark before 3.5.2
(`signed_zero_keys` false) and every non-floating child, since an integral or
decimal value cast to double is never `-0.0`. Each state row now carries an
extra list offset plus one `i64`, and each final group keeps one `usize` per
merged partial. For a high-cardinality `GROUP BY` with about one value per
group per partition, I expect this to roughly double the state shuffle bytes
and the final accumulator's memory (estimate, not measured). Could the serde
set the flag only for `FloatType | DoubleType` children, and the accumulators
skip the ends (emitting an empty `partial_ends`) when it is false?
##########
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:
These three tests repeat cases the SQL files already check against Spark.
`picks_the_zero_spark_iterates_first` uses the same rows and percentile as the
first query in `percentile_zero_tie.sql`. `follows_the_table_as_it_grows` is
the 500-key `test_percentile_zero_tie_large` case plus the small grouped table.
`keeps_the_sign_of_a_lone_zero` is `test_percentile_signed_zero` in
`percentile.sql`. Could they be removed, keeping the spill, partial and
replay-gating tests that SQL cannot pin?
##########
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:
The replay is allocated outside the memory pool and is large next to the
tracked values. `Option<f64>` takes 16 bytes per slot, a table holds 1.43 to
2.86 slots per distinct key, and `partial`, `deserialized` and `table` are all
alive inside the loop at lines 527-531. That is roughly 70 to 140 bytes per
distinct value against the 8 that `size()` reports (my arithmetic, not
measured). Storing the key bits in a `Vec<u64>` with `u64::MAX` as the empty
marker would halve the slot size and drop the repeated `key_bits(existing)` per
probe. `u64::MAX` is never a key, because NaNs canonicalize to `0x7ff8...` and
float keys fit in 32 bits. Dropping `partial` once `deserialized` is built
would also remove one table from the peak.
##########
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:
Per-partial replay ranks `-0.0` first in both arrival orders. Replaying the
rows as one partial ranks `0.0` first only when the 15-row file is merged
first. With the 5-row file first, it also ranks `-0.0` first. So this case only
catches a regression to the flattened replay for one shuffle order. Could the
comment say so? The Rust comment at `percentile.rs` line 699 says the reverse
("when the second is merged first"), while the test there flattens the first
partial 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]
