neilconway commented on code in PR #23827:
URL: https://github.com/apache/datafusion/pull/23827#discussion_r3651289281
##########
datafusion/functions-aggregate/src/min_max.rs:
##########
@@ -790,72 +785,83 @@ 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),
+ deque: VecDeque::with_capacity(capacity),
+ push_seq: 0,
+ pop_seq: 0,
}
}
/// 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().map(|(_, val)| val)
+ }
+
+ #[inline]
+ fn check_invariants(&self) {
+ debug_assert!(self.pop_seq <= self.push_seq);
+ debug_assert!(
+ self.deque
+ .front()
+ .is_none_or(|&(front_seq, _)| front_seq >= self.pop_seq)
+ );
+
+ #[cfg(debug_assertions)]
+ {
+ let mut iter = self.deque.iter();
+ if let Some(mut prev) = iter.next() {
+ for curr in iter {
+ debug_assert!(prev.0 < curr.0);
+ debug_assert!(prev.1 < curr.1);
+ prev = curr;
+ }
+ }
Review Comment:
I don't think an assertion that is this expensive is warranted, even for
debug-only builds -- I'd just omit it.
##########
datafusion/functions-aggregate/src/min_max.rs:
##########
@@ -909,71 +916,82 @@ impl<T: Clone + PartialOrd> MovingMax<T> {
#[inline]
pub fn with_capacity(capacity: usize) -> Self {
Self {
- push_stack: Vec::with_capacity(capacity),
- pop_stack: Vec::with_capacity(capacity),
+ deque: VecDeque::with_capacity(capacity),
+ push_seq: 0,
+ pop_seq: 0,
}
}
/// Returns the maximum of the sliding window or `None` if the window is
empty.
#[inline]
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 }),
+ self.deque.front().map(|(_, val)| val)
+ }
+
+ #[inline]
+ fn check_invariants(&self) {
+ debug_assert!(self.pop_seq <= self.push_seq);
+ debug_assert!(
+ self.deque
+ .front()
+ .is_none_or(|&(front_seq, _)| front_seq >= self.pop_seq)
+ );
+
+ #[cfg(debug_assertions)]
+ {
+ let mut iter = self.deque.iter();
+ if let Some(mut prev) = iter.next() {
+ for curr in iter {
+ debug_assert!(prev.0 < curr.0);
+ debug_assert!(prev.1 > curr.1);
+ prev = curr;
+ }
+ }
}
}
/// 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((_, max)) => {
- if val < *max {
- (val, max.clone())
- } else {
- (val.clone(), val)
- }
- }
- None => (val.clone(), val),
- });
+ let seq = self.push_seq;
+ self.push_seq += 1;
+ while self.deque.back().is_some_and(|back_val| back_val.1 <= val) {
+ self.deque.pop_back();
+ }
+ self.deque.push_back((seq, val));
+
+ self.check_invariants();
}
- /// Removes and returns the last value of the sliding window.
+ /// Removes the oldest value from the sliding window.
Review Comment:
Can we add a comment on the behavior for empty windows?
--
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]
