(datafusion) branch gh-readonly-queue/main/pr-23899-e660118996630bf7173b31a67e97dee97dac1fe4 created (now 88365ddd62)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-23899-e660118996630bf7173b31a67e97dee97dac1fe4
in repository https://gitbox.apache.org/repos/asf/datafusion.git


      at 88365ddd62 chore: adjust `size` accounting for `min_max` (#23899)

This branch includes the following new commits:

     new 88365ddd62 chore: adjust `size` accounting for `min_max` (#23899)

The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org

(datafusion) 01/01: chore: adjust `size` accounting for `min_max` (#23899)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23899-e660118996630bf7173b31a67e97dee97dac1fe4
in repository https://gitbox.apache.org/repos/asf/datafusion.git

commit 88365ddd62b17c1eabd20ed0b064f626f9e77686
Author: Oleks V <co...@users.noreply.github.com>
AuthorDate: Sat Jul 25 20:27:25 2026 -0700

    chore: adjust `size` accounting for `min_max` (#23899)
    
    ## Which issue does this PR close?
    
    <!--
    We generally require a GitHub issue to be filed for all bug fixes and
    enhancements and this helps us generate change logs for our releases.
    You can link an issue to this PR using the GitHub syntax. For example
    `Closes #123` indicates that this PR will close issue #123.
    -->
    
    - Part of #23393 .
    
       ## Rationale for this change
    
    `SlidingMinAccumulator::size` and `SlidingMaxAccumulator::size` only
    reported
    the stack size of their `ScalarValue` field plus its heap payload,
    ignoring the
    memory held by the underlying `MovingMin` / `MovingMax` sliding-window
    buffers.
    For windowed `MIN`/`MAX` over string or list data, the two per-element
    stacks
    can hold megabytes of `ScalarValue` payload that the memory pool was
    never
      told about, understating accumulator memory usage.
    
      ## What changes are included in this PR?
    
    - Add a private `heap_size(elem_heap)` method to `MovingMin<T>` and
    `MovingMax<T>`
    that reports the two stack buffers' capacity in bytes plus each stored
        element's heap payload.
    - Factor the shared implementation into a `moving_stacks_heap_size` free
    helper
        so the two types stay in sync.
      - Include the buffer bytes in `SlidingMinAccumulator::size` and
        `SlidingMaxAccumulator::size` via the new method.
    
      ## Are these changes tested?
    
    Yes. Two new unit tests in
    `datafusion/functions-aggregate/src/min_max.rs`:
    
    - `moving_min_max_heap_size_i32` — fixed-width `T`, verifies buffer-only
        accounting with and without pushed elements.
    - `moving_min_max_heap_size_counts_elems` — `T = String`, verifies each
    of the
    two slots in a `(T, T)` pair contributes independently to the heap
    payload
        (mirroring the two independent `Clone`s made by `push`).
---
 datafusion/functions-aggregate/src/min_max.rs | 83 ++++++++++++++++++++++++++-
 1 file changed, 80 insertions(+), 3 deletions(-)

diff --git a/datafusion/functions-aggregate/src/min_max.rs b/datafusion/functions-aggregate/src/min_max.rs
index f4eaaab853..1a3179170c 100644
--- a/datafusion/functions-aggregate/src/min_max.rs
+++ b/datafusion/functions-aggregate/src/min_max.rs
@@ -52,7 +52,7 @@ use datafusion_expr::{
 use datafusion_expr::{GroupsAccumulator, StatisticsArgs};
 use datafusion_macros::user_doc;
 use half::f16;
-use std::mem::size_of_val;
+use std::mem::{size_of, size_of_val};
 use std::ops::Deref;
 
 fn get_min_max_result_type(input_types: &[DataType]) -> Result<Vec<DataType>> {
@@ -433,7 +433,9 @@ impl Accumulator for SlidingMaxAccumulator {
     }
 
     fn size(&self) -> usize {
-        size_of_val(self) - size_of_val(&self.max) + self.max.size()
+        size_of_val(self) - size_of_val(&self.max)
+            + self.max.size()
+            + self.moving_max.heap_size(|sv| sv.size() - size_of_val(sv))
     }
 }
 
@@ -721,7 +723,9 @@ impl Accumulator for SlidingMinAccumulator {
     }
 
     fn size(&self) -> usize {
-        size_of_val(self) - size_of_val(&self.min) + self.min.size()
+        size_of_val(self) - size_of_val(&self.min)
+            + self.min.size()
+            + self.moving_min.heap_size(|sv| sv.size() - size_of_val(sv))
     }
 }
 
@@ -857,6 +861,30 @@ impl<T: Clone + PartialOrd> MovingMin<T> {
     pub fn is_empty(&self) -> bool {
         self.len() == 0
     }
+
+    /// Heap bytes owned by the two stack buffers plus each stored `T`'s
+    /// heap payload as reported by `elem_heap`. Excludes `size_of::<Self>()`.
+    #[inline]
+    fn heap_size(&self, elem_heap: impl Fn(&T) -> usize) -> usize {
+        moving_stacks_heap_size(&self.push_stack, &self.pop_stack, elem_heap)
+    }
+}
+
+/// Shared implementation for [`MovingMin::heap_size`] and
+/// [`MovingMax::heap_size`]. Both share the same two-stack layout.
+#[inline]
+fn moving_stacks_heap_size<T>(
+    push_stack: &Vec<(T, T)>,
+    pop_stack: &Vec<(T, T)>,
+    elem_heap: impl Fn(&T) -> usize,
+) -> usize {
+    let buffers = (push_stack.capacity() + pop_stack.capacity()) * size_of::<(T, T)>();
+    let elems: usize = push_stack
+        .iter()
+        .chain(pop_stack.iter())
+        .map(|(a, b)| elem_heap(a) + elem_heap(b))
+        .sum();
+    buffers + elems
 }
 
 /// Keep track of the maximum value in a sliding window.
@@ -975,6 +1003,13 @@ impl<T: Clone + PartialOrd> MovingMax<T> {
     pub fn is_empty(&self) -> bool {
         self.len() == 0
     }
+
+    /// Heap bytes owned by the two stack buffers plus each stored `T`'s
+    /// heap payload as reported by `elem_heap`. Excludes `size_of::<Self>()`.
+    #[inline]
+    fn heap_size(&self, elem_heap: impl Fn(&T) -> usize) -> usize {
+        moving_stacks_heap_size(&self.push_stack, &self.pop_stack, elem_heap)
+    }
 }
 
 make_udaf_expr_and_func!(
@@ -1213,6 +1248,48 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn moving_min_max_heap_size_i32() {
+        // Fixed-width `T` has no per-element heap payload, so `heap_size`
+        // reports exactly the two stack buffers' capacity in bytes.
+        let mut moving_min = MovingMin::<i32>::with_capacity(4);
+        let mut moving_max = MovingMax::<i32>::with_capacity(4);
+        let elem = |_: &i32| 0;
+
+        // Both stacks are `with_capacity(4)`, so total slots = 8.
+        let buffer_only = 2 * 4 * size_of::<(i32, i32)>();
+        assert_eq!(moving_min.heap_size(elem), buffer_only);
+        assert_eq!(moving_max.heap_size(elem), buffer_only);
+
+        for i in 0..3 {
+            moving_min.push(i);
+            moving_max.push(i);
+        }
+        // Elements sit inside the pre-allocated buffers, so still buffer-only.
+        assert_eq!(moving_min.heap_size(elem), buffer_only);
+        assert_eq!(moving_max.heap_size(elem), buffer_only);
+    }
+
+    #[test]
+    fn moving_min_max_heap_size_counts_elems() {
+        // Each buffered slot is a `(T, T)` pair, so a stored element is
+        // visited twice by `heap_size` — mirrors two independent `Clone`s.
+        let mut moving_min = MovingMin::<String>::with_capacity(2);
+        let mut moving_max = MovingMax::<String>::with_capacity(2);
+        let elem = |s: &String| s.capacity();
+
+        moving_min.push("abcdef".to_string());
+        moving_max.push("abcdef".to_string());
+
+        // Both `push_stack` and `pop_stack` allocate `capacity` slots.
+        let buffers = 2 * 2 * size_of::<(String, String)>();
+        // 2 slots per stored element (value + running extremum) times the
+        // per-element heap payload from `elem`.
+        let elems = 2 * 6;
+        assert_eq!(moving_min.heap_size(elem), buffers + elems);
+        assert_eq!(moving_max.heap_size(elem), buffers + elems);
+    }
+
     #[test]
     fn test_min_max_coerce_types() {
         // the coerced types is same with input types


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org