fix: push OFFSET through CoalescePartitionsExec to DataSourceExec

When a single parquet file is split into multiple byte-range partitions,
CoalescePartitionsExec sits between GlobalLimitExec and DataSourceExec.
Previously, the optimizer would keep the GlobalLimitExec for the skip,
meaning offset wasn't pushed to the parquet reader.

Now we try to push the offset through the combining operator to its
DataSourceExec child, which uses a shared Arc<AtomicUsize> counter to
coordinate offset consumption across partitions. This eliminates the
GlobalLimitExec entirely for supported sources.

Result on 14GB hits.parquet: OFFSET 99M LIMIT 5 from 3.2s → 29ms.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: zhuqi-lucas

datafusion/datasource-parquet/src/source.rs

@@ -294,6 +294,11 @@ pub struct ParquetSource {
     /// so we still need to sort them after reading, so the reverse scan is inexact.
     /// Used to optimize ORDER BY ... DESC on sorted data.
     reverse_row_groups: bool,
+    /// Shared remaining offset counter across all partitions.
+    /// Initialized from `FileScanConfig.offset` in `create_morselizer`;
+    /// all partitions clone the same Arc so offset consumption is
+    /// coordinated atomically.
+    remaining_offset: Arc<std::sync::atomic::AtomicUsize>,
 }
 
 impl ParquetSource {
@@ -319,6 +324,7 @@ impl ParquetSource {
             #[cfg(feature = "parquet_encryption")]
             encryption_factory: None,
             reverse_row_groups: false,
+            remaining_offset: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
         }
     }
 
@@ -553,16 +559,23 @@ impl FileSource for ParquetSource {
             .as_ref()
             .map(|time_unit| parse_coerce_int96_string(time_unit.as_str()).unwrap());
 
+        // Initialize the shared offset from base_config on first partition.
+        // All partitions share the same Arc; scanning hasn't started yet
+        // (streams are created before poll_next), so no race with consumers.
+        if let Some(offset) = base_config.offset {
+            self.remaining_offset
+                .store(offset, std::sync::atomic::Ordering::SeqCst);
+        }
+        let remaining_offset = Arc::clone(&self.remaining_offset);
+
         Ok(Box::new(ParquetMorselizer {
             partition_index: partition,
             projection: self.projection.clone(),
             batch_size: self
                 .batch_size
                 .expect("Batch size must set before creating ParquetMorselizer"),
             limit: base_config.limit,
-            remaining_offset: Arc::new(std::sync::atomic::AtomicUsize::new(
-                base_config.offset.unwrap_or(0),
-            )),
+            remaining_offset,
             preserve_order: base_config.preserve_order,
             predicate: self.predicate.clone(),
             table_schema: self.table_schema.clone(),

datafusion/physical-optimizer/src/limit_pushdown.rs

@@ -216,11 +216,25 @@ pub fn pushdown_limit_helper(
             // fetch info to plan if possible. If not, we must add a limit node
             // with the information from the global state.
             let mut new_plan = plan_with_fetch;
-            // Execution plans can't (yet) handle skip, so if we have one,
-            // we still need to add a global limit
             if global_state.skip > 0 {
-                new_plan =
-                    add_global_limit(new_plan, global_state.skip, global_state.fetch);
+                // Try to push offset through the combining operator
+                // to its child (DataSourceExec).
+                let offset_pushed = {
+                    let children = new_plan.children();
+                    children.len() == 1
+                        && children[0].with_offset(global_state.skip).is_some()
+                };
+                if offset_pushed {
+                    let child = new_plan.children()[0]
+                        .with_offset(global_state.skip)
+                        .unwrap();
+                    let fallback = Arc::clone(&new_plan);
+                    new_plan =
+                        new_plan.with_new_children(vec![child]).unwrap_or(fallback);
+                } else {
+                    new_plan =
+                        add_global_limit(new_plan, global_state.skip, global_state.fetch);
+                }
             }
             global_state.fetch = skip_and_fetch;
             global_state.skip = 0;

datafusion/sqllogictest/test_files/sort_pushdown.slt

@@ -2374,6 +2374,7 @@ SELECT * FROM tn_offset LIMIT 5 OFFSET 10;
 15 150
 
 # Test N.8: OFFSET with multi-partition (target_partitions=4)
+# Small file stays as 1 group, but offset is still pushed down correctly.
 statement ok
 SET datafusion.execution.target_partitions = 4;