zhangfengcdt commented on code in PR #6928:
URL: https://github.com/apache/datafusion-comet/pull/6928#discussion_r4285082084
##########
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));
Review Comment:
Fixed by modeling the merge stages. The state now carries `partial_depths`
next to `partial_ends`, and the serde sets a `merge_stage` flag for
`PartialMerge`, so each stage's partials are nested under one node and replayed
as Spark's `OpenHashMap` would merge them. Your 22-row example gives `+0.0` and
is checked in `nests_the_partials_of_each_merge_stage` and the new
`percentile_zero_tie.sql` query with `count(DISTINCT k)`. One remaining gap: a
spill inside a merge stage nests one level deeper than Spark; documented in the
code.
##########
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:
Documented as a limit rather than modeled. Early emit only happens on a
reservation failure, and at that point Spark's own sort-based fallback changes
its answer too, so matching it would need task identity per state row. The doc
comment lists this with the sort-based fallback.
--
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]
