fix: address PR review comments for StringView/BinaryView GC spill

- Return post-GC sliced size from append_batch so callers use the
  correct post-GC size for memory accounting (fixes cetra3's
  CHANGES_REQUESTED: max_record_batch_size was measured pre-GC in
  sort.rs and spill_manager.rs)
- Fix incorrect comment claiming Arrow gc() is a no-op; it always
  allocates new compact buffers
- Add comment in should_gc_view_array explaining why we sum
  data_buffers directly instead of using get_buffer_memory_size()
- Enhance append_batch doc comment with GC rationale per reviewer request
- Reduce row counts in heavy GC tests

Author: EeshanBembi

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

@@ -429,10 +429,9 @@ impl ExternalSorter {
             })?;
 
         for batch in batches_to_spill {
-            in_progress_file.append_batch(&batch)?;
+            let gc_sliced_size = in_progress_file.append_batch(&batch)?;
 
-            *max_record_batch_size =
-                (*max_record_batch_size).max(batch.get_sliced_size()?);
+            *max_record_batch_size = (*max_record_batch_size).max(gc_sliced_size);
         }
 
         assert_or_internal_err!(

datafusion/physical-plan/src/spill/in_progress_spill_file.rs

@@ -24,7 +24,10 @@ use arrow::array::RecordBatch;
 use datafusion_common::exec_datafusion_err;
 use datafusion_execution::disk_manager::RefCountedTempFile;
 
-use super::{IPCStreamWriter, gc_view_arrays, spill_manager::SpillManager};
+use super::{
+    IPCStreamWriter, gc_view_arrays,
+    spill_manager::{GetSlicedSize, SpillManager},
+};
 
 /// Represents an in-progress spill file used for writing `RecordBatch`es to disk, created by `SpillManager`.
 /// Caller is able to use this struct to incrementally append in-memory batches to
@@ -50,13 +53,18 @@ impl InProgressSpillFile {
     }
 
     /// Appends a `RecordBatch` to the spill file, initializing the writer if necessary.
-    /// Performs garbage collection on StringView/BinaryView arrays to reduce spill file size.
+    ///
+    /// Before writing, performs GC on StringView/BinaryView arrays to compact backing
+    /// buffers. When a view array is sliced, it still references the original full buffers,
+    /// causing massive spill files without GC (see issue #19414: 820MB → 33MB after GC).
+    ///
+    /// Returns the post-GC sliced memory size of the batch for memory accounting.
     ///
     /// # Errors
     /// - Returns an error if the file is not active (has been finalized)
     /// - Returns an error if appending would exceed the disk usage limit configured
     ///   by `max_temp_directory_size` in `DiskManager`
-    pub fn append_batch(&mut self, batch: &RecordBatch) -> Result<()> {
+    pub fn append_batch(&mut self, batch: &RecordBatch) -> Result<usize> {
         if self.in_progress_file.is_none() {
             return Err(exec_datafusion_err!(
                 "Append operation failed: No active in-progress file. The file may have already been finalized."
@@ -106,7 +114,7 @@ impl InProgressSpillFile {
                 unreachable!() // Already checked inside current function
             }
         }
-        Ok(())
+        gc_batch.get_sliced_size()
     }
 
     pub fn flush(&mut self) -> Result<()> {

datafusion/physical-plan/src/spill/mod.rs

@@ -440,36 +440,24 @@ pub(crate) fn gc_view_arrays(batch: &RecordBatch) -> Result<RecordBatch> {
     Ok(RecordBatch::try_new(batch.schema(), new_columns)?)
 }
 
-/// Determines whether a view array should be garbage collected.
+/// Determines whether a view array should be garbage collected before spilling.
 ///
-/// This function checks if:
-/// 1. The array has data buffers (non-inline strings/binaries)
-/// 2. The total buffer memory exceeds a threshold
-/// 3. The array appears to be sliced (has potential memory waste)
+/// Arrow's `gc()` always allocates new compact buffers (it is never a no-op), so we
+/// check here to skip the allocation cost when data buffers are small.
 ///
-/// The Arrow `gc()` method itself is a no-op for arrays that don't benefit from GC,
-/// but we still check here to avoid the overhead of calling gc() unnecessarily.
-///
-/// # Memory threshold rationale
-///
-/// We use a 10KB threshold based on:
-/// - Small arrays (< 10KB) have negligible memory impact even if sliced
-/// - GC has CPU overhead that isn't worth it for small arrays
-/// - This threshold captures pathological cases (e.g., ClickBench: 820MB -> 33MB)
-///   while avoiding unnecessary GC on small working sets
+/// We check `data_buffers` capacity rather than `array.get_buffer_memory_size()` because
+/// the latter includes the views buffer (16 bytes × n_rows), which scales with row count
+/// regardless of string content. We want to trigger GC only when non-inline string data
+/// is substantial enough to justify the copy.
 fn should_gc_view_array<T: ByteViewType>(array: &GenericByteViewArray<T>) -> bool {
     const MIN_BUFFER_SIZE_FOR_GC: usize = 10 * 1024; // 10KB threshold
 
     let data_buffers = array.data_buffers();
     if data_buffers.is_empty() {
-        // All strings/binaries are inlined (< 12 bytes), no GC needed
         return false;
     }
 
-    // Calculate total buffer memory
     let total_buffer_size: usize = data_buffers.iter().map(|b| b.capacity()).sum();
-
-    // Only GC if buffers exceed threshold
     total_buffer_size > MIN_BUFFER_SIZE_FOR_GC
 }
 
@@ -1035,7 +1023,7 @@ mod tests {
     fn test_gc_string_view_before_spill() -> Result<()> {
         use arrow::array::StringViewArray;
 
-        let strings: Vec<String> = (0..1000)
+        let strings: Vec<String> = (0..200)
             .map(|i| {
                 if i % 2 == 0 {
                     "short_string".to_string()
@@ -1056,7 +1044,7 @@ mod tests {
             Arc::clone(&schema),
             vec![Arc::new(string_array) as ArrayRef],
         )?;
-        let sliced_batch = batch.slice(0, 100);
+        let sliced_batch = batch.slice(0, 20);
         let gc_batch = gc_view_arrays(&sliced_batch)?;
 
         assert_eq!(gc_batch.num_rows(), sliced_batch.num_rows());
@@ -1069,7 +1057,7 @@ mod tests {
     fn test_gc_binary_view_before_spill() -> Result<()> {
         use arrow::array::BinaryViewArray;
 
-        let binaries: Vec<Vec<u8>> = (0..1000)
+        let binaries: Vec<Vec<u8>> = (0..200)
             .map(|i| {
                 if i % 2 == 0 {
                     vec![1, 2, 3, 4]
@@ -1091,7 +1079,7 @@ mod tests {
             Arc::clone(&schema),
             vec![Arc::new(binary_array) as ArrayRef],
         )?;
-        let sliced_batch = batch.slice(0, 100);
+        let sliced_batch = batch.slice(0, 20);
         let gc_batch = gc_view_arrays(&sliced_batch)?;
 
         assert_eq!(gc_batch.num_rows(), sliced_batch.num_rows());
@@ -1161,7 +1149,7 @@ mod tests {
 
     #[test]
     fn test_verify_gc_triggers_for_sliced_arrays() -> Result<()> {
-        let strings: Vec<String> = (0..1000)
+        let strings: Vec<String> = (0..200)
             .map(|i| {
                 format!(
                     "http://example.com/very/long/path/that/exceeds/inline/threshold/{i}"
@@ -1181,7 +1169,7 @@ mod tests {
             vec![Arc::new(string_array.clone()) as ArrayRef],
         )?;
 
-        let sliced = batch.slice(0, 100);
+        let sliced = batch.slice(0, 20);
 
         let sliced_array = sliced
             .column(0)
@@ -1208,7 +1196,7 @@ mod tests {
         use arrow::array::StringViewArray;
         use std::fs;
 
-        let num_rows = 5000;
+        let num_rows = 1000;
         let mut strings = Vec::with_capacity(num_rows);
 
         for i in 0..num_rows {
@@ -1284,7 +1272,7 @@ mod tests {
 
     #[test]
     fn test_spill_with_and_without_gc_comparison() -> Result<()> {
-        let num_rows = 2000;
+        let num_rows = 400;
         let strings: Vec<String> = (0..num_rows)
             .map(|i| {
                 format!(
@@ -1303,7 +1291,7 @@ mod tests {
         let batch =
             RecordBatch::try_new(schema, vec![Arc::new(string_array) as ArrayRef])?;
 
-        let sliced_batch = batch.slice(0, 200);
+        let sliced_batch = batch.slice(0, 40);
 
         let array_without_gc = sliced_batch
             .column(0)

datafusion/physical-plan/src/spill/spill_manager.rs

@@ -23,7 +23,6 @@ use crate::{common::spawn_buffered, metrics::SpillMetrics};
 use arrow::array::StringViewArray;
 use arrow::datatypes::SchemaRef;
 use arrow::record_batch::RecordBatch;
-use datafusion_common::utils::memory::get_record_batch_memory_size;
 use datafusion_common::{DataFusionError, Result, config::SpillCompression};
 use datafusion_execution::SendableRecordBatchStream;
 use datafusion_execution::disk_manager::RefCountedTempFile;
@@ -127,10 +126,8 @@ impl SpillManager {
             if borrowed.num_rows() == 0 {
                 return Ok(());
             }
-            in_progress_file.append_batch(borrowed)?;
-
-            max_record_batch_size =
-                max_record_batch_size.max(get_record_batch_memory_size(borrowed));
+            let gc_sliced_size = in_progress_file.append_batch(borrowed)?;
+            max_record_batch_size = max_record_batch_size.max(gc_sliced_size);
             Result::<_, DataFusionError>::Ok(())
         })?;
 
@@ -153,9 +150,9 @@ impl SpillManager {
 
         while let Some(batch) = stream.next().await {
             let batch = batch?;
-            in_progress_file.append_batch(&batch)?;
+            let gc_sliced_size = in_progress_file.append_batch(&batch)?;
 
-            max_record_batch_size = max_record_batch_size.max(batch.get_sliced_size()?);
+            max_record_batch_size = max_record_batch_size.max(gc_sliced_size);
         }
 
         let file = in_progress_file.finish()?;