@@ -66,7 +66,7 @@ use datafusion_execution::memory_pool::{
 use datafusion_execution::runtime_env::RuntimeEnv;
 use datafusion_physical_expr::LexOrdering;
 use datafusion_physical_expr::PhysicalExpr;
-use datafusion_physical_expr::expressions::{DynamicFilterPhysicalExpr, lit};
+use datafusion_physical_expr::expressions::{Column, DynamicFilterPhysicalExpr, lit};
 
 use futures::{Stream, StreamExt, TryStreamExt};
 use log::{debug, trace};
@@ -784,13 +784,13 @@ impl ExternalSorter {
         let (sizes, shared_input_size) = self.sort_run_sizes(&runs)?;
         let shared_input =
             shared_input_size.map(|size| Arc::new(self.reservation.split(size)));
-        let batch_size = self.merge_run_batch_size();
         let jobs =
             runs.into_iter()
                 .zip(sizes)
                 .enumerate()
                 .map(move |(index, (batch, size))| {
                     let reservation = self.reservation.split(size);
+                    let batch_size = self.merge_run_batch_size(&batch)?;
                     let sort = self.sort_run(
                         batch,
                         reservation,
@@ -978,11 +978,11 @@ impl ExternalSorter {
         let (sizes, shared_input_size) = self.sort_run_sizes(&runs)?;
         let shared_input =
             shared_input_size.map(|size| Arc::new(self.reservation.split(size)));
-        let batch_size = self.merge_run_batch_size();
         runs.into_iter()
             .zip(sizes)
             .map(|(run, size)| {
                 let reservation = self.reservation.split(size);
+                let batch_size = self.merge_run_batch_size(&run)?;
                 self.sort_batch_stream(run, reservation, shared_input.clone(), batch_size)
             })
             .collect()
@@ -1038,12 +1038,6 @@ impl ExternalSorter {
         let expressions = self.expr.clone();
         let merge_pool = Arc::clone(&self.merge_pool);
         let elapsed_compute = self.metrics.baseline.elapsed_compute().clone();
-        // Split batches of views sharing bytes would each be charged for them
-        let batch_size = if batch_size < self.batch_size && views_share_bytes(&batch) {
-            self.batch_size
-        } else {
-            batch_size
-        };
         // Set when `merge_run_batch_size` split this run
         let split = batch_size < self.batch_size;
 
@@ -1096,15 +1090,17 @@ impl ExternalSorter {
         })
     }
 
-    /// Batch size of the sorted runs that feed a merge. A merge in the row format
-    /// holds the 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`. Runs keep `batch_size`
-    /// when their batches would share buffers other than the data of top-level
-    /// view columns, such as dictionary values, since every batch is charged for
-    /// them. `sort_run` also keeps it for a run whose views share bytes.
-    fn merge_run_batch_size(&self) -> usize {
+    /// Batch size of the sorted run `run` when it feeds a merge. A merge in the
+    /// row format holds the keys of every run's current batch at once. The sort
+    /// reserves about twice the run, so the keys fit next to the sorted copy if
+    /// they are at most as large as the run. Otherwise, with a memory limit, the
+    /// run uses 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`.
+    ///
+    /// Runs keep `batch_size` when their batches would share buffers other than
+    /// the data of top-level view columns, such as dictionary values, or when the
+    /// views of `run` share bytes, since every batch is charged for them.
+    fn merge_run_batch_size(&self, run: &RecordBatch) -> Result<usize> {
         let limited = !matches!(
             self.runtime.memory_pool.memory_limit(),
             MemoryLimit::Infinite
@@ -1114,11 +1110,34 @@ impl ExternalSorter {
             sorted_copy_keeps_buffers(data_type)
                 && !matches!(data_type, DataType::Utf8View | DataType::BinaryView)
         });
-        if limited && !shares_buffers && merges_in_row_format(&self.expr, &self.schema) {
-            (self.batch_size / 8).max(1024).min(self.batch_size)
+        if limited
+            && !shares_buffers
+            && merges_in_row_format(&self.expr, &self.schema)
+            && self.encoded_keys_may_exceed(run)?
+            && !views_share_bytes(run)
+        {
+            Ok((self.batch_size / 8).max(1024).min(self.batch_size))
         } else {
-            self.batch_size
+            Ok(self.batch_size)
+        }
+    }
+
+    /// Returns whether the row-format keys of `run` may be larger than `run`. A
+    /// computed key counts as larger, since sizing it would evaluate it again.
+    fn encoded_keys_may_exceed(&self, run: &RecordBatch) -> Result<bool> {
+        // An 8-byte offset per row, then a null byte and the bytes of each key
+        let mut encoded = run.num_rows() * size_of::<usize>();
+        for sort_expr in self.expr.iter() {
+            let Some(column) = sort_expr.expr.downcast_ref::<Column>() else {
+                return Ok(true);
+            };
+            encoded += run.num_rows()
+                + run
+                    .column(column.index())
+                    .to_data()
+                    .get_slice_memory_size()?;
         }
+        Ok(encoded > run.get_sliced_size()?)
     }
 
     /// If this sort may spill, pre-allocates
 
@@ -685,10 +685,11 @@ async fn test_sort_preparation_keeps_output_reserved() -> Result<()> {
 
 /// Merging sorted runs in the row format holds the keys of every run's current
 /// batch. Two `Int32` keys encode to 18 bytes per row against 8 bytes in the
-/// batches, and a single `FixedSizeBinary(16)` key also merges in the row
-/// format, so merging full-size runs needs more memory than the sort reserves
-/// while buffering them. The `Utf8View` and dictionary payloads keep buffers
-/// shared after sorting, which must not be charged once per split batch.
+/// batches, and a single `FixedSizeBinary(16)` key and two long `Utf8View` keys
+/// also encode to more than their runs, so merging full-size runs needs more
+/// memory than the sort reserves while buffering them. The payloads make runs
+/// larger than their keys or keep buffers shared after sorting, so these runs
+/// keep `batch_size` and must not be charged once per batch either.
 async fn check_merge_keys_larger_than_batches() -> Result<()> {
     let rows = 8192;
     let int_keys = |batch: i32| -> Vec<ArrayRef> {
@@ -711,7 +712,9 @@ async fn check_merge_keys_larger_than_batches() -> Result<()> {
         DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8));
     for payload in [DataType::Utf8View, dictionary] {
         let fields = [&int_fields[..], &[("c", payload.clone())]].concat();
-        check_merge_keys(&fields, 2, 16, true, |batch| {
+        // Runs of this `Utf8View` payload are not split, as on `main`
+        let within_pool = payload != DataType::Utf8View;
+        check_merge_keys(&fields, 2, 16, within_pool, |batch| {
             let payload = cast(&strings(batch), &payload)?;
             Ok([int_keys(batch), vec![payload]].concat())
         })
@@ -830,6 +833,52 @@ async fn test_merge_keys_larger_than_batches_multithreaded() -> Result<()> {
     check_merge_keys_larger_than_batches().await
 }
 
+/// Only runs whose encoded keys are larger than the run are split.
+#[test]
+fn test_merge_run_batch_size_follows_key_size() -> Result<()> {
+    let batch_size = 8192;
+    let rows = batch_size as i32;
+    let check = |payload: Option<ArrayRef>, expected: usize| -> Result<()> {
+        let mut columns: Vec<(&str, ArrayRef)> = vec![
+            ("a", Arc::new(Int64Array::from_iter_values(0..rows as i64))),
+            ("b", Arc::new(Int64Array::from_iter_values(0..rows as i64))),
+        ];
+        columns.extend(payload.map(|payload| ("c", payload)));
+        let batch = RecordBatch::try_from_iter(columns)?;
+        let ordering = LexOrdering::new((0..2).map(|i| {
+            PhysicalSortExpr::new_default(Arc::new(Column::new(
+                batch.schema().field(i).name(),
+                i,
+            )))
+        }))
+        .expect("two sort keys");
+        let runtime = RuntimeEnvBuilder::new()
+            .with_memory_limit(1024 * 1024 * 1024, 1.0)
+            .build_arc()?;
+        let sorter = ExternalSorter::new(
+            0,
+            batch.schema(),
+            ordering,
+            batch_size,
+            0,
+            0,
+            SpillCompression::Uncompressed,
+            &ExecutionPlanMetricsSet::new(),
+            runtime,
+        )?;
+        assert_eq!(sorter.merge_run_batch_size(&batch)?, expected);
+        Ok(())
+    };
+    // Encoded keys: 26 bytes a row, against 16 bytes a row
+    check(None, batch_size / 8)?;
+    // Against 16 + 16 bytes a row: inlined views add more than the keys need
+    let short: StringViewArray = (0..rows).map(|i| Some(format!("{i}"))).collect();
+    check(Some(Arc::new(short)), batch_size)?;
+    // Against 16 + 16 + 64 bytes a row
+    let long: StringViewArray = (0..rows).map(|i| Some(format!("{i:0>64}"))).collect();
+    check(Some(Arc::new(long)), batch_size)
+}
+
 #[tokio::test]
 async fn test_single_batch_spill_preserves_workspace_after_limit_decreases() -> Result<()>
 {
 
@@ -833,15 +833,17 @@ const VIEW_SIZE_BYTES: usize = 16;
 ///
 /// # When GC is triggered
 ///
-/// View GC is only performed when data buffers exceed a threshold (currently 10KB).
-/// This balances memory savings against the CPU overhead of garbage collection.
-/// Small view buffers are passed through unchanged since the GC overhead would exceed
-/// any memory savings.
+/// View GC runs when data buffers above a threshold (currently 10KB) hold more
+/// bytes than the views use. This balances memory savings against the CPU
+/// overhead of garbage collection. It also runs at any size when several data
+/// buffers remain and the views share no bytes, since each buffer read back
+/// from the spill is charged for the whole IPC message.
 ///
 /// # Performance considerations
 ///
 /// - If no arrays need compaction, the original batch is cloned cheaply
-/// - View GC is skipped for small buffers to avoid unnecessary CPU overhead
+/// - View GC is skipped for a single small data buffer to avoid unnecessary CPU
+///   overhead
 /// - Nested container types are traversed recursively so view arrays inside
 ///   `List`, `Map`, `Union`, `Dictionary`, and other child-bearing arrays are compacted too
 /// - Traversal uses physical children, not logical reachability. For example,
@@ -2706,6 +2708,24 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_gc_keeps_shared_views_in_two_buffers() -> Result<()> {
+        use arrow::array::AsArray;
+        use arrow::compute::concat;
+
+        let (_, first) = shared_dictionary_views(1000, 8, 1024)?;
+        let (_, second) = shared_dictionary_views(1000, 8, 1024)?;
+        let merged = concat(&[&first as &dyn Array, &second])?;
+        let merged = merged.as_string_view();
+
+        // `gc()` would copy 16384 values, 8x what the two dictionaries hold
+        let compacted = compact_view_array(merged).expect("buffer entries repeat");
+        assert_eq!(compacted.data_buffers().len(), 2);
+        assert_eq!(&compacted, merged);
+
+        Ok(())
+    }
+
     #[test]
     fn test_gc_keeps_shared_dictionary_buffer() -> Result<()> {
         let (dictionary, merged) = shared_dictionary_views(1000, 8, 1024)?;