Repository navigation
[core] Reclaim consumed merge batches with pending output - #1390
suxiaogang223 wants to merge 1 commit into
Conversation
Fixes apache#1386. Preserve and remap output references when compacting source buffers before the output batch fills.
JeonDaehong
left a comment
JeonDaehong
left a comment
There was a problem hiding this comment.
Thanks for picking this up. The remap of pending indices looks right to me, and the sort_merge tests pass with it. My comments are about when it compacts. Measured on c7i.8xlarge, release build, on f91fdb8.
| // There can be at most one source batch per cursor and per pending | ||
| // output row. A factor of two amortizes compaction while bounding | ||
| // unrelated batches across a long sequence of omitted keys. | ||
| let referenced_batch_bound = num_streams.saturating_add(output_indices.len()); |
There was a problem hiding this comment.
The bound counts pending rows, not batches, so with full-size batches it rarely fires. I ran two streams with 1024-row batches through this path: keys 0..1000 kept, then 1000..204800 deleted, batch size 1024. Main and this PR both peak at 399 of 400 source batches alive. A bound on live batches instead (2x the buffer length after the last compaction) brings it to 6.
That bound has a cost though: aggregation reads got 24% slower and partial-update 27% (2M keys, 3 overlapping rounds), since every materialized row is its own buffer entry and keeps resetting the count. With this PR as is, deduplicate and aggregation are unchanged, and partial-update is about 1% slower (1542–1552 vs 1527–1534 ms, 3 alternating rounds). Counting only Source entries might avoid both, but I haven't measured that.
| // Two cursors and at most three pending source rows need only a small | ||
| // buffer. Unrelated consumed batches must not accumulate with deletes. | ||
| assert!( | ||
| peak.load(AtomicOrdering::Relaxed) <= 16, |
There was a problem hiding this comment.
The test uses one-row batches, where rows and batches are the same count. Could it also cover multi-row batches, e.g. 1024 rows with the kept keys filling most of an output batch? That's the case the bound misses.
| } | ||
| if output_indices.is_empty() { | ||
| compact_batch_buffer(&mut batch_buffer, &mut stream_batch_idx, &cursors); | ||
| // There can be at most one source batch per cursor and per pending |
There was a problem hiding this comment.
Nit on the wording: materialized results are buffer entries too, one per pending row for aggregation and partial-update, so this is a bound on referenced entries rather than source batches. That's also why the bound grows with every merged key on those engines.
| @@ -1872,6 +1887,10 @@ fn compact_batch_buffer( | |||
|
|
|||
| *batch_buffer = new_buffer; | |||
There was a problem hiding this comment.
Now that compaction can run before the output is full, this clones every live entry on each pass, which is one Arc bump per column per entry. Taking the old buffer with std::mem::take and moving the live entries over would avoid that; with many pending materialized rows on a wide table it adds up. Might be part of the ~1% on partial-update, though I haven't isolated it.
| use std::sync::Arc; | ||
|
|
||
| #[tokio::test] | ||
| async fn pending_output_releases_unreferenced_source_batches() { |
There was a problem hiding this comment.
The lookup path (changelog-producer = lookup) also pushes a one-row batch for a key without a base row, and when that key's result is omitted nothing references it afterwards. With this change those get reclaimed too, where main kept them for as long as output rows were pending. Could a case with a LookupProvider cover that?
Purpose
Linked issue: close #1386
A partially filled output batch keeps
output_indicesnonempty. If many later keys are omitted because their winning rows are retracts, consumed source batches remain inbatch_bufferuntil the output fills or input ends. Reclaim unrelated batches while keeping both cursor and pending-output references valid.Brief change log
Tests
cargo +1.95.0 test --locked -p paimon --lib table::sort_merge::tests: 42 passed.cargo +1.95.0 test --locked -p paimon --lib: 4,860 passed, 6 ignored.-D warnings -A clippy::nonminimal_bool. Strict Clippy's six existing nonminimal-bool errors were reproduced on unmodified main; no source lint suppressions were added.cargo +1.95.0 fmt --all -- --checkandgit diff --check: passed.Local Cargo builds used debug info disabled and incremental compilation disabled. The regression measures retained source-array ownership, not RSS. This change does not establish a byte budget for all merge state or address a single key with arbitrarily many versions.
API and Format
No public API or storage-format changes. Pending output retains its original rows and order; output batch sizing is unchanged.
Documentation
Update the buffer ownership and compaction comments beside the implementation.
AI assistance: Codex helped develop the implementation, regression, and this description.