You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/19 12:26:51 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-6370-004b8b4ca2613209910bd88fed2ecc00c6b94b54 created (now c7e6986f74)
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-6370-004b8b4ca2613209910bd88fed2ecc00c6b94b54
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at c7e6986f74 fix: count grouped integer SUM state in the aggregate's memory reservation (#6370)
This branch includes the following new commits:
new c7e6986f74 fix: count grouped integer SUM state in the aggregate's memory reservation (#6370)
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-comet) 01/01: fix: count grouped integer SUM state in the aggregate's memory reservation (#6370)
Posted by 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-6370-004b8b4ca2613209910bd88fed2ecc00c6b94b54
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit c7e6986f740d7715b53b2e81d49f3ad5733f09fa
Author: Andy Grove <ag...@apache.org>
AuthorDate: Mon Oct 19 09:12:56 2026 +0000
fix: count grouped integer SUM state in the aggregate's memory reservation (#6370)
The three SumInt*GroupsAccumulator impls returned size_of_val(self) from
GroupsAccumulator::size(), which covers the Vec headers but not the
per-group sums (16 bytes per group, plus has_all_nulls for Try).
DataFusion sizes a hash aggregate's reservation from these sizes, so
grouped integer sums reserved far less than they held. Count the
capacity of the state vectors, as SumDecimalGroupsAccumulator does.
The Accumulator impl for SparkBloomFilter had the same mistake and left
out its bit array, so include that too.
---
native/spark-expr/src/agg_funcs/sum_int.rs | 46 ++++++++++++++++++++--
.../src/bloom_filter/bloom_filter_agg.rs | 11 +++++-
.../spark-expr/src/bloom_filter/spark_bit_array.rs | 5 +++
.../src/bloom_filter/spark_bloom_filter.rs | 5 +++
4 files changed, 63 insertions(+), 4 deletions(-)
diff --git a/native/spark-expr/src/agg_funcs/sum_int.rs b/native/spark-expr/src/agg_funcs/sum_int.rs
index d3e367af72..5be284cc8b 100644
--- a/native/spark-expr/src/agg_funcs/sum_int.rs
+++ b/native/spark-expr/src/agg_funcs/sum_int.rs
@@ -541,7 +541,7 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorLegacy {
}
fn size(&self) -> usize {
- std::mem::size_of_val(self)
+ self.sums.capacity() * std::mem::size_of::<Option<i64>>()
}
}
@@ -696,7 +696,7 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorAnsi {
}
fn size(&self) -> usize {
- std::mem::size_of_val(self)
+ self.sums.capacity() * std::mem::size_of::<Option<i64>>()
}
}
@@ -904,7 +904,8 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorTry {
}
fn size(&self) -> usize {
- std::mem::size_of_val(self)
+ self.sums.capacity() * std::mem::size_of::<Option<i64>>()
+ + self.has_all_nulls.capacity() * std::mem::size_of::<bool>()
}
}
@@ -1029,6 +1030,45 @@ mod tests {
assert_eq!(acc.evaluate().unwrap(), ScalarValue::Int64(Some(60)));
}
+ /// DataFusion sizes a grouped aggregate's memory reservation from `GroupsAccumulator::size()`,
+ /// so it has to include the per-group state, not just the struct holding the `Vec` headers.
+ fn assert_size_covers_group_state(acc: &mut dyn GroupsAccumulator, bytes_per_group: usize) {
+ const NUM_GROUPS: usize = 1_000_000;
+ let values: ArrayRef = Arc::new(Int64Array::from_iter_values(0..NUM_GROUPS as i64));
+ let group_indices: Vec<usize> = (0..NUM_GROUPS).collect();
+ acc.update_batch(&[values], &group_indices, None, NUM_GROUPS)
+ .unwrap();
+ assert!(
+ acc.size() >= NUM_GROUPS * bytes_per_group,
+ "size() reported {} bytes for {NUM_GROUPS} groups",
+ acc.size()
+ );
+ }
+
+ #[test]
+ fn test_legacy_size_counts_group_state() {
+ assert_size_covers_group_state(
+ &mut SumIntGroupsAccumulatorLegacy::new(),
+ std::mem::size_of::<Option<i64>>(),
+ );
+ }
+
+ #[test]
+ fn test_ansi_size_counts_group_state() {
+ assert_size_covers_group_state(
+ &mut SumIntGroupsAccumulatorAnsi::new(),
+ std::mem::size_of::<Option<i64>>(),
+ );
+ }
+
+ #[test]
+ fn test_try_size_counts_group_state() {
+ assert_size_covers_group_state(
+ &mut SumIntGroupsAccumulatorTry::new(),
+ std::mem::size_of::<Option<i64>>() + std::mem::size_of::<bool>(),
+ );
+ }
+
/// Spark reports an integral SUM overflow as `long overflow` with the `try_add` suggestion,
/// on the update and merge paths of both the scalar and the grouped ANSI accumulators.
#[test]
diff --git a/native/spark-expr/src/bloom_filter/bloom_filter_agg.rs b/native/spark-expr/src/bloom_filter/bloom_filter_agg.rs
index adce71233a..2b50b40013 100644
--- a/native/spark-expr/src/bloom_filter/bloom_filter_agg.rs
+++ b/native/spark-expr/src/bloom_filter/bloom_filter_agg.rs
@@ -173,7 +173,7 @@ impl Accumulator for SparkBloomFilter {
}
fn size(&self) -> usize {
- std::mem::size_of_val(self)
+ std::mem::size_of_val(self) + self.heap_size()
}
fn state(&mut self) -> Result<Vec<ScalarValue>> {
@@ -222,4 +222,13 @@ mod tests {
ScalarValue::Binary(Some(_))
));
}
+
+ /// `size()` has to include the bit array the filter allocates up front, not only the struct.
+ #[test]
+ fn size_counts_bit_array() {
+ let num_bits = 1 << 20;
+ let num_hash = spark_bloom_filter::optimal_num_hash_functions(100, num_bits);
+ let acc = SparkBloomFilter::new(SparkBloomFilterVersion::V1, num_hash, num_bits, 0);
+ assert!(acc.size() >= num_bits as usize / 8);
+ }
}
diff --git a/native/spark-expr/src/bloom_filter/spark_bit_array.rs b/native/spark-expr/src/bloom_filter/spark_bit_array.rs
index 214a832c06..86130b66d4 100644
--- a/native/spark-expr/src/bloom_filter/spark_bit_array.rs
+++ b/native/spark-expr/src/bloom_filter/spark_bit_array.rs
@@ -62,6 +62,11 @@ impl SparkBitArray {
self.data.len()
}
+ /// Bytes allocated for the word buffer.
+ pub fn heap_size(&self) -> usize {
+ self.data.capacity() * std::mem::size_of::<u64>()
+ }
+
pub fn data(&self) -> &[u64] {
&self.data
}
diff --git a/native/spark-expr/src/bloom_filter/spark_bloom_filter.rs b/native/spark-expr/src/bloom_filter/spark_bloom_filter.rs
index 114b6be404..2c08274a99 100644
--- a/native/spark-expr/src/bloom_filter/spark_bloom_filter.rs
+++ b/native/spark-expr/src/bloom_filter/spark_bloom_filter.rs
@@ -278,6 +278,11 @@ impl SparkBloomFilter {
self.bits.cardinality()
}
+ /// Bytes allocated for the bit array.
+ pub fn heap_size(&self) -> usize {
+ self.bits.heap_size()
+ }
+
pub fn state_as_bytes(&self) -> Vec<u8> {
self.spark_serialization()
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org