sunchao commented on code in PR #25428:
URL: https://github.com/apache/datafusion/pull/25428#discussion_r4106856911


##########
datafusion/physical-plan/src/sorts/multi_level_merge.rs:
##########
@@ -393,6 +401,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
+                    // reservation lifetime or admission under contention.
+                    let needed = self.sorted_spill_files.len() + 1;
+                    if needed < sorted_spill_files.len() {
+                        let (first, first_limit) = &sorted_spill_files[0];
+                        // A wider pending run or a smaller output batch limit 
can
+                        // change the next pass's fan-in and add intermediate 
work.
+                        // Apply this sizing heuristic only to uniform run 
budgets.
+                        let uniform_runs = sorted_spill_files
+                            .iter()
+                            .chain(&self.sorted_spill_files)
+                            .all(|(file, limit)| {
+                                file.max_record_batch_memory
+                                    == first.max_record_batch_memory
+                                    && limit == first_limit

Review Comment:
   Addressed in 
[`af8d793bad`](https://github.com/apache/datafusion/commit/af8d793bad8daf05190746db9ff25b1b055f3aa5).
 The nine 64-row runs / 128-row output regression fails on the previous head: 
sizing rewrites 576 rows versus 448 without sizing. Both modes now rewrite 448 
rows, with exact sorted output and released memory/spill files. The full-batch 
case still reduces rewritten rows from 896 to 384.
   
   I kept the row information private instead of adding a field to 
`SortedSpillFile`. The aggregate records the smallest maximum batch row count 
across its original runs; any run shorter than the target batch size disables 
sizing for that replay. Intermediate and split writes also check their actual 
maximum batch rows against the effective output limit, and permanently disable 
sizing when a new run lacks a full-sized batch. The existing 
uniform-memory/limit guard remains.
   
   Validation passed: the eight sizing cases, 12,161 Rust tests and all 524 SQL 
files in the extended suite, strict all-feature/all-target Clippy, and the full 
lint/documentation checks. I also replaced the PR benchmark section with fresh 
exact-base-versus-this-head results and their limitations.
   



-- 
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]

Reply via email to