jayzhan211 commented on PR #24392:
URL: https://github.com/apache/datafusion/pull/24392#issuecomment-5619174254
@lyne7-sc , here is a suggestion:
**Retained payload pins the source batch's whole view/dictionary buffer**
`store_batch` replaces `ScalarValue::compacted()` with `copy_array_data`:
```rust
// Detach the stored payload from potentially oversized backing buffers.
let values = make_array(copy_array_data(&values.to_data()));
```
`copy_array_data` (`MutableArrayData`) compacts an offset-sliced array, but
it does **not** compact `Utf8View`/`BinaryView` variadic buffers or dictionary
value buffers. `ScalarValue::compacted()` did both — `compact_view_buffers()`
calls `.gc()`, and `try_from_array` on a dictionary yields a single unpacked
value rather than the dictionary.
Measured `acc.size()` after retaining **one row**, this branch vs `main`:
| payload | main | this PR |
|---|---|---|
| `Utf8View`, 1 row from a 10-row batch | 1,012 | 17,494 |
| `Utf8View`, 1 row from a 10,000-row batch | 1,012 | 509,014 |
| `Dictionary(Int32,Utf8)`, dict of 1,000 | 1,131 | 38,110 |
| `Dictionary(Int32,Utf8)`, dict of 10,000 | 1,131 | 303,518 |
That cost is per stored batch per accumulator, and ordered `ARRAY_AGG` gets
one accumulator **per group** through `GroupsAccumulatorAdapter`
(`groups_accumulator_supported` is false when `order_bys` is non-empty).
`Utf8View` is the default for Parquet string columns, so `SELECT g,
array_agg(str_col ORDER BY k) FROM parquet GROUP BY g` now reports hundreds of
KB per group to the memory pool and keeps the source batches alive for the
whole aggregation — spurious spills / `ResourcesExhausted` in exactly the
workload shape #20788 is about.
Suggested fix — compact the two lossy encodings after the copy:
```rust
// Detach the stored payload from potentially oversized backing buffers.
let values = make_array(copy_array_data(&values.to_data()));
+let values = compact_payload(values)?;
```
```rust
/// `copy_array_data` compacts offset-sliced buffers but keeps view variadic
/// buffers and dictionary values whole, so a single retained row can pin an
/// entire input batch. Compact those explicitly.
fn compact_payload(array: ArrayRef) -> Result<ArrayRef> {
match array.data_type() {
DataType::Utf8View => Ok(Arc::new(array.as_string_view().gc())),
DataType::BinaryView => Ok(Arc::new(array.as_binary_view().gc())),
DataType::Dictionary(_, value_type) => {
// Round-tripping rebuilds the dictionary from only the
referenced values.
let target = array.data_type().clone();
let flat = cast(array.as_ref(), value_type)?;
Ok(cast(flat.as_ref(), &target)?)
}
// List/LargeList/Struct/FixedSizeList need to recurse into children;
// `ScalarValue::compact_view_buffers` already implements exactly
this
// walk — consider making it `pub` and reusing it here instead.
_ => Ok(array),
}
}
```
Please also restore a guard for this. `does_not_over_account_memory_ordered`
used to carry `// without compaction, the size is 17112`; that comment was
dropped and the test still only covers `List<Utf8>`. Something like:
```rust
#[test]
fn stored_payload_does_not_pin_source_buffers() -> Result<()> {
for n in [10usize, 10_000] {
let vals: Vec<String> =
(0..n).map(|i|
format!("this-is-a-long-string-value-{i:08}")).collect();
let arr = StringViewArray::from(vals);
let mut acc = ordered_accumulator(
DataType::Utf8View, DataType::Int64,
SortOptions::new(false, false), false, false,
)?;
acc.update_batch(&[
Arc::new(arr.slice(3, 1)),
Arc::new(Int64Array::from(vec![1i64])),
])?;
// size must not scale with the source batch
assert!(acc.size() < 2_000, "n={n}: size={}", acc.size());
}
Ok(())
}
```
and the same shape for `Dictionary(Int32, Utf8)` with dictionaries of 100
and 10,000 entries.
--
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]
