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/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;