@@ -15,14 +15,26 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::collections::VecDeque;
+use std::collections::{HashSet, VecDeque};
 use std::sync::Arc;
 
-use crate::PartitionedFile;
 use crate::file_groups::FileGroup;
 use crate::file_scan_config::FileScanConfig;
+use crate::{FileRange, PartitionedFile};
+use datafusion_physical_expr::utils::collect_columns;
 use parking_lot::Mutex;
 
+/// Minimum size of a morsel produced by lazy splitting, in bytes of data the
+/// scan actually reads (i.e. projected, compressed bytes).
+///
+/// Each piece pays a fixed cost to open (metadata load, pruning, stream
+/// setup), which smaller pieces amortize poorly. Since work items are split
+/// by whole-file byte ranges, this floor is translated into file-range bytes
+/// by dividing by the estimated fraction of columns the scan reads (see
+/// [`projected_fraction`]): reading 1 of 10 columns makes the file-range
+/// floor 10x larger so a morsel still carries ~this many bytes of real work.
+const MORSEL_MIN_PROJECTED_SIZE: i64 = 1024 * 1024;
+
 /// Source of work for `ScanState`.
 ///
 /// Streams that may share work across siblings use [`WorkSource::Shared`],
@@ -70,16 +82,53 @@ pub(crate) struct SharedWorkSource {
 
 #[derive(Debug, Default)]
 pub(super) struct SharedWorkSourceInner {
-    files: Mutex<VecDeque<PartitionedFile>>,
+    state: Mutex<VecDeque<PartitionedFile>>,
+    /// Number of sibling streams consuming from this queue. `0` disables
+    /// tail splitting of popped byte ranges.
+    consumers: usize,
+    /// Minimum byte-range size of a tail-split morsel
+    /// ([`MORSEL_MIN_PROJECTED_SIZE`] scaled by the projected column
+    /// fraction).
+    min_morsel_size: i64,
+}
+
+fn range_size(file: &PartitionedFile) -> i64 {
+    file.range.as_ref().map_or(0, |r| r.end - r.start)
 }
 
 impl SharedWorkSource {
-    /// Create a shared work source containing the provided unopened files.
+    /// Create a shared work source containing the provided unopened files,
+    /// without lazy morsel splitting.
+    #[cfg(test)]
     pub(crate) fn new(files: impl IntoIterator<Item = PartitionedFile>) -> Self {
-        let files = files.into_iter().collect();
+        Self::with_consumers(files, 0)
+    }
+
+    /// Create a shared work source consumed by `consumers` sibling streams
+    /// that lazily splits popped byte ranges into morsels sized to the
+    /// remaining work.
+    #[cfg(test)]
+    pub(crate) fn with_consumers(
+        files: impl IntoIterator<Item = PartitionedFile>,
+        consumers: usize,
+    ) -> Self {
+        Self::with_consumers_and_min_morsel_size(
+            files,
+            consumers,
+            MORSEL_MIN_PROJECTED_SIZE,
+        )
+    }
+
+    fn with_consumers_and_min_morsel_size(
+        files: impl IntoIterator<Item = PartitionedFile>,
+        consumers: usize,
+        min_morsel_size: i64,
+    ) -> Self {
         Self {
             inner: Arc::new(SharedWorkSourceInner {
-                files: Mutex::new(files),
+                state: Mutex::new(files.into_iter().collect()),
+                consumers,
+                min_morsel_size,
             }),
         }
     }
@@ -97,13 +146,182 @@ impl SharedWorkSource {
             .cloned()
             .collect();
         let files = config.file_source.reorder_files(files);
-        Self::new(files)
+        // A morsel should carry ~MORSEL_MIN_PROJECTED_SIZE bytes of data the
+        // scan actually reads; scale the file-range floor by the fraction of
+        // columns referenced.
+        let min_morsel_size =
+            (MORSEL_MIN_PROJECTED_SIZE as f64 / projected_fraction(config)) as i64;
+        Self::with_consumers_and_min_morsel_size(
+            files,
+            config.file_groups.len(),
+            min_morsel_size,
+        )
     }
 
     /// Pop the next file from the shared work queue.
     ///
     /// Returns `None` if the queue is empty
+    ///
+    /// # Tail morsel splitting
+    ///
+    /// While the queue is deep, popped work items are returned exactly as
+    /// the planner sized them. Only at the tail of the scan — when a pop
+    /// would leave the queue empty — is the final byte range split into
+    /// `min_morsel_size` morsels (the excess halves are pushed back), so
+    /// sibling streams that finish their pieces early steal a share of the
+    /// last piece instead of idling behind one straggler.
     fn pop_front(&self) -> Option<PartitionedFile> {
-        self.inner.files.lock().pop_front()
+        let mut files = self.inner.state.lock();
+        let mut file = files.pop_front()?;
+
+        if self.inner.consumers > 0 && files.is_empty() {
+            while range_size(&file) >= 2 * self.inner.min_morsel_size {
+                let second_half = split_off_second_half(&mut file)
+                    .expect("file with a non-empty byte range is splittable");
+                files.push_front(second_half);
+            }
+        }
+        Some(file)
+    }
+}
+
+/// Estimate the fraction of a file's bytes this scan reads, as the number of
+/// distinct file columns referenced by its projection and filter over the
+/// file schema width. This is a column-count approximation (columns are
+/// assumed equally wide), used to translate the projected-byte morsel floor
+/// into file-range bytes.
+fn projected_fraction(config: &FileScanConfig) -> f64 {
+    let num_fields = config
+        .file_source
+        .table_schema()
+        .file_schema()
+        .fields()
+        .len();
+    if num_fields == 0 {
+        return 1.0;
+    }
+    let Some(projection) = config.file_source.projection() else {
+        return 1.0;
+    };
+    let mut columns = HashSet::new();
+    for expr in projection.iter() {
+        columns.extend(collect_columns(&expr.expr));
+    }
+    if let Some(filter) = config.file_source.filter() {
+        columns.extend(collect_columns(&filter));
+    }
+    // Count at least one column (a bare `count(*)` still reads something)
+    // and at most the file schema width (the projection may also reference
+    // partition or virtual columns).
+    (columns.len().clamp(1, num_fields) as f64) / num_fields as f64
+}
+
+/// Shrink `file` to the first half of its byte range and return a clone
+/// covering the second half. Returns `None` (leaving `file` untouched) for
+/// unranged files, which the planner has not marked as byte-splittable.
+fn split_off_second_half(file: &mut PartitionedFile) -> Option<PartitionedFile> {
+    let FileRange { start, end } = *file.range.as_ref()?;
+    let mid = start + (end - start) / 2;
+    let mut second_half = file.clone();
+    second_half.range = Some(FileRange { start: mid, end });
+    file.range = Some(FileRange { start, end: mid });
+    Some(second_half)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const MIB: i64 = 1024 * 1024;
+
+    fn ranged_file(size: i64) -> PartitionedFile {
+        PartitionedFile::new("test.parquet".to_string(), size as u64).with_range(0, size)
+    }
+
+    /// Pop everything and return the byte ranges in pop order.
+    fn drain(source: &SharedWorkSource) -> Vec<FileRange> {
+        std::iter::from_fn(|| source.pop_front().map(|f| f.range.unwrap())).collect()
+    }
+
+    #[test]
+    fn lazy_split_tiles_range_into_morsels() {
+        let source = SharedWorkSource::with_consumers([ranged_file(8 * MIB)], 2);
+        let ranges = drain(&source);
+
+        // The single range was split into several morsels
+        assert!(ranges.len() > 1, "expected splitting, got {ranges:?}");
+        // Morsels are contiguous, ascending, and tile [0, 8MiB) exactly
+        assert_eq!(ranges.first().unwrap().start, 0);
+        assert_eq!(ranges.last().unwrap().end, 8 * MIB);
+        for pair in ranges.windows(2) {
+            assert_eq!(pair[0].end, pair[1].start, "gap or overlap: {ranges:?}");
+        }
+        // No morsel is smaller than the floor
+        for r in &ranges {
+            assert!(
+                r.end - r.start >= MORSEL_MIN_PROJECTED_SIZE,
+                "too small: {ranges:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn no_split_without_consumers() {
+        let source = SharedWorkSource::new([ranged_file(8 * MIB)]);
+        assert_eq!(
+            drain(&source),
+            vec![FileRange {
+                start: 0,
+                end: 8 * MIB
+            }]
+        );
+    }
+
+    #[test]
+    fn deep_backlog_pops_whole_ranges() {
+        // With plenty of remaining work relative to the consumer count, the
+        // fair share exceeds the range size and pieces are not split.
+        let files = (0..8).map(|_| ranged_file(8 * MIB));
+        let source = SharedWorkSource::with_consumers(files, 1);
+        let first = source.pop_front().unwrap().range.unwrap();
+        assert_eq!(
+            first,
+            FileRange {
+                start: 0,
+                end: 8 * MIB
+            }
+        );
+    }
+
+    #[test]
+    fn morsels_shrink_toward_the_tail() {
+        let files = (0..4).map(|i| {
+            PartitionedFile::new("test.parquet".to_string(), 32 * MIB as u64)
+                .with_range(i * 8 * MIB, (i + 1) * 8 * MIB)
+        });
+        let source = SharedWorkSource::with_consumers(files, 4);
+        let ranges = drain(&source);
+        assert!(ranges.len() > 4, "expected splitting, got {ranges:?}");
+        let first = ranges.first().unwrap();
+        let last = ranges.last().unwrap();
+        assert!(
+            last.end - last.start < first.end - first.start,
+            "tail morsels should be smaller: {ranges:?}"
+        );
+    }
+
+    #[test]
+    fn unranged_files_are_not_split() {
+        let file = PartitionedFile::new("test.parquet".to_string(), 8 * MIB as u64);
+        let source = SharedWorkSource::with_consumers([file], 4);
+        let popped = source.pop_front().unwrap();
+        assert_eq!(popped.range, None);
+        assert!(source.pop_front().is_none());
+    }
+
+    #[test]
+    fn small_ranges_are_not_split() {
+        let source = SharedWorkSource::with_consumers([ranged_file(MIB)], 4);
+        assert_eq!(drain(&source), vec![FileRange { start: 0, end: MIB }]);
     }
 }