jayzhan211 commented on code in PR #26953:
URL: https://github.com/apache/datafusion/pull/26953#discussion_r4430699232
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -1078,6 +1087,23 @@ impl ExternalSorter {
})
}
+ /// Batch size of the sorted runs that feed a merge. A multi-column merge
holds
+ /// the row-format keys of every run's current batch at once, and these
keys can
+ /// be larger than the batches. With a memory limit, runs therefore use an
eighth
+ /// of `batch_size`, but at least 1024 rows to keep the per-batch merge
cost
+ /// small. The merge output still uses `batch_size`.
+ fn merge_run_batch_size(&self) -> usize {
Review Comment:
With a non-inlined `Utf8View` column, each of the 8 chunks from
`sort_batch_chunked` keeps the input's full view data buffer, and `sort_run`
adds up `get_record_batch_memory_size` per chunk. With logging added to
`sort_run`, one run had `input=1228800 reserved=2457600 total_sorted=8454144`,
so it asks for about 8x the payload. This test passes on main and fails here
with `Failed to allocate additional 5.7 MB for ExternalSorter[0]`:
```rs
let schema = Arc::new(Schema::new(vec![
Field::new("a", DataType::Int32, false),
Field::new("b", DataType::Int32, false),
Field::new("c", DataType::Utf8View, false),
]));
// 64 batches of 8192 rows; c = format!("{i:0>64}"); sort by (a, b)
let capacity = MERGE_BYTES + 16 *
get_reserved_bytes_for_record_batch(&batches[0])?;
let runtime = RuntimeEnvBuilder::new()
.with_memory_pool(Arc::new(GreedyMemoryPool::new(capacity)))
.build_arc()?;
```
Running `crate::spill::gc_view_arrays` on the chunks when
`sorted_batches.len() > 1` fixed it locally. Three exact-accounting tests in
`spill_tests.rs` then need new expected numbers. Could you add this case as a
test with the fix?
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -1078,6 +1087,23 @@ impl ExternalSorter {
})
}
+ /// Batch size of the sorted runs that feed a merge. A multi-column merge
holds
+ /// the row-format keys of every run's current batch at once, and these
keys can
+ /// be larger than the batches. With a memory limit, runs therefore use an
eighth
+ /// of `batch_size`, but at least 1024 rows to keep the per-batch merge
cost
+ /// small. The merge output still uses `batch_size`.
+ fn merge_run_batch_size(&self) -> usize {
+ let limited = !matches!(
+ self.runtime.memory_pool.memory_limit(),
+ MemoryLimit::Infinite
+ );
+ if limited && self.expr.len() > 1 {
Review Comment:
`expr.len() > 1` doesn't match when the merge uses the row format:
`StreamingMergeBuilder::build` also uses `RowCursorStream` for a single key
that isn't primitive or a plain string or binary type. A single
`FixedSizeBinary(16)` key (UUIDs) still fails the same way. Could the gate
follow the merge's choice, ideally as a helper next to `build` so the two stay
in sync, with a test for a single `FixedSizeBinary` key? With this, that test
passes and the existing sort tests still pass:
```rs
fn merges_in_row_format(expr: &LexOrdering, schema: &Schema) -> bool {
if expr.len() > 1 {
return true;
}
let Ok(data_type) = expr[0].expr.data_type(schema) else {
return true;
};
!(data_type.is_primitive()
|| matches!(
data_type,
DataType::Utf8
| DataType::Utf8View
| DataType::LargeUtf8
| DataType::Binary
| DataType::LargeBinary
))
}
```
--
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]
