neilconway commented on code in PR #23827: URL: https://github.com/apache/datafusion/pull/23827#discussion_r3639279750
########## datafusion/functions-aggregate/src/min_max.rs: ########## Review Comment: This comment needs updating. ########## datafusion/functions-aggregate/benches/max_latency.rs: ########## @@ -0,0 +1,223 @@ +// 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. + +use datafusion_common::ScalarValue; +use rand::Rng; +use std::collections::VecDeque; +use std::time::{Duration, Instant}; + +// === Old Two-Stack Queue Implementation === +#[derive(Debug)] +pub struct TwoStackMax<T> { + push_stack: Vec<(T, T)>, + pop_stack: Vec<(T, T)>, +} + +impl<T: Clone + PartialOrd> TwoStackMax<T> { + pub fn new() -> Self { + Self { + push_stack: Vec::new(), + pop_stack: Vec::new(), + } + } + + pub fn push(&mut self, val: T) { + self.push_stack.push(match self.push_stack.last() { + Some((_, max)) => { + if val < *max { + (val, max.clone()) + } else { + (val.clone(), val) + } + } + None => (val.clone(), val), + }); + } + + pub fn pop(&mut self) -> Option<T> { + if self.pop_stack.is_empty() { + match self.push_stack.pop() { + Some((val, _)) => { + let mut last = (val.clone(), val); + self.pop_stack.push(last.clone()); + while let Some((val, _)) = self.push_stack.pop() { + let max = if last.1 > val { + last.1.clone() + } else { + val.clone() + }; + last = (val.clone(), max); + self.pop_stack.push(last.clone()); + } + } + None => return None, + } + } + self.pop_stack.pop().map(|(val, _)| val) + } +} + +// === Production Monotonic Deque Implementation === +#[derive(Debug)] +pub struct MonotonicMax<T> { + fifo: VecDeque<T>, + deque: VecDeque<T>, +} + +impl<T: Clone + PartialOrd> MonotonicMax<T> { + pub fn new() -> Self { + Self { + fifo: VecDeque::new(), + deque: VecDeque::new(), + } + } + + pub fn push(&mut self, val: T) { + self.fifo.push_back(val.clone()); + while self.deque.back().map_or(false, |back_val| *back_val < val) { + self.deque.pop_back(); + } + self.deque.push_back(val); + } + + pub fn pop(&mut self) -> Option<T> { + if let Some(popped) = self.fifo.pop_front() { + if self + .deque + .front() + .map_or(false, |front_val| *front_val == popped) + { + self.deque.pop_front(); + } + Some(popped) + } else { + None + } + } +} + +// === Generator Utilities === +fn generate_random_i64(size: usize) -> Vec<i64> { + let mut rng = rand::thread_rng(); + (0..size).map(|_| rng.gen_range(0..1000000)).collect() +} + +fn generate_random_strings(size: usize) -> Vec<String> { + let mut rng = rand::thread_rng(); + (0..size) + .map(|_| { + let len = rng.gen_range(10..40); + (0..len) + .map(|_| rng.gen_range(b'a'..=b'z') as char) + .collect() + }) + .collect() +} + +fn main() { + let dataset_size = 5000000; // 5M elements + println!("Generating raw inputs of size {}...", dataset_size); + let i64_raw = generate_random_i64(dataset_size); + let str_raw = generate_random_strings(dataset_size); + + let scalar_int_data: Vec<ScalarValue> = i64_raw + .iter() + .map(|&val| ScalarValue::Int64(Some(val))) + .collect(); + let scalar_utf8_data: Vec<ScalarValue> = str_raw + .iter() + .map(|val| ScalarValue::Utf8(Some(val.clone()))) + .collect(); + + let datasets = vec![ + ("scalar_int64", scalar_int_data), + ("scalar_utf8_string", scalar_utf8_data), + ]; + + println!("Starting end-to-end latency measurements..."); + for (label, data) in datasets { + println!("\n=================================================="); + println!("DATATYPE: {}", label); + println!("=================================================="); + + for window_size in [1000, 10000, 100000] { + println!(" --- Window Size: {} ---", window_size); + + // Two-Stack Max Latency + { + let mut q = TwoStackMax::new(); + let mut max_push_time = Duration::ZERO; + let mut max_pop_time = Duration::ZERO; + + let start_total = Instant::now(); + for (i, val) in data.iter().enumerate() { + let t0 = Instant::now(); + q.push(val.clone()); + let t_push = t0.elapsed(); + if t_push > max_push_time { + max_push_time = t_push; + } + + if i >= window_size { + let t0 = Instant::now(); + q.pop(); + let t_pop = t0.elapsed(); + if t_pop > max_pop_time { + max_pop_time = t_pop; + } + } + } + let total_duration = start_total.elapsed(); + println!( + " Two-Stack Queue: Total Time = {:?}, Max Push = {:?}, Max Pop = {:?}", + total_duration, max_push_time, max_pop_time + ); + } + + // Monotonic Deque Max Latency + { + let mut q = MonotonicMax::new(); + let mut max_push_time = Duration::ZERO; + let mut max_pop_time = Duration::ZERO; + + let start_total = Instant::now(); + for (i, val) in data.iter().enumerate() { + let t0 = Instant::now(); + q.push(val.clone()); + let t_push = t0.elapsed(); + if t_push > max_push_time { + max_push_time = t_push; + } + + if i >= window_size { + let t0 = Instant::now(); + q.pop(); + let t_pop = t0.elapsed(); + if t_pop > max_pop_time { + max_pop_time = t_pop; + } + } + } + let total_duration = start_total.elapsed(); + println!( + " Monotonic Deque: Total Time = {:?}, Max Push = {:?}, Max Pop = {:?}", + total_duration, max_push_time, max_pop_time + ); + } + } + } +} Review Comment: It is useful to report these statistics in the PR discussion, but I think we should omit the file from the PR itself. ########## datafusion/functions-aggregate/src/min_max.rs: ########## @@ -790,72 +790,55 @@ impl<T: Clone + PartialOrd> MovingMin<T> { #[inline] pub fn with_capacity(capacity: usize) -> Self { Self { - push_stack: Vec::with_capacity(capacity), - pop_stack: Vec::with_capacity(capacity), + fifo: std::collections::VecDeque::with_capacity(capacity), + deque: std::collections::VecDeque::with_capacity(capacity), } } /// Returns the minimum of the sliding window or `None` if the window is /// empty. #[inline] pub fn min(&self) -> Option<&T> { - match (self.push_stack.last(), self.pop_stack.last()) { - (None, None) => None, - (Some((_, min)), None) => Some(min), - (None, Some((_, min))) => Some(min), - (Some((_, a)), Some((_, b))) => Some(if a < b { a } else { b }), - } + self.deque.front() } /// Pushes a new element into the sliding window. #[inline] pub fn push(&mut self, val: T) { - self.push_stack.push(match self.push_stack.last() { - Some((_, min)) => { - if val > *min { - (val, min.clone()) - } else { - (val.clone(), val) - } - } - None => (val.clone(), val), - }); + self.fifo.push_back(val.clone()); + while self.deque.back().map_or(false, |back_val| *back_val > val) { + self.deque.pop_back(); + } + self.deque.push_back(val); } /// Removes and returns the last value of the sliding window. #[inline] pub fn pop(&mut self) -> Option<T> { Review Comment: I notice that we don't actually use the return value of `pop` in the sliding window accumulators. If we stop having `pop` return a value, I wonder if we can simplify the implementation -- e.g., get rid of `fifo` and just use a single `deque` that stores candidates in the window and their positions / sequence numbers. Something like ``` structure MovingMax<T>: candidates : deque of (seq, value) # front = oldest; values strictly decreasing push_seq : integer = 0 # total elements ever pushed pop_seq : integer = 0 # total elements ever popped # window = elements with seq in [pop_seq, push_seq) ``` On push(), we evict all the candidates that are dominated by the newly pushed value, if any. On pop(), if the popped element is the front of `candidates` (based on sequence number, not value), remove it. ########## datafusion/functions-aggregate/benches/sliding_min_max.rs: ########## @@ -0,0 +1,232 @@ +// 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. + +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use datafusion_common::ScalarValue; +use rand::Rng; +use std::collections::VecDeque; + +// === Old Two-Stack Queue Implementation === +#[derive(Debug)] +pub struct TwoStackMax<T> { + push_stack: Vec<(T, T)>, + pop_stack: Vec<(T, T)>, +} Review Comment: Generally, we like benchmarks to: 1. Directly test the production implementation, not a copy of it. If necessary we can refactor the production code to make this possible. 2. Only test the current version of the code, not prior versions or alternative implementations. We'd then establish the performance delta of the PR by comparing the benchmark results with and without the optimization applied, rather than having both old and new algorithms in the benchmark code that is committed to main. ########## datafusion/functions-aggregate/benches/sliding_min_max.rs: ########## @@ -0,0 +1,232 @@ +// 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. + +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use datafusion_common::ScalarValue; +use rand::Rng; +use std::collections::VecDeque; + +// === Old Two-Stack Queue Implementation === +#[derive(Debug)] +pub struct TwoStackMax<T> { + push_stack: Vec<(T, T)>, + pop_stack: Vec<(T, T)>, +} + +impl<T: Clone + PartialOrd> TwoStackMax<T> { + pub fn new() -> Self { + Self { + push_stack: Vec::new(), + pop_stack: Vec::new(), + } + } + + pub fn max(&self) -> Option<&T> { + match (self.push_stack.last(), self.pop_stack.last()) { + (None, None) => None, + (Some((_, max)), None) => Some(max), + (None, Some((_, max))) => Some(max), + (Some((_, a)), Some((_, b))) => Some(if a > b { a } else { b }), + } + } + + pub fn push(&mut self, val: T) { + self.push_stack.push(match self.push_stack.last() { + Some((_, max)) => { + if val < *max { + (val, max.clone()) + } else { + (val.clone(), val) + } + } + None => (val.clone(), val), + }); + } + + pub fn pop(&mut self) -> Option<T> { + if self.pop_stack.is_empty() { + match self.push_stack.pop() { + Some((val, _)) => { + let mut last = (val.clone(), val); + self.pop_stack.push(last.clone()); + while let Some((val, _)) = self.push_stack.pop() { + let max = if last.1 > val { + last.1.clone() + } else { + val.clone() + }; + last = (val.clone(), max); + self.pop_stack.push(last.clone()); + } + } + None => return None, + } + } + self.pop_stack.pop().map(|(val, _)| val) + } +} + +// === Production Monotonic Deque Implementation (storing T directly) === +#[derive(Debug)] +pub struct MonotonicMax<T> { + fifo: VecDeque<T>, + deque: VecDeque<T>, +} + +impl<T: Clone + PartialOrd> MonotonicMax<T> { + pub fn new() -> Self { + Self { + fifo: VecDeque::new(), + deque: VecDeque::new(), + } + } + + pub fn max(&self) -> Option<&T> { + self.deque.front() + } + + pub fn push(&mut self, val: T) { + self.fifo.push_back(val.clone()); + while self.deque.back().map_or(false, |back_val| *back_val < val) { + self.deque.pop_back(); + } + self.deque.push_back(val); + } + + pub fn pop(&mut self) -> Option<T> { + if let Some(popped) = self.fifo.pop_front() { + if self + .deque + .front() + .map_or(false, |front_val| *front_val == popped) + { + self.deque.pop_front(); + } + Some(popped) + } else { + None + } + } +} + +// === Generator Utilities === +fn generate_random_i64(size: usize) -> Vec<i64> { + let mut rng = rand::thread_rng(); Review Comment: We should see the RNG to ensure determinism; see how the other benchmarks work as an example. Also `rand::rngs::StdRng` not `thread_rng`. ########## datafusion/functions-aggregate/src/min_max.rs: ########## @@ -790,72 +790,55 @@ impl<T: Clone + PartialOrd> MovingMin<T> { #[inline] pub fn with_capacity(capacity: usize) -> Self { Self { - push_stack: Vec::with_capacity(capacity), - pop_stack: Vec::with_capacity(capacity), + fifo: std::collections::VecDeque::with_capacity(capacity), Review Comment: `use std::collections::VecDeque` would be a bit more readable. -- 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]
