From 29952cce28902c7aad4faf6776576811b88e015b Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Mon, 28 Sep 2026 14:15:04 +0530 Subject: [PATCH 01/13] fix(physical-plan): replace approximate map_size accounting with exact HashTable::allocation_size() in GroupValuesPrimitive --- .../group_values/single_group_by/primitive.rs | 89 ++++++++++++++++++- 1 file changed, 88 insertions(+), 1 deletion(-) diff --git a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs index 21b62457e3831..23b2ae43f7673 100644 --- a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs +++ b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs @@ -178,7 +178,7 @@ where } fn size(&self) -> usize { - self.map.capacity() * size_of::<(usize, u64)>() + self.values.allocated_size() + self.map.allocation_size() + self.values.allocated_size() } fn is_empty(&self) -> bool { @@ -322,4 +322,91 @@ mod tests { Ok(()) } + + #[test] + fn test_exact_hash_table_allocation_accounting() -> Result<()> { + let mut group_values = GroupValuesPrimitive::::new(DataType::Int32); + + // 1. Initial state: map is unallocated + assert_eq!(group_values.map.capacity(), 0); + assert_eq!(group_values.map.allocation_size(), 0); + assert_eq!( + group_values.size(), + group_values.values.allocated_size() + group_values.map.allocation_size() + ); + + // 2. Insert keys and observe table allocation growth + let num_distinct = 500; + let arr: ArrayRef = + Arc::new(Int32Array::from_iter_values(0..num_distinct as i32)); + let mut groups = vec![]; + group_values.intern(&[arr], &mut groups)?; + assert_eq!(groups.len(), num_distinct as usize); + + let table_bytes = group_values.map.allocation_size(); + assert!(group_values.map.capacity() >= num_distinct as usize); + assert!(table_bytes > 0); + + // Hashbrown's layout contains entry slots, control bytes, and trailing group bytes. + // It must strictly exceed the naive entry-only size + let entry_only_bytes = + group_values.map.capacity() * std::mem::size_of::<(usize, u64)>(); + assert!( + table_bytes > entry_only_bytes, + "allocation_size ({table_bytes}) must exceed entry-only capacity ({entry_only_bytes})" + ); + + assert_eq!( + group_values.size(), + group_values.values.allocated_size() + table_bytes + ); + + // 3. Partial emit (EmitTo::First) + let emit_count = 200; + let emitted = group_values.emit(EmitTo::First(emit_count))?; + assert_eq!(emitted[0].len(), emit_count); + assert_eq!(group_values.len(), (num_distinct as usize) - emit_count); + + // Retained capacity is still charged until the table releases it + assert_eq!(group_values.map.allocation_size(), table_bytes); + assert_eq!( + group_values.size(), + group_values.values.allocated_size() + table_bytes + ); + + // 4. Emit all (EmitTo::All) + let remaining_count = group_values.len(); + let emitted_all = group_values.emit(EmitTo::All)?; + assert_eq!(emitted_all[0].len(), remaining_count); + assert_eq!(group_values.len(), 0); + + // Clear preserves table capacity + assert_eq!(group_values.map.allocation_size(), table_bytes); + + // 5. clear_shrink: releases/shrinks map capacity + group_values.clear_shrink(16); + let shrunken_table_bytes = group_values.map.allocation_size(); + assert!( + shrunken_table_bytes < table_bytes, + "clear_shrink should shrink table from {table_bytes} to {shrunken_table_bytes}" + ); + assert_eq!( + group_values.size(), + group_values.values.allocated_size() + shrunken_table_bytes + ); + + // 6. Reinsert new distinct values into the shrunken table + let new_arr: ArrayRef = Arc::new(Int32Array::from_iter_values( + (num_distinct as i32)..(num_distinct as i32 * 2), + )); + group_values.intern(&[new_arr], &mut groups)?; + assert_eq!(groups.len(), num_distinct as usize); + assert!(group_values.map.allocation_size() > shrunken_table_bytes); + assert_eq!( + group_values.size(), + group_values.values.allocated_size() + group_values.map.allocation_size() + ); + + Ok(()) + } } From 472fd7d5095258bf1fe8bbd75d10158cab88df51 Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Tue, 29 Sep 2026 22:31:30 +0530 Subject: [PATCH 02/13] fix(physical-plan): use imported size_of instead of fully-qualified std::mem::size_of in test The test added in the exact HashTable accounting fix used std::mem::size_of::<(usize, u64)>(), but size_of is already in scope via use std::mem::size_of at the top of the file. This trips -D unused-qualifications on Linux CI (which runs with -D warnings), causing the linux build test job to fail. Use the bare size_of::<(usize, u64)>() to stay consistent with the import and satisfy the lint. --- .../src/aggregates/group_values/single_group_by/primitive.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs index 23b2ae43f7673..76f90bb107b84 100644 --- a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs +++ b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs @@ -349,8 +349,7 @@ mod tests { // Hashbrown's layout contains entry slots, control bytes, and trailing group bytes. // It must strictly exceed the naive entry-only size - let entry_only_bytes = - group_values.map.capacity() * std::mem::size_of::<(usize, u64)>(); + let entry_only_bytes = group_values.map.capacity() * size_of::<(usize, u64)>(); assert!( table_bytes > entry_only_bytes, "allocation_size ({table_bytes}) must exceed entry-only capacity ({entry_only_bytes})" From 0ee077255c63cc502fa75ffc1fe1bc414087775d Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Thu, 1 Oct 2026 14:27:06 +0530 Subject: [PATCH 03/13] fix(test): correct initial-state assertions in allocation accounting test GroupValuesPrimitive::new() calls HashTable::with_capacity(128), which immediately reserves backing storage. The previous assertions treated the map as unallocated (capacity == 0, allocation_size == 0), causing the test to fail on the very first check. Instead, capture the actual nonzero initial allocation_size() returned by the constructor and assert that: - map.capacity() >= 128 (matching the with_capacity argument) - allocation_size() > 0 (pre-allocation is real) - size() == values.allocated_size() + initial_map_bytes This makes the focused regression test pass while still exercising the exact accounting that the broader fix (using allocation_size() instead of the naive capacity * size_of formula) is meant to validate. --- .../group_values/single_group_by/primitive.rs | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs index 76f90bb107b84..6d085594944ff 100644 --- a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs +++ b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs @@ -327,12 +327,23 @@ mod tests { fn test_exact_hash_table_allocation_accounting() -> Result<()> { let mut group_values = GroupValuesPrimitive::::new(DataType::Int32); - // 1. Initial state: map is unallocated - assert_eq!(group_values.map.capacity(), 0); - assert_eq!(group_values.map.allocation_size(), 0); + // 1. Initial state: the constructor calls HashTable::with_capacity(128), + // so the map already has a nonzero retained allocation. Capture it + // and verify that size() accounts for it correctly. + let initial_map_bytes = group_values.map.allocation_size(); + assert!( + group_values.map.capacity() >= 128, + "expected initial capacity >= 128, got {}", + group_values.map.capacity(), + ); + assert!( + initial_map_bytes > 0, + "expected nonzero initial allocation_size, got 0", + ); assert_eq!( group_values.size(), - group_values.values.allocated_size() + group_values.map.allocation_size() + group_values.values.allocated_size() + initial_map_bytes, + "size() must include the pre-allocated map bytes", ); // 2. Insert keys and observe table allocation growth From d91db6374855fb6d6cbd6e3a4d7e7342a13ead56 Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Thu, 1 Oct 2026 15:59:16 +0530 Subject: [PATCH 04/13] fix(clippy): move size_of import into test module After replacing the naive capacity * size_of formula with HashTable::allocation_size() in fn size(), size_of is no longer used in production code. The module-level import then triggers an unused-imports warning that clippy -D warnings promotes to an error on the lib target. Move the import into mod tests where it is actually consumed, so the lib and test targets are both clean. --- .../src/aggregates/group_values/single_group_by/primitive.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs index 6d085594944ff..3ee597f1c1153 100644 --- a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs +++ b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs @@ -31,7 +31,6 @@ use half::f16; use hashbrown::hash_table::HashTable; #[cfg(not(feature = "force_hash_collisions"))] use std::hash::BuildHasher; -use std::mem::size_of; use std::sync::Arc; /// A trait to allow hashing of floating point numbers @@ -283,6 +282,7 @@ mod tests { use arrow::array::{ArrayRef, Int32Array}; use arrow::datatypes::DataType; use datafusion_expr::EmitTo; + use std::mem::size_of; use std::sync::Arc; /// Mirror of the `EmitTo::take_needed` regression test, applied to the From a22792aee2c07b280ce3db0365738f931b2ea80c Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Thu, 1 Oct 2026 16:46:07 +0530 Subject: [PATCH 05/13] fix(clippy): remove redundant imports from test module ArrayRef, DataType, EmitTo, and Arc are already imported in the parent module and brought into scope via 'use super::*'. This caused clippy to flag them as unused imports on the test target. --- .../src/aggregates/group_values/single_group_by/primitive.rs | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs index 3ee597f1c1153..ab369b9850ca0 100644 --- a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs +++ b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs @@ -279,11 +279,8 @@ where mod tests { use super::*; use arrow::array::types::Int32Type; - use arrow::array::{ArrayRef, Int32Array}; - use arrow::datatypes::DataType; - use datafusion_expr::EmitTo; + use arrow::array::Int32Array; use std::mem::size_of; - use std::sync::Arc; /// Mirror of the `EmitTo::take_needed` regression test, applied to the /// concrete `GroupValuesPrimitive` accumulator. From d4d0442d0c84cfeb65f56224db4c4218944704d6 Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Thu, 1 Oct 2026 16:57:20 +0530 Subject: [PATCH 06/13] fix(test): restore accidentally removed test imports in primitive.rs The previous commit removed ArrayRef, DataType, EmitTo, and Arc from mod tests under the incorrect assumption that use super::* re-exports private use declarations from the parent module. In Rust, glob imports only propagate pub items, so each of these must be imported explicitly. All four types are used directly in the test function bodies: - ArrayRef: type annotation on let arr bindings - DataType: passed to GroupValuesPrimitive::new(DataType::Int32) - EmitTo: EmitTo::First(n) and EmitTo::All arm in emit() calls - Arc: Arc::new(...) wrapping Int32Array values Removing them caused a compile error on the test target. --- .../src/aggregates/group_values/single_group_by/primitive.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs index ab369b9850ca0..3ee597f1c1153 100644 --- a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs +++ b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs @@ -279,8 +279,11 @@ where mod tests { use super::*; use arrow::array::types::Int32Type; - use arrow::array::Int32Array; + use arrow::array::{ArrayRef, Int32Array}; + use arrow::datatypes::DataType; + use datafusion_expr::EmitTo; use std::mem::size_of; + use std::sync::Arc; /// Mirror of the `EmitTo::take_needed` regression test, applied to the /// concrete `GroupValuesPrimitive` accumulator. From d7648458d1ff23d1ce8a9c62f33a754f46cd6580 Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Thu, 1 Oct 2026 17:14:03 +0530 Subject: [PATCH 07/13] test(sqllogictest): relax hardcoded spill_count in aggregate_memory_spill.slt The exact spill count depends on when the memory budget is crossed, which shifts with any improvement to memory accounting precision. Switching GroupValuesPrimitive::size() from the naive capacity * size_of formula to HashTable::allocation_size() reports a slightly larger (more accurate) figure, which moves the spill trigger and changes the per-query spill count. The test comment already states 'spill_count metric must be > 0': the intent is to verify that spilling happened, not to pin to a specific count. Use spill_count= to accept any positive value, consistent with how Case G already handles its non-deterministic counts. --- .../test_files/aggregate_memory_spill.slt | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt b/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt index 48622127bce99..1a3987441faff 100644 --- a/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt +++ b/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt @@ -61,7 +61,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[spill_count=7,] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[spill_count=] # --- Case B: multi-column GROUP BY (is_single() = false) --- @@ -87,7 +87,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000), v@0 * 13 % 100000 as t.v * Int64(13) % Int64(100000)], aggr=[sum(t.v)], metrics=[spill_count=7,] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000), v@0 * 13 % 100000 as t.v * Int64(13) % Int64(100000)], aggr=[sum(t.v)], metrics=[spill_count=] # --- Case C: DISTINCT aggregate under memory limit --- @@ -113,7 +113,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[group_alias_0@0 as group_alias_0], aggr=[count(alias1)], metrics=[spill_count=7,] +04)------AggregateExec: mode=Single, gby=[group_alias_0@0 as group_alias_0], aggr=[count(alias1)], metrics=[spill_count=] # --- Case D: multiple aggregates (sum/min/max) under memory limit --- @@ -139,7 +139,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v), min(t.v), max(t.v)], metrics=[spill_count=7,] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v), min(t.v), max(t.v)], metrics=[spill_count=] # --- Case E: avg() aggregate (Float64 output) under memory limit --- @@ -165,7 +165,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[avg(t.v)], metrics=[spill_count=7,] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[avg(t.v)], metrics=[spill_count=] # --- Case F: array_agg() aggregate (growable state) under memory limit --- @@ -191,7 +191,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[array_agg(t.v)], metrics=[spill_count=7,] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[array_agg(t.v)], metrics=[spill_count=] # --- Case G: partial/final aggregation under memory limit --- From bcafb9304128897a2183209ca29993b52aeca5b2 Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Wed, 7 Oct 2026 18:54:12 +0530 Subject: [PATCH 08/13] test(sqllogictest): update peak_mem_used for GroupValuesPrimitive in explain_analyze.slt --- datafusion/sqllogictest/test_files/explain_analyze.slt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/datafusion/sqllogictest/test_files/explain_analyze.slt b/datafusion/sqllogictest/test_files/explain_analyze.slt index 89567a920de68..bb1ac16c5fee3 100644 --- a/datafusion/sqllogictest/test_files/explain_analyze.slt +++ b/datafusion/sqllogictest/test_files/explain_analyze.slt @@ -500,7 +500,7 @@ GROUP BY k; ---- Plan with Metrics 01)ProjectionExec: expr=[k@0 as k, count(Int64(1))@1 as count(*)], metrics=[output_bytes=1056.0 B] -02)--AggregateExec: mode=Single, gby=[k@0 as k], aggr=[count(Int64(1))], metrics=[output_bytes=1056.0 B, spilled_bytes=0.0 B, peak_mem_used=9.4 KB] +02)--AggregateExec: mode=Single, gby=[k@0 as k], aggr=[count(Int64(1))], metrics=[output_bytes=1056.0 B, spilled_bytes=0.0 B, peak_mem_used=10.9 KB] 03)----ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=32.0 B] 04)------DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] From 3a4d08432bd2132f5c377fe484269c9c09b5577e Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Wed, 7 Oct 2026 19:30:00 +0530 Subject: [PATCH 09/13] test(physical-plan): avoid partial branch and format-string coverage misses in primitive test --- .../group_values/single_group_by/primitive.rs | 42 ++++++------------- 1 file changed, 13 insertions(+), 29 deletions(-) diff --git a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs index ab746491d12a3..816b020c39fb9 100644 --- a/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs +++ b/datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs @@ -327,7 +327,7 @@ mod tests { } #[test] - fn test_exact_hash_table_allocation_accounting() -> Result<()> { + fn test_exact_hash_table_allocation_accounting() { let mut group_values = GroupValuesPrimitive::::new(DataType::Int32); let self_size = size_of::>(); @@ -335,26 +335,18 @@ mod tests { // so the map already has a nonzero retained allocation. Capture it // and verify that size() accounts for it correctly. let initial_map_bytes = group_values.map.allocation_size(); - assert!( - group_values.map.capacity() >= 128, - "expected initial capacity >= 128, got {}", - group_values.map.capacity(), - ); - assert!( - initial_map_bytes > 0, - "expected nonzero initial allocation_size, got 0", - ); + assert!(group_values.map.capacity() >= 128); + assert!(initial_map_bytes > 0); assert_eq!( group_values.size(), self_size + group_values.values.allocated_size() + initial_map_bytes, - "size() must include the pre-allocated map bytes", ); // 2. Insert keys and observe table allocation growth let num_distinct = 500_i32; let arr: ArrayRef = Arc::new(Int32Array::from_iter_values(0..num_distinct)); let mut groups = vec![]; - group_values.intern(&[arr], &mut groups)?; + group_values.intern(&[arr], &mut groups).unwrap(); assert_eq!(groups.len(), num_distinct as usize); let table_bytes = group_values.map.allocation_size(); @@ -364,19 +356,16 @@ mod tests { // Hashbrown's layout contains entry slots, control bytes, and trailing group bytes. // It must strictly exceed the naive entry-only size let entry_only_bytes = group_values.map.capacity() * size_of::<(usize, u64)>(); - assert!( - table_bytes > entry_only_bytes, - "allocation_size ({table_bytes}) must exceed entry-only capacity ({entry_only_bytes})" - ); + assert!(table_bytes > entry_only_bytes); assert_eq!( group_values.size(), - self_size + group_values.values.allocated_size() + table_bytes + self_size + group_values.values.allocated_size() + table_bytes, ); // 3. Partial emit (EmitTo::First) let emit_count = 200; - let emitted = group_values.emit(EmitTo::First(emit_count))?; + let emitted = group_values.emit(EmitTo::First(emit_count)).unwrap(); assert_eq!(emitted[0].len(), emit_count); assert_eq!(group_values.len(), (num_distinct as usize) - emit_count); @@ -384,12 +373,12 @@ mod tests { assert_eq!(group_values.map.allocation_size(), table_bytes); assert_eq!( group_values.size(), - self_size + group_values.values.allocated_size() + table_bytes + self_size + group_values.values.allocated_size() + table_bytes, ); // 4. Emit all (EmitTo::All) let remaining_count = group_values.len(); - let emitted_all = group_values.emit(EmitTo::All)?; + let emitted_all = group_values.emit(EmitTo::All).unwrap(); assert_eq!(emitted_all[0].len(), remaining_count); assert_eq!(group_values.len(), 0); @@ -399,30 +388,25 @@ mod tests { // 5. clear_shrink: releases/shrinks map capacity group_values.clear_shrink(16); let shrunken_table_bytes = group_values.map.allocation_size(); - assert!( - shrunken_table_bytes < table_bytes, - "clear_shrink should shrink table from {table_bytes} to {shrunken_table_bytes}" - ); + assert!(shrunken_table_bytes < table_bytes); assert_eq!( group_values.size(), - self_size + group_values.values.allocated_size() + shrunken_table_bytes + self_size + group_values.values.allocated_size() + shrunken_table_bytes, ); // 6. Reinsert new distinct values into the shrunken table let new_arr: ArrayRef = Arc::new(Int32Array::from_iter_values( num_distinct..(num_distinct * 2), )); - group_values.intern(&[new_arr], &mut groups)?; + group_values.intern(&[new_arr], &mut groups).unwrap(); assert_eq!(groups.len(), num_distinct as usize); assert!(group_values.map.allocation_size() > shrunken_table_bytes); assert_eq!( group_values.size(), self_size + group_values.values.allocated_size() - + group_values.map.allocation_size() + + group_values.map.allocation_size(), ); - - Ok(()) } #[test] From edb75126c2345a85eec18a9b9b17530884aaa16e Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Fri, 9 Oct 2026 18:20:15 +0530 Subject: [PATCH 10/13] fix(physical-plan): update aggregate_memory_spill tests with robust spilled_rows check --- .../test_files/aggregate_memory_spill.slt | 24 +++++++++---------- 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt b/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt index 1873428056b5e..7922f4e2816be 100644 --- a/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt +++ b/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt @@ -50,7 +50,7 @@ FROM ( ---- 100000 5000050000 -# Assert spill happened, the `spill_count` metric must be > 0 +# Assert spill happened, the `spilled_rows` metric must be > 0 query TT EXPLAIN ANALYZE SELECT count(*), sum(total) @@ -61,7 +61,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[spill_count=] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[spilled_rows=K,] # --- Case B: multi-column GROUP BY (is_single() = false) --- @@ -76,7 +76,7 @@ FROM ( ---- 100000 5000050000 -# Assert spill happened, the `spill_count` metric must be > 0 +# Assert spill happened, the `spilled_rows` metric must be > 0 query TT EXPLAIN ANALYZE SELECT count(*), sum(total) @@ -87,7 +87,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000), v@0 * 13 % 100000 as t.v * Int64(13) % Int64(100000)], aggr=[sum(t.v)], metrics=[spill_count=] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000), v@0 * 13 % 100000 as t.v * Int64(13) % Int64(100000)], aggr=[sum(t.v)], metrics=[spilled_rows=K,] # --- Case C: DISTINCT aggregate under memory limit --- @@ -102,7 +102,7 @@ FROM ( ---- 100000 100000 -# Assert spill happened, the `spill_count` metric must be > 0 +# Assert spill happened, the `spilled_rows` metric must be > 0 query TT EXPLAIN ANALYZE SELECT count(*), sum(d) @@ -113,7 +113,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[group_alias_0@0 as group_alias_0], aggr=[count(alias1)], metrics=[spill_count=] +04)------AggregateExec: mode=Single, gby=[group_alias_0@0 as group_alias_0], aggr=[count(alias1)], metrics=[spilled_rows=K,] # --- Case D: multiple aggregates (sum/min/max) under memory limit --- @@ -128,7 +128,7 @@ FROM ( ---- 100000 5000050000 1 100000 -# Assert spill happened, the `spill_count` metric must be > 0 +# Assert spill happened, the `spilled_rows` metric must be > 0 query TT EXPLAIN ANALYZE SELECT count(*), sum(s), min(mn), max(mx) @@ -139,7 +139,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v), min(t.v), max(t.v)], metrics=[spill_count=] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v), min(t.v), max(t.v)], metrics=[spilled_rows=K,] # --- Case E: avg() aggregate (Float64 output) under memory limit --- @@ -154,7 +154,7 @@ FROM ( ---- 100000 1 100000 -# Assert spill happened, the `spill_count` metric must be > 0 +# Assert spill happened, the `spilled_rows` metric must be > 0 query TT EXPLAIN ANALYZE SELECT count(*), min(a), max(a) @@ -165,7 +165,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[avg(t.v)], metrics=[spill_count=] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[avg(t.v)], metrics=[spilled_rows=K,] # --- Case F: array_agg() aggregate (growable state) under memory limit --- @@ -180,7 +180,7 @@ FROM ( ---- 100000 100000 -# Assert spill happened, the `spill_count` metric must be > 0 +# Assert spill happened, the `spilled_rows` metric must be > 0 query TT EXPLAIN ANALYZE SELECT count(*), sum(l) @@ -191,7 +191,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[array_agg(t.v)], metrics=[spill_count=] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[array_agg(t.v)], metrics=[spilled_rows=K,] # --- Case G: partial/final aggregation under memory limit --- From e29da7cd7681b1f3ccaaa1de0eafcab971dbe2ae Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Tue, 13 Oct 2026 16:18:39 +0530 Subject: [PATCH 11/13] fix(sqllogictest): use dual-anchor pattern for aggregate spill assertions The previous spilled_rows=K pattern had a false-positive risk: the leading wildcard could skip past spilled_rows= entirely and match the K suffix in output_rows=100.0 K from a different operator. This meant the assertion could pass even when spilled_rows=0. Replace with a two-anchor approach on the AggregateExec line: metrics=[spill_count=, spilled_rows= K,] How this is robust: - spill_count= is a literal anchor, so the leading wildcard cannot skip past it to a later metric or a different operator line. - spilled_rows= K relies on the fact that DataFusion only emits the K suffix for row counts >= 1000. When no spilling occurs, the output is spilled_rows=0 (no K-suffix), which cannot satisfy the K, literal, so the assertion correctly fails. Apply the same dual-anchor pattern to Case G (FinalPartitioned) and replace the extraneous RESET for a setting never SET in this file with the correct RESET datafusion.execution.target_partitions. --- .../test_files/aggregate_memory_spill.slt | 36 +++++++++++-------- 1 file changed, 22 insertions(+), 14 deletions(-) diff --git a/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt b/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt index 7922f4e2816be..b22edc6dcea4d 100644 --- a/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt +++ b/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt @@ -50,7 +50,10 @@ FROM ( ---- 100000 5000050000 -# Assert spill happened, the `spilled_rows` metric must be > 0 +# Assert spill happened: anchor on `spill_count=` so the leading wildcard +# cannot skip to a later metric, then verify `spilled_rows` has a K-suffix +# (only emitted for counts >= 1000). `spilled_rows=0` has no K-suffix, so the +# pattern fails when spilling does not occur. query TT EXPLAIN ANALYZE SELECT count(*), sum(total) @@ -61,7 +64,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[spilled_rows=K,] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[spill_count=, spilled_rows= K,] # --- Case B: multi-column GROUP BY (is_single() = false) --- @@ -76,7 +79,8 @@ FROM ( ---- 100000 5000050000 -# Assert spill happened, the `spilled_rows` metric must be > 0 +# Assert spill happened: spill_count anchors the search; spilled_rows K-suffix +# only appears for counts >= 1000, failing when no spill occurs. query TT EXPLAIN ANALYZE SELECT count(*), sum(total) @@ -87,7 +91,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000), v@0 * 13 % 100000 as t.v * Int64(13) % Int64(100000)], aggr=[sum(t.v)], metrics=[spilled_rows=K,] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000), v@0 * 13 % 100000 as t.v * Int64(13) % Int64(100000)], aggr=[sum(t.v)], metrics=[spill_count=, spilled_rows= K,] # --- Case C: DISTINCT aggregate under memory limit --- @@ -102,7 +106,8 @@ FROM ( ---- 100000 100000 -# Assert spill happened, the `spilled_rows` metric must be > 0 +# Assert spill happened: spill_count anchors the search; spilled_rows K-suffix +# only appears for counts >= 1000, failing when no spill occurs. query TT EXPLAIN ANALYZE SELECT count(*), sum(d) @@ -113,7 +118,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[group_alias_0@0 as group_alias_0], aggr=[count(alias1)], metrics=[spilled_rows=K,] +04)------AggregateExec: mode=Single, gby=[group_alias_0@0 as group_alias_0], aggr=[count(alias1)], metrics=[spill_count=, spilled_rows= K,] # --- Case D: multiple aggregates (sum/min/max) under memory limit --- @@ -128,7 +133,8 @@ FROM ( ---- 100000 5000050000 1 100000 -# Assert spill happened, the `spilled_rows` metric must be > 0 +# Assert spill happened: spill_count anchors the search; spilled_rows K-suffix +# only appears for counts >= 1000, failing when no spill occurs. query TT EXPLAIN ANALYZE SELECT count(*), sum(s), min(mn), max(mx) @@ -139,7 +145,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v), min(t.v), max(t.v)], metrics=[spilled_rows=K,] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v), min(t.v), max(t.v)], metrics=[spill_count=, spilled_rows= K,] # --- Case E: avg() aggregate (Float64 output) under memory limit --- @@ -154,7 +160,8 @@ FROM ( ---- 100000 1 100000 -# Assert spill happened, the `spilled_rows` metric must be > 0 +# Assert spill happened: spill_count anchors the search; spilled_rows K-suffix +# only appears for counts >= 1000, failing when no spill occurs. query TT EXPLAIN ANALYZE SELECT count(*), min(a), max(a) @@ -165,7 +172,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[avg(t.v)], metrics=[spilled_rows=K,] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[avg(t.v)], metrics=[spill_count=, spilled_rows= K,] # --- Case F: array_agg() aggregate (growable state) under memory limit --- @@ -180,7 +187,8 @@ FROM ( ---- 100000 100000 -# Assert spill happened, the `spilled_rows` metric must be > 0 +# Assert spill happened: spill_count anchors the search; spilled_rows K-suffix +# only appears for counts >= 1000, failing when no spill occurs. query TT EXPLAIN ANALYZE SELECT count(*), sum(l) @@ -191,7 +199,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[array_agg(t.v)], metrics=[spilled_rows=K,] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[array_agg(t.v)], metrics=[spill_count=, spilled_rows= K,] # --- Case G: partial/final aggregation under memory limit --- @@ -232,7 +240,7 @@ FROM ( ) ---- -06)----------AggregateExec: mode=FinalPartitioned, gby=[t.v * Int64(7) % Int64(100000)@0 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[spilled_rows=K,] +06)----------AggregateExec: mode=FinalPartitioned, gby=[t.v * Int64(7) % Int64(100000)@0 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[spill_count=, spilled_rows= K,] 08)--------------AggregateExec: mode=Partial, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[early_emit_count=] @@ -248,4 +256,4 @@ statement ok RESET datafusion.execution.batch_size statement ok -RESET datafusion.catalog.create_default_catalog_and_schema +RESET datafusion.execution.target_partitions From 7e3dd7cb23f8531f36d7a06f9872ca851ddfccc0 Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Tue, 13 Oct 2026 23:53:25 +0530 Subject: [PATCH 12/13] test(sqllogictest): restore RESET for catalog setting in aggregate_memory_spill.slt --- datafusion/sqllogictest/test_files/aggregate_memory_spill.slt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt b/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt index b22edc6dcea4d..ccf5ab1c19da5 100644 --- a/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt +++ b/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt @@ -256,4 +256,4 @@ statement ok RESET datafusion.execution.batch_size statement ok -RESET datafusion.execution.target_partitions +RESET datafusion.catalog.create_default_catalog_and_schema From bd90343205de63605e389f8a8a7d574d8d5c54eb Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Wed, 14 Oct 2026 18:07:32 +0530 Subject: [PATCH 13/13] test(sqllogictest): assert spilled_rows=100.0 K in aggregate_memory_spill.slt The previous pattern used K, to verify a positive spill count, but AggregateExec emits output_rows=100.0 K, before spill_count in its metrics string. That makes the leading wildcard capable of consuming output_rows=100.0 K, and the test would pass even when spill_count=0 -- the false-positive @kosiew identified. Fix: match spilled_bytes=, between spill_count and spilled_rows, and assert the literal spilled_rows=100.0 K, for Cases A-F. The workload writes 100k groups to disk on every spill, so the formatter always produces 100.0 K when spilling occurred and 0 (no K suffix) when it did not. --- .../test_files/aggregate_memory_spill.slt | 36 ++++++++----------- 1 file changed, 15 insertions(+), 21 deletions(-) diff --git a/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt b/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt index ccf5ab1c19da5..79d1213ce4316 100644 --- a/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt +++ b/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt @@ -50,10 +50,9 @@ FROM ( ---- 100000 5000050000 -# Assert spill happened: anchor on `spill_count=` so the leading wildcard -# cannot skip to a later metric, then verify `spilled_rows` has a K-suffix -# (only emitted for counts >= 1000). `spilled_rows=0` has no K-suffix, so the -# pattern fails when spilling does not occur. +# Assert spill happened: the aggregate spills all 100k rows, so +# spilled_rows is formatted as "100.0 K". This value cannot appear +# when spill_count=0 because spilled_rows would then be 0 (no K suffix). query TT EXPLAIN ANALYZE SELECT count(*), sum(total) @@ -64,7 +63,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[spill_count=, spilled_rows= K,] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[spill_count=, spilled_bytes=, spilled_rows=100.0 K,] # --- Case B: multi-column GROUP BY (is_single() = false) --- @@ -79,8 +78,7 @@ FROM ( ---- 100000 5000050000 -# Assert spill happened: spill_count anchors the search; spilled_rows K-suffix -# only appears for counts >= 1000, failing when no spill occurs. +# Assert spill happened (same reasoning as Case A). query TT EXPLAIN ANALYZE SELECT count(*), sum(total) @@ -91,7 +89,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000), v@0 * 13 % 100000 as t.v * Int64(13) % Int64(100000)], aggr=[sum(t.v)], metrics=[spill_count=, spilled_rows= K,] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000), v@0 * 13 % 100000 as t.v * Int64(13) % Int64(100000)], aggr=[sum(t.v)], metrics=[spill_count=, spilled_bytes=, spilled_rows=100.0 K,] # --- Case C: DISTINCT aggregate under memory limit --- @@ -106,8 +104,7 @@ FROM ( ---- 100000 100000 -# Assert spill happened: spill_count anchors the search; spilled_rows K-suffix -# only appears for counts >= 1000, failing when no spill occurs. +# Assert spill happened (same reasoning as Case A). query TT EXPLAIN ANALYZE SELECT count(*), sum(d) @@ -118,7 +115,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[group_alias_0@0 as group_alias_0], aggr=[count(alias1)], metrics=[spill_count=, spilled_rows= K,] +04)------AggregateExec: mode=Single, gby=[group_alias_0@0 as group_alias_0], aggr=[count(alias1)], metrics=[spill_count=, spilled_bytes=, spilled_rows=100.0 K,] # --- Case D: multiple aggregates (sum/min/max) under memory limit --- @@ -133,8 +130,7 @@ FROM ( ---- 100000 5000050000 1 100000 -# Assert spill happened: spill_count anchors the search; spilled_rows K-suffix -# only appears for counts >= 1000, failing when no spill occurs. +# Assert spill happened (same reasoning as Case A). query TT EXPLAIN ANALYZE SELECT count(*), sum(s), min(mn), max(mx) @@ -145,7 +141,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v), min(t.v), max(t.v)], metrics=[spill_count=, spilled_rows= K,] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v), min(t.v), max(t.v)], metrics=[spill_count=, spilled_bytes=, spilled_rows=100.0 K,] # --- Case E: avg() aggregate (Float64 output) under memory limit --- @@ -160,8 +156,7 @@ FROM ( ---- 100000 1 100000 -# Assert spill happened: spill_count anchors the search; spilled_rows K-suffix -# only appears for counts >= 1000, failing when no spill occurs. +# Assert spill happened (same reasoning as Case A). query TT EXPLAIN ANALYZE SELECT count(*), min(a), max(a) @@ -172,7 +167,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[avg(t.v)], metrics=[spill_count=, spilled_rows= K,] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[avg(t.v)], metrics=[spill_count=, spilled_bytes=, spilled_rows=100.0 K,] # --- Case F: array_agg() aggregate (growable state) under memory limit --- @@ -187,8 +182,7 @@ FROM ( ---- 100000 100000 -# Assert spill happened: spill_count anchors the search; spilled_rows K-suffix -# only appears for counts >= 1000, failing when no spill occurs. +# Assert spill happened (same reasoning as Case A). query TT EXPLAIN ANALYZE SELECT count(*), sum(l) @@ -199,7 +193,7 @@ FROM ( ) ---- -04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[array_agg(t.v)], metrics=[spill_count=, spilled_rows= K,] +04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[array_agg(t.v)], metrics=[spill_count=, spilled_bytes=, spilled_rows=100.0 K,] # --- Case G: partial/final aggregation under memory limit --- @@ -240,7 +234,7 @@ FROM ( ) ---- -06)----------AggregateExec: mode=FinalPartitioned, gby=[t.v * Int64(7) % Int64(100000)@0 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[spill_count=, spilled_rows= K,] +06)----------AggregateExec: mode=FinalPartitioned, gby=[t.v * Int64(7) % Int64(100000)@0 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[spill_count=, spilled_bytes=, spilled_rows= K,] 08)--------------AggregateExec: mode=Partial, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[early_emit_count=]