ExternalSorter: add pool-relative pre-merge spill trigger (0.66)

The self-relative trigger (`sorter.used() > pool_limit / num_workers`)
catches early-finisher partitions but misses the common Q18 pattern:
all partitions transition with their own footprint just below the
fair share (e.g. 470 MiB each on 8 GiB / 16 workers where self_share
is 512 MiB), yet the pool itself is at 7+ GiB because other consumers
are loaded. Self-relative never fires; the merge enters with no
headroom.

Port the pool-relative trigger from the original
`reclaimer-sortmerge-hashagg` branch with a 0.66 threshold (vs. the
old 0.33). On an 8 GiB pool that's ~5.3 GiB — high enough that we
don't spill spuriously on light queries, but well below the
"merge-can't-grow" zone Q18 hits at ~7 GiB.

Now: `sorter.used() > self_share || pool_used > pool_threshold`.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: JanKaul

datafusion/physical-plan/src/sorts/sort.rs

@@ -77,6 +77,15 @@ use datafusion_physical_expr::expressions::{DynamicFilterPhysicalExpr, lit};
 use futures::{StreamExt, TryStreamExt};
 use log::{debug, trace};
 
+/// Fraction of the memory pool above which a sort partition will
+/// proactively spill its in-memory batches before entering the
+/// non-reclaimable merge phase, regardless of how much that one
+/// partition is holding. Complements the self-relative check
+/// (`sorter.used() > pool_limit / num_workers`): catches the case
+/// where this partition is small but the pool is hot — the merge
+/// would still benefit from any bytes we can release now.
+const PRE_MERGE_SPILL_POOL_THRESHOLD: f64 = 0.66;
+
 struct ExternalSorterMetrics {
     /// metrics
     baseline: BaselineMetrics,
@@ -1595,15 +1604,26 @@ impl ExecutionPlan for SortExec {
                             handle_reclaim_request(&mut sorter, resp_tx).await?;
                         }
                         // Proactive spill before entering the
-                        // non-reclaimable merge phase. If this
-                        // partition is still holding more than its
-                        // fair share (`pool_limit / num_workers`),
-                        // dump the in-memory batches so siblings can
-                        // use the freed bytes. After this, `sort()`
-                        // takes the multi-level-merge path with a
-                        // tiny resident footprint instead of
-                        // becoming a "fat monument" that holds memory
-                        // for the rest of the query.
+                        // non-reclaimable merge phase. Two signals,
+                        // either fires:
+                        //
+                        // - Self-relative: this partition alone holds
+                        //   more than its fair per-worker share
+                        //   (`limit / num_workers`). Catches early
+                        //   finishers whose own footprint is the
+                        //   problem even when `pool.reserved()`
+                        //   hasn't yet risen.
+                        //
+                        // - Pool-relative: total pool usage is past
+                        //   the `PRE_MERGE_SPILL_POOL_THRESHOLD`
+                        //   fraction. Catches the "I'm small but the
+                        //   pool is hot" case — small partitions
+                        //   transition with the pool already
+                        //   saturated by upstream consumers, and the
+                        //   upcoming merge would have no headroom
+                        //   to grow into. Spilling now releases the
+                        //   batches before the merge commits to its
+                        //   non-reclaimable footprint.
                         if !sorter.in_mem_batches.is_empty()
                             && let MemoryLimit::Finite(limit) =
                                 sorter.runtime.memory_pool.memory_limit()
@@ -1613,7 +1633,11 @@ impl ExecutionPlan for SortExec {
                                 .num_workers()
                                 .max(1);
                             let self_share = limit / num_workers;
-                            if sorter.used() > self_share {
+                            let pool_threshold =
+                                (limit as f64 * PRE_MERGE_SPILL_POOL_THRESHOLD) as usize;
+                            let pool_used = sorter.runtime.memory_pool.reserved();
+                            if sorter.used() > self_share || pool_used > pool_threshold
+                            {
                                 sorter.sort_and_spill_in_mem_batches().await?;
                             }
                         }