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/07/24 12:20:22 UTC
(datafusion) branch gh-readonly-queue/main/pr-23838-5d1c3cdee976895dd80d28e520a0b31bf2bdf21b created (now 7dfeeb0412)
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-23838-5d1c3cdee976895dd80d28e520a0b31bf2bdf21b
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 7dfeeb0412 test (slt): add memory-limited aggregation sqllogictests (#23838)
This branch includes the following new commits:
new 7dfeeb0412 test (slt): add memory-limited aggregation sqllogictests (#23838)
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: test (slt): add memory-limited aggregation sqllogictests (#23838)
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-23838-5d1c3cdee976895dd80d28e520a0b31bf2bdf21b
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 7dfeeb041211907c487285e2756d438d2af7d28b
Author: Naman Modi <na...@gmail.com>
AuthorDate: Fri Jul 24 17:49:55 2026 +0530
test (slt): add memory-limited aggregation sqllogictests (#23838)
## Which issue does this PR close?
- Part of #22710.
## Rationale for this change
There's almost no slt coverage for grouped aggregation under a memory
limit, where the aggregate spills to disk and re-groups the spilled
state. #23657 covers the ordered path & this covers the unordered/hash
path.
## What changes are included in this PR?
- Adds `aggregate_memory_limit.slt` in which the high-cardinality "GROUP
BY" is run under a 1M limit, so the hash aggregate spills. The group key
is scrambled with `(v * 7) % 100000` (a bijection, so still 100000
groups) to keep the input unsorted; otherwise it takes the streaming
path and never spills.
- Scoped to a single partition (`target_partitions = 1`), so the spill
happens in one aggregate operator with no repartition.
- Cases cover different accumulator states: single-column, multi-column,
count(DISTINCT), sum/min/max, avg (widening), array_agg (growable).
## Are these changes tested?
This PR is tests. All pass locally.
## Are there any user-facing changes?
No.
---
.../test_files/aggregate_memory_spill.slt | 228 +++++++++++++++++++++
1 file changed, 228 insertions(+)
diff --git a/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt b/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt
new file mode 100644
index 0000000000..7615209255
--- /dev/null
+++ b/datafusion/sqllogictest/test_files/aggregate_memory_spill.slt
@@ -0,0 +1,228 @@
+# 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.
+
+# Memory-limited (spilling) grouped hash aggregation.
+#
+# High-cardinality GROUP BY under a tight memory limit: the aggregate spills to
+# disk, re-groups the spilled state, and must still return the right answer.
+#
+# The group key is scrambled with `(v * 7) % 100000` because generate_series is
+# sorted, which would take the streaming path that never spills. gcd(7, 100000)
+# = 1, so it's a bijection over 1..100000. Still 100000 groups, just unsorted,
+# so the hash table grows and spills.
+#
+# Each query aggregates over the grouped result, so the expected output is one
+# row. sum(1..100000) = 5000050000, and every v lands in one group, so the
+# per-group sums always add back to that total.
+
+# Single partition keeps the aggregation in one operator (no repartition).
+statement ok
+SET datafusion.execution.target_partitions = 1
+
+statement ok
+SET datafusion.runtime.memory_limit = '1M'
+
+# --- Case A: single-column high-cardinality GROUP BY ---
+query II
+SELECT count(*), sum(total)
+FROM (
+ SELECT (v * 7) % 100000 AS k, sum(v) AS total
+ FROM generate_series(1, 100000) AS t(v)
+ GROUP BY (v * 7) % 100000
+)
+----
+100000 5000050000
+
+# Prove the inner aggregate actually spills (else these tests would silently stop covering the spill path).
+# Only `spill_count` is pinned; the other metrics vary per run.
+query TT
+EXPLAIN ANALYZE
+SELECT count(*), sum(total)
+FROM (
+ SELECT (v * 7) % 100000 AS k, sum(v) AS total
+ FROM generate_series(1, 100000) AS t(v)
+ GROUP BY (v * 7) % 100000
+)
+----
+Plan with Metrics
+01)ProjectionExec: expr=[count(Int64(1))@0 as count(*), sum(total)@1 as sum(total)], metrics=[<slt:ignore>]
+02)--AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1)), sum(total)], metrics=[<slt:ignore>]
+03)----ProjectionExec: expr=[sum(t.v)@1 as total], metrics=[<slt:ignore>]
+04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[<slt:ignore>spill_count=9,<slt:ignore>]
+05)--------ProjectionExec: expr=[value@0 as v], metrics=[<slt:ignore>]
+06)----------LazyMemoryExec: partitions=1, batch_generators=[generate_series: start=1, end=100000, batch_size=8192], metrics=[<slt:ignore>]
+
+# --- Case B: multi-column GROUP BY (is_single() = false) ---
+# Both keys are bijections of v, so each (a, b) pair is unique: 100000 groups.
+query II
+SELECT count(*), sum(total)
+FROM (
+ SELECT (v * 7) % 100000 AS a, (v * 13) % 100000 AS b, sum(v) AS total
+ FROM generate_series(1, 100000) AS t(v)
+ GROUP BY (v * 7) % 100000, (v * 13) % 100000
+)
+----
+100000 5000050000
+
+# Assert this case spills too.
+query TT
+EXPLAIN ANALYZE
+SELECT count(*), sum(total)
+FROM (
+ SELECT (v * 7) % 100000 AS a, (v * 13) % 100000 AS b, sum(v) AS total
+ FROM generate_series(1, 100000) AS t(v)
+ GROUP BY (v * 7) % 100000, (v * 13) % 100000
+)
+----
+Plan with Metrics
+01)ProjectionExec: expr=[count(Int64(1))@0 as count(*), sum(total)@1 as sum(total)], metrics=[<slt:ignore>]
+02)--AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1)), sum(total)], metrics=[<slt:ignore>]
+03)----ProjectionExec: expr=[sum(t.v)@2 as total], metrics=[<slt:ignore>]
+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=[<slt:ignore>spill_count=11,<slt:ignore>]
+05)--------ProjectionExec: expr=[value@0 as v], metrics=[<slt:ignore>]
+06)----------LazyMemoryExec: partitions=1, batch_generators=[generate_series: start=1, end=100000, batch_size=8192], metrics=[<slt:ignore>]
+
+# --- Case C: DISTINCT aggregate under memory limit ---
+# One distinct value per group, so each count(DISTINCT v) = 1.
+query II
+SELECT count(*), sum(d)
+FROM (
+ SELECT (v * 7) % 100000 AS k, count(DISTINCT v) AS d
+ FROM generate_series(1, 100000) AS t(v)
+ GROUP BY (v * 7) % 100000
+)
+----
+100000 100000
+
+# Assert this case spills too.
+query TT
+EXPLAIN ANALYZE
+SELECT count(*), sum(d)
+FROM (
+ SELECT (v * 7) % 100000 AS k, count(DISTINCT v) AS d
+ FROM generate_series(1, 100000) AS t(v)
+ GROUP BY (v * 7) % 100000
+)
+----
+Plan with Metrics
+01)ProjectionExec: expr=[count(Int64(1))@0 as count(*), sum(d)@1 as sum(d)], metrics=[<slt:ignore>]
+02)--AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1)), sum(d)], metrics=[<slt:ignore>]
+03)----ProjectionExec: expr=[count(alias1)@1 as d], metrics=[<slt:ignore>]
+04)------AggregateExec: mode=Single, gby=[group_alias_0@0 as group_alias_0], aggr=[count(alias1)], metrics=[<slt:ignore>spill_count=18,<slt:ignore>]
+05)--------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as group_alias_0, v@0 as alias1], aggr=[], ordering_mode=Sorted, metrics=[<slt:ignore>]
+06)----------ProjectionExec: expr=[value@0 as v], metrics=[<slt:ignore>]
+07)------------LazyMemoryExec: partitions=1, batch_generators=[generate_series: start=1, end=100000, batch_size=8192], metrics=[<slt:ignore>]
+
+# --- Case D: multiple aggregates (sum/min/max) under memory limit ---
+# Each group holds a single v, so min(v) = max(v) = v within the group.
+query IIII
+SELECT count(*), sum(s), min(mn), max(mx)
+FROM (
+ SELECT (v * 7) % 100000 AS k, sum(v) AS s, min(v) AS mn, max(v) AS mx
+ FROM generate_series(1, 100000) AS t(v)
+ GROUP BY (v * 7) % 100000
+)
+----
+100000 5000050000 1 100000
+
+# Assert this case spills too.
+query TT
+EXPLAIN ANALYZE
+SELECT count(*), sum(s), min(mn), max(mx)
+FROM (
+ SELECT (v * 7) % 100000 AS k, sum(v) AS s, min(v) AS mn, max(v) AS mx
+ FROM generate_series(1, 100000) AS t(v)
+ GROUP BY (v * 7) % 100000
+)
+----
+Plan with Metrics
+01)ProjectionExec: expr=[count(Int64(1))@0 as count(*), sum(s)@1 as sum(s), min(mn)@2 as min(mn), max(mx)@3 as max(mx)], metrics=[<slt:ignore>]
+02)--AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1)), sum(s), min(mn), max(mx)], metrics=[<slt:ignore>]
+03)----ProjectionExec: expr=[sum(t.v)@1 as s, min(t.v)@2 as mn, max(t.v)@3 as mx], metrics=[<slt:ignore>]
+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=[<slt:ignore>spill_count=27,<slt:ignore>]
+05)--------ProjectionExec: expr=[value@0 as v], metrics=[<slt:ignore>]
+06)----------LazyMemoryExec: partitions=1, batch_generators=[generate_series: start=1, end=100000, batch_size=8192], metrics=[<slt:ignore>]
+
+# --- Case E: avg() aggregate (Float64 output) under memory limit ---
+# Each group holds a single v, so avg(v) = v within the group.
+query IRR
+SELECT count(*), min(a), max(a)
+FROM (
+ SELECT (v * 7) % 100000 AS k, avg(v) AS a
+ FROM generate_series(1, 100000) AS t(v)
+ GROUP BY (v * 7) % 100000
+)
+----
+100000 1 100000
+
+# Assert this case spills too.
+query TT
+EXPLAIN ANALYZE
+SELECT count(*), min(a), max(a)
+FROM (
+ SELECT (v * 7) % 100000 AS k, avg(v) AS a
+ FROM generate_series(1, 100000) AS t(v)
+ GROUP BY (v * 7) % 100000
+)
+----
+Plan with Metrics
+01)ProjectionExec: expr=[count(Int64(1))@0 as count(*), min(a)@1 as min(a), max(a)@2 as max(a)], metrics=[<slt:ignore>]
+02)--AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1)), min(a), max(a)], metrics=[<slt:ignore>]
+03)----ProjectionExec: expr=[avg(t.v)@1 as a], metrics=[<slt:ignore>]
+04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[avg(t.v)], metrics=[<slt:ignore>spill_count=11,<slt:ignore>]
+05)--------ProjectionExec: expr=[value@0 as v], metrics=[<slt:ignore>]
+06)----------LazyMemoryExec: partitions=1, batch_generators=[generate_series: start=1, end=100000, batch_size=8192], metrics=[<slt:ignore>]
+
+# --- Case F: array_agg() aggregate (growable state) under memory limit ---
+# Each group holds a single v, so array_length(array_agg(v)) = 1.
+query II
+SELECT count(*), sum(l)
+FROM (
+ SELECT (v * 7) % 100000 AS k, array_length(array_agg(v)) AS l
+ FROM generate_series(1, 100000) AS t(v)
+ GROUP BY (v * 7) % 100000
+)
+----
+100000 100000
+
+# Assert this case spills too.
+query TT
+EXPLAIN ANALYZE
+SELECT count(*), sum(l)
+FROM (
+ SELECT (v * 7) % 100000 AS k, array_length(array_agg(v)) AS l
+ FROM generate_series(1, 100000) AS t(v)
+ GROUP BY (v * 7) % 100000
+)
+----
+Plan with Metrics
+01)ProjectionExec: expr=[count(Int64(1))@0 as count(*), sum(l)@1 as sum(l)], metrics=[<slt:ignore>]
+02)--AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1)), sum(l)], metrics=[<slt:ignore>]
+03)----ProjectionExec: expr=[array_length(array_agg(t.v)@1) as l], metrics=[<slt:ignore>]
+04)------AggregateExec: mode=Single, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[array_agg(t.v)], metrics=[<slt:ignore>spill_count=10,<slt:ignore>]
+05)--------ProjectionExec: expr=[value@0 as v], metrics=[<slt:ignore>]
+06)----------LazyMemoryExec: partitions=1, batch_generators=[generate_series: start=1, end=100000, batch_size=8192], metrics=[<slt:ignore>]
+
+# Restore settings to slt runner defaults
+statement ok
+RESET datafusion.runtime.memory_limit
+
+statement ok
+SET datafusion.execution.target_partitions = 4
+
+statement ok
+RESET datafusion.catalog.create_default_catalog_and_schema
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org