sunchao commented on code in PR #25428:
URL: https://github.com/apache/datafusion/pull/25428#discussion_r4112144194
##########
datafusion/physical-plan/src/sorts/multi_level_merge.rs:
##########
@@ -393,6 +411,35 @@ impl MultiLevelMergeBuilder {
}
};
+ if self.size_intermediate_merges
+ && self.sorted_streams.is_empty()
+ && !self.sorted_spill_files.is_empty()
+ {
+ // If one intermediate merge can leave a final pass with
the
+ // admitted fan-in, merge only the runs needed to get
there.
+ // Keep the admitted reservation so sizing does not change
Review Comment:
Fixed in
[78c4b7b](https://github.com/apache/datafusion/commit/78c4b7bdb05df36583f6a2f81d2bc6098fd34030).
A trimmed pass now keeps the existing buffer reservation in the builder
through intermediate EOF and spill completion, then transfers it to the next
admission.
There is one additional constraint: ordinary admission initially requests
buffers **plus equal replay headroom**, then releases the headroom. Retaining
the buffer grant alone therefore cannot secure the original `k` inputs. The
sizing calculation now budgets final buffers and headroom within that existing
grant, without reserving extra bytes. With eight full runs and initial fan-in
seven, it merges six and leaves three for final replay; nine runs now keep the
ordinary seven-input selection. The existing width guards remain; a wider
variable-width output can still change the subsequent budget.
The competing-consumer test reproduces your nine-run case on the previous
head: 1,152 versus 896 rewritten rows at subsequent fan-in three, and 2,304
versus 1,152 at fan-in two. Both replay-headroom release and intermediate EOF
reproduce it. The fix passes both boundaries; the eight-run case still improves
from 896 to 768 rows. Tests also check exact ordered output and
reservation/file cleanup, including dropping the intermediate stream and
builder.
I checked the earlier 64 MiB trace too: all four trims immediately precede
final 15-input merges. The exceptional builder initially admitted nine inputs
without trimming, accounting for its additional merge. The reason that initial
admission stopped at nine was not logged, so that trace cannot establish its
cause.
Both optional builder arguments now match, and I clarified the final-replay
docs. Validation passed: 12,172 Rust tests, all 524 SQL files, strict clippy,
formatting, and full repository lint/docs. The description now contains fresh
exact-base/current-head results: traced rewrites decrease 2.24% at 40 MiB and
23.00% at 56 MiB, with all actual trims reaching final replay. It also retains
the adverse diagnostic counters and variable timings; those timings do not
reliably quantify a latency improvement.
##########
datafusion/physical-plan/src/sorts/multi_level_merge.rs:
##########
@@ -985,6 +1033,148 @@ mod tests {
)
}
+ #[rstest::rstest]
+ #[case::full_batches(9, 128, 3)]
+ #[case::unchanged_selection(13, 128, 7)]
+ #[case::short_batches(9, 64, 7)]
+ #[tokio::test]
+ async fn intermediate_merge_sizing_preserves_final_output(
Review Comment:
Added in
[78c4b7b](https://github.com/apache/datafusion/commit/78c4b7bdb05df36583f6a2f81d2bc6098fd34030):
a pool wrapper with a real competing reservation, triggered by `shrink` at
either replay-headroom release or intermediate EOF. Eight cases cover
eight/nine full runs and subsequent fan-in two/three, comparing sized and
ordinary intermediate row writes and exact sorted values. The four nine-run
cases fail on the previous head and pass with the fix. A separate builder-drop
case checks retained-reservation lifetime and cleanup. All nine cases passed in
the full workspace run.
--
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]
