You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "andygrove (via GitHub)" <gi...@apache.org> on 2026/09/29 01:12:46 UTC
[PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
andygrove opened a new pull request, #25853:
URL: https://github.com/apache/datafusion/pull/25853
## Which issue does this PR close?
- Closes #25851.
## Rationale for this change
A grouped aggregate whose groups carry large state, such as `array_agg` over a low-cardinality key, fails with `ResourcesExhausted` after it spills, even though each group fits in memory and the same data spread over many small groups succeeds.
Spilling writes the aggregate state in batches of at most `batch_size` rows, whatever their size in bytes. With fewer groups than `batch_size`, every group lands in one batch, and everything after the spill has to hold that batch at once:
- With a single spill file, replay reads the batch directly and its table must hold every group at once. In the reproducer from the issue, that is 225 MB of state in a 128 MB pool, although each group is about 28 MB.
- With several spill files, the merge reserves twice the largest batch of each file. When that does not fit, it tries to split the file, which first reserves twice the batch it is about to split.
## What changes are included in this PR?
- Aggregate spill files are written in batches of at most `batch_size` rows and about 1 MiB (`SPILL_BATCH_TARGET_BYTES`). The number of rows per batch comes from the average row size of the spilled state, and is at least one. With the default `batch_size` of 8192, only rows averaging more than 128 bytes are written in smaller batches, so ordinary spills are unchanged.
- The merge that replays the spill files uses the smallest number of rows per spilled batch as its batch size. Otherwise it would combine the rows of the small spilled batches back into one large batch before replay.
This applies to every stream built on `AggregateSpill` (hash and ordered, final and single-stage). The legacy `GroupedHashAggregateStream`, used only when `datafusion.execution.enable_migration_aggregate` is `false`, is unchanged.
## What is the testing strategy for this PR?
- New Case H in `aggregate_memory_spill.slt`: `array_agg` over 32 groups of about 2 MB each with a 24 MB limit. It runs once with two partitions (the final aggregation spills, and replay reads back one run per partition, as in the issue) and once with one partition (the single-stage aggregation spills five runs, and replay merges them). `EXPLAIN ANALYZE` assertions check that each spilled. Both queries fail on `main` with `ResourcesExhausted` and pass with this change. Reverting only the merge batch size makes the single-partition query fail again.
- Unit test for how the rows per spilled batch are chosen: small rows keep `batch_size`, rows of a quarter of the target are written four to a batch, and a row larger than the target is written alone.
With `datafusion-cli -m 128M`, the reproducer from the issue now returns 16 groups and 3,000,000 values. It previously failed on every run. The variant from the issue's additional context, with partial aggregation skipped, also succeeds now.
`external_aggr` shows no change on any of its 8 cases, since its rows are far below the target. On a spilling `array_agg` over 200,000 groups of about 2.5 KB each, interleaved runs are 0.93x the time of `main` at 512 MB and 0.97x at 1 GB. At 512 MB, `main` also failed 7 of 15 runs with `ResourcesExhausted` and this change failed none. Spilling aggregates with narrow rows are unchanged (0.99x to 1.00x).
## Are there any user-facing changes?
Aggregations whose groups have large state can now complete under memory limits where they previously failed after spilling. Spill files of rows averaging more than 128 bytes (with the default `batch_size`) contain more, smaller batches. No API or configuration changes.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-6384133846
π€ Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/25853#issuecomment-6384101936)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c6384101936-3827-npbmq 6.12.105+ #1 SMP Sat Sep 5 07:54:11 UTC 2026 aarch64 GNU/Linux`
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
Comparing fix-issue-258521 (3289c00a6bfdf14b9ad444d7ff47b2b9364b1be2) to 1da3cf0 (merge-base) [diff](https://github.com/apache/datafusion/compare/1da3cf049fa1a7a708acc2c29eba4f2e9dece8db..3289c00a6bfdf14b9ad444d7ff47b2b9364b1be2)
<details><summary>Run configuration</summary>
```yaml
run benchmark external_aggr
```
</details>
Results will be posted here when complete
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#discussion_r4139961214
##########
datafusion/physical-plan/src/aggregates/spill.rs:
##########
@@ -280,6 +300,7 @@ impl AggregateSpill {
.with_expressions(&spill_expr)
.with_metrics(baseline_metrics.intermediate())
.with_batch_size(batch_size)
+ .with_max_batch_bytes(Some(SPILL_BATCH_TARGET_BYTES))
.with_reservation(merge_reservation)
.with_replay_headroom()
Review Comment:
Just a rough intuition: weβve accumulated quite a few special cases to make memory-limited execution work. There may be an opportunity to unify and simplify some of them.
##########
datafusion/physical-plan/src/aggregates/spill.rs:
##########
@@ -290,11 +311,141 @@ impl AggregateSpill {
partition,
merged,
&InputOrderMode::Sorted,
- baseline_metrics.clone(),
+ baseline_metrics.intermediate(),
metrics,
None,
reservation,
)?;
- Ok(replay.into_stream())
+ // Replay emits the groups of each merged batch, which may be few
+ let output = coalesce(
+ replay.into_stream(),
+ batch_size,
+ SPILL_BATCH_TARGET_BYTES,
+ baseline_metrics.elapsed_compute().clone(),
+ );
+ Ok(Box::pin(ObservedStream::new(
+ output,
+ baseline_metrics.clone(),
+ None,
+ )))
+ }
+}
+
+/// Combines consecutive small batches of `input` into batches of up to
+/// `batch_size` rows and `max_bytes`. Batches of at least half that size are
+/// passed on as they are.
+fn coalesce(
Review Comment:
IIRC, producing smaller batches is an internal issue for ordered aggregation, and it would be better solved within the operator rather than by adding a wrapper here.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-6222064846
@EmilyMatt I prototyped your suggestion on current main to check it: before inserting a batch, the final and single-stage hash aggregates spill their table if the table plus the batch would exceed a budget. I also tried inserting the input one row at a time, so that a large input batch can't force a large spill either.
On the reproducer from #25851 (`-m 128M`):
- Spilling before inserting never triggers. Each final partition receives a single 80-row batch that already holds the state of all 8 of its groups, so the table is empty when it arrives.
- With one-row inserts and budgets of 8, 16 and 32 MB, each partition writes 80 runs of about 2.4 MB, well within what reading back 2x needs. It still fails. The merge's intermediate passes emit batches of `batch_size` rows, and with fewer rows than that, each pass writes everything it merged as one batch. The runs grow from 2.4 MB to 112 MB, and splitting the 112 MB batch reserves twice its size: `Failed to allocate additional 225.1 MB`.
- The two-partition query in Case H (24 MB) fails at every budget I tried.
This PR passes both.
When the input arrives in small batches, as in the single-stage queries of Case H, spilling at 2β4 MB does let them pass on main, because the merge's split path eventually shrinks the batches of the runs it splits. But replay then runs one row per batch: the last Case H query spills 50 times instead of 6, emits 340K batches instead of 75, and takes about 1.1 s against 0.1 s with this PR.
So the size of the batches replay has to hold is set by how spill files and merges cut rows into batches, not by when the aggregate spills. That is what this PR changes. Reserving before inserting would still be worth doing on its own, since today the aggregate can overshoot its budget by a whole input batch before it spills, which is what #25606 tracks.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "EmilyMatt (via GitHub)" <gi...@apache.org>.
EmilyMatt commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-5929231866
While this can be a useful mechanism, I don't think it's the right solution for the problem at hand, I think the aggregate estimations can and should be improved, this ties us into behaviour that can be both bad for perf, and complex to maintain (and is also not very accurate since it uses averages)
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "EmilyMatt (via GitHub)" <gi...@apache.org>.
EmilyMatt commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-5990311644
> @EmilyMatt I agree the aggregate's estimates are worth improving, since they decide when it spills and whether it keeps room to sort and write the spill. But they don't fix this issue, which happens after the spill, when the spilled state is read back.
>
> A spill file is read back one batch at a time, and spilled batches are bounded only by `batch_size` rows. With fewer groups than `batch_size`, all the groups of a spill land in one batch, however early the aggregate spills, and the replay's ordered aggregate, which cannot spill, has to hold that whole batch. In the reproducer in #25851, each final partition receives one input batch that already holds all of its state, so the final aggregate spills on that first batch and writes its 8 groups as one 184 MB batch. Reading it back needs 225 MB in a 128 MB pool, although any one group fits. No reservation changes the size of that batch. With several spill files, the merge has the same problem: it combines rows until it has `batch_size` of them, so it rebuilds batches of large groups even from small spilled batches.
>
> On the other points:
>
> * Performance: the TPC-H and TPC-DS runs above show no slowdowns, `external_aggr` is not slower, and merges without a byte limit only check that none is set.
>
> * Maintenance: the byte limit is crate-private and opt-in, and only aggregate spilling sets it.
>
> * Accuracy: spilling cuts batches by the size of each row (`SpilledRowSizes`), not by averages. Only the merge estimates rows by the average row size of their batch. It never takes more rows from a batch than the batch holds, so it can underestimate an output batch by at most the input batches it holds, which are about 1 MiB each unless they hold a single larger row.
My point is that no 184MB should be spilled. for a 128MB pool we should be spilling by a calculation that would allow the 2x readback expected(so a much earlier spill), and this should be calculated before inserting a new input batch, and if not possible - spill first, and only then insert the new batch into the cleared state.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-5901513632
π€ Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/25853#issuecomment-5901451278)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c5901451278-2936-vrrh4 6.12.94+ #1 SMP Wed Aug 19 07:47:20 UTC 2026 aarch64 GNU/Linux`
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
Comparing fix-issue-258521 (832a7eb5c708f1ca2aa71e008201660203db45b4) to bdad988 (merge-base) [diff](https://github.com/apache/datafusion/compare/bdad988f23f4f8a45a71409e7d7f24a326ca909c..832a7eb5c708f1ca2aa71e008201660203db45b4)
<details><summary>Run configuration</summary>
```yaml
run benchmark tpch
```
</details>
Results will be posted here when complete
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-6384102134
run benchmark spill_views
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-5901608535
π€ Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/25853#issuecomment-5901451278)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB)
Comparing fix-issue-258521 (832a7eb5c708f1ca2aa71e008201660203db45b4) to bdad988 (merge-base) [diff](https://github.com/apache/datafusion/compare/bdad988f23f4f8a45a71409e7d7f24a326ca909c..832a7eb5c708f1ca2aa71e008201660203db45b4)
<details><summary>Run configuration</summary>
```yaml
run benchmark tpcds
```
</details>
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
<details><summary>Details</summary>
<p>
```
Comparing HEAD and fix-issue-258521
--------------------
Benchmark tpcds_sf1.json
--------------------
βββββββββββββ³βββββββββββββ³βββββββββββββββββββ³ββββββββββββββββ
β Query β HEAD β fix-issue-258521 β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β QQuery 1 β 5.93 ms β 5.90 ms β no change β
β QQuery 2 β 82.03 ms β 81.12 ms β no change β
β QQuery 3 β 29.36 ms β 29.32 ms β no change β
β QQuery 4 β 495.20 ms β 482.79 ms β no change β
β QQuery 5 β 51.88 ms β 52.15 ms β no change β
β QQuery 6 β 37.23 ms β 36.51 ms β no change β
β QQuery 7 β 76.16 ms β 76.14 ms β no change β
β QQuery 8 β 37.64 ms β 37.11 ms β no change β
β QQuery 9 β 53.57 ms β 53.19 ms β no change β
β QQuery 10 β 62.78 ms β 63.13 ms β no change β
β QQuery 11 β 307.85 ms β 299.48 ms β no change β
β QQuery 12 β 30.39 ms β 30.10 ms β no change β
β QQuery 13 β 119.97 ms β 118.77 ms β no change β
β QQuery 14 β 397.66 ms β 401.60 ms β no change β
β QQuery 15 β 59.32 ms β 59.32 ms β no change β
β QQuery 16 β 7.02 ms β 7.00 ms β no change β
β QQuery 17 β 79.28 ms β 82.07 ms β no change β
β QQuery 18 β 108.22 ms β 109.79 ms β no change β
β QQuery 19 β 42.53 ms β 43.10 ms β no change β
β QQuery 20 β 36.59 ms β 37.20 ms β no change β
β QQuery 21 β 17.51 ms β 18.10 ms β no change β
β QQuery 22 β 68.10 ms β 69.47 ms β no change β
β QQuery 23 β 329.75 ms β 333.15 ms β no change β
β QQuery 24 β 205.04 ms β 206.61 ms β no change β
β QQuery 25 β 108.53 ms β 110.64 ms β no change β
β QQuery 26 β 49.52 ms β 49.63 ms β no change β
β QQuery 27 β 6.48 ms β 6.44 ms β no change β
β QQuery 28 β 57.39 ms β 58.11 ms β no change β
β QQuery 29 β 96.98 ms β 96.48 ms β no change β
β QQuery 30 β 33.83 ms β 33.37 ms β no change β
β QQuery 31 β 112.65 ms β 113.55 ms β no change β
β QQuery 32 β 21.65 ms β 21.58 ms β no change β
β QQuery 33 β 38.38 ms β 38.47 ms β no change β
β QQuery 34 β 10.56 ms β 10.63 ms β no change β
β QQuery 35 β 76.65 ms β 78.50 ms β no change β
β QQuery 36 β 6.35 ms β 6.31 ms β no change β
β QQuery 37 β 7.41 ms β 7.37 ms β no change β
β QQuery 38 β 64.07 ms β 65.03 ms β no change β
β QQuery 39 β 76.09 ms β 77.06 ms β no change β
β QQuery 40 β 24.51 ms β 25.36 ms β no change β
β QQuery 41 β 11.64 ms β 11.92 ms β no change β
β QQuery 42 β 24.61 ms β 24.76 ms β no change β
β QQuery 43 β 5.20 ms β 5.18 ms β no change β
β QQuery 44 β 9.35 ms β 9.24 ms β no change β
β QQuery 45 β 41.00 ms β 42.23 ms β no change β
β QQuery 46 β 12.42 ms β 12.48 ms β no change β
β QQuery 47 β 243.79 ms β 235.83 ms β no change β
β QQuery 48 β 96.50 ms β 97.33 ms β no change β
β QQuery 49 β 71.67 ms β 71.84 ms β no change β
β QQuery 50 β 58.89 ms β 59.47 ms β no change β
β QQuery 51 β 95.71 ms β 92.40 ms β no change β
β QQuery 52 β 24.69 ms β 24.82 ms β no change β
β QQuery 53 β 29.77 ms β 29.77 ms β no change β
β QQuery 54 β 55.18 ms β 55.54 ms β no change β
β QQuery 55 β 23.88 ms β 24.43 ms β no change β
β QQuery 56 β 38.78 ms β 39.93 ms β no change β
β QQuery 57 β 173.30 ms β 174.92 ms β no change β
β QQuery 58 β 110.13 ms β 111.93 ms β no change β
β QQuery 59 β 116.75 ms β 118.70 ms β no change β
β QQuery 60 β 40.04 ms β 39.80 ms β no change β
β QQuery 61 β 12.06 ms β 11.98 ms β no change β
β QQuery 62 β 45.24 ms β 45.49 ms β no change β
β QQuery 63 β 29.98 ms β 30.13 ms β no change β
β QQuery 64 β 366.62 ms β 369.03 ms β no change β
β QQuery 65 β 132.12 ms β 133.48 ms β no change β
β QQuery 66 β 78.19 ms β 79.17 ms β no change β
β QQuery 67 β 261.86 ms β 259.29 ms β no change β
β QQuery 68 β 12.25 ms β 12.24 ms β no change β
β QQuery 69 β 57.24 ms β 57.99 ms β no change β
β QQuery 70 β 104.50 ms β 105.03 ms β no change β
β QQuery 71 β 36.04 ms β 35.40 ms β no change β
β QQuery 72 β 1874.42 ms β 1747.79 ms β +1.07x faster β
β QQuery 73 β 10.72 ms β 9.90 ms β +1.08x faster β
β QQuery 74 β 177.92 ms β 161.55 ms β +1.10x faster β
β QQuery 75 β 142.42 ms β 140.59 ms β no change β
β QQuery 76 β 34.70 ms β 34.60 ms β no change β
β QQuery 77 β 61.24 ms β 60.44 ms β no change β
β QQuery 78 β 167.64 ms β 163.41 ms β no change β
β QQuery 79 β 67.60 ms β 66.08 ms β no change β
β QQuery 80 β 97.89 ms β 95.53 ms β no change β
β QQuery 81 β 27.07 ms β 26.32 ms β no change β
β QQuery 82 β 17.11 ms β 16.80 ms β no change β
β QQuery 83 β 35.10 ms β 33.60 ms β no change β
β QQuery 84 β 30.67 ms β 29.69 ms β no change β
β QQuery 85 β 104.49 ms β 103.57 ms β no change β
β QQuery 86 β 26.30 ms β 25.97 ms β no change β
β QQuery 87 β 65.72 ms β 62.82 ms β no change β
β QQuery 88 β 62.71 ms β 60.93 ms β no change β
β QQuery 89 β 35.72 ms β 35.10 ms β no change β
β QQuery 90 β 17.38 ms β 17.02 ms β no change β
β QQuery 91 β 45.61 ms β 44.78 ms β no change β
β QQuery 92 β 31.04 ms β 30.21 ms β no change β
β QQuery 93 β 50.00 ms β 49.48 ms β no change β
β QQuery 94 β 39.81 ms β 38.97 ms β no change β
β QQuery 95 β 82.21 ms β 81.64 ms β no change β
β QQuery 96 β 24.17 ms β 23.99 ms β no change β
β QQuery 97 β 52.06 ms β 51.51 ms β no change β
β QQuery 98 β 44.24 ms β 43.05 ms β no change β
β QQuery 99 β 66.79 ms β 66.36 ms β no change β
βββββββββββββ΄βββββββββββββ΄βββββββββββββββββββ΄ββββββββββββββββ
βββββββββββββββββββββββββββββββββββ³ββββββββββββ
β Benchmark Summary β β
β‘ββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 9444.23ms β
β Total Time (fix-issue-258521) β 9275.09ms β
β Average Time (HEAD) β 95.40ms β
β Average Time (fix-issue-258521) β 93.69ms β
β Queries Faster β 3 β
β Queries Slower β 0 β
β Queries with No Change β 96 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββ΄ββββββββββββ
Distribution per query (min / mean Β±stddev / max):
Comparing HEAD and fix-issue-258521
--------------------
Benchmark tpcds_sf1.json
--------------------
βββββββββββββ³ββββββββββββββββββββββββββββββββββββββββ³ββββββββββββββββββββββββββββββββββββββββ³ββββββββββββββββ
β Query β HEAD β fix-issue-258521 β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β QQuery 1 β 5.93 / 6.46 Β±0.90 / 8.26 ms β 5.90 / 6.44 Β±0.93 / 8.30 ms β no change β
β QQuery 2 β 82.03 / 82.34 Β±0.32 / 82.92 ms β 81.12 / 81.37 Β±0.17 / 81.65 ms β no change β
β QQuery 3 β 29.36 / 29.74 Β±0.29 / 30.23 ms β 29.32 / 29.51 Β±0.15 / 29.75 ms β no change β
β QQuery 4 β 495.20 / 506.69 Β±7.27 / 515.25 ms β 482.79 / 488.66 Β±3.68 / 494.32 ms β no change β
β QQuery 5 β 51.88 / 52.78 Β±0.49 / 53.33 ms β 52.15 / 52.54 Β±0.28 / 52.99 ms β no change β
β QQuery 6 β 37.23 / 37.46 Β±0.27 / 37.97 ms β 36.51 / 37.26 Β±0.42 / 37.73 ms β no change β
β QQuery 7 β 76.16 / 76.79 Β±0.46 / 77.41 ms β 76.14 / 76.57 Β±0.31 / 77.04 ms β no change β
β QQuery 8 β 37.64 / 38.10 Β±0.53 / 39.13 ms β 37.11 / 39.13 Β±2.56 / 44.02 ms β no change β
β QQuery 9 β 53.57 / 56.10 Β±1.60 / 57.88 ms β 53.19 / 55.42 Β±1.25 / 56.66 ms β no change β
β QQuery 10 β 62.78 / 63.37 Β±0.38 / 63.86 ms β 63.13 / 63.43 Β±0.28 / 63.86 ms β no change β
β QQuery 11 β 307.85 / 310.97 Β±2.55 / 314.66 ms β 299.48 / 305.40 Β±3.37 / 309.75 ms β no change β
β QQuery 12 β 30.39 / 30.57 Β±0.15 / 30.80 ms β 30.10 / 30.32 Β±0.14 / 30.52 ms β no change β
β QQuery 13 β 119.97 / 120.34 Β±0.27 / 120.58 ms β 118.77 / 120.02 Β±0.92 / 121.41 ms β no change β
β QQuery 14 β 397.66 / 402.24 Β±2.95 / 405.32 ms β 401.60 / 403.88 Β±2.46 / 407.81 ms β no change β
β QQuery 15 β 59.32 / 60.36 Β±0.70 / 61.33 ms β 59.32 / 62.13 Β±2.96 / 67.86 ms β no change β
β QQuery 16 β 7.02 / 7.19 Β±0.13 / 7.32 ms β 7.00 / 7.15 Β±0.09 / 7.26 ms β no change β
β QQuery 17 β 79.28 / 80.25 Β±0.70 / 81.14 ms β 82.07 / 83.02 Β±0.84 / 84.57 ms β no change β
β QQuery 18 β 108.22 / 109.95 Β±1.89 / 113.37 ms β 109.79 / 112.86 Β±2.94 / 118.13 ms β no change β
β QQuery 19 β 42.53 / 43.01 Β±0.38 / 43.51 ms β 43.10 / 43.32 Β±0.20 / 43.61 ms β no change β
β QQuery 20 β 36.59 / 37.06 Β±0.35 / 37.64 ms β 37.20 / 37.58 Β±0.22 / 37.88 ms β no change β
β QQuery 21 β 17.51 / 17.86 Β±0.25 / 18.26 ms β 18.10 / 19.09 Β±1.67 / 22.41 ms β 1.07x slower β
β QQuery 22 β 68.10 / 68.68 Β±0.53 / 69.66 ms β 69.47 / 69.85 Β±0.42 / 70.64 ms β no change β
β QQuery 23 β 329.75 / 332.80 Β±2.15 / 335.43 ms β 333.15 / 342.60 Β±6.34 / 352.65 ms β no change β
β QQuery 24 β 205.04 / 207.78 Β±2.72 / 212.96 ms β 206.61 / 211.81 Β±4.42 / 219.90 ms β no change β
β QQuery 25 β 108.53 / 110.93 Β±1.60 / 112.91 ms β 110.64 / 111.54 Β±0.77 / 112.86 ms β no change β
β QQuery 26 β 49.52 / 50.07 Β±0.30 / 50.44 ms β 49.63 / 50.56 Β±0.50 / 51.05 ms β no change β
β QQuery 27 β 6.48 / 6.65 Β±0.16 / 6.95 ms β 6.44 / 6.64 Β±0.22 / 7.00 ms β no change β
β QQuery 28 β 57.39 / 60.28 Β±1.97 / 62.00 ms β 58.11 / 63.25 Β±3.69 / 69.54 ms β no change β
β QQuery 29 β 96.98 / 97.94 Β±1.25 / 100.25 ms β 96.48 / 97.87 Β±1.11 / 99.81 ms β no change β
β QQuery 30 β 33.83 / 34.25 Β±0.27 / 34.62 ms β 33.37 / 33.70 Β±0.42 / 34.51 ms β no change β
β QQuery 31 β 112.65 / 115.09 Β±2.54 / 119.52 ms β 113.55 / 114.83 Β±0.72 / 115.78 ms β no change β
β QQuery 32 β 21.65 / 21.76 Β±0.09 / 21.87 ms β 21.58 / 22.00 Β±0.30 / 22.39 ms β no change β
β QQuery 33 β 38.38 / 38.75 Β±0.36 / 39.36 ms β 38.47 / 38.88 Β±0.30 / 39.25 ms β no change β
β QQuery 34 β 10.56 / 10.77 Β±0.12 / 10.91 ms β 10.63 / 10.84 Β±0.14 / 11.00 ms β no change β
β QQuery 35 β 76.65 / 77.62 Β±1.15 / 79.87 ms β 78.50 / 79.89 Β±0.86 / 80.88 ms β no change β
β QQuery 36 β 6.35 / 6.44 Β±0.09 / 6.56 ms β 6.31 / 6.93 Β±1.15 / 9.23 ms β 1.08x slower β
β QQuery 37 β 7.41 / 8.22 Β±1.53 / 11.29 ms β 7.37 / 7.56 Β±0.15 / 7.78 ms β +1.09x faster β
β QQuery 38 β 64.07 / 65.13 Β±0.66 / 66.06 ms β 65.03 / 66.25 Β±0.95 / 67.86 ms β no change β
β QQuery 39 β 76.09 / 76.68 Β±0.45 / 77.46 ms β 77.06 / 78.57 Β±1.26 / 80.24 ms β no change β
β QQuery 40 β 24.51 / 26.47 Β±3.07 / 32.60 ms β 25.36 / 25.97 Β±0.51 / 26.82 ms β no change β
β QQuery 41 β 11.64 / 11.79 Β±0.11 / 11.95 ms β 11.92 / 12.13 Β±0.13 / 12.30 ms β no change β
β QQuery 42 β 24.61 / 25.52 Β±0.65 / 26.42 ms β 24.76 / 25.25 Β±0.46 / 26.03 ms β no change β
β QQuery 43 β 5.20 / 5.32 Β±0.07 / 5.39 ms β 5.18 / 5.29 Β±0.08 / 5.41 ms β no change β
β QQuery 44 β 9.35 / 9.52 Β±0.11 / 9.67 ms β 9.24 / 9.41 Β±0.14 / 9.66 ms β no change β
β QQuery 45 β 41.00 / 41.82 Β±0.57 / 42.53 ms β 42.23 / 42.36 Β±0.13 / 42.58 ms β no change β
β QQuery 46 β 12.42 / 12.93 Β±0.61 / 14.05 ms β 12.48 / 12.72 Β±0.23 / 13.08 ms β no change β
β QQuery 47 β 243.79 / 249.11 Β±3.58 / 254.02 ms β 235.83 / 241.70 Β±6.37 / 252.74 ms β no change β
β QQuery 48 β 96.50 / 98.13 Β±1.31 / 100.04 ms β 97.33 / 99.18 Β±2.36 / 103.82 ms β no change β
β QQuery 49 β 71.67 / 72.30 Β±0.79 / 73.86 ms β 71.84 / 72.24 Β±0.43 / 73.01 ms β no change β
β QQuery 50 β 58.89 / 59.34 Β±0.24 / 59.53 ms β 59.47 / 62.48 Β±4.88 / 72.18 ms β 1.05x slower β
β QQuery 51 β 95.71 / 99.48 Β±4.30 / 107.34 ms β 92.40 / 96.62 Β±2.38 / 98.91 ms β no change β
β QQuery 52 β 24.69 / 25.00 Β±0.41 / 25.81 ms β 24.82 / 25.17 Β±0.31 / 25.67 ms β no change β
β QQuery 53 β 29.77 / 29.96 Β±0.17 / 30.25 ms β 29.77 / 29.90 Β±0.14 / 30.16 ms β no change β
β QQuery 54 β 55.18 / 57.07 Β±2.44 / 61.90 ms β 55.54 / 58.73 Β±3.99 / 66.35 ms β no change β
β QQuery 55 β 23.88 / 24.30 Β±0.35 / 24.93 ms β 24.43 / 25.52 Β±1.06 / 27.42 ms β 1.05x slower β
β QQuery 56 β 38.78 / 39.77 Β±0.57 / 40.45 ms β 39.93 / 40.35 Β±0.22 / 40.52 ms β no change β
β QQuery 57 β 173.30 / 176.70 Β±2.43 / 180.17 ms β 174.92 / 177.24 Β±1.79 / 180.38 ms β no change β
β QQuery 58 β 110.13 / 112.26 Β±2.20 / 115.66 ms β 111.93 / 113.20 Β±1.00 / 114.86 ms β no change β
β QQuery 59 β 116.75 / 118.21 Β±1.39 / 120.84 ms β 118.70 / 119.71 Β±1.21 / 122.04 ms β no change β
β QQuery 60 β 40.04 / 40.92 Β±0.56 / 41.49 ms β 39.80 / 40.72 Β±0.60 / 41.50 ms β no change β
β QQuery 61 β 12.06 / 12.16 Β±0.06 / 12.24 ms β 11.98 / 12.08 Β±0.06 / 12.16 ms β no change β
β QQuery 62 β 45.24 / 45.43 Β±0.20 / 45.75 ms β 45.49 / 45.82 Β±0.32 / 46.27 ms β no change β
β QQuery 63 β 29.98 / 30.75 Β±0.85 / 32.37 ms β 30.13 / 30.65 Β±0.44 / 31.18 ms β no change β
β QQuery 64 β 366.62 / 371.26 Β±3.47 / 375.34 ms β 369.03 / 374.45 Β±3.51 / 379.27 ms β no change β
β QQuery 65 β 132.12 / 135.12 Β±4.24 / 143.45 ms β 133.48 / 137.13 Β±2.75 / 141.58 ms β no change β
β QQuery 66 β 78.19 / 79.38 Β±0.61 / 79.92 ms β 79.17 / 81.01 Β±2.52 / 86.02 ms β no change β
β QQuery 67 β 261.86 / 270.39 Β±7.32 / 282.06 ms β 259.29 / 265.18 Β±4.84 / 270.95 ms β no change β
β QQuery 68 β 12.25 / 12.47 Β±0.17 / 12.69 ms β 12.24 / 12.34 Β±0.11 / 12.51 ms β no change β
β QQuery 69 β 57.24 / 58.00 Β±0.49 / 58.60 ms β 57.99 / 58.45 Β±0.41 / 59.04 ms β no change β
β QQuery 70 β 104.50 / 107.58 Β±4.03 / 115.31 ms β 105.03 / 108.30 Β±3.80 / 114.89 ms β no change β
β QQuery 71 β 36.04 / 37.23 Β±1.49 / 40.08 ms β 35.40 / 36.18 Β±0.64 / 37.16 ms β no change β
β QQuery 72 β 1874.42 / 1917.60 Β±38.91 / 1968.82 ms β 1747.79 / 1800.56 Β±50.69 / 1870.92 ms β +1.06x faster β
β QQuery 73 β 10.72 / 11.35 Β±0.35 / 11.72 ms β 9.90 / 10.34 Β±0.49 / 11.03 ms β +1.10x faster β
β QQuery 74 β 177.92 / 185.48 Β±7.56 / 198.97 ms β 161.55 / 166.93 Β±6.20 / 178.56 ms β +1.11x faster β
β QQuery 75 β 142.42 / 146.88 Β±3.99 / 152.91 ms β 140.59 / 143.19 Β±2.63 / 148.21 ms β no change β
β QQuery 76 β 34.70 / 35.30 Β±0.53 / 36.27 ms β 34.60 / 35.20 Β±0.47 / 35.98 ms β no change β
β QQuery 77 β 61.24 / 64.12 Β±3.97 / 71.99 ms β 60.44 / 61.60 Β±1.40 / 64.26 ms β no change β
β QQuery 78 β 167.64 / 170.75 Β±4.27 / 178.98 ms β 163.41 / 165.95 Β±1.79 / 168.61 ms β no change β
β QQuery 79 β 67.60 / 70.96 Β±4.25 / 79.28 ms β 66.08 / 66.43 Β±0.39 / 67.17 ms β +1.07x faster β
β QQuery 80 β 97.89 / 99.96 Β±1.86 / 103.13 ms β 95.53 / 96.99 Β±1.67 / 100.16 ms β no change β
β QQuery 81 β 27.07 / 27.24 Β±0.11 / 27.39 ms β 26.32 / 26.66 Β±0.23 / 26.92 ms β no change β
β QQuery 82 β 17.11 / 17.25 Β±0.13 / 17.42 ms β 16.80 / 16.99 Β±0.14 / 17.15 ms β no change β
β QQuery 83 β 35.10 / 38.18 Β±5.43 / 49.03 ms β 33.60 / 34.08 Β±0.40 / 34.71 ms β +1.12x faster β
β QQuery 84 β 30.67 / 31.43 Β±0.71 / 32.40 ms β 29.69 / 29.98 Β±0.18 / 30.22 ms β no change β
β QQuery 85 β 104.49 / 106.03 Β±1.06 / 107.64 ms β 103.57 / 107.39 Β±4.11 / 115.18 ms β no change β
β QQuery 86 β 26.30 / 26.60 Β±0.20 / 26.90 ms β 25.97 / 26.13 Β±0.17 / 26.45 ms β no change β
β QQuery 87 β 65.72 / 68.87 Β±4.05 / 76.72 ms β 62.82 / 63.46 Β±0.53 / 64.39 ms β +1.09x faster β
β QQuery 88 β 62.71 / 62.88 Β±0.17 / 63.09 ms β 60.93 / 61.39 Β±0.31 / 61.89 ms β no change β
β QQuery 89 β 35.72 / 35.99 Β±0.29 / 36.50 ms β 35.10 / 35.35 Β±0.23 / 35.66 ms β no change β
β QQuery 90 β 17.38 / 17.47 Β±0.09 / 17.58 ms β 17.02 / 17.61 Β±0.90 / 19.38 ms β no change β
β QQuery 91 β 45.61 / 47.77 Β±2.61 / 52.70 ms β 44.78 / 45.29 Β±0.46 / 46.08 ms β +1.05x faster β
β QQuery 92 β 31.04 / 31.99 Β±1.61 / 35.19 ms β 30.21 / 30.65 Β±0.32 / 31.14 ms β no change β
β QQuery 93 β 50.00 / 50.99 Β±0.72 / 52.16 ms β 49.48 / 50.44 Β±0.70 / 51.58 ms β no change β
β QQuery 94 β 39.81 / 40.16 Β±0.44 / 41.03 ms β 38.97 / 40.93 Β±2.23 / 45.21 ms β no change β
β QQuery 95 β 82.21 / 83.19 Β±1.05 / 85.22 ms β 81.64 / 82.28 Β±0.56 / 83.22 ms β no change β
β QQuery 96 β 24.17 / 24.48 Β±0.18 / 24.74 ms β 23.99 / 24.33 Β±0.20 / 24.60 ms β no change β
β QQuery 97 β 52.06 / 53.54 Β±0.99 / 55.10 ms β 51.51 / 52.14 Β±1.03 / 54.18 ms β no change β
β QQuery 98 β 44.24 / 44.89 Β±0.64 / 46.12 ms β 43.05 / 44.52 Β±1.85 / 47.98 ms β no change β
β QQuery 99 β 66.79 / 68.47 Β±2.99 / 74.44 ms β 66.36 / 66.90 Β±0.45 / 67.70 ms β no change β
βββββββββββββ΄ββββββββββββββββββββββββββββββββββββββββ΄ββββββββββββββββββββββββββββββββββββββββ΄ββββββββββββββββ
βββββββββββββββββββββββββββββββββββ³ββββββββββββ
β Benchmark Summary β β
β‘ββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 9635.01ms β
β Total Time (fix-issue-258521) β 9469.85ms β
β Average Time (HEAD) β 97.32ms β
β Average Time (fix-issue-258521) β 95.66ms β
β Queries Faster β 8 β
β Queries Slower β 4 β
β Queries with No Change β 87 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββ΄ββββββββββββ
```
</p>
</details>
<details><summary>Resource Usage</summary>
**tpcds β base (merge-base)**
| Metric | Value |
|--------|-------|
| Wall time | 50.0s |
| Peak memory | 2.0 GiB |
| Avg memory | 1.4 GiB |
| CPU user | 209.6s |
| CPU sys | 6.0s |
| Peak spill | 0 B |
**tpcds β branch**
| Metric | Value |
|--------|-------|
| Wall time | 50.0s |
| Peak memory | 2.2 GiB |
| Avg memory | 1.5 GiB |
| CPU user | 201.1s |
| CPU sys | 5.9s |
| Peak spill | 0 B |
</details>
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-5981949099
Addressed the review feedback in b7cd9113c:
- @comphead: the `IncrementalSortIterator` docs now say a chunk can be shorter than `batch_size`, `view_bytes` applies the view size rule in one place, and `row_bytes` is now `avg_row_bytes`. The inline suggestions are in the threads.
- @2010YOUY01: removed the coalescing of replay output.
`external_aggr` against main after these changes: no query is slower, 7 are unchanged and 1 is 1.06x faster.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#discussion_r4139564667
##########
datafusion/sqllogictest/test_files/aggregate_memory_spill.slt:
##########
@@ -233,6 +233,110 @@ FROM (
08)--------------AggregateExec: mode=Partial, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[<slt:ignore>early_emit_count=<slt:ignore>]
<slt:ignore>
+# --- Case H: a few groups with large state ---
+# The array_agg states of all 32 groups do not fit in the memory limit, but a
+# single group does. Spilling must write the groups in batches that can be
+# read back within the limit, instead of one batch holding every group.
+# See https://github.com/apache/datafusion/issues/25851
+statement ok
+RESET datafusion.runtime.max_spill_merge_fan_in
+
+statement ok
+SET datafusion.execution.batch_size = 8192
+
+statement ok
+SET datafusion.runtime.memory_limit = '24M'
+
+# Partial/final aggregation: the final aggregation spills the partial states of
+# its partition.
+statement ok
+SET datafusion.execution.target_partitions = 2
+
+query II
+SELECT count(*), sum(n)
+FROM (
+ SELECT v % 32 AS k, cardinality(array_agg(concat(CAST(v AS VARCHAR), repeat('x', 100)))) AS n
+ FROM generate_series(1, 500000) AS t(v)
+ GROUP BY v % 32
+)
+----
+32 500000
+
+# Assert the final aggregation spilled, the `spilled_bytes` metric must be in MB
+query TT
+EXPLAIN ANALYZE
+SELECT count(*), sum(n)
+FROM (
+ SELECT v % 32 AS k, cardinality(array_agg(concat(CAST(v AS VARCHAR), repeat('x', 100)))) AS n
+ FROM generate_series(1, 500000) AS t(v)
+ GROUP BY v % 32
+)
+----
+<slt:ignore>
+06)----------AggregateExec: mode=FinalPartitioned, gby=[t.v % Int64(32)@0 as t.v % Int64(32)], aggr=[<slt:ignore>], metrics=[<slt:ignore>spilled_bytes=<slt:ignore> MB, spilled_rows=<slt:ignore>]
+07)------------RepartitionExec: <slt:ignore>
+
+# Single-stage aggregation: spills several runs as input arrives, which replay
+# merges.
+statement ok
+SET datafusion.execution.target_partitions = 1
+
+query II
+SELECT count(*), sum(n)
+FROM (
+ SELECT v % 32 AS k, cardinality(array_agg(concat(CAST(v AS VARCHAR), repeat('x', 100)))) AS n
+ FROM generate_series(1, 500000) AS t(v)
+ GROUP BY v % 32
+)
+----
+32 500000
+
+# Assert spill happened, the `spilled_bytes` metric must be in MB
+query TT
+EXPLAIN ANALYZE
+SELECT count(*), sum(n)
+FROM (
+ SELECT v % 32 AS k, cardinality(array_agg(concat(CAST(v AS VARCHAR), repeat('x', 100)))) AS n
+ FROM generate_series(1, 500000) AS t(v)
+ GROUP BY v % 32
+)
+----
+<slt:ignore>
+04)------AggregateExec: mode=Single, gby=[v@0 % 32 as t.v % Int64(32)], aggr=[<slt:ignore>], metrics=[<slt:ignore>spilled_bytes=<slt:ignore> MB, spilled_rows=<slt:ignore>]
+05)--------ProjectionExec: <slt:ignore>
+
+# 14 large groups, adjacent in key order, spilled together with 62,500 small
Review Comment:
Every other case in this file follows its query with an `EXPLAIN ANALYZE` check that it spilled. This query and the next one have none, so if a later planner or memory change stopped them from spilling, they would keep passing without covering #25851.
##########
datafusion/physical-plan/src/sorts/streaming_merge.rs:
##########
@@ -326,6 +338,121 @@ mod tests {
ExecutionPlanMetricsSet, SpillMetrics,
};
+ /// Merges five 1000 byte rows, one per batch, with 1000 small rows in
+ /// batches of 100, either from spill files or from in-memory streams.
+ async fn merge_wide_and_narrow_rows(
+ from_spill_files: bool,
+ max_batch_bytes: Option<usize>,
+ ) -> Result<Vec<RecordBatch>> {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("key", DataType::Int32, false),
+ Field::new("value", DataType::Utf8, false),
+ ]));
+ let batch = |keys: std::ops::Range<i32>, value_len: usize| {
+ let values = keys.clone().map(|_| "x".repeat(value_len));
+ RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(Int32Array::from_iter_values(keys)),
+ Arc::new(StringArray::from_iter_values(values)),
+ ],
+ )
+ .unwrap()
+ };
+ let wide: Vec<_> = (0..5).map(|key| batch(key..key + 1, 1000)).collect();
+ let narrow: Vec<_> = (0..10)
+ .map(|i| batch(10 + i * 100..10 + (i + 1) * 100, 1))
+ .collect();
+
+ let sort: LexOrdering =
+ [PhysicalSortExpr::new_default(col("key", &schema)?)].into();
+ let mut builder = StreamingMergeBuilder::new()
+ .with_schema(Arc::clone(&schema))
+ .with_expressions(&sort)
+ .with_metrics(BaselineMetrics::new(&ExecutionPlanMetricsSet::new(), 0))
+ .with_batch_size(100)
+ .with_max_batch_bytes(max_batch_bytes)
+ .with_bypass_mempool();
+ if from_spill_files {
+ let spill_manager = SpillManager::new(
+ Arc::new(TaskContext::default()).runtime_env(),
+ SpillMetrics::new(&ExecutionPlanMetricsSet::new(), 0),
+ Arc::clone(&schema),
+ );
+ let spill = |batches: &[RecordBatch]| SortedSpillFile {
+ file: spill_manager
+ .spill_record_batch_and_finish(batches, "test")
+ .unwrap()
+ .unwrap(),
+ max_record_batch_memory: batches
+ .iter()
+ .map(RecordBatch::get_array_memory_size)
+ .max()
+ .unwrap(),
+ };
+ builder = builder
+ .with_sorted_spill_files(vec![spill(&wide), spill(&narrow)])
+ .with_spill_manager(spill_manager);
+ } else {
+ let stream = |batches: Vec<RecordBatch>| {
+ Box::pin(RecordBatchStreamAdapter::new(
+ Arc::clone(&schema),
+ futures::stream::iter(batches.into_iter().map(Ok)),
+ )) as SendableRecordBatchStream
+ };
+ builder = builder.with_streams(vec![stream(wide), stream(narrow)]);
+ }
+ let batches = collect(builder.build()?).await?;
+
+ let keys: Vec<i32> = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_primitive::<Int32Type>()
+ .values()
+ .to_vec()
+ })
+ .collect();
+ let expected: Vec<i32> = (0..5).chain(10..1010).collect();
+ assert_eq!(keys, expected);
+ Ok(batches)
+ }
+
+ #[tokio::test]
+ async fn test_max_batch_bytes_limits_merged_batches() -> Result<()> {
+ let num_rows = |batches: &[RecordBatch]| -> Vec<usize> {
+ batches.iter().map(RecordBatch::num_rows).collect()
+ };
+ for from_spill_files in [true, false] {
Review Comment:
The byte cap is only set by aggregate replay, which always merges spill files. Each multi-level pass (`create_new_merge_sort`) then builds this same `SortPreservingMergeStream` over the spill-file streams, so the `false` run covers no code path that the `true` run misses. Keeping only the spill-file case would drop the `from_spill_files` flag and the in-memory branch of `merge_wide_and_narrow_rows`.
##########
datafusion/physical-plan/src/spill/spilled_row_sizes.rs:
##########
@@ -0,0 +1,426 @@
+// 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.
+
+//! Per-row sizes of a batch once it is written to a spill file.
+
+use arrow::array::{Array, AsArray, RecordBatch};
+use arrow::buffer::{OffsetBuffer, ScalarBuffer};
+use arrow::datatypes::DataType;
+use arrow_data::MAX_INLINE_VIEW_LEN;
+
+use super::VIEW_SIZE_BYTES;
+
+/// Estimates how many bytes each row of a batch takes once written to a spill
+/// file, where view arrays are compacted (see [`super::gc_view_arrays`]).
+///
+/// A row counts its fixed-width values, the offsets and bytes of its
+/// variable-width values, and the elements of its nested values. Buffers that
+/// every row shares, such as validity bitmaps and dictionary values, are not
+/// counted.
+pub(crate) struct SpilledRowSizes {
+ /// Bytes that every row takes in the fixed-width columns
+ fixed: usize,
+ /// The other columns
+ variable: Vec<ColumnSize>,
+ /// See [`Self::max_row`]
+ max_row: usize,
+}
+
+impl SpilledRowSizes {
+ pub(crate) fn new(batch: &RecordBatch) -> Self {
+ let mut fixed = 0;
+ let mut variable = vec![];
+ for column in batch.columns() {
+ let column = ColumnSize::new(column.as_ref());
+ match column.fixed() {
+ Some(width) => fixed += width,
+ None => variable.push(column),
+ }
+ }
+ let max_row = variable.iter().fold(fixed, |bytes, column| {
+ bytes.saturating_add(column.max_row())
+ });
+ Self {
+ fixed,
+ variable,
+ max_row,
+ }
+ }
+
+ /// Returns the size in bytes of row `row`.
+ pub(crate) fn row(&self, row: usize) -> usize {
+ self.variable.iter().fold(self.fixed, |bytes, column| {
+ bytes + column.rows(row, row + 1)
+ })
+ }
+
+ /// Returns the size in bytes of every row, if they all have the same size.
+ pub(crate) fn fixed(&self) -> Option<usize> {
+ self.variable.is_empty().then_some(self.fixed)
+ }
+
+ /// Returns an upper bound on the size in bytes of any row.
+ pub(crate) fn max_row(&self) -> usize {
+ self.max_row
+ }
+}
+
+/// Size model of one column, see [`SpilledRowSizes`].
+enum ColumnSize {
+ /// Every row takes the same number of bytes.
+ Fixed(usize),
+ /// An offset per row, plus the bytes between consecutive offsets.
+ Bytes(Offsets),
+ /// A view per row, plus the bytes of values too long to be inlined.
+ Views(ScalarBuffer<u128>),
+ /// An offset per row, plus the elements between consecutive offsets.
+ List(Offsets, Box<ColumnSize>),
+ /// A fixed number of elements per row.
+ FixedSizeList(usize, Box<ColumnSize>),
+ /// A row of each field.
+ Struct(Vec<ColumnSize>),
+}
+
+impl ColumnSize {
+ fn new(array: &dyn Array) -> Self {
+ match array.data_type() {
+ DataType::Utf8 => Self::Bytes(array.as_string::<i32>().offsets().into()),
+ DataType::LargeUtf8 => Self::Bytes(array.as_string::<i64>().offsets().into()),
+ DataType::Binary => Self::Bytes(array.as_binary::<i32>().offsets().into()),
+ DataType::LargeBinary => {
+ Self::Bytes(array.as_binary::<i64>().offsets().into())
+ }
+ DataType::Utf8View => Self::Views(array.as_string_view().views().clone()),
+ DataType::BinaryView => Self::Views(array.as_binary_view().views().clone()),
+ DataType::List(_) => {
+ let list = array.as_list::<i32>();
+ Self::List(list.offsets().into(), Box::new(Self::new(list.values())))
+ }
+ DataType::LargeList(_) => {
+ let list = array.as_list::<i64>();
+ Self::List(list.offsets().into(), Box::new(Self::new(list.values())))
+ }
+ DataType::Map(_, _) => {
+ let map = array.as_map();
+ Self::List(map.offsets().into(), Box::new(Self::new(map.entries())))
+ }
+ DataType::FixedSizeList(_, size) => Self::FixedSizeList(
+ *size as usize,
+ Box::new(Self::new(array.as_fixed_size_list().values())),
+ ),
+ DataType::Struct(_) => Self::Struct(
+ array
+ .as_struct()
+ .columns()
+ .iter()
+ .map(|field| Self::new(field.as_ref()))
+ .collect(),
+ ),
+ DataType::Dictionary(key, _) => {
+ Self::Fixed(key.primitive_width().unwrap_or_default())
+ }
+ DataType::FixedSizeBinary(size) => Self::Fixed(*size as usize),
+ DataType::Null | DataType::Boolean => Self::Fixed(0),
+ data_type => Self::Fixed(data_type.primitive_width().unwrap_or_else(|| {
+ // Other layouts are spread evenly over their rows
+ let bytes = array.to_data().get_slice_memory_size().unwrap_or_default();
+ bytes / array.len().max(1)
+ })),
+ }
+ }
+
+ /// Returns the size in bytes of every row, if they all have the same size.
+ fn fixed(&self) -> Option<usize> {
+ match self {
+ Self::Fixed(width) => Some(*width),
+ Self::FixedSizeList(size, values) => values.fixed().map(|width| size * width),
+ Self::Struct(fields) => fields.iter().map(Self::fixed).sum(),
+ Self::Bytes(_) | Self::Views(_) | Self::List(_, _) => None,
+ }
+ }
+
+ /// Returns an upper bound on the size in bytes of a row.
+ fn max_row(&self) -> usize {
+ match self {
+ Self::Fixed(width) => *width,
+ Self::Bytes(offsets) => offsets.width() + offsets.max_len(),
+ Self::Views(views) => {
+ let max_len = views
+ .iter()
+ .map(|&view| view as u32)
+ .filter(|&len| len > MAX_INLINE_VIEW_LEN)
+ .max()
+ .unwrap_or_default();
+ VIEW_SIZE_BYTES + max_len as usize
+ }
+ Self::List(offsets, values) => offsets
+ .width()
+ .saturating_add(offsets.max_len().saturating_mul(values.max_row())),
+ Self::FixedSizeList(size, values) => size.saturating_mul(values.max_row()),
+ Self::Struct(fields) => fields
+ .iter()
+ .fold(0, |bytes, field| bytes.saturating_add(field.max_row())),
+ }
+ }
+
+ /// Returns the size in bytes of rows `start..end`.
+ fn rows(&self, start: usize, end: usize) -> usize {
+ let num_rows = end - start;
+ match self {
+ Self::Fixed(width) => num_rows * width,
+ Self::Bytes(offsets) => {
+ num_rows * offsets.width() + offsets.get(end) - offsets.get(start)
+ }
+ Self::Views(views) => views[start..end]
+ .iter()
+ .map(|&view| {
+ let len = view as u32;
+ if len > MAX_INLINE_VIEW_LEN {
+ VIEW_SIZE_BYTES + len as usize
+ } else {
+ VIEW_SIZE_BYTES
+ }
+ })
+ .sum(),
+ Self::List(offsets, values) => {
+ num_rows * offsets.width()
+ + values.rows(offsets.get(start), offsets.get(end))
+ }
+ Self::FixedSizeList(size, values) => values.rows(start * size, end * size),
+ Self::Struct(fields) => {
+ fields.iter().map(|field| field.rows(start, end)).sum()
+ }
+ }
+ }
+}
+
+/// The offsets of a variable-width or list column.
+enum Offsets {
+ Small(ScalarBuffer<i32>),
+ Large(ScalarBuffer<i64>),
+}
+
+impl Offsets {
+ /// Returns the size in bytes of one offset.
+ fn width(&self) -> usize {
+ match self {
+ Self::Small(_) => size_of::<i32>(),
+ Self::Large(_) => size_of::<i64>(),
+ }
+ }
+
+ fn get(&self, index: usize) -> usize {
+ match self {
+ Self::Small(offsets) => offsets[index] as usize,
+ Self::Large(offsets) => offsets[index] as usize,
+ }
+ }
+
+ /// Returns the largest distance between consecutive offsets.
+ fn max_len(&self) -> usize {
+ match self {
+ Self::Small(offsets) => offsets
+ .windows(2)
+ .map(|pair| (pair[1] - pair[0]) as usize)
+ .max(),
+ Self::Large(offsets) => offsets
+ .windows(2)
+ .map(|pair| (pair[1] - pair[0]) as usize)
+ .max(),
+ }
+ .unwrap_or_default()
+ }
+}
+
+impl From<&OffsetBuffer<i32>> for Offsets {
+ fn from(offsets: &OffsetBuffer<i32>) -> Self {
+ Self::Small(offsets.inner().clone())
+ }
+}
+
+impl From<&OffsetBuffer<i64>> for Offsets {
+ fn from(offsets: &OffsetBuffer<i64>) -> Self {
+ Self::Large(offsets.inner().clone())
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow::array::{
+ ArrayRef, BooleanArray, DictionaryArray, FixedSizeBinaryArray, Int32Array,
+ Int64Array, ListArray, StringArray, StringViewArray, StructArray,
+ };
+ use arrow::buffer::Buffer;
+ use arrow::datatypes::{Field, Int32Type, Int64Type};
+ use std::sync::Arc;
+
+ fn batch(columns: Vec<ArrayRef>) -> RecordBatch {
+ RecordBatch::try_from_iter(
+ columns
+ .into_iter()
+ .enumerate()
+ .map(|(i, column)| (format!("c{i}"), column)),
+ )
+ .unwrap()
+ }
+
+ fn row_sizes(columns: Vec<ArrayRef>) -> Vec<usize> {
+ let batch = batch(columns);
+ let sizes = SpilledRowSizes::new(&batch);
+ (0..batch.num_rows()).map(|row| sizes.row(row)).collect()
+ }
+
+ #[test]
+ fn fixed_width_and_offsets() {
+ let ints: ArrayRef = Arc::new(Int64Array::from(vec![1, 2]));
+ let strings: ArrayRef = Arc::new(StringArray::from(vec!["a", "bbbb"]));
+ // 8 bytes of Int64, plus a 4 byte offset and the string bytes
+ assert_eq!(row_sizes(vec![ints, strings]), vec![13, 16]);
+ }
+
+ #[test]
+ fn rows_of_fixed_width_columns_have_one_size() {
+ let ints: ArrayRef = Arc::new(Int64Array::from(vec![1, 2]));
+ let fixed: ArrayRef = Arc::new(FixedSizeBinaryArray::new(
+ 3,
+ Buffer::from(vec![0u8; 6]),
+ None,
+ ));
+ let strings: ArrayRef = Arc::new(StringArray::from(vec!["a", "bbbb"]));
+ let sizes = SpilledRowSizes::new(&batch(vec![Arc::clone(&ints), fixed]));
+ assert_eq!(sizes.fixed(), Some(11));
+ assert_eq!(
+ SpilledRowSizes::new(&batch(vec![ints, strings])).fixed(),
+ None
+ );
+ }
+
+ #[test]
+ fn largest_row_bounds_every_row() {
Review Comment:
This test builds the same view and list arrays as `views_count_their_bytes_unless_inlined` and `lists_count_their_elements`, and nearly the same batch as `fixed_width_and_offsets`. Its `max_row` asserts could move into those three tests, next to the `row` asserts for the same arrays.
`other_types` and `rows_of_fixed_width_columns_have_one_size` also build the same `FixedSizeBinary(3)` array, so they could be one test.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-5901590467
π€ Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/25853#issuecomment-5901451278)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB)
Comparing fix-issue-258521 (832a7eb5c708f1ca2aa71e008201660203db45b4) to bdad988 (merge-base) [diff](https://github.com/apache/datafusion/compare/bdad988f23f4f8a45a71409e7d7f24a326ca909c..832a7eb5c708f1ca2aa71e008201660203db45b4)
<details><summary>Run configuration</summary>
```yaml
run benchmark tpch
```
</details>
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
<details><summary>Details</summary>
<p>
```
Comparing HEAD and fix-issue-258521
--------------------
Benchmark tpch_sf1.json
--------------------
βββββββββββββ³βββββββββββ³βββββββββββββββββββ³ββββββββββββ
β Query β HEAD β fix-issue-258521 β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β QQuery 1 β 40.74 ms β 40.75 ms β no change β
β QQuery 2 β 18.58 ms β 18.72 ms β no change β
β QQuery 3 β 28.37 ms β 28.20 ms β no change β
β QQuery 4 β 17.03 ms β 17.40 ms β no change β
β QQuery 5 β 34.71 ms β 35.52 ms β no change β
β QQuery 6 β 15.99 ms β 16.37 ms β no change β
β QQuery 7 β 41.10 ms β 42.25 ms β no change β
β QQuery 8 β 40.41 ms β 40.12 ms β no change β
β QQuery 9 β 48.74 ms β 50.68 ms β no change β
β QQuery 10 β 39.87 ms β 40.19 ms β no change β
β QQuery 11 β 12.95 ms β 13.14 ms β no change β
β QQuery 12 β 20.78 ms β 21.19 ms β no change β
β QQuery 13 β 39.15 ms β 38.89 ms β no change β
β QQuery 14 β 24.30 ms β 24.33 ms β no change β
β QQuery 15 β 30.86 ms β 30.79 ms β no change β
β QQuery 16 β 13.96 ms β 14.13 ms β no change β
β QQuery 17 β 72.60 ms β 71.77 ms β no change β
β QQuery 18 β 60.93 ms β 59.89 ms β no change β
β QQuery 19 β 31.86 ms β 31.85 ms β no change β
β QQuery 20 β 31.98 ms β 31.70 ms β no change β
β QQuery 21 β 55.41 ms β 54.53 ms β no change β
β QQuery 22 β 13.77 ms β 13.45 ms β no change β
βββββββββββββ΄βββββββββββ΄βββββββββββββββββββ΄ββββββββββββ
βββββββββββββββββββββββββββββββββββ³βββββββββββ
β Benchmark Summary β β
β‘βββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 734.10ms β
β Total Time (fix-issue-258521) β 735.87ms β
β Average Time (HEAD) β 33.37ms β
β Average Time (fix-issue-258521) β 33.45ms β
β Queries Faster β 0 β
β Queries Slower β 0 β
β Queries with No Change β 22 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββ΄βββββββββββ
Distribution per query (min / mean Β±stddev / max):
Comparing HEAD and fix-issue-258521
--------------------
Benchmark tpch_sf1.json
--------------------
βββββββββββββ³βββββββββββββββββββββββββββββββββ³βββββββββββββββββββββββββββββββββ³ββββββββββββββββ
β Query β HEAD β fix-issue-258521 β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β QQuery 1 β 40.74 / 41.65 Β±0.96 / 43.23 ms β 40.75 / 41.70 Β±0.98 / 43.33 ms β no change β
β QQuery 2 β 18.58 / 18.96 Β±0.44 / 19.73 ms β 18.72 / 18.84 Β±0.20 / 19.24 ms β no change β
β QQuery 3 β 28.37 / 28.87 Β±0.47 / 29.72 ms β 28.20 / 28.77 Β±0.29 / 29.01 ms β no change β
β QQuery 4 β 17.03 / 18.19 Β±0.87 / 19.40 ms β 17.40 / 17.52 Β±0.07 / 17.62 ms β no change β
β QQuery 5 β 34.71 / 35.08 Β±0.26 / 35.35 ms β 35.52 / 35.94 Β±0.40 / 36.68 ms β no change β
β QQuery 6 β 15.99 / 16.13 Β±0.10 / 16.31 ms β 16.37 / 16.93 Β±1.08 / 19.08 ms β no change β
β QQuery 7 β 41.10 / 42.73 Β±1.26 / 44.10 ms β 42.25 / 43.04 Β±0.88 / 44.70 ms β no change β
β QQuery 8 β 40.41 / 40.57 Β±0.15 / 40.82 ms β 40.12 / 40.69 Β±0.31 / 41.07 ms β no change β
β QQuery 9 β 48.74 / 49.94 Β±1.01 / 51.80 ms β 50.68 / 51.22 Β±0.59 / 52.13 ms β no change β
β QQuery 10 β 39.87 / 40.01 Β±0.16 / 40.30 ms β 40.19 / 41.19 Β±1.15 / 43.43 ms β no change β
β QQuery 11 β 12.95 / 13.30 Β±0.24 / 13.58 ms β 13.14 / 13.56 Β±0.48 / 14.47 ms β no change β
β QQuery 12 β 20.78 / 21.77 Β±0.80 / 22.77 ms β 21.19 / 22.36 Β±1.08 / 24.27 ms β no change β
β QQuery 13 β 39.15 / 40.54 Β±2.00 / 44.50 ms β 38.89 / 39.94 Β±1.10 / 41.97 ms β no change β
β QQuery 14 β 24.30 / 24.56 Β±0.20 / 24.82 ms β 24.33 / 24.55 Β±0.16 / 24.81 ms β no change β
β QQuery 15 β 30.86 / 31.25 Β±0.70 / 32.64 ms β 30.79 / 31.03 Β±0.14 / 31.20 ms β no change β
β QQuery 16 β 13.96 / 14.12 Β±0.12 / 14.33 ms β 14.13 / 14.36 Β±0.23 / 14.78 ms β no change β
β QQuery 17 β 72.60 / 73.71 Β±0.78 / 74.77 ms β 71.77 / 72.72 Β±1.10 / 74.81 ms β no change β
β QQuery 18 β 60.93 / 61.79 Β±1.09 / 63.89 ms β 59.89 / 60.39 Β±0.46 / 61.01 ms β no change β
β QQuery 19 β 31.86 / 33.80 Β±1.56 / 35.57 ms β 31.85 / 32.02 Β±0.15 / 32.27 ms β +1.06x faster β
β QQuery 20 β 31.98 / 32.21 Β±0.27 / 32.73 ms β 31.70 / 31.93 Β±0.33 / 32.57 ms β no change β
β QQuery 21 β 55.41 / 56.29 Β±0.99 / 57.93 ms β 54.53 / 55.11 Β±0.73 / 56.51 ms β no change β
β QQuery 22 β 13.77 / 14.08 Β±0.21 / 14.38 ms β 13.45 / 13.76 Β±0.24 / 14.04 ms β no change β
βββββββββββββ΄βββββββββββββββββββββββββββββββββ΄βββββββββββββββββββββββββββββββββ΄ββββββββββββββββ
βββββββββββββββββββββββββββββββββββ³βββββββββββ
β Benchmark Summary β β
β‘βββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 749.54ms β
β Total Time (fix-issue-258521) β 747.60ms β
β Average Time (HEAD) β 34.07ms β
β Average Time (fix-issue-258521) β 33.98ms β
β Queries Faster β 1 β
β Queries Slower β 0 β
β Queries with No Change β 21 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββ΄βββββββββββ
```
</p>
</details>
<details><summary>Resource Usage</summary>
**tpch β base (merge-base)**
| Metric | Value |
|--------|-------|
| Wall time | 5.0s |
| Peak memory | 1.3 GiB |
| Avg memory | 499.1 MiB |
| CPU user | 20.8s |
| CPU sys | 1.8s |
| Peak spill | 0 B |
**tpch β branch**
| Metric | Value |
|--------|-------|
| Wall time | 5.0s |
| Peak memory | 1.3 GiB |
| Avg memory | 502.6 MiB |
| CPU user | 20.8s |
| CPU sys | 1.8s |
| Peak spill | 0 B |
</details>
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#discussion_r4178384017
##########
datafusion/physical-plan/src/spill/spilled_row_sizes.rs:
##########
@@ -0,0 +1,426 @@
+// 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.
+
+//! Per-row sizes of a batch once it is written to a spill file.
+
+use arrow::array::{Array, AsArray, RecordBatch};
+use arrow::buffer::{OffsetBuffer, ScalarBuffer};
+use arrow::datatypes::DataType;
+use arrow_data::MAX_INLINE_VIEW_LEN;
+
+use super::VIEW_SIZE_BYTES;
+
+/// Estimates how many bytes each row of a batch takes once written to a spill
+/// file, where view arrays are compacted (see [`super::gc_view_arrays`]).
+///
+/// A row counts its fixed-width values, the offsets and bytes of its
+/// variable-width values, and the elements of its nested values. Buffers that
+/// every row shares, such as validity bitmaps and dictionary values, are not
+/// counted.
+pub(crate) struct SpilledRowSizes {
+ /// Bytes that every row takes in the fixed-width columns
+ fixed: usize,
+ /// The other columns
+ variable: Vec<ColumnSize>,
+ /// See [`Self::max_row`]
+ max_row: usize,
+}
+
+impl SpilledRowSizes {
+ pub(crate) fn new(batch: &RecordBatch) -> Self {
+ let mut fixed = 0;
+ let mut variable = vec![];
+ for column in batch.columns() {
+ let column = ColumnSize::new(column.as_ref());
+ match column.fixed() {
+ Some(width) => fixed += width,
+ None => variable.push(column),
+ }
+ }
+ let max_row = variable.iter().fold(fixed, |bytes, column| {
+ bytes.saturating_add(column.max_row())
+ });
+ Self {
+ fixed,
+ variable,
+ max_row,
+ }
+ }
+
+ /// Returns the size in bytes of row `row`.
+ pub(crate) fn row(&self, row: usize) -> usize {
+ self.variable.iter().fold(self.fixed, |bytes, column| {
+ bytes + column.rows(row, row + 1)
+ })
+ }
+
+ /// Returns the size in bytes of every row, if they all have the same size.
+ pub(crate) fn fixed(&self) -> Option<usize> {
+ self.variable.is_empty().then_some(self.fixed)
+ }
+
+ /// Returns an upper bound on the size in bytes of any row.
+ pub(crate) fn max_row(&self) -> usize {
+ self.max_row
+ }
+}
+
+/// Size model of one column, see [`SpilledRowSizes`].
+enum ColumnSize {
+ /// Every row takes the same number of bytes.
+ Fixed(usize),
+ /// An offset per row, plus the bytes between consecutive offsets.
+ Bytes(Offsets),
+ /// A view per row, plus the bytes of values too long to be inlined.
+ Views(ScalarBuffer<u128>),
+ /// An offset per row, plus the elements between consecutive offsets.
+ List(Offsets, Box<ColumnSize>),
+ /// A fixed number of elements per row.
+ FixedSizeList(usize, Box<ColumnSize>),
+ /// A row of each field.
+ Struct(Vec<ColumnSize>),
+}
+
+impl ColumnSize {
+ fn new(array: &dyn Array) -> Self {
+ match array.data_type() {
+ DataType::Utf8 => Self::Bytes(array.as_string::<i32>().offsets().into()),
+ DataType::LargeUtf8 => Self::Bytes(array.as_string::<i64>().offsets().into()),
+ DataType::Binary => Self::Bytes(array.as_binary::<i32>().offsets().into()),
+ DataType::LargeBinary => {
+ Self::Bytes(array.as_binary::<i64>().offsets().into())
+ }
+ DataType::Utf8View => Self::Views(array.as_string_view().views().clone()),
+ DataType::BinaryView => Self::Views(array.as_binary_view().views().clone()),
+ DataType::List(_) => {
+ let list = array.as_list::<i32>();
+ Self::List(list.offsets().into(), Box::new(Self::new(list.values())))
+ }
+ DataType::LargeList(_) => {
+ let list = array.as_list::<i64>();
+ Self::List(list.offsets().into(), Box::new(Self::new(list.values())))
+ }
+ DataType::Map(_, _) => {
+ let map = array.as_map();
+ Self::List(map.offsets().into(), Box::new(Self::new(map.entries())))
+ }
+ DataType::FixedSizeList(_, size) => Self::FixedSizeList(
+ *size as usize,
+ Box::new(Self::new(array.as_fixed_size_list().values())),
+ ),
+ DataType::Struct(_) => Self::Struct(
+ array
+ .as_struct()
+ .columns()
+ .iter()
+ .map(|field| Self::new(field.as_ref()))
+ .collect(),
+ ),
+ DataType::Dictionary(key, _) => {
+ Self::Fixed(key.primitive_width().unwrap_or_default())
+ }
+ DataType::FixedSizeBinary(size) => Self::Fixed(*size as usize),
+ DataType::Null | DataType::Boolean => Self::Fixed(0),
+ data_type => Self::Fixed(data_type.primitive_width().unwrap_or_else(|| {
+ // Other layouts are spread evenly over their rows
+ let bytes = array.to_data().get_slice_memory_size().unwrap_or_default();
+ bytes / array.len().max(1)
+ })),
+ }
+ }
+
+ /// Returns the size in bytes of every row, if they all have the same size.
+ fn fixed(&self) -> Option<usize> {
+ match self {
+ Self::Fixed(width) => Some(*width),
+ Self::FixedSizeList(size, values) => values.fixed().map(|width| size * width),
+ Self::Struct(fields) => fields.iter().map(Self::fixed).sum(),
+ Self::Bytes(_) | Self::Views(_) | Self::List(_, _) => None,
+ }
+ }
+
+ /// Returns an upper bound on the size in bytes of a row.
+ fn max_row(&self) -> usize {
+ match self {
+ Self::Fixed(width) => *width,
+ Self::Bytes(offsets) => offsets.width() + offsets.max_len(),
+ Self::Views(views) => {
+ let max_len = views
+ .iter()
+ .map(|&view| view as u32)
+ .filter(|&len| len > MAX_INLINE_VIEW_LEN)
+ .max()
+ .unwrap_or_default();
+ VIEW_SIZE_BYTES + max_len as usize
+ }
+ Self::List(offsets, values) => offsets
+ .width()
+ .saturating_add(offsets.max_len().saturating_mul(values.max_row())),
+ Self::FixedSizeList(size, values) => size.saturating_mul(values.max_row()),
+ Self::Struct(fields) => fields
+ .iter()
+ .fold(0, |bytes, field| bytes.saturating_add(field.max_row())),
+ }
+ }
+
+ /// Returns the size in bytes of rows `start..end`.
+ fn rows(&self, start: usize, end: usize) -> usize {
+ let num_rows = end - start;
+ match self {
+ Self::Fixed(width) => num_rows * width,
+ Self::Bytes(offsets) => {
+ num_rows * offsets.width() + offsets.get(end) - offsets.get(start)
+ }
+ Self::Views(views) => views[start..end]
+ .iter()
+ .map(|&view| {
+ let len = view as u32;
+ if len > MAX_INLINE_VIEW_LEN {
+ VIEW_SIZE_BYTES + len as usize
+ } else {
+ VIEW_SIZE_BYTES
+ }
+ })
+ .sum(),
+ Self::List(offsets, values) => {
+ num_rows * offsets.width()
+ + values.rows(offsets.get(start), offsets.get(end))
+ }
+ Self::FixedSizeList(size, values) => values.rows(start * size, end * size),
+ Self::Struct(fields) => {
+ fields.iter().map(|field| field.rows(start, end)).sum()
+ }
+ }
+ }
+}
+
+/// The offsets of a variable-width or list column.
+enum Offsets {
+ Small(ScalarBuffer<i32>),
+ Large(ScalarBuffer<i64>),
+}
+
+impl Offsets {
+ /// Returns the size in bytes of one offset.
+ fn width(&self) -> usize {
+ match self {
+ Self::Small(_) => size_of::<i32>(),
+ Self::Large(_) => size_of::<i64>(),
+ }
+ }
+
+ fn get(&self, index: usize) -> usize {
+ match self {
+ Self::Small(offsets) => offsets[index] as usize,
+ Self::Large(offsets) => offsets[index] as usize,
+ }
+ }
+
+ /// Returns the largest distance between consecutive offsets.
+ fn max_len(&self) -> usize {
+ match self {
+ Self::Small(offsets) => offsets
+ .windows(2)
+ .map(|pair| (pair[1] - pair[0]) as usize)
+ .max(),
+ Self::Large(offsets) => offsets
+ .windows(2)
+ .map(|pair| (pair[1] - pair[0]) as usize)
+ .max(),
+ }
+ .unwrap_or_default()
+ }
+}
+
+impl From<&OffsetBuffer<i32>> for Offsets {
+ fn from(offsets: &OffsetBuffer<i32>) -> Self {
+ Self::Small(offsets.inner().clone())
+ }
+}
+
+impl From<&OffsetBuffer<i64>> for Offsets {
+ fn from(offsets: &OffsetBuffer<i64>) -> Self {
+ Self::Large(offsets.inner().clone())
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow::array::{
+ ArrayRef, BooleanArray, DictionaryArray, FixedSizeBinaryArray, Int32Array,
+ Int64Array, ListArray, StringArray, StringViewArray, StructArray,
+ };
+ use arrow::buffer::Buffer;
+ use arrow::datatypes::{Field, Int32Type, Int64Type};
+ use std::sync::Arc;
+
+ fn batch(columns: Vec<ArrayRef>) -> RecordBatch {
+ RecordBatch::try_from_iter(
+ columns
+ .into_iter()
+ .enumerate()
+ .map(|(i, column)| (format!("c{i}"), column)),
+ )
+ .unwrap()
+ }
+
+ fn row_sizes(columns: Vec<ArrayRef>) -> Vec<usize> {
+ let batch = batch(columns);
+ let sizes = SpilledRowSizes::new(&batch);
+ (0..batch.num_rows()).map(|row| sizes.row(row)).collect()
+ }
+
+ #[test]
+ fn fixed_width_and_offsets() {
+ let ints: ArrayRef = Arc::new(Int64Array::from(vec![1, 2]));
+ let strings: ArrayRef = Arc::new(StringArray::from(vec!["a", "bbbb"]));
+ // 8 bytes of Int64, plus a 4 byte offset and the string bytes
+ assert_eq!(row_sizes(vec![ints, strings]), vec![13, 16]);
+ }
+
+ #[test]
+ fn rows_of_fixed_width_columns_have_one_size() {
+ let ints: ArrayRef = Arc::new(Int64Array::from(vec![1, 2]));
+ let fixed: ArrayRef = Arc::new(FixedSizeBinaryArray::new(
+ 3,
+ Buffer::from(vec![0u8; 6]),
+ None,
+ ));
+ let strings: ArrayRef = Arc::new(StringArray::from(vec!["a", "bbbb"]));
+ let sizes = SpilledRowSizes::new(&batch(vec![Arc::clone(&ints), fixed]));
+ assert_eq!(sizes.fixed(), Some(11));
+ assert_eq!(
+ SpilledRowSizes::new(&batch(vec![ints, strings])).fixed(),
+ None
+ );
+ }
+
+ #[test]
+ fn largest_row_bounds_every_row() {
Review Comment:
Done in b7cd9113c. `row_sizes` now returns the `max_row` bound along with the row sizes, so each test checks both for the same arrays, and `largest_row_bounds_every_row` is gone. `other_types` is folded into `rows_of_fixed_width_columns_have_one_size`.
##########
datafusion/physical-plan/src/sorts/streaming_merge.rs:
##########
@@ -326,6 +338,121 @@ mod tests {
ExecutionPlanMetricsSet, SpillMetrics,
};
+ /// Merges five 1000 byte rows, one per batch, with 1000 small rows in
+ /// batches of 100, either from spill files or from in-memory streams.
+ async fn merge_wide_and_narrow_rows(
+ from_spill_files: bool,
+ max_batch_bytes: Option<usize>,
+ ) -> Result<Vec<RecordBatch>> {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("key", DataType::Int32, false),
+ Field::new("value", DataType::Utf8, false),
+ ]));
+ let batch = |keys: std::ops::Range<i32>, value_len: usize| {
+ let values = keys.clone().map(|_| "x".repeat(value_len));
+ RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(Int32Array::from_iter_values(keys)),
+ Arc::new(StringArray::from_iter_values(values)),
+ ],
+ )
+ .unwrap()
+ };
+ let wide: Vec<_> = (0..5).map(|key| batch(key..key + 1, 1000)).collect();
+ let narrow: Vec<_> = (0..10)
+ .map(|i| batch(10 + i * 100..10 + (i + 1) * 100, 1))
+ .collect();
+
+ let sort: LexOrdering =
+ [PhysicalSortExpr::new_default(col("key", &schema)?)].into();
+ let mut builder = StreamingMergeBuilder::new()
+ .with_schema(Arc::clone(&schema))
+ .with_expressions(&sort)
+ .with_metrics(BaselineMetrics::new(&ExecutionPlanMetricsSet::new(), 0))
+ .with_batch_size(100)
+ .with_max_batch_bytes(max_batch_bytes)
+ .with_bypass_mempool();
+ if from_spill_files {
+ let spill_manager = SpillManager::new(
+ Arc::new(TaskContext::default()).runtime_env(),
+ SpillMetrics::new(&ExecutionPlanMetricsSet::new(), 0),
+ Arc::clone(&schema),
+ );
+ let spill = |batches: &[RecordBatch]| SortedSpillFile {
+ file: spill_manager
+ .spill_record_batch_and_finish(batches, "test")
+ .unwrap()
+ .unwrap(),
+ max_record_batch_memory: batches
+ .iter()
+ .map(RecordBatch::get_array_memory_size)
+ .max()
+ .unwrap(),
+ };
+ builder = builder
+ .with_sorted_spill_files(vec![spill(&wide), spill(&narrow)])
+ .with_spill_manager(spill_manager);
+ } else {
+ let stream = |batches: Vec<RecordBatch>| {
+ Box::pin(RecordBatchStreamAdapter::new(
+ Arc::clone(&schema),
+ futures::stream::iter(batches.into_iter().map(Ok)),
+ )) as SendableRecordBatchStream
+ };
+ builder = builder.with_streams(vec![stream(wide), stream(narrow)]);
+ }
+ let batches = collect(builder.build()?).await?;
+
+ let keys: Vec<i32> = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_primitive::<Int32Type>()
+ .values()
+ .to_vec()
+ })
+ .collect();
+ let expected: Vec<i32> = (0..5).chain(10..1010).collect();
+ assert_eq!(keys, expected);
+ Ok(batches)
+ }
+
+ #[tokio::test]
+ async fn test_max_batch_bytes_limits_merged_batches() -> Result<()> {
+ let num_rows = |batches: &[RecordBatch]| -> Vec<usize> {
+ batches.iter().map(RecordBatch::num_rows).collect()
+ };
+ for from_spill_files in [true, false] {
Review Comment:
Done in b7cd9113c. The test now merges only spill files, without the `from_spill_files` flag or the in-memory branch.
##########
datafusion/sqllogictest/test_files/aggregate_memory_spill.slt:
##########
@@ -233,6 +233,110 @@ FROM (
08)--------------AggregateExec: mode=Partial, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[<slt:ignore>early_emit_count=<slt:ignore>]
<slt:ignore>
+# --- Case H: a few groups with large state ---
+# The array_agg states of all 32 groups do not fit in the memory limit, but a
+# single group does. Spilling must write the groups in batches that can be
+# read back within the limit, instead of one batch holding every group.
+# See https://github.com/apache/datafusion/issues/25851
+statement ok
+RESET datafusion.runtime.max_spill_merge_fan_in
+
+statement ok
+SET datafusion.execution.batch_size = 8192
+
+statement ok
+SET datafusion.runtime.memory_limit = '24M'
+
+# Partial/final aggregation: the final aggregation spills the partial states of
+# its partition.
+statement ok
+SET datafusion.execution.target_partitions = 2
+
+query II
+SELECT count(*), sum(n)
+FROM (
+ SELECT v % 32 AS k, cardinality(array_agg(concat(CAST(v AS VARCHAR), repeat('x', 100)))) AS n
+ FROM generate_series(1, 500000) AS t(v)
+ GROUP BY v % 32
+)
+----
+32 500000
+
+# Assert the final aggregation spilled, the `spilled_bytes` metric must be in MB
+query TT
+EXPLAIN ANALYZE
+SELECT count(*), sum(n)
+FROM (
+ SELECT v % 32 AS k, cardinality(array_agg(concat(CAST(v AS VARCHAR), repeat('x', 100)))) AS n
+ FROM generate_series(1, 500000) AS t(v)
+ GROUP BY v % 32
+)
+----
+<slt:ignore>
+06)----------AggregateExec: mode=FinalPartitioned, gby=[t.v % Int64(32)@0 as t.v % Int64(32)], aggr=[<slt:ignore>], metrics=[<slt:ignore>spilled_bytes=<slt:ignore> MB, spilled_rows=<slt:ignore>]
+07)------------RepartitionExec: <slt:ignore>
+
+# Single-stage aggregation: spills several runs as input arrives, which replay
+# merges.
+statement ok
+SET datafusion.execution.target_partitions = 1
+
+query II
+SELECT count(*), sum(n)
+FROM (
+ SELECT v % 32 AS k, cardinality(array_agg(concat(CAST(v AS VARCHAR), repeat('x', 100)))) AS n
+ FROM generate_series(1, 500000) AS t(v)
+ GROUP BY v % 32
+)
+----
+32 500000
+
+# Assert spill happened, the `spilled_bytes` metric must be in MB
+query TT
+EXPLAIN ANALYZE
+SELECT count(*), sum(n)
+FROM (
+ SELECT v % 32 AS k, cardinality(array_agg(concat(CAST(v AS VARCHAR), repeat('x', 100)))) AS n
+ FROM generate_series(1, 500000) AS t(v)
+ GROUP BY v % 32
+)
+----
+<slt:ignore>
+04)------AggregateExec: mode=Single, gby=[v@0 % 32 as t.v % Int64(32)], aggr=[<slt:ignore>], metrics=[<slt:ignore>spilled_bytes=<slt:ignore> MB, spilled_rows=<slt:ignore>]
+05)--------ProjectionExec: <slt:ignore>
+
+# 14 large groups, adjacent in key order, spilled together with 62,500 small
Review Comment:
Done in b7cd9113c. Both queries now have an `EXPLAIN ANALYZE` check that `spilled_bytes` is in MB. Locally they spill 7 and 5 times.
##########
datafusion/physical-plan/src/aggregates/spill.rs:
##########
@@ -290,11 +311,141 @@ impl AggregateSpill {
partition,
merged,
&InputOrderMode::Sorted,
- baseline_metrics.clone(),
+ baseline_metrics.intermediate(),
metrics,
None,
reservation,
)?;
- Ok(replay.into_stream())
+ // Replay emits the groups of each merged batch, which may be few
+ let output = coalesce(
+ replay.into_stream(),
+ batch_size,
+ SPILL_BATCH_TARGET_BYTES,
+ baseline_metrics.elapsed_compute().clone(),
+ );
+ Ok(Box::pin(ObservedStream::new(
+ output,
+ baseline_metrics.clone(),
+ None,
+ )))
+ }
+}
+
+/// Combines consecutive small batches of `input` into batches of up to
+/// `batch_size` rows and `max_bytes`. Batches of at least half that size are
+/// passed on as they are.
+fn coalesce(
Review Comment:
Removed in b7cd9113c, so replay returns the `OrderedFinalAggregateStream` output as on main. It emits the groups that each input batch completes, so even on main replay emits about `batch_size / runs` rows per batch. Agreed that combining them belongs in the operator, I'll follow up separately.
##########
datafusion/physical-plan/src/aggregates/spill.rs:
##########
@@ -280,6 +300,7 @@ impl AggregateSpill {
.with_expressions(&spill_expr)
.with_metrics(baseline_metrics.intermediate())
.with_batch_size(batch_size)
+ .with_max_batch_bytes(Some(SPILL_BATCH_TARGET_BYTES))
.with_reservation(merge_reservation)
.with_replay_headroom()
Review Comment:
Agreed. The spill merge now has three options that only aggregate spilling sets (`with_replay_headroom`, `with_intermediate_merge_sizing` and `with_max_batch_bytes`). I'd rather revisit them together in a follow-up than grow this PR.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-6384238616
π€ Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/25853#issuecomment-6384101936)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB)
Comparing fix-issue-258521 (3289c00a6bfdf14b9ad444d7ff47b2b9364b1be2) to 1da3cf0 (merge-base) [diff](https://github.com/apache/datafusion/compare/1da3cf049fa1a7a708acc2c29eba4f2e9dece8db..3289c00a6bfdf14b9ad444d7ff47b2b9364b1be2)
<details><summary>Run configuration</summary>
```yaml
run benchmark external_aggr
```
</details>
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
<details><summary>Details</summary>
<p>
```
Comparing HEAD and fix-issue-258521
--------------------
Benchmark external_aggr.json
--------------------
ββββββββββββββββ³ββββββββββββ³βββββββββββββββββββ³ββββββββββββ
β Query β HEAD β fix-issue-258521 β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Q1(64.0 MB) β 56.41 ms β 55.59 ms β no change β
β Q1(32.0 MB) β 50.44 ms β 49.23 ms β no change β
β Q1(16.0 MB) β 50.28 ms β 50.11 ms β no change β
β Q2(512.0 MB) β 241.92 ms β 253.00 ms β no change β
β Q2(256.0 MB) β 234.64 ms β 240.13 ms β no change β
β Q2(128.0 MB) β 206.75 ms β 209.50 ms β no change β
β Q2(64.0 MB) β 202.74 ms β 206.70 ms β no change β
β Q2(32.0 MB) β 264.67 ms β 266.90 ms β no change β
ββββββββββββββββ΄ββββββββββββ΄βββββββββββββββββββ΄ββββββββββββ
βββββββββββββββββββββββββββββββββββ³ββββββββββββ
β Benchmark Summary β β
β‘ββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 1307.85ms β
β Total Time (fix-issue-258521) β 1331.16ms β
β Average Time (HEAD) β 163.48ms β
β Average Time (fix-issue-258521) β 166.39ms β
β Queries Faster β 0 β
β Queries Slower β 0 β
β Queries with No Change β 8 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββ΄ββββββββββββ
Distribution per query (min / mean Β±stddev / max):
Comparing HEAD and fix-issue-258521
--------------------
Benchmark external_aggr.json
--------------------
ββββββββββββββββ³βββββββββββββββββββββββββββββββββββββ³βββββββββββββββββββββββββββββββββββββ³ββββββββββββ
β Query β HEAD β fix-issue-258521 β Change β
β‘βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Q1(64.0 MB) β 56.41 / 58.77 Β±2.27 / 62.52 ms β 55.59 / 59.07 Β±5.36 / 69.70 ms β no change β
β Q1(32.0 MB) β 50.44 / 52.06 Β±1.46 / 54.31 ms β 49.23 / 52.69 Β±2.02 / 54.76 ms β no change β
β Q1(16.0 MB) β 50.28 / 50.85 Β±0.60 / 51.94 ms β 50.11 / 52.27 Β±2.13 / 56.29 ms β no change β
β Q2(512.0 MB) β 241.92 / 251.00 Β±8.59 / 266.96 ms β 253.00 / 260.70 Β±7.20 / 273.34 ms β no change β
β Q2(256.0 MB) β 234.64 / 247.95 Β±13.27 / 268.65 ms β 240.13 / 253.05 Β±15.83 / 282.60 ms β no change β
β Q2(128.0 MB) β 206.75 / 212.86 Β±6.55 / 225.13 ms β 209.50 / 215.98 Β±6.12 / 226.78 ms β no change β
β Q2(64.0 MB) β 202.74 / 210.11 Β±10.05 / 229.92 ms β 206.70 / 209.31 Β±2.69 / 213.98 ms β no change β
β Q2(32.0 MB) β 264.67 / 267.63 Β±2.77 / 272.90 ms β 266.90 / 272.51 Β±3.20 / 275.78 ms β no change β
ββββββββββββββββ΄βββββββββββββββββββββββββββββββββββββ΄βββββββββββββββββββββββββββββββββββββ΄ββββββββββββ
βββββββββββββββββββββββββββββββββββ³ββββββββββββ
β Benchmark Summary β β
β‘ββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 1351.22ms β
β Total Time (fix-issue-258521) β 1375.58ms β
β Average Time (HEAD) β 168.90ms β
β Average Time (fix-issue-258521) β 171.95ms β
β Queries Faster β 0 β
β Queries Slower β 0 β
β Queries with No Change β 8 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββ΄ββββββββββββ
```
</p>
</details>
<details><summary>Memory Pool Peaks</summary>
Peak `MemoryPool` reservation per query β what DataFusion's accounting believes it reserved. Recorded only for benchmarks that write a results JSON (the `dfbench` suites, and the suites `bench.sh` runs through the Criterion SQL harness since apache/datafusion#25644), and only under a memory limit: `DATAFUSION_RUNTIME_MEMORY_LIMIT`, or one the suite sets itself.
Base: `1da3cf0 (merge-base)` | Changed: `fix-issue-258521`
**`external_aggr`**
| Query | Base | Changed | Change |
| --- | --- | --- | --- |
| 1(64.0 MB) | 29.3 MiB | 29.3 MiB | +0.0% |
| 1(32.0 MB) | 14.3 MiB | 15.3 MiB | +7.5% |
| 1(16.0 MB) | 8.9 MiB | 8.9 MiB | +0.0% |
| 2(512.0 MB) | 136.8 MiB | 136.3 MiB | -0.4% |
| 2(256.0 MB) | 98.0 MiB | 97.9 MiB | -0.2% |
| 2(128.0 MB) | 49.3 MiB | 49.2 MiB | -0.1% |
| 2(64.0 MB) | 32.5 MiB | 32.5 MiB | +0.0% |
| 2(32.0 MB) | 18.2 MiB | 17.7 MiB | -2.7% |
**Pool accounting vs. process RSS**
Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
| Benchmark | Side | Max pool peak | Peak RSS | Gap | RSS / pool |
| --- | --- | --- | --- | --- | --- |
| external_aggr | base (`1da3cf0 (merge-base)`) | 136.8 MiB | 453.5 MiB | 316.6 MiB | 3.3Γ |
| external_aggr | changed (`fix-issue-258521`) | 136.3 MiB | 458.2 MiB | 322.0 MiB | 3.4Γ |
</details>
<details><summary>Resource Usage</summary>
**external_aggr β base (merge-base)**
| Metric | Value |
|--------|-------|
| Wall time | 105.0s |
| Peak memory | 453.5 MiB |
| Avg memory | 27.0 MiB |
| CPU user | 21.9s |
| CPU sys | 3.6s |
| Peak spill | 93.1 MiB |
**external_aggr β branch**
| Metric | Value |
|--------|-------|
| Wall time | 110.0s |
| Peak memory | 458.2 MiB |
| Avg memory | 23.4 MiB |
| CPU user | 21.7s |
| CPU sys | 3.8s |
| Peak spill | 65.1 MiB |
</details>
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "andygrove (via GitHub) via github" <gi...@datafusion.apache.org>.
andygrove commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-6723039337
Filed the remaining dictionary case (values that repeat within a group but not across groups, with a large dictionary) as #27240, with a reproducer. It fails the same way as before this PR's dictionary change did, so I'd keep it out of this PR and handle it as a follow-up, alongside the consolidation in #26711.
@2010YOUY01 @jayzhan211 thanks for the design notes. This PR stays a bounded fix for #25851, and the design questions (one byte limit for every caller, removing the three workarounds) belong under #26711.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-5901520896
π€ Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/25853#issuecomment-5901451278)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c5901451278-2937-rrdhm 6.12.94+ #1 SMP Wed Aug 19 07:47:20 UTC 2026 aarch64 GNU/Linux`
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
Comparing fix-issue-258521 (832a7eb5c708f1ca2aa71e008201660203db45b4) to bdad988 (merge-base) [diff](https://github.com/apache/datafusion/compare/bdad988f23f4f8a45a71409e7d7f24a326ca909c..832a7eb5c708f1ca2aa71e008201660203db45b4)
<details><summary>Run configuration</summary>
```yaml
run benchmark tpcds
```
</details>
Results will be posted here when complete
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-6223976313
Merged main into the branch. Beyond the conflict, this needed two changes:
- `VIEW_SIZE_BYTES` is defined for all builds again (ea8d37dd5). #23565 made it test-only, but `SpilledRowSizes` uses it.
- Replay compacts the view arrays of nested columns in merged batches (3289c00a6). With main merged, the Case H query with 14 adjacent large groups failed with `ResourcesExhausted` at 24 MB, which I bisected to #23565. A merged batch references the whole data buffers of the spilled batches its rows come from, and replay keeps the `array_agg` lists of each group until the group completes, so it held about twice the state of each group. Since #23565, spilling no longer compacts view arrays whose buffers are fully used, which was enough to push this query over the limit. Top-level columns are left as they are, since group keys and the accumulators of top-level states typically copy the values they keep.
@comphead both came after your approval, in case you want to look at 3289c00a6.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "EmilyMatt (via GitHub)" <gi...@apache.org>.
EmilyMatt commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-6248566015
What I see:
1. During the actual emit, for a brief instance, we are holding both the output arrays, and the state, meaning peak is still double, we need to have an actual estimation for the emitted size, I have a POC for this, can open a PR if you want, then you'll be able to use it to calculate the expected peak.
2. remaining_groups.get_array_memory_size() only calculates the *data*, not the capacity, and since you add another spill trigger, we could be doing more of these emits where the the Vec's capacities are larger than data and so we emit more than we actually reserved and are releaseing, unfortunately I don't see any solution for this other than calling shrink_to_fit() *for each array* after emitting it(so we peak at each array's 2x size, rather than the whole batch's), adds an extra copy, but greatly reduces memory footprint.
3. try_reserve_hash_table_growth still lets a batch through on failed reservation, instead we should slice it and try again, since our estimation depends on the the input array sizes, but also on the sliced data sizes. so it's possible that we can safely process the batch by simply doing the extra spill :woman_shrugging:
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "andygrove (via GitHub) via github" <gi...@datafusion.apache.org>.
andygrove commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-6662614104
@alamb @2010YOUY01 do you have any additional feedback, or can we merge this one (and maybe consider for 56.0.0)?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "jayzhan211 (via GitHub) via github" <gi...@datafusion.apache.org>.
jayzhan211 commented on code in PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#discussion_r4435947546
##########
datafusion/sqllogictest/test_files/aggregate_memory_spill.slt:
##########
@@ -239,6 +239,146 @@ FROM (
08)--------------AggregateExec: mode=Partial, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[<slt:ignore>early_emit_count=<slt:ignore>]
<slt:ignore>
+# --- Case H: a few groups with large state ---
Review Comment:
Optional: every Case H query uses `array_agg`, so only the list-of-views path is covered end to end. A `string_agg` case exercises top-level bytes state; it fails on main and passes with this PR at `-m 128M`:
```sql
SET datafusion.execution.target_partitions = 2;
SELECT count(*), sum(n) FROM (
SELECT value % 16 AS k, length(string_agg(concat(CAST(value AS VARCHAR), repeat('x', 100)), ',')) AS n
FROM generate_series(1, 3000000)
GROUP BY value % 16);
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-5883502938
## [Codecov](https://app.codecov.io/gh/apache/datafusion/pull/25853?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `82.75862%` with `5 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 82.57%. Comparing base ([`bdad988`](https://app.codecov.io/gh/apache/datafusion/commit/bdad988f23f4f8a45a71409e7d7f24a326ca909c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`d541333`](https://app.codecov.io/gh/apache/datafusion/commit/d541333b1b2d7f72bb85f8507bcf607183a0fe9b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/datafusion/pull/25853?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [datafusion/physical-plan/src/aggregates/spill.rs](https://app.codecov.io/gh/apache/datafusion/pull/25853?src=pr&el=tree&filepath=datafusion%2Fphysical-plan%2Fsrc%2Faggregates%2Fspill.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1wbGFuL3NyYy9hZ2dyZWdhdGVzL3NwaWxsLnJz) | 82.75% | [0 Missing and 5 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/25853?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #25853 +/- ##
==========================================
- Coverage 82.57% 82.57% -0.01%
==========================================
Files 1142 1142
Lines 440861 440886 +25
Branches 440861 440886 +25
==========================================
+ Hits 364047 364060 +13
- Misses 54818 54823 +5
- Partials 21996 22003 +7
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/datafusion/pull/25853?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "andygrove (via GitHub) via github" <gi...@datafusion.apache.org>.
andygrove commented on code in PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#discussion_r4436292566
##########
datafusion/physical-plan/src/spill/spilled_row_sizes.rs:
##########
@@ -0,0 +1,398 @@
+// 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.
+
+//! Per-row sizes of a batch once it is written to a spill file.
+
+use arrow::array::{Array, AsArray, RecordBatch};
+use arrow::buffer::{OffsetBuffer, ScalarBuffer};
+use arrow::datatypes::DataType;
+use arrow_data::MAX_INLINE_VIEW_LEN;
+
+use super::VIEW_SIZE_BYTES;
+
+/// Estimates how many bytes each row of a batch takes once written to a spill
+/// file, where view arrays are compacted (see [`super::gc_view_arrays`]).
+///
+/// A row counts its fixed-width values, the offsets and bytes of its
+/// variable-width values, and the elements of its nested values. Buffers that
+/// every row shares, such as validity bitmaps and dictionary values, are not
+/// counted. Each view counts the bytes of its value, so views that share bytes
+/// are counted as if compaction copied them, although it can write shared bytes
+/// only once.
+pub(crate) struct SpilledRowSizes {
+ /// Bytes that every row takes in the fixed-width columns
+ fixed: usize,
+ /// The other columns
+ variable: Vec<ColumnSize>,
+ /// See [`Self::max_row`]
+ max_row: usize,
+}
+
+impl SpilledRowSizes {
+ pub(crate) fn new(batch: &RecordBatch) -> Self {
+ let mut fixed = 0;
+ let mut variable = vec![];
+ for column in batch.columns() {
+ let column = ColumnSize::new(column.as_ref());
+ match column.fixed() {
+ Some(width) => fixed += width,
+ None => variable.push(column),
+ }
+ }
+ let max_row = variable.iter().fold(fixed, |bytes, column| {
+ bytes.saturating_add(column.max_row())
+ });
+ Self {
+ fixed,
+ variable,
+ max_row,
+ }
+ }
+
+ /// Returns the size in bytes of row `row`.
+ pub(crate) fn row(&self, row: usize) -> usize {
+ self.variable.iter().fold(self.fixed, |bytes, column| {
+ bytes + column.rows(row, row + 1)
+ })
+ }
+
+ /// Returns the size in bytes of every row, if they all have the same size.
+ pub(crate) fn fixed(&self) -> Option<usize> {
+ self.variable.is_empty().then_some(self.fixed)
+ }
+
+ /// Returns an upper bound on the size in bytes of any row.
+ pub(crate) fn max_row(&self) -> usize {
+ self.max_row
+ }
+}
+
+/// Size model of one column, see [`SpilledRowSizes`].
+enum ColumnSize {
+ /// Every row takes the same number of bytes.
+ Fixed(usize),
+ /// An offset per row, plus the bytes between consecutive offsets.
+ Bytes(Offsets),
+ /// A view per row, plus the bytes of values too long to be inlined.
+ Views(ScalarBuffer<u128>),
+ /// An offset per row, plus the elements between consecutive offsets.
+ List(Offsets, Box<ColumnSize>),
+ /// A fixed number of elements per row.
+ FixedSizeList(usize, Box<ColumnSize>),
+ /// A row of each field.
+ Struct(Vec<ColumnSize>),
+}
+
+impl ColumnSize {
+ fn new(array: &dyn Array) -> Self {
+ match array.data_type() {
+ DataType::Utf8 => Self::Bytes(array.as_string::<i32>().offsets().into()),
+ DataType::LargeUtf8 => Self::Bytes(array.as_string::<i64>().offsets().into()),
+ DataType::Binary => Self::Bytes(array.as_binary::<i32>().offsets().into()),
+ DataType::LargeBinary => {
+ Self::Bytes(array.as_binary::<i64>().offsets().into())
+ }
+ DataType::Utf8View => Self::Views(array.as_string_view().views().clone()),
+ DataType::BinaryView => Self::Views(array.as_binary_view().views().clone()),
+ DataType::List(_) => {
+ let list = array.as_list::<i32>();
+ Self::List(list.offsets().into(), Box::new(Self::new(list.values())))
+ }
+ DataType::LargeList(_) => {
+ let list = array.as_list::<i64>();
+ Self::List(list.offsets().into(), Box::new(Self::new(list.values())))
+ }
+ DataType::Map(_, _) => {
+ let map = array.as_map();
+ Self::List(map.offsets().into(), Box::new(Self::new(map.entries())))
+ }
+ DataType::FixedSizeList(_, size) => Self::FixedSizeList(
+ *size as usize,
+ Box::new(Self::new(array.as_fixed_size_list().values())),
+ ),
+ DataType::Struct(_) => Self::Struct(
+ array
+ .as_struct()
+ .columns()
+ .iter()
+ .map(|field| Self::new(field.as_ref()))
+ .collect(),
+ ),
+ DataType::Dictionary(key, _) => {
+ Self::Fixed(key.primitive_width().unwrap_or_default())
+ }
Review Comment:
Thanks, that reproduces here, and so does a single-stage version (`target_partitions = 1`), which fails with `Failed to allocate additional 233.6 MB`. Fixed in c5cc3af12, but differently from the diff above:
- `normalized_keys()` asserts that the dictionary has values, so it panics on the empty dictionary that a batch of null lists holds.
- Compacting the dictionaries of every chunk makes repeated values expensive to spill. The `array_agg` state dictionary holds a copy of each input batch's dictionary, and each chunk references a different subset of those copies, so each chunk writes its own. On `array_agg` over 100 distinct 200-byte values (2M rows, 100K groups, 24 MB limit), the diff above spills 371 MB, against 14.7 MB on main, where every chunk shares the dictionary and the spill file holds it once.
So `SpilledRowSizes` now decides per dictionary. When fewer than two keys reference each value on average, a row counts the values its keys reference, and `IncrementalSortIterator` compacts each chunk to those values. Other dictionaries count only their keys and stay shared, as on main. Replay also compacts the dictionaries of nested columns, as it does their views, since a merged batch can hold the dictionaries of all the batches its rows come from. The compaction only runs in aggregate spilling, so the spill files of other operators don't change.
Your query and the 1000-byte variant now pass at 128 MB, as do single-stage and `max_spill_merge_fan_in = 2` versions, and the low-cardinality case above spills 10.5 MB and 14.7 MB (20- and 200-byte values), as on main. Case I now has a version of your query at the file's 24 MB limit, with `target_partitions` 2 and 1, and a query whose values repeat within each group, which needs the replay compaction.
This still leaves one case as on main: a large dictionary whose values repeat within a group but not across groups. For example, `array_agg` over 5M rows whose values each repeat 10 times in one group still fails at 24 and 64 MB. Compacting would be cheap there, but telling it apart from values that groups share needs counting distinct values per row, which I'd leave for a follow-up.
##########
datafusion/sqllogictest/test_files/aggregate_memory_spill.slt:
##########
@@ -239,6 +239,146 @@ FROM (
08)--------------AggregateExec: mode=Partial, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[<slt:ignore>early_emit_count=<slt:ignore>]
<slt:ignore>
+# --- Case H: a few groups with large state ---
Review Comment:
Added in c5cc3af12, as a single-stage query. At the file's 24 MB limit with 500K rows, the two-partition version already passes on main, while the single-stage one fails there with `Failed to allocate additional 48.3 MB`.
##########
datafusion/sqllogictest/test_files/aggregate_memory_spill.slt:
##########
@@ -239,6 +239,146 @@ FROM (
08)--------------AggregateExec: mode=Partial, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[<slt:ignore>early_emit_count=<slt:ignore>]
<slt:ignore>
+# --- Case H: a few groups with large state ---
+# The array_agg states of all 32 groups do not fit in the memory limit, but a
+# single group does. Spilling must write the groups in batches that can be
+# read back within the limit, instead of one batch holding every group.
Review Comment:
It comes from the Case I queries, but not from the spilling changes. Run on its own, the first Case I query (two partitions) peaks at 103.6 MB allocated with 10.9 KB reserved, sampled when a partial aggregate changes its reservation. With a 2 GB limit, where nothing spills and none of this PR's code runs, it's the same (107 MB). With `target_partitions = 1` and 2 GB, the drift is 103.5 MB with 48 B reserved, sampled when the aggregate releases its reservation after emitting. So it's memory that the aggregation holds outside any reservation, such as the batches an aggregate has emitted and the next operator hasn't reserved, which is what #25650 tracks.
##########
datafusion/physical-plan/src/spill/spilled_row_sizes.rs:
##########
@@ -0,0 +1,398 @@
+// 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.
+
+//! Per-row sizes of a batch once it is written to a spill file.
+
+use arrow::array::{Array, AsArray, RecordBatch};
+use arrow::buffer::{OffsetBuffer, ScalarBuffer};
+use arrow::datatypes::DataType;
+use arrow_data::MAX_INLINE_VIEW_LEN;
+
+use super::VIEW_SIZE_BYTES;
+
+/// Estimates how many bytes each row of a batch takes once written to a spill
+/// file, where view arrays are compacted (see [`super::gc_view_arrays`]).
+///
+/// A row counts its fixed-width values, the offsets and bytes of its
+/// variable-width values, and the elements of its nested values. Buffers that
+/// every row shares, such as validity bitmaps and dictionary values, are not
+/// counted. Each view counts the bytes of its value, so views that share bytes
+/// are counted as if compaction copied them, although it can write shared bytes
+/// only once.
+pub(crate) struct SpilledRowSizes {
Review Comment:
Agreed. Nothing in it is specific to aggregation, since it sizes rows by how they're written to a spill file, but only aggregate spilling uses it so far. A sort could use it the same way to bound its spill batches. Happy to consolidate it in the #26711 follow-up.
##########
datafusion/physical-plan/src/sorts/merge.rs:
##########
@@ -146,11 +156,22 @@ impl<C: CursorValues> SortPreservingMergeStream<C> {
poll_reset_epochs: vec![0; stream_count],
loser_tree: vec![],
batch_size,
+ max_batch_bytes: None,
+ avg_row_bytes: vec![0; stream_count],
+ in_progress_bytes: 0,
fetch,
produced: 0,
}
}
+ /// Also emits a batch once its rows take about `max_batch_bytes`, as
Review Comment:
Only aggregate spilling needs it so far. `batch_size` bounds rows, and a row of aggregate state can hold a whole group, such as an `array_agg` list, so the batches that spilling writes and replay merges can be far larger than their row count suggests. A global `max_batch_bytes` like `batch_size` makes sense to me, but every operator that builds batches would need to estimate row sizes, as this PR does for the two that aggregate spilling uses: `IncrementalSortIterator` sizes each row as it is spilled, and the merge uses the average row size of each input batch. I kept the option crate-private so it can become a config later without an API change. It could be part of #26711.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "andygrove (via GitHub) via github" <gi...@datafusion.apache.org>.
andygrove commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-6560189742
Merged main again and addressed the latest reviews:
- Main added its own Case H to `aggregate_memory_spill.slt`, so this PR's case is now Case I.
- Dictionary-typed state (@jayzhan211): c5cc3af12, details in the review thread. It sizes and compacts dictionaries whose values are mostly unique, and leaves the others shared as on main.
- `string_agg` case added, and @alamb's questions answered inline.
@comphead, the dictionary handling is new since your approval, in case you want to look at c5cc3af12.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "jayzhan211 (via GitHub) via github" <gi...@datafusion.apache.org>.
jayzhan211 commented on code in PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#discussion_r4435947546
##########
datafusion/sqllogictest/test_files/aggregate_memory_spill.slt:
##########
@@ -239,6 +239,146 @@ FROM (
08)--------------AggregateExec: mode=Partial, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[<slt:ignore>early_emit_count=<slt:ignore>]
<slt:ignore>
+# --- Case H: a few groups with large state ---
Review Comment:
Optional: every Case H query uses `array_agg`, so only the list-of-views path is covered end to end. A `string_agg` case exercises top-level bytes state; it fails on main and passes with this PR at `-m 128M`:
```sql
SELECT count(*), sum(n) FROM (
SELECT value % 16 AS k, length(string_agg(concat(CAST(value AS VARCHAR), repeat('x', 100)), ',')) AS n
FROM generate_series(1, 3000000)
GROUP BY value % 16);
```
##########
datafusion/physical-plan/src/spill/spilled_row_sizes.rs:
##########
@@ -0,0 +1,398 @@
+// 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.
+
+//! Per-row sizes of a batch once it is written to a spill file.
+
+use arrow::array::{Array, AsArray, RecordBatch};
+use arrow::buffer::{OffsetBuffer, ScalarBuffer};
+use arrow::datatypes::DataType;
+use arrow_data::MAX_INLINE_VIEW_LEN;
+
+use super::VIEW_SIZE_BYTES;
+
+/// Estimates how many bytes each row of a batch takes once written to a spill
+/// file, where view arrays are compacted (see [`super::gc_view_arrays`]).
+///
+/// A row counts its fixed-width values, the offsets and bytes of its
+/// variable-width values, and the elements of its nested values. Buffers that
+/// every row shares, such as validity bitmaps and dictionary values, are not
+/// counted. Each view counts the bytes of its value, so views that share bytes
+/// are counted as if compaction copied them, although it can write shared bytes
+/// only once.
+pub(crate) struct SpilledRowSizes {
+ /// Bytes that every row takes in the fixed-width columns
+ fixed: usize,
+ /// The other columns
+ variable: Vec<ColumnSize>,
+ /// See [`Self::max_row`]
+ max_row: usize,
+}
+
+impl SpilledRowSizes {
+ pub(crate) fn new(batch: &RecordBatch) -> Self {
+ let mut fixed = 0;
+ let mut variable = vec![];
+ for column in batch.columns() {
+ let column = ColumnSize::new(column.as_ref());
+ match column.fixed() {
+ Some(width) => fixed += width,
+ None => variable.push(column),
+ }
+ }
+ let max_row = variable.iter().fold(fixed, |bytes, column| {
+ bytes.saturating_add(column.max_row())
+ });
+ Self {
+ fixed,
+ variable,
+ max_row,
+ }
+ }
+
+ /// Returns the size in bytes of row `row`.
+ pub(crate) fn row(&self, row: usize) -> usize {
+ self.variable.iter().fold(self.fixed, |bytes, column| {
+ bytes + column.rows(row, row + 1)
+ })
+ }
+
+ /// Returns the size in bytes of every row, if they all have the same size.
+ pub(crate) fn fixed(&self) -> Option<usize> {
+ self.variable.is_empty().then_some(self.fixed)
+ }
+
+ /// Returns an upper bound on the size in bytes of any row.
+ pub(crate) fn max_row(&self) -> usize {
+ self.max_row
+ }
+}
+
+/// Size model of one column, see [`SpilledRowSizes`].
+enum ColumnSize {
+ /// Every row takes the same number of bytes.
+ Fixed(usize),
+ /// An offset per row, plus the bytes between consecutive offsets.
+ Bytes(Offsets),
+ /// A view per row, plus the bytes of values too long to be inlined.
+ Views(ScalarBuffer<u128>),
+ /// An offset per row, plus the elements between consecutive offsets.
+ List(Offsets, Box<ColumnSize>),
+ /// A fixed number of elements per row.
+ FixedSizeList(usize, Box<ColumnSize>),
+ /// A row of each field.
+ Struct(Vec<ColumnSize>),
+}
+
+impl ColumnSize {
+ fn new(array: &dyn Array) -> Self {
+ match array.data_type() {
+ DataType::Utf8 => Self::Bytes(array.as_string::<i32>().offsets().into()),
+ DataType::LargeUtf8 => Self::Bytes(array.as_string::<i64>().offsets().into()),
+ DataType::Binary => Self::Bytes(array.as_binary::<i32>().offsets().into()),
+ DataType::LargeBinary => {
+ Self::Bytes(array.as_binary::<i64>().offsets().into())
+ }
+ DataType::Utf8View => Self::Views(array.as_string_view().views().clone()),
+ DataType::BinaryView => Self::Views(array.as_binary_view().views().clone()),
+ DataType::List(_) => {
+ let list = array.as_list::<i32>();
+ Self::List(list.offsets().into(), Box::new(Self::new(list.values())))
+ }
+ DataType::LargeList(_) => {
+ let list = array.as_list::<i64>();
+ Self::List(list.offsets().into(), Box::new(Self::new(list.values())))
+ }
+ DataType::Map(_, _) => {
+ let map = array.as_map();
+ Self::List(map.offsets().into(), Box::new(Self::new(map.entries())))
+ }
+ DataType::FixedSizeList(_, size) => Self::FixedSizeList(
+ *size as usize,
+ Box::new(Self::new(array.as_fixed_size_list().values())),
+ ),
+ DataType::Struct(_) => Self::Struct(
+ array
+ .as_struct()
+ .columns()
+ .iter()
+ .map(|field| Self::new(field.as_ref()))
+ .collect(),
+ ),
+ DataType::Dictionary(key, _) => {
+ Self::Fixed(key.primitive_width().unwrap_or_default())
+ }
Review Comment:
Dictionary-typed state still hits #25851: a dictionary row is sized by its key only, and `take` keeps every dictionary value, so a `List<Dictionary>` state ends up in one chunk, and even a one-row chunk carries the whole dictionary. Since Comet passes dictionaries, I think this belongs in this PR.
Fails on this PR with `Failed to allocate additional 160.4 MB` (`datafusion-cli -m 128M`):
```sql
SET datafusion.execution.target_partitions = 2;
SELECT count(*), sum(n) FROM (
SELECT k, cardinality(array_agg(v)) AS n FROM (
SELECT value % 16 AS k,
arrow_cast(concat(CAST(value AS VARCHAR), repeat('x', 100)), 'Dictionary(Int32, Utf8)') AS v
FROM generate_series(1, 3000000))
GROUP BY k);
```
I tried the change below locally. The query above returns 16 groups and 3,000,000 values, and so does a variant with 1000-byte values over 300,000 rows, which still fails with only the GC part. `dictionaries_count_their_keys` needs updating. The GC in `gc_array` applies to every spill; if you'd rather keep sort unchanged, it could run only on the chunks `IncrementalSortIterator` yields when `max_batch_bytes` is set (I haven't tested that variant).
```diff
--- a/datafusion/physical-plan/src/spill/spilled_row_sizes.rs
+++ b/datafusion/physical-plan/src/spill/spilled_row_sizes.rs
/// A row of each field.
Struct(Vec<ColumnSize>),
+ /// A key per row, plus the dictionary value it references.
+ Dictionary(usize, Vec<usize>, Box<ColumnSize>),
}
@@ fn new
DataType::Dictionary(key, _) => {
- Self::Fixed(key.primitive_width().unwrap_or_default())
+ let dictionary = array.as_any_dictionary();
+ Self::Dictionary(
+ key.primitive_width().unwrap_or_default(),
+ dictionary.normalized_keys(),
+ Box::new(Self::new(dictionary.values().as_ref())),
+ )
}
@@ fn fixed
- Self::Bytes(_) | Self::Views(_) | Self::List(_, _) => None,
+ Self::Bytes(_)
+ | Self::Views(_)
+ | Self::List(_, _)
+ | Self::Dictionary(_, _, _) => None,
@@ fn max_row
+ Self::Dictionary(width, _, values) => width.saturating_add(values.max_row()),
@@ fn rows
+ Self::Dictionary(width, keys, values) => keys[start..end]
+ .iter()
+ .map(|&key| width + values.rows(key, key + 1))
+ .sum(),
--- a/datafusion/physical-plan/src/spill/mod.rs
+++ b/datafusion/physical-plan/src/spill/mod.rs
@@ fn gc_array
+ DataType::Dictionary(_, _) => {
+ let dictionary = array.as_any_dictionary();
+ let gc =
+ arrow_select::dictionary::garbage_collect_any_dictionary(dictionary)?;
+ let shrunk =
+ gc.as_any_dictionary().values().len() < dictionary.values().len();
+ let (values, values_mutated) = gc_array(gc.as_any_dictionary().values())?;
+ if !shrunk && !values_mutated {
+ return Ok((Arc::clone(array), false));
+ }
+ Ok((gc.as_any_dictionary().with_values(values), true))
+ }
_ => gc_array_children(array),
```
(This also needs `arrow-select = { workspace = true }` in `datafusion/physical-plan/Cargo.toml`.) Could you add the query above to Case H as well?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#discussion_r4367160490
##########
datafusion/physical-plan/src/aggregates/spill.rs:
##########
@@ -280,6 +300,7 @@ impl AggregateSpill {
.with_expressions(&spill_expr)
.with_metrics(baseline_metrics.intermediate())
.with_batch_size(batch_size)
+ .with_max_batch_bytes(Some(SPILL_BATCH_TARGET_BYTES))
.with_reservation(merge_reservation)
.with_replay_headroom()
Review Comment:
Filed
- https://github.com/apache/datafusion/issues/26711
I think once we clean up the old aggregate code some of the older configs will go away
- https://github.com/apache/datafusion/issues/25902
##########
datafusion/physical-plan/src/sorts/merge.rs:
##########
@@ -146,11 +156,22 @@ impl<C: CursorValues> SortPreservingMergeStream<C> {
poll_reset_epochs: vec![0; stream_count],
loser_tree: vec![],
batch_size,
+ max_batch_bytes: None,
+ avg_row_bytes: vec![0; stream_count],
+ in_progress_bytes: 0,
fetch,
produced: 0,
}
}
+ /// Also emits a batch once its rows take about `max_batch_bytes`, as
Review Comment:
why is this an optional setting on just the merge stream? It seems like maybe we would want a `max_batch_bytes` as a global config setting, similar to `batchs_size` (though it would take quite a bit of effort to thread it through)
##########
datafusion/physical-plan/src/spill/spilled_row_sizes.rs:
##########
@@ -0,0 +1,398 @@
+// 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.
+
+//! Per-row sizes of a batch once it is written to a spill file.
+
+use arrow::array::{Array, AsArray, RecordBatch};
+use arrow::buffer::{OffsetBuffer, ScalarBuffer};
+use arrow::datatypes::DataType;
+use arrow_data::MAX_INLINE_VIEW_LEN;
+
+use super::VIEW_SIZE_BYTES;
+
+/// Estimates how many bytes each row of a batch takes once written to a spill
+/// file, where view arrays are compacted (see [`super::gc_view_arrays`]).
+///
+/// A row counts its fixed-width values, the offsets and bytes of its
+/// variable-width values, and the elements of its nested values. Buffers that
+/// every row shares, such as validity bitmaps and dictionary values, are not
+/// counted. Each view counts the bytes of its value, so views that share bytes
+/// are counted as if compaction copied them, although it can write shared bytes
+/// only once.
+pub(crate) struct SpilledRowSizes {
Review Comment:
This seems oddly specific for aggregation, but I think we can consolidate as part of a follow on
##########
datafusion/sqllogictest/test_files/aggregate_memory_spill.slt:
##########
@@ -239,6 +239,146 @@ FROM (
08)--------------AggregateExec: mode=Partial, gby=[v@0 * 7 % 100000 as t.v * Int64(7) % Int64(100000)], aggr=[sum(t.v)], metrics=[<slt:ignore>early_emit_count=<slt:ignore>]
<slt:ignore>
+# --- Case H: a few groups with large state ---
+# The array_agg states of all 32 groups do not fit in the memory limit, but a
+# single group does. Spilling must write the groups in batches that can be
+# read back within the limit, instead of one batch holding every group.
Review Comment:
When i ran this test locally
```shell
cargo test --test sqllogictests -- aggregate_memory_spill
```
It produced several warnings about peak memory drift
```
warning: `datafusion-sqllogictest` (test "sqllogictests") generated 1 warning
Finished `test` profile [unoptimized + debuginfo] target(s) in 1m 10s
Running bin/sqllogictests.rs (target/debug/deps/sqllogictests-616efeba2509d40d)
Running with 16 test threads (available parallelism: 16)
Completed 1 test files in 8 seconds Peak memory drift: drift=102.7 MB allocated=102.7 MB reserved=10.9 KB pool=aggregate_memory_spill.slt consumer=PartialHashAggregateStream[0]
Test files ran concurrently, so the file and consumer above can be wrong. Run with --test-threads 1 to attribute drift to one file.
```
Is that expected?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "2010YOUY01 (via GitHub) via github" <gi...@datafusion.apache.org>.
2010YOUY01 commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-6700552632
> @alamb @2010YOUY01 do you have any additional feedback, or can we merge this one (and maybe consider for 56.0.0)?
I don't fully understand the underlying logic, so I can't offer concrete feedback. Just sharing an additional perspective here. I have no objections and trust the reviewers' approval judgment.
**Suggestion:** If anyone has a good understanding of the overall sort spilling design, it would be helpful to add a top-level comment summarizing how the existing mechanisms fit together.
**Concern:** Given the PR history in fixing sort spilling, I have the impression that more fixes may actually introduce more bugs rather than make spilling more stable. The process feels a bit like gradient descent programmingβfixing corner cases one by one, the review also focus on letting AI to identify more corner cases, rather than refining a clear high-level design. This makes it hard to tell whether we're converging toward a stable implementation.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "jayzhan211 (via GitHub) via github" <gi...@datafusion.apache.org>.
jayzhan211 commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-6702291445
I put together [this page](https://claude.ai/artifact/FREAr6m99WzAH8HPW7t5yw) to help me understand the overall sort spilling design. It has diagrams, code links and measurements, in case it helps others here.
When memory runs out, `SortExec` and hash aggregates write sorted runs to disk, and `MultiLevelMergeBuilder` merges them in as many passes as memory allows. Before each pass it reserves 2 to 4 times the largest batch of each run, then bypasses the memory pool. Batches are cut by row count, so nothing limits that largest batch. Three workarounds exist only because of that gap: re-spilling a run with its batches halved (#22945), per-run row limits (#23286), and a scan and retry for runs that can't shrink (#25383).
This PR adds a byte limit for aggregates, at the spill writer and the merge output. `SortExec` and those three workarounds don't change.
Remaining work, as untested ideas that could go under #26711:
- Apply one byte limit to every caller, including `SortExec`, and derive it from the memory budget instead of a constant. Drafts #25910 and #26987 go in this direction.
- Once that's in place, delete the three workarounds.
- Track the merge's real memory instead of bypassing the pool after admission.
- Let replay release a finished group before it takes the next one, so it peaks at one large group instead of two.
- Test that a query passing at one memory limit also passes at any larger limit. The page has a counterexample.
Please correct anything I got wrong.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-5981949254
@EmilyMatt I agree the aggregate's estimates are worth improving, since they decide when it spills and whether it keeps room to sort and write the spill. But they don't fix this issue, which happens after the spill, when the spilled state is read back.
A spill file is read back one batch at a time, and spilled batches are bounded only by `batch_size` rows. With fewer groups than `batch_size`, all the groups of a spill land in one batch, however early the aggregate spills, and the replay's ordered aggregate, which cannot spill, has to hold that whole batch. In the reproducer in #25851, each final partition receives one input batch that already holds all of its state, so the final aggregate spills on that first batch and writes its 8 groups as one 184 MB batch. Reading it back needs 225 MB in a 128 MB pool, although any one group fits. No reservation changes the size of that batch. With several spill files, the merge has the same problem: it combines rows until it has `batch_size` of them, so it rebuilds batches of large groups even from small spilled batches.
On the other points:
- Performance: the TPC-H and TPC-DS runs above show no slowdowns, `external_aggr` is not slower, and merges without a byte limit only check that none is set.
- Maintenance: the byte limit is crate-private and opt-in, and only aggregate spilling sets it.
- Accuracy: spilling cuts batches by the size of each row (`SpilledRowSizes`), not by averages. Only the merge estimates rows by the average row size of their batch. It never takes more rows from a batch than the batch holds, so it can underestimate an output batch by at most the input batches it holds, which are about 1 MiB each unless they hold a single larger row.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-5901451278
run benchmark tpch tpcds
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-6384441468
π€ Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/25853#issuecomment-6384102134)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB)
Comparing fix-issue-258521 (3289c00a6bfdf14b9ad444d7ff47b2b9364b1be2) to 1da3cf0 (merge-base) [diff](https://github.com/apache/datafusion/compare/1da3cf049fa1a7a708acc2c29eba4f2e9dece8db..3289c00a6bfdf14b9ad444d7ff47b2b9364b1be2)
<details><summary>Run configuration</summary>
```yaml
run benchmark spill_views
```
</details>
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
<details><summary>Details</summary>
<p>
```
group HEAD fix-issue-258521
----- ---- ----------------
spill_views/q01_sort_string_1_distinct_repeated 1.01 68.0Β±0.62ms ? ?/sec 1.00 67.4Β±0.30ms ? ?/sec
spill_views/q02_sort_string_1000_distinct_repeated 1.00 66.9Β±0.34ms ? ?/sec 1.01 67.3Β±0.43ms ? ?/sec
spill_views/q03_sort_binary_1000_distinct_repeated 1.00 67.4Β±0.35ms ? ?/sec 1.00 67.6Β±0.39ms ? ?/sec
spill_views/q04_sort_string_all_distinct_distinct 1.00 121.8Β±0.77ms ? ?/sec 1.00 122.0Β±0.73ms ? ?/sec
spill_views/q05_group_by_string_all_distinct_distinct 1.02 134.0Β±4.79ms ? ?/sec 1.00 130.9Β±5.90ms ? ?/sec
```
</p>
</details>
<details><summary>Memory Pool Peaks</summary>
Peak `MemoryPool` reservation per query β what DataFusion's accounting believes it reserved. Recorded only for benchmarks that write a results JSON (the `dfbench` suites, and the suites `bench.sh` runs through the Criterion SQL harness since apache/datafusion#25644), and only under a memory limit: `DATAFUSION_RUNTIME_MEMORY_LIMIT`, or one the suite sets itself.
Base: `1da3cf0 (merge-base)` | Changed: `fix-issue-258521`
**`spill_views`**
| Query | Base | Changed | Change |
| --- | --- | --- | --- |
| spill_views/q01_sort_string_1_distinct_repeated | 40.6 MiB | 40.6 MiB | +0.0% |
| spill_views/q02_sort_string_1000_distinct_repeated | 39.8 MiB | 39.8 MiB | +0.0% |
| spill_views/q03_sort_binary_1000_distinct_repeated | 39.8 MiB | 39.8 MiB | +0.0% |
| spill_views/q04_sort_string_all_distinct_distinct | 102.6 MiB | 102.6 MiB | +0.0% |
| spill_views/q05_group_by_string_all_distinct_distinct | 96.0 MiB | 96.0 MiB | +0.0% |
**Pool accounting vs. process RSS**
Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
| Benchmark | Side | Max pool peak | Peak RSS | Gap | RSS / pool |
| --- | --- | --- | --- | --- | --- |
| spill_views | base (`1da3cf0 (merge-base)`) | 102.6 MiB | 772.3 MiB | 669.8 MiB | 7.5Γ |
| spill_views | changed (`fix-issue-258521`) | 102.6 MiB | 783.2 MiB | 680.6 MiB | 7.6Γ |
</details>
<details><summary>Resource Usage</summary>
**spill_views β base (merge-base)**
| Metric | Value |
|--------|-------|
| Wall time | 745.2s |
| Peak memory | 772.3 MiB |
| Avg memory | 33.0 MiB |
| CPU user | 76.8s |
| CPU sys | 25.0s |
| Peak spill | 84.2 MiB |
**spill_views β branch**
| Metric | Value |
|--------|-------|
| Wall time | 750.3s |
| Peak memory | 783.2 MiB |
| Avg memory | 34.0 MiB |
| CPU user | 80.0s |
| CPU sys | 25.3s |
| Peak spill | 84.2 MiB |
</details>
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-6384134072
π€ Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/25853#issuecomment-6384102134)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c6384102134-3826-4wv2t 6.12.105+ #1 SMP Sat Sep 5 07:54:11 UTC 2026 aarch64 GNU/Linux`
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
Comparing fix-issue-258521 (3289c00a6bfdf14b9ad444d7ff47b2b9364b1be2) to 1da3cf0 (merge-base) [diff](https://github.com/apache/datafusion/compare/1da3cf049fa1a7a708acc2c29eba4f2e9dece8db..3289c00a6bfdf14b9ad444d7ff47b2b9364b1be2)
<details><summary>Run configuration</summary>
```yaml
run benchmark spill_views
```
</details>
Results will be posted here when complete
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-6384101936
run benchmark external_aggr
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "andygrove (via GitHub) via github" <gi...@datafusion.apache.org>.
andygrove merged PR #25853:
URL: https://github.com/apache/datafusion/pull/25853
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size aggregate spill batches by bytes as well as rows [datafusion]
Posted by "andygrove (via GitHub) via github" <gi...@datafusion.apache.org>.
andygrove commented on PR #25853:
URL: https://github.com/apache/datafusion/pull/25853#issuecomment-6723587376
Pushed 1132fb770: the merge of main from today failed clippy, because main now disallows `DictionaryArray::with_values` in `clippy.toml`. The call in `gc_array` runs on a dictionary that was just garbage collected, so it holds no more values than rows. It now has an `#[expect(clippy::disallowed_methods)]` saying so, as `map_dictionary_values` does. No other change.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org