You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "TinyMurky (via GitHub)" <gi...@apache.org> on 2026/09/19 09:56:48 UTC
[PR] perf(functions-aggregate): coalesce small ordered ARRAY_AGG batches [datafusion]
TinyMurky opened a new pull request, #25497:
URL: https://github.com/apache/datafusion/pull/25497
## Which issue does this PR close?
- Closes #25199
## Rationale for this change
#24392 changed OrderSensitiveArrayAggAccumulator to retain payloads as Arrow arrays instead of converting each value to a ScalarValue. This significantly reduced memory usage for reasonably sized input batches.
However, the accumulator currently stores one ArrayRef per update_batch() call. Each array has a fixed cost from its ArrayData, buffer allocation, and Arc, regardless of how many rows it contains.
Ordered `ARRAY_AGG` does not support a native `GroupsAccumulator` because `groups_accumulator_supported` requires `order_bys` to be empty. Grouped execution therefore falls back to `GroupsAccumulatorAdapter`, which calls `update_batch()` once per group per input batch.
With a high-cardinality `GROUP BY`, these calls often contain only one or two rows. Without this change, each call retains a separate `ArrayRef`, so the accumulator pays the fixed per-array allocation cost for many small Arrow arrays, increasing the per-row memory footprint.
## What changes are included in this PR?
- Coalesce consecutive small ordered ARRAY_AGG payload batches up to 64 rows.
- Add tests covering:
- coalescing exactly up to the threshold;
- creating a new batch after the threshold is reached;
- entry indices across coalesced and newly created batches;
- ordering across multiple coalesced batches;
- coalescing payloads received through partial-state merge_batch().
- Add ordered ARRAY_AGG benchmarks covering:
- 1, 8, 64, and 2,048 rows per update_batch();
- random input;
- preordered input.
### Retained memory
Retained memory was measured using `Accumulator::size()` after inserting 2,048 Int64 payloads with an Int64 ordering key and before calling evaluate(). The same measurement code was used for the baseline after [#24392](https://github.com/apache/datafusion/pull/24392) and for this change.
| Rows/update | Batches | Total retained (with this change) | Bytes/row (with this change) | Total retained (baseline [#24392](https://github.com/apache/datafusion/pull/24392)| Bytes/row (baseline [#24392](https://github.com/apache/datafusion/pull/24392)) |
| ----------: | ------: | --------------------------------: | ---------------------------: | ----------------------------------------------------------------------------------- | ------------------------------------------------------------------------------ |
| 1 | 32 | 104,701 B | 51.12 B/row | 445,181 B | 217.37 B/row |
| 8 | 32 | 90,365 B | 44.12 B/row | 115,453 B | 56.37 B/row |
| 64 | 32 | 88,573 B | 43.25 B/row | 88,573 B | 43.25 B/row |
| 2048 | 1 | 84,901 B | 41.46 B/row | 84,901 B | 41.46 B/row |
The worst-case one-row update footprint decreases from 217.37 B/row to 51.12 B/row. Inputs already at or above the 64-row coalescing threshold retain the existing memory footprint.
## Are these changes tested?
Added unit tests covering:
- coalescing small batches exactly up to the threshold;
- creating a new batch when the threshold would be exceeded;
- preserving the correct batch_idx and row_idx for entries;
- sorting values across multiple coalesced batches;
- coalescing small partial-state payloads passed through merge_batch().
Added Criterion benchmarks for ordered ARRAY_AGG using random and preordered input with 1, 8, 64, and 2,048 rows per update_batch().
The following commands have been executed and passed:
- `cargo test -p datafusion-functions-aggregate --lib array_agg::tests`
- `cargo bench -p datafusion-functions-aggregate --bench array_agg --no-run`
- `cargo bench -p datafusion-functions-aggregate --bench array_agg --ordered_array_agg`
- `cargo test --profile=ci --test sqllogictests`
- `cargo test -p datafusion`
- `cargo test -p datafusion-cli`
## Are there any user-facing changes?
No
--
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] perf(functions-aggregate): coalesce small ordered ARRAY_AGG batches [datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #25497:
URL: https://github.com/apache/datafusion/pull/25497#discussion_r4109876396
##########
datafusion/functions-aggregate/src/array_agg.rs:
##########
Review Comment:
Each coalesced payload is copied twice: `copy_array_data` + `compact_payload` detach it, then `concat` copies tail + payload again. On the PR's own bench (interleaved runs, base 696eaf58d vs this branch) that is a regression on exactly the inputs this PR targets:
| case | main | PR |
|---|---|---|
| i64 ordered, 1 row/update | 645 µs | 903 µs (+40%) |
| i64 random, 1 row/update | 710 µs | 970 µs (+35%) |
| i64 ordered, 8 rows/update | 90 µs | 122 µs (+35%) |
For primitive, boolean and offset byte types, `concat` already produces fresh buffers, so the detach copy can be skipped when coalescing. A prototype of that gives 391 µs / 465 µs on the 1-row cases and 64 µs / 136 µs on 8 rows, faster than main, and all `array_agg` tests pass. View and dictionary types still need the detach first: their raw input buffers over-report `get_buffer_memory_size`, so they would never pass the byte cap.
```rs
let row_count = values.len();
let concat_copies = values.data_type().is_primitive()
|| matches!(
values.data_type(),
DataType::Boolean
| DataType::Utf8
| DataType::LargeUtf8
| DataType::Binary
| DataType::LargeBinary
);
let can_coalesce = concat_copies
&& self.batches.last().is_some_and(|last| {
last.len() + row_count <= ORDERED_ARRAY_AGG_COALESCE_ROWS
&& last.get_buffer_memory_size() + values.get_buffer_memory_size()
<= ORDERED_ARRAY_AGG_COALESCE_BYTES
});
// `concat` copies fixed-layout payloads, so detaching first is only
// needed when the payload is stored as-is or may share buffers.
let values = if can_coalesce {
values
} else {
compact_payload(make_array(copy_array_data(&values.to_data())))?
};
```
Then branch on `can_coalesce` instead of re-checking the tail. Can you add the before/after numbers to the PR description?
##########
datafusion/functions-aggregate/src/array_agg.rs:
##########
@@ -1531,11 +1539,37 @@ impl OrderSensitiveArrayAggAccumulator {
}
let start = self.entries.len();
- let batch_idx = self.batches.len();
- self.batches.push(values);
- self.entries.extend(
- (0..row_count).map(|row_idx| OrderedArrayAggEntry { batch_idx, row_idx }),
- );
+ let (batch_idx, row_offset) = match self.batches.last() {
+ Some(last_batch)
+ if last_batch.len() + row_count <= ORDERED_ARRAY_AGG_COALESCE_ROWS
+ && last_batch.get_buffer_memory_size()
+ + values.get_buffer_memory_size()
+ <= ORDERED_ARRAY_AGG_COALESCE_BYTES =>
+ {
+ let merged =
Review Comment:
For Utf8View/BinaryView, `concat` → `GenericByteViewBuilder::append_array` clones every input's data buffers instead of copying the bytes. After 64 one-row updates the coalesced tail is 1 array with **64 data buffers**, so each row still pays for its own `Buffer` + `Arc<Bytes>` allocation. That is the fixed per-row overhead this PR is meant to remove, and `size()` can't see it. Running `compact_payload` on the merged array gets it down to 1 buffer. The extra copy is bounded by the 4 KiB cap.
```diff
- let merged =
- arrow::compute::concat(&[last_batch.as_ref(), values.as_ref()])?;
+ // View concat shares the inputs' data buffers; GC folds them
+ // into one so the tail does not keep a buffer per update.
+ let merged = compact_payload(arrow::compute::concat(&[
+ last_batch.as_ref(),
+ values.as_ref(),
+ ])?)?;
```
The Utf8View test could also assert this:
```rs
assert_eq!(acc.batches[0].as_string_view().data_buffers().len(), 1);
```
Fine to handle in a follow-up if you'd rather keep this PR focused.
I checked this with a local probe (64 single-row Utf8View updates): 64 data buffers on the PR branch, 1 with compact_payload after the concat. I haven't run the suggested extra assertion as part of the PR's own Utf8View 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] perf(functions-aggregate): coalesce small ordered ARRAY_AGG batches [datafusion]
Posted by "TinyMurky (via GitHub)" <gi...@apache.org>.
TinyMurky commented on PR #25497:
URL: https://github.com/apache/datafusion/pull/25497#issuecomment-5830309356
Hello @jayzhan211 !
Thank you for spending time reviewing my pr!
I have added the following tests to the original pr.
First I added `ordered_array_agg_does_not_coalesce_large_payloads` as you mention above.
https://github.com/apache/datafusion/blob/d6c4d8b5e4ddd33313ce9eb7b4c7104a8bb5fd29/datafusion/functions-aggregate/src/array_agg.rs#L3035-L3059
Then I separated `ordered_array_agg_coalesces_dictionary_and_view_payloads` into two different tests:
- `ordered_array_agg_sorts_across_coalesced_utf8_view_batches`
https://github.com/apache/datafusion/blob/d6c4d8b5e4ddd33313ce9eb7b4c7104a8bb5fd29/datafusion/functions-aggregate/src/array_agg.rs#L3061-L3108
- `ordered_array_agg_sorts_across_coalesced_dictionary_batches`
https://github.com/apache/datafusion/blob/d6c4d8b5e4ddd33313ce9eb7b4c7104a8bb5fd29/datafusion/functions-aggregate/src/array_agg.rs#L3110-L3163
Could you review these tests? Thank you!
--
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] perf(functions-aggregate): coalesce small ordered ARRAY_AGG batches [datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 merged PR #25497:
URL: https://github.com/apache/datafusion/pull/25497
--
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] perf(functions-aggregate): coalesce small ordered ARRAY_AGG batches [datafusion]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #25497:
URL: https://github.com/apache/datafusion/pull/25497#issuecomment-5741018098
## [Codecov](https://app.codecov.io/gh/apache/datafusion/pull/25497?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 `90.54054%` with `7 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 82.38%. Comparing base ([`d477a2b`](https://app.codecov.io/gh/apache/datafusion/commit/d477a2b038730647bef258cb98e58978b0c74699?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`7cdd7b5`](https://app.codecov.io/gh/apache/datafusion/commit/7cdd7b5a9b682054d24da48dc7d5fb0d1a9468c7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on main.
| [Files with missing lines](https://app.codecov.io/gh/apache/datafusion/pull/25497?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/functions-aggregate/src/array\_agg.rs](https://app.codecov.io/gh/apache/datafusion/pull/25497?src=pr&el=tree&filepath=datafusion%2Ffunctions-aggregate%2Fsrc%2Farray_agg.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9mdW5jdGlvbnMtYWdncmVnYXRlL3NyYy9hcnJheV9hZ2cucnM=) | 90.54% | [3 Missing and 4 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/25497?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 #25497 +/- ##
========================================
Coverage 82.37% 82.38%
========================================
Files 1138 1138
Lines 433505 433803 +298
Branches 433505 433803 +298
========================================
+ Hits 357102 357385 +283
- Misses 54850 54856 +6
- Partials 21553 21562 +9
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/datafusion/pull/25497?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] perf(functions-aggregate): coalesce small ordered ARRAY_AGG batches [datafusion]
Posted by "TinyMurky (via GitHub)" <gi...@apache.org>.
TinyMurky commented on code in PR #25497:
URL: https://github.com/apache/datafusion/pull/25497#discussion_r4111977161
##########
datafusion/functions-aggregate/src/array_agg.rs:
##########
Review Comment:
I had run benchmark on following commits and updated the PR description :
- 6a792c67 : `main`
- 8588db18 : this pr before updating detach copy skip
- 19564841 : this pr after updating change
<!-- obsidian -->
Benchmark | main 6a792c67 (µs) | PR 8588db18 (µs) | PR 19564841 (µs)
-- | -- | -- | --
i64 ordered，1 row/update | 1,527.20 | 3,494.10 | 779.37
i64 random，1 row/update | 2,207.60 | 2,051.30 | 1,052.50
i64 ordered，8 rows/update | 221.42 | 258.73 | 133.85
i64 random，8 rows/update | 454.68 | 494.41 | 363.10
i64 ordered，64 rows/update | 50.37 | 50.37 | 50.44
i64 random，64 rows/update | 258.03 | 258.29 | 262.51
i64 ordered，2,048 rows/update | 18.68 | 19.09 | 19.00
i64 random，2,048 rows/update | 228.52 | 234.06 | 236.29
Utf8 4 KiB，1 row/update（update_batches only） | 2,417.90 | 2,442.10 | 2,642.90
the code that I edited in the new commit was as below:
https://github.com/apache/datafusion/blob/19564841c0df254acc31ccced5142733c3157059/datafusion/functions-aggregate/src/array_agg.rs#L1526-L1579
--
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] perf(functions-aggregate): coalesce small ordered ARRAY_AGG batches [datafusion]
Posted by "TinyMurky (via GitHub)" <gi...@apache.org>.
TinyMurky commented on PR #25497:
URL: https://github.com/apache/datafusion/pull/25497#issuecomment-5799343041
Hello @jayzhan211 !
Sorry for the delay, I've finished the change, and add the new benchmark to test the 4 KiB byte cap of `OrderedArrayAgg`. Could you review tthe change?
The 4 KiB byte cap reduced the wide Utf8 update-only benchmark from 34.45 ms with the row-only cap to 2.09 ms, reducing runtime by approximately 93.9%.
Benchmarks below show the differents between old commit (092743dce) and new commit (0ddf5e8e5)
```
ordered_array_agg i64 ordered input, 1 rows per update_batch
time: [1.5908 ms 1.6032 ms 1.6175 ms]
change: [−38.237% −36.692% −35.285%] (p = 0.00 < 0.05)
Performance has improved.
Found 11 outliers among 100 measurements (11.00%)
3 (3.00%) high mild
8 (8.00%) high severe
Benchmarking ordered_array_agg i64 random input, 1 rows per update_batch: Warming up for 3.0000 s
Warning: Unable to complete 100 samples in 5.0s. You may wish to increase target time to 9.4s, enable flat sampling, or reduce sample count to 50.
Benchmarking ordered_array_agg i64 random input, 1 rows per update_batch: Collecting 100 samples in estimated 9.ordered_array_agg i64 random input, 1 rows per update_batch
time: [1.8412 ms 1.8520 ms 1.8631 ms]
change: [−47.695% −43.259% −38.684%] (p = 0.00 < 0.05)
Performance has improved.
Found 7 outliers among 100 measurements (7.00%)
4 (4.00%) high mild
3 (3.00%) high severe
Benchmarking ordered_array_agg i64 ordered input, 8 rows per update_batch: Collecting 100 samples in estimated 5ordered_array_agg i64 ordered input, 8 rows per update_batch
time: [231.63 µs 233.90 µs 236.66 µs]
change: [−50.086% −46.184% −42.206%] (p = 0.00 < 0.05)
Performance has improved.
Found 16 outliers among 100 measurements (16.00%)
6 (6.00%) high mild
10 (10.00%) high severe
Benchmarking ordered_array_agg i64 random input, 8 rows per update_batch: Collecting 100 samples in estimated 6.ordered_array_agg i64 random input, 8 rows per update_batch
time: [430.89 µs 432.56 µs 434.36 µs]
change: [−52.435% −48.834% −45.128%] (p = 0.00 < 0.05)
Performance has improved.
Found 9 outliers among 100 measurements (9.00%)
6 (6.00%) high mild
3 (3.00%) high severe
Benchmarking ordered_array_agg i64 ordered input, 64 rows per update_batch: Collecting 100 samples in estimated ordered_array_agg i64 ordered input, 64 rows per update_batch
time: [43.170 µs 43.354 µs 43.552 µs]
change: [−55.102% −50.657% −45.909%] (p = 0.00 < 0.05)
Performance has improved.
Found 15 outliers among 100 measurements (15.00%)
1 (1.00%) low mild
7 (7.00%) high mild
7 (7.00%) high severe
Benchmarking ordered_array_agg i64 random input, 64 rows per update_batch: Collecting 100 samples in estimated 5ordered_array_agg i64 random input, 64 rows per update_batch
time: [233.26 µs 234.71 µs 236.45 µs]
change: [−46.101% −42.135% −38.055%] (p = 0.00 < 0.05)
Performance has improved.
Found 13 outliers among 100 measurements (13.00%)
7 (7.00%) high mild
6 (6.00%) high severe
Benchmarking ordered_array_agg i64 ordered input, 2048 rows per update_batch: Collecting 100 samples in estimateordered_array_agg i64 ordered input, 2048 rows per update_batch
time: [16.337 µs 16.380 µs 16.435 µs]
change: [−56.556% −52.611% −48.422%] (p = 0.00 < 0.05)
Performance has improved.
Found 12 outliers among 100 measurements (12.00%)
2 (2.00%) high mild
10 (10.00%) high severe
Benchmarking ordered_array_agg i64 random input, 2048 rows per update_batch: Collecting 100 samples in estimatedordered_array_agg i64 random input, 2048 rows per update_batch
time: [205.26 µs 207.21 µs 210.25 µs]
change: [−39.842% −38.164% −36.577%] (p = 0.00 < 0.05)
Performance has improved.
Found 9 outliers among 100 measurements (9.00%)
5 (5.00%) high mild
4 (4.00%) high severe
Benchmarking ordered_array_agg utf8 4 KiB, 1 row per update_batch: Collecting 100 samples in estimated 5.0868 s ordered_array_agg utf8 4 KiB, 1 row per update_batch
time: [2.0722 ms 2.0891 ms 2.1080 ms]
change: [−89.963% −88.934% −87.805%] (p = 0.00 < 0.05)
Performance has improved.
Found 15 outliers among 100 measurements (15.00%)
2 (2.00%) high mild
13 (13.00%) high severe
```
--
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] perf(functions-aggregate): coalesce small ordered ARRAY_AGG batches [datafusion]
Posted by "TinyMurky (via GitHub)" <gi...@apache.org>.
TinyMurky commented on PR #25497:
URL: https://github.com/apache/datafusion/pull/25497#issuecomment-5741245345
I used this function to test the retained memories
```rust
/// To calculate how maay bytes per row do `OrderSensitiveArrayAggAccumulator` need in different rows per update_batch
#[test]
#[ignore = "manual retained-memory measurement"]
fn report_ordered_array_agg_retained_memory() -> Result<()> {
use arrow::array::Int64Array;
const TOTAL_ROWS: usize = 2048;
for rows_per_update in [1, 8, 64, TOTAL_ROWS] {
let mut accumulator = ordered_accumulator(
DataType::Int64,
DataType::Int64,
SortOptions::new(false, false),
false, // input is not declared preordered
false, // not reversed
)?;
for offset in (0..TOTAL_ROWS).step_by(rows_per_update) {
let len = rows_per_update.min(TOTAL_ROWS - offset);
let values = (offset..offset + len)
.map(|value| value as i64)
.collect::<Vec<_>>();
let payload = Arc::new(Int64Array::from(values)) as ArrayRef;
accumulator.update_batch(&[Arc::clone(&payload), payload])?;
}
let total_bytes = accumulator.size();
let bytes_per_row = total_bytes as f64 / TOTAL_ROWS as f64;
eprintln!(
"rows/update={rows_per_update:>4}, \
batches={:>4}, \
total={total_bytes:>8} B, \
bytes/row={bytes_per_row:.2}",
accumulator.batches.len(),
);
}
Ok(())
}
```
--
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] perf(functions-aggregate): coalesce small ordered ARRAY_AGG batches [datafusion]
Posted by "TinyMurky (via GitHub)" <gi...@apache.org>.
TinyMurky commented on code in PR #25497:
URL: https://github.com/apache/datafusion/pull/25497#discussion_r4085505002
##########
datafusion/functions-aggregate/src/array_agg.rs:
##########
@@ -1535,11 +1539,34 @@ impl OrderSensitiveArrayAggAccumulator {
}
let start = self.entries.len();
- let batch_idx = self.batches.len();
- self.batches.push(values);
- self.entries.extend(
- (0..row_count).map(|row_idx| OrderedArrayAggEntry { batch_idx, row_idx }),
- );
+ let (batch_idx, row_offset) = match self.batches.last() {
+ Some(last_batch)
+ if last_batch.len() + row_count <= ORDERED_ARRAY_AGG_COALESCE_ROWS =>
Review Comment:
I have add the following codes
https://github.com/apache/datafusion/blob/0ddf5e8e579b9e3566cc19e34f1d41d436a58fc9/datafusion/functions-aggregate/src/array_agg.rs#L1541-L1548
--
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] perf(functions-aggregate): coalesce small ordered ARRAY_AGG batches [datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #25497:
URL: https://github.com/apache/datafusion/pull/25497#discussion_r4053379692
##########
datafusion/functions-aggregate/src/array_agg.rs:
##########
@@ -1535,11 +1539,34 @@ impl OrderSensitiveArrayAggAccumulator {
}
let start = self.entries.len();
- let batch_idx = self.batches.len();
- self.batches.push(values);
- self.entries.extend(
- (0..row_count).map(|row_idx| OrderedArrayAggEntry { batch_idx, row_idx }),
- );
+ let (batch_idx, row_offset) = match self.batches.last() {
+ Some(last_batch)
+ if last_batch.len() + row_count <= ORDERED_ARRAY_AGG_COALESCE_ROWS =>
Review Comment:
Row-only threshold → every 1-row update re-copies the whole tail batch (≤63 rows) regardless of width. Measured, 2048 × 1-row `update_batch`, Utf8 payload, update only, main → PR: 16 KB strings 1.25 → 17.0 ms (13.7×) for 1.2% less `size()`; 256 KB strings 12.3 → 288 ms (23×) for 0.07%. Per-array overhead saved is only ~170 B, so coalescing should stop once the tail is no longer tiny.
With the cap below: 16 KB 1.69 ms, 256 KB 18.9 ms, Int64 `size()` unchanged (104,661 vs 445,141 on main).
```diff
+/// Skip coalescing once the tail batch is large enough that the fixed
+/// per-array overhead is negligible relative to the payload.
+const ORDERED_ARRAY_AGG_COALESCE_BYTES: usize = 4096;
+
let (batch_idx, row_offset) = match self.batches.last() {
Some(last_batch)
- if last_batch.len() + row_count <= ORDERED_ARRAY_AGG_COALESCE_ROWS =>
+ if last_batch.len() + row_count <= ORDERED_ARRAY_AGG_COALESCE_ROWS
+ && last_batch.get_array_memory_size()
+ + values.get_array_memory_size()
+ <= ORDERED_ARRAY_AGG_COALESCE_BYTES =>
{
```
Please add a wide-payload case (e.g. 4 KB Utf8, 1 row per `update_batch`) to the new bench. Separately, the PR's own bench shows +29–42% at 1 row/update and +18–35% at 8 rows/update for Int64 vs main (`concat` fixed cost, not memcpy) — worth stating in the description as the CPU-for-memory trade
--
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] perf(functions-aggregate): coalesce small ordered ARRAY_AGG batches [datafusion]
Posted by "TinyMurky (via GitHub)" <gi...@apache.org>.
TinyMurky commented on code in PR #25497:
URL: https://github.com/apache/datafusion/pull/25497#discussion_r4111992703
##########
datafusion/functions-aggregate/src/array_agg.rs:
##########
@@ -1531,11 +1539,37 @@ impl OrderSensitiveArrayAggAccumulator {
}
let start = self.entries.len();
- let batch_idx = self.batches.len();
- self.batches.push(values);
- self.entries.extend(
- (0..row_count).map(|row_idx| OrderedArrayAggEntry { batch_idx, row_idx }),
- );
+ let (batch_idx, row_offset) = match self.batches.last() {
+ Some(last_batch)
+ if last_batch.len() + row_count <= ORDERED_ARRAY_AGG_COALESCE_ROWS
+ && last_batch.get_buffer_memory_size()
+ + values.get_buffer_memory_size()
+ <= ORDERED_ARRAY_AGG_COALESCE_BYTES =>
+ {
+ let merged =
Review Comment:
I have edited the code as below for seperating `Utf8View`/`BinaryView` for one more `compact_payload`
https://github.com/apache/datafusion/blob/c92bbc919cf78ab3d48b72f1040d1a681ba7d4c8/datafusion/functions-aggregate/src/array_agg.rs#L1584-L1596
And I added a new `BinaryView` test to test if it can be coalesced same as `Utf8View`
https://github.com/apache/datafusion/blob/19564841c0df254acc31ccced5142733c3157059/datafusion/functions-aggregate/src/array_agg.rs#L3158-L3201
--
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] perf(functions-aggregate): coalesce small ordered ARRAY_AGG batches [datafusion]
Posted by "TinyMurky (via GitHub)" <gi...@apache.org>.
TinyMurky commented on code in PR #25497:
URL: https://github.com/apache/datafusion/pull/25497#discussion_r4111992703
##########
datafusion/functions-aggregate/src/array_agg.rs:
##########
@@ -1531,11 +1539,37 @@ impl OrderSensitiveArrayAggAccumulator {
}
let start = self.entries.len();
- let batch_idx = self.batches.len();
- self.batches.push(values);
- self.entries.extend(
- (0..row_count).map(|row_idx| OrderedArrayAggEntry { batch_idx, row_idx }),
- );
+ let (batch_idx, row_offset) = match self.batches.last() {
+ Some(last_batch)
+ if last_batch.len() + row_count <= ORDERED_ARRAY_AGG_COALESCE_ROWS
+ && last_batch.get_buffer_memory_size()
+ + values.get_buffer_memory_size()
+ <= ORDERED_ARRAY_AGG_COALESCE_BYTES =>
+ {
+ let merged =
Review Comment:
I have edited the code as below for seperating `Utf8View`/`BinaryView` for one more `compact_payload`
https://github.com/apache/datafusion/blob/19564841c0df254acc31ccced5142733c3157059/datafusion/functions-aggregate/src/array_agg.rs#L1574-L1589
And I added a new `BinaryView` test to test if it can be coalesced same as `Utf8View`
https://github.com/apache/datafusion/blob/19564841c0df254acc31ccced5142733c3157059/datafusion/functions-aggregate/src/array_agg.rs#L3158-L3201
--
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] perf(functions-aggregate): coalesce small ordered ARRAY_AGG batches [datafusion]
Posted by "TinyMurky (via GitHub)" <gi...@apache.org>.
TinyMurky commented on code in PR #25497:
URL: https://github.com/apache/datafusion/pull/25497#discussion_r4111977161
##########
datafusion/functions-aggregate/src/array_agg.rs:
##########
Review Comment:
I had run benchmark on following commits and updated the PR description :
- 6a792c67 : `main`
- 8588db18 : this pr before updating detach copy skip
- 19564841 : this pr after updating change
<!-- obsidian -->
Benchmark | main 6a792c67 (µs) | PR 8588db18 (µs) | PR 19564841 (µs)
-- | -- | -- | --
i64 ordered，1 row/update | 1,527.20 | 3,494.10 | 779.37
i64 random，1 row/update | 2,207.60 | 2,051.30 | 1,052.50
i64 ordered，8 rows/update | 221.42 | 258.73 | 133.85
i64 random，8 rows/update | 454.68 | 494.41 | 363.10
i64 ordered，64 rows/update | 50.37 | 50.37 | 50.44
i64 random，64 rows/update | 258.03 | 258.29 | 262.51
i64 ordered，2,048 rows/update | 18.68 | 19.09 | 19.00
i64 random，2,048 rows/update | 228.52 | 234.06 | 236.29
Utf8 4 KiB，1 row/update（update_batches only） | 2,417.90 | 2,442.10 | 2,642.90
the code that I edited in the new commit was as below，and it would skip `compact_view_after_concat` too, because the `compact_payload()` would be called during small batches merged if `DataType` is `Utf8View` or `BinaryView` and it followed small batches condition.
https://github.com/apache/datafusion/blob/c92bbc919cf78ab3d48b72f1040d1a681ba7d4c8/datafusion/functions-aggregate/src/array_agg.rs#L1527-L1583
--
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] perf(functions-aggregate): coalesce small ordered ARRAY_AGG batches [datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on PR #25497:
URL: https://github.com/apache/datafusion/pull/25497#issuecomment-5814129630
The new tests only coalesce Int64, but `concat` works differently for Dictionary (it merges the dictionary values) and Utf8View (it appends data buffers). Please add one case for each that coalesces past 64 rows and checks the evaluated order. It's fine to handle this in a follow-up.
```rs
#[test]
fn ordered_array_agg_coalesces_dictionary_and_view_payloads() -> Result<()> {
use arrow::array::{DictionaryArray, Int64Array, StringViewArray};
use arrow::datatypes::Int8Type;
let dictionary_type =
DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Utf8));
for payload_type in [dictionary_type, DataType::Utf8View] {
let mut acc = ordered_accumulator(
payload_type.clone(),
DataType::Int64,
SortOptions::new(false, false),
false,
false,
)?;
// Longer than 12 bytes so view payloads are stored in data buffers.
let value = |i: usize| format!("payload-value-{i:03}");
for i in (0..100).rev() {
let values: ArrayRef = match payload_type {
DataType::Utf8View => {
Arc::new(StringViewArray::from_iter_values([value(i)]))
}
_ => Arc::new(
[value(i).as_str()]
.into_iter()
.collect::<DictionaryArray<Int8Type>>(),
),
};
let ordering = Arc::new(Int64Array::from(vec![i64::try_from(i).unwrap()]))
as ArrayRef;
acc.update_batch(&[values, ordering])?;
}
assert_eq!(
acc.batches
.iter()
.map(|batch| batch.len())
.collect::<Vec<_>>(),
vec![64, 36],
);
let ScalarValue::List(result) = acc.evaluate()? else {
return internal_err!("expect list");
};
let values = cast(result.values(), &DataType::Utf8)?;
let values = values.as_string::<i32>();
for i in 0..100 {
assert_eq!(values.value(i), value(i));
}
}
Ok(())
}
```
--
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] perf(functions-aggregate): coalesce small ordered ARRAY_AGG batches [datafusion]
Posted by "TinyMurky (via GitHub)" <gi...@apache.org>.
TinyMurky commented on PR #25497:
URL: https://github.com/apache/datafusion/pull/25497#issuecomment-5854401827
Thank you!
--
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] perf(functions-aggregate): coalesce small ordered ARRAY_AGG batches [datafusion]
Posted by "TinyMurky (via GitHub)" <gi...@apache.org>.
TinyMurky commented on PR #25497:
URL: https://github.com/apache/datafusion/pull/25497#issuecomment-5847930013
Hello @jayzhan211 !
Thank you for reviewing once again!
I have added the detach skip logic and the call for`compact_payload()` that only when `DataType` is `Utf8View` or `BinaryView` only.
--
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] perf(functions-aggregate): coalesce small ordered ARRAY_AGG batches [datafusion]
Posted by "TinyMurky (via GitHub)" <gi...@apache.org>.
TinyMurky commented on code in PR #25497:
URL: https://github.com/apache/datafusion/pull/25497#discussion_r4111977161
##########
datafusion/functions-aggregate/src/array_agg.rs:
##########
Review Comment:
I had run benchmark on following commits and updated the PR description :
- 6a792c67 : `main`
- 8588db18 : this pr before updating detach copy skip
- 19564841 : this pr after updating change
<!-- obsidian -->
Benchmark | main 6a792c67 (µs) | PR 8588db18 (µs) | PR 19564841 (µs)
-- | -- | -- | --
i64 ordered，1 row/update | 1,527.20 | 3,494.10 | 779.37
i64 random，1 row/update | 2,207.60 | 2,051.30 | 1,052.50
i64 ordered，8 rows/update | 221.42 | 258.73 | 133.85
i64 random，8 rows/update | 454.68 | 494.41 | 363.10
i64 ordered，64 rows/update | 50.37 | 50.37 | 50.44
i64 random，64 rows/update | 258.03 | 258.29 | 262.51
i64 ordered，2,048 rows/update | 18.68 | 19.09 | 19.00
i64 random，2,048 rows/update | 228.52 | 234.06 | 236.29
Utf8 4 KiB，1 row/update（update_batches only） | 2,417.90 | 2,442.10 | 2,642.90
the code that I edited in the new commit was as below，and it would skip `compact_view_after_concat` too, because the `compact_payload()` would be called during small batches merged if `DataType` is `Utf8View` or `BinaryView` and it followed small batches conditions.
https://github.com/apache/datafusion/blob/c92bbc919cf78ab3d48b72f1040d1a681ba7d4c8/datafusion/functions-aggregate/src/array_agg.rs#L1527-L1583
--
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] perf(functions-aggregate): coalesce small ordered ARRAY_AGG batches [datafusion]
Posted by "TinyMurky (via GitHub)" <gi...@apache.org>.
TinyMurky commented on code in PR #25497:
URL: https://github.com/apache/datafusion/pull/25497#discussion_r4085505002
##########
datafusion/functions-aggregate/src/array_agg.rs:
##########
@@ -1535,11 +1539,34 @@ impl OrderSensitiveArrayAggAccumulator {
}
let start = self.entries.len();
- let batch_idx = self.batches.len();
- self.batches.push(values);
- self.entries.extend(
- (0..row_count).map(|row_idx| OrderedArrayAggEntry { batch_idx, row_idx }),
- );
+ let (batch_idx, row_offset) = match self.batches.last() {
+ Some(last_batch)
+ if last_batch.len() + row_count <= ORDERED_ARRAY_AGG_COALESCE_ROWS =>
Review Comment:
I have add the following codes
https://github.com/apache/datafusion/blob/0ddf5e8e579b9e3566cc19e34f1d41d436a58fc9/datafusion/functions-aggregate/src/array_agg.rs#L1541-L1548
And added following benchmarks to test the 4 KiB cap
https://github.com/apache/datafusion/blob/0ddf5e8e579b9e3566cc19e34f1d41d436a58fc9/datafusion/functions-aggregate/benches/array_agg.rs#L447-L502
--
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