@@ -306,6 +306,8 @@ pub(crate) struct IncrementalSortIterator {
     /// See [`Self::with_max_batch_bytes`].
     max_batch_bytes: Option<usize>,
     indices: Option<UInt32Array>,
+    /// Row sizes of `batch`, measured with `indices` when `max_batch_bytes` is set
+    row_sizes: Option<SpilledRowSizes>,
     cursor: usize,
 }
 
@@ -322,6 +324,7 @@ impl IncrementalSortIterator {
             max_batch_bytes: None,
             cursor: 0,
             indices: None,
+            row_sizes: None,
         }
     }
 
@@ -335,15 +338,19 @@ impl IncrementalSortIterator {
 
     /// Returns how many of the rows at `indices` fit in `max_batch_bytes`.
     fn rows_within(&self, indices: &[u32]) -> usize {
-        let Some(max_batch_bytes) = self.max_batch_bytes else {
+        let (Some(max_batch_bytes), Some(row_sizes)) =
+            (self.max_batch_bytes, &self.row_sizes)
+        else {
             return indices.len();
         };
-        let row_sizes = SpilledRowSizes::new(&self.batch);
         if let Some(row_bytes) = row_sizes.fixed() {
             return (max_batch_bytes / row_bytes.max(1))
                 .max(1)
                 .min(indices.len());
         }
+        if row_sizes.max_row().saturating_mul(indices.len()) <= max_batch_bytes {
+            return indices.len();
+        }
         let mut bytes = 0;
         for (rows, &row) in indices.iter().enumerate() {
             bytes += row_sizes.row(row as usize);
@@ -380,6 +387,9 @@ impl Iterator for IncrementalSortIterator {
                     Err(e) => return Some(Err(e.into())),
                 };
                 self.indices = Some(indices);
+                if self.max_batch_bytes.is_some() {
+                    self.row_sizes = Some(SpilledRowSizes::new(&self.batch));
+                }
 
                 // Call again, this time it will hit the Some(indices) branch and return the first batch
                 self.next()
@@ -404,6 +414,7 @@ impl Iterator for IncrementalSortIterator {
                     let schema = self.batch.schema();
                     let _ = mem::replace(&mut self.batch, RecordBatch::new_empty(schema));
                     self.indices = None;
+                    self.row_sizes = None;
                 }
 
                 // Return the new batch
 
@@ -31,27 +31,51 @@ use super::VIEW_SIZE_BYTES;
 /// variable-width values, and the elements of its nested values. Buffers that
 /// every row shares, such as validity bitmaps and dictionary values, are not
 /// counted.
-pub(crate) struct SpilledRowSizes(ColumnSize);
+pub(crate) struct SpilledRowSizes {
+    /// Bytes that every row takes in the fixed-width columns
+    fixed: usize,
+    /// The other columns
+    variable: Vec<ColumnSize>,
+    /// See [`Self::max_row`]
+    max_row: usize,
+}
 
 impl SpilledRowSizes {
     pub(crate) fn new(batch: &RecordBatch) -> Self {
-        Self(ColumnSize::Struct(
-            batch
-                .columns()
-                .iter()
-                .map(|column| ColumnSize::new(column.as_ref()))
-                .collect(),
-        ))
+        let mut fixed = 0;
+        let mut variable = vec![];
+        for column in batch.columns() {
+            let column = ColumnSize::new(column.as_ref());
+            match column.fixed() {
+                Some(width) => fixed += width,
+                None => variable.push(column),
+            }
+        }
+        let max_row = variable.iter().fold(fixed, |bytes, column| {
+            bytes.saturating_add(column.max_row())
+        });
+        Self {
+            fixed,
+            variable,
+            max_row,
+        }
     }
 
     /// Returns the size in bytes of row `row`.
     pub(crate) fn row(&self, row: usize) -> usize {
-        self.0.rows(row, row + 1)
+        self.variable.iter().fold(self.fixed, |bytes, column| {
+            bytes + column.rows(row, row + 1)
+        })
     }
 
     /// Returns the size in bytes of every row, if they all have the same size.
     pub(crate) fn fixed(&self) -> Option<usize> {
-        self.0.fixed()
+        self.variable.is_empty().then_some(self.fixed)
+    }
+
+    /// Returns an upper bound on the size in bytes of any row.
+    pub(crate) fn max_row(&self) -> usize {
+        self.max_row
     }
 }
 
@@ -129,6 +153,30 @@ impl ColumnSize {
         }
     }
 
+    /// Returns an upper bound on the size in bytes of a row.
+    fn max_row(&self) -> usize {
+        match self {
+            Self::Fixed(width) => *width,
+            Self::Bytes(offsets) => offsets.width() + offsets.max_len(),
+            Self::Views(views) => {
+                let max_len = views
+                    .iter()
+                    .map(|&view| view as u32)
+                    .filter(|&len| len > MAX_INLINE_VIEW_LEN)
+                    .max()
+                    .unwrap_or_default();
+                VIEW_SIZE_BYTES + max_len as usize
+            }
+            Self::List(offsets, values) => offsets
+                .width()
+                .saturating_add(offsets.max_len().saturating_mul(values.max_row())),
+            Self::FixedSizeList(size, values) => size.saturating_mul(values.max_row()),
+            Self::Struct(fields) => fields
+                .iter()
+                .fold(0, |bytes, field| bytes.saturating_add(field.max_row())),
+        }
+    }
+
     /// Returns the size in bytes of rows `start..end`.
     fn rows(&self, start: usize, end: usize) -> usize {
         let num_rows = end - start;
@@ -181,6 +229,21 @@ impl Offsets {
             Self::Large(offsets) => offsets[index] as usize,
         }
     }
+
+    /// Returns the largest distance between consecutive offsets.
+    fn max_len(&self) -> usize {
+        match self {
+            Self::Small(offsets) => offsets
+                .windows(2)
+                .map(|pair| (pair[1] - pair[0]) as usize)
+                .max(),
+            Self::Large(offsets) => offsets
+                .windows(2)
+                .map(|pair| (pair[1] - pair[0]) as usize)
+                .max(),
+        }
+        .unwrap_or_default()
+    }
 }
 
 impl From<&OffsetBuffer<i32>> for Offsets {
@@ -247,6 +310,36 @@ mod tests {
         );
     }
 
+    #[test]
+    fn largest_row_bounds_every_row() {
+        let ints: ArrayRef = Arc::new(Int64Array::from(vec![1, 2, 3]));
+        let strings: ArrayRef = Arc::new(StringArray::from(vec!["a", "bbbb", ""]));
+        // 8 bytes of Int64, plus a 4 byte offset and the longest string
+        assert_eq!(
+            SpilledRowSizes::new(&batch(vec![ints, strings])).max_row(),
+            16
+        );
+
+        let long = "a string longer than twelve";
+        let views: ArrayRef = Arc::new(StringViewArray::from(vec!["short", long]));
+        assert_eq!(
+            SpilledRowSizes::new(&batch(vec![views])).max_row(),
+            16 + long.len()
+        );
+
+        // An offset plus the most elements in a row, each of the largest size
+        let list: ArrayRef =
+            Arc::new(ListArray::from_iter_primitive::<Int64Type, _, _>(vec![
+                Some(vec![Some(1), Some(2), Some(3)]),
+                Some(vec![]),
+                Some(vec![Some(4)]),
+            ]));
+        assert_eq!(
+            SpilledRowSizes::new(&batch(vec![list])).max_row(),
+            4 + 3 * 8
+        );
+    }
+
     #[test]
     fn views_count_their_bytes_unless_inlined() {
         let long = "a string longer than twelve";