andygrove commented on code in PR #4801: URL: https://github.com/apache/datafusion-comet/pull/4801#discussion_r3552843073
########## native/spark-expr/src/agg_funcs/quantile_summaries.rs: ########## @@ -0,0 +1,440 @@ +// 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. + +//! A bit-for-bit port of Spark's `QuantileSummaries` (Greenwald-Khanna), the +//! sketch behind `approx_percentile` / `percentile_approx`. Kept free of any +//! DataFusion dependency so it can be unit-tested in isolation. +//! +//! Reference: `org.apache.spark.sql.catalyst.util.QuantileSummaries`. + +use std::collections::VecDeque; + +/// A single sampled statistic: the value, its minimum rank jump `g`, and the +/// maximum span of the rank `delta`. +#[derive(Debug, Clone, PartialEq)] +pub struct Stats { + pub value: f64, + pub g: i64, + pub delta: i64, +} + +#[derive(Debug, Clone)] +pub struct QuantileSummaries { + compress_threshold: usize, + relative_error: f64, + sampled: Vec<Stats>, + count: i64, + compressed: bool, + head_sampled: Vec<f64>, +} + +impl QuantileSummaries { + pub const DEFAULT_COMPRESS_THRESHOLD: usize = 10000; + pub const DEFAULT_HEAD_SIZE: usize = 50000; + + /// Mirrors Spark's `PercentileDigest` ctor which builds a summary with + /// `compressed = true`. + pub fn new(compress_threshold: usize, relative_error: f64) -> Self { + Self { + compress_threshold, + relative_error, + sampled: Vec::new(), + count: 0, + compressed: true, + head_sampled: Vec::new(), + } + } + + pub fn count(&self) -> i64 { + self.count + } + + /// Heap bytes held by this summary (excluding the struct itself). + pub fn heap_size(&self) -> usize { + self.sampled.capacity() * std::mem::size_of::<Stats>() + + self.head_sampled.capacity() * std::mem::size_of::<f64>() + } + + pub fn insert(&mut self, x: f64) { + self.head_sampled.push(x); + self.compressed = false; + if self.head_sampled.len() >= Self::DEFAULT_HEAD_SIZE { + self.with_head_buffer_inserted(); + if self.sampled.len() >= self.compress_threshold { + self.compress(); + } + } + } + + fn with_head_buffer_inserted(&mut self) { + if self.head_sampled.is_empty() { + return; + } + let mut current_count = self.count; + let mut sorted = std::mem::take(&mut self.head_sampled); + // Spark relies on `Array[Double].sorted`; use total ordering so the port + // is deterministic even in the presence of NaN (Spark's typical inputs + // are NaN-free). + sorted.sort_by(|a, b| a.total_cmp(b)); + + let mut new_samples: Vec<Stats> = Vec::new(); + let mut sample_idx = 0usize; + let mut ops_idx = 0usize; + while ops_idx < sorted.len() { + let current_sample = sorted[ops_idx]; + while sample_idx < self.sampled.len() + && self.sampled[sample_idx].value <= current_sample + { + new_samples.push(self.sampled[sample_idx].clone()); + sample_idx += 1; + } + current_count += 1; + let delta = if new_samples.is_empty() + || (sample_idx == self.sampled.len() && ops_idx == sorted.len() - 1) + { + 0 + } else { + // Spark uses `.toInt` here (unlike `.toLong` in merge/compress); we use i64 Review Comment: Confirmed: 3.4.3 / 3.5.8 / 4.0.1 / 4.1.0-preview2 all use `.toLong` on this line, matching our i64, so there is no divergence. Replaced the comment with that fact in cc1ab34. ########## native/spark-expr/src/agg_funcs/quantile_summaries.rs: ########## @@ -0,0 +1,440 @@ +// 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. + +//! A bit-for-bit port of Spark's `QuantileSummaries` (Greenwald-Khanna), the Review Comment: Rewrote the module header to spell out what is load-bearing for bit-identity (the 50000-value flush boundary, the integer arithmetic, the backward compress traversal, the merge interleave and delta adjustment, and the query ceil-rank sweep) and to state that the data-structure choices are free to be optimized. ########## native/spark-expr/src/agg_funcs/quantile_summaries.rs: ########## @@ -0,0 +1,440 @@ +// 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. + +//! A bit-for-bit port of Spark's `QuantileSummaries` (Greenwald-Khanna), the +//! sketch behind `approx_percentile` / `percentile_approx`. Kept free of any +//! DataFusion dependency so it can be unit-tested in isolation. +//! +//! Reference: `org.apache.spark.sql.catalyst.util.QuantileSummaries`. + +use std::collections::VecDeque; + +/// A single sampled statistic: the value, its minimum rank jump `g`, and the +/// maximum span of the rank `delta`. +#[derive(Debug, Clone, PartialEq)] Review Comment: Derived `Copy` on `Stats` and dropped the clones in `with_head_buffer_inserted`, `compress_immut`, and `merge`. -- 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]
