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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
