Try and improve the perf of natural file splits in DF (#7609)

## Summary

Try and re-capture some of the performance we lost in #7591, only doing
the extra work when its actually required.

---------

Signed-off-by: Adam Gutglick <adam@spiraldb.com>

Author: AdamGS

vortex-datafusion/src/persistent/opener.rs

@@ -307,13 +307,7 @@ impl FileOpener for VortexOpener {
                 }
             };
 
-            let natural_split_ranges = natural_split_ranges_for_file(
-                natural_split_ranges.as_ref(),
-                &file.object_meta.location,
-                &layout_reader,
-            )?;
-
-            let mut scan_builder = ScanBuilder::new(session.clone(), layout_reader);
+            let mut scan_builder = ScanBuilder::new(session.clone(), Arc::clone(&layout_reader));
 
             if let Some(extensions) = file.extensions
                 && let Some(vortex_plan) = extensions.downcast_ref::<VortexAccessPlan>()
@@ -328,16 +322,25 @@ impl FileOpener for VortexOpener {
                     end: u64::try_from(file_range.end)
                         .map_err(|_| exec_datafusion_err!("Vortex file range end is negative"))?,
                 };
-
-                let Some(row_range) = split_aligned_row_range(
-                    byte_range,
-                    file.object_meta.size,
-                    natural_split_ranges.as_ref(),
-                ) else {
-                    return Ok(stream::empty().boxed());
-                };
-
-                scan_builder = scan_builder.with_row_range(row_range);
+                if byte_range.start != 0 || byte_range.end != file.object_meta.size {
+                    // Full-file scans already cover every natural split. Only translate the
+                    // byte range back into row boundaries when DataFusion has trimmed the file.
+                    let natural_split_ranges = natural_split_ranges_for_file(
+                        natural_split_ranges.as_ref(),
+                        &file.object_meta.location,
+                        &layout_reader,
+                    )?;
+
+                    let Some(row_range) = split_aligned_row_range(
+                        byte_range,
+                        file.object_meta.size,
+                        natural_split_ranges.as_ref(),
+                    ) else {
+                        return Ok(stream::empty().boxed());
+                    };
+
+                    scan_builder = scan_builder.with_row_range(row_range);
+                }
             }
 
             let filter = filter
@@ -477,6 +480,8 @@ fn compute_natural_split_ranges(layout_reader: &dyn LayoutReader) -> DFResult<Ar
 }
 
 /// Translate a DataFusion byte range to the contiguous natural split ranges it owns.
+/// Most splits are assigned by midpoint, but the leading split stays with the range that owns
+/// byte 0 so a tiny first byte range still claims the first rows.
 fn split_aligned_row_range(
     byte_range: Range<u64>,
     total_size: u64,
@@ -491,10 +496,13 @@ fn split_aligned_row_range(
         return None;
     }
 
-    let mut owned_splits = split_ranges.iter().filter(|split_range| {
-        let midpoint_byte = split_midpoint_to_byte(split_range, row_count, total_size);
-        byte_range.contains(&midpoint_byte)
-    });
+    let mut owned_splits = split_ranges
+        .iter()
+        .enumerate()
+        .filter_map(|(idx, split_range)| {
+            let assignment_byte = split_assignment_byte(idx, split_range, row_count, total_size);
+            byte_range.contains(&assignment_byte).then_some(split_range)
+        });
 
     let first_split = owned_splits.next()?;
     let mut row_range = first_split.start..first_split.end;
@@ -505,6 +513,21 @@ fn split_aligned_row_range(
     Some(row_range)
 }
 
+fn split_assignment_byte(
+    idx: usize,
+    split_range: &Range<u64>,
+    row_count: u64,
+    total_size: u64,
+) -> u64 {
+    if idx == 0 && split_range.start == 0 {
+        // Byte 0 is the only stable representative for the leading split. A midpoint can fall
+        // into the next DataFusion byte range and leave the first range with no rows to read.
+        0
+    } else {
+        split_midpoint_to_byte(split_range, row_count, total_size)
+    }
+}
+
 fn split_midpoint_to_byte(split_range: &Range<u64>, row_count: u64, total_size: u64) -> u64 {
     let midpoint_row = split_range.start + (split_range.end - split_range.start) / 2;
     let midpoint_byte = (u128::from(midpoint_row) * u128::from(total_size)) / u128::from(row_count);
@@ -566,6 +589,7 @@ mod tests {
     #[case(3..7, 10, vec![0..2, 2..5, 5..10], Some(2..5))]
     #[case(1..8, 10, vec![0..1, 1..9, 9..10], Some(1..9))]
     #[case(1..4, 16, vec![0..1, 1..2, 2..3, 3..4], None)]
+    #[case(0..1, 10, vec![0..2, 2..10], Some(0..2))]
     fn test_split_aligned_row_range(
         #[case] byte_range: Range<u64>,
         #[case] total_size: u64,