Push down offset to TableScan

datafusion/catalog-listing/src/table.rs

@@ -53,6 +53,7 @@ use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory;
 use datafusion_physical_expr_common::sort_expr::LexOrdering;
 use datafusion_physical_plan::ExecutionPlan;
 use datafusion_physical_plan::empty::EmptyExec;
+use datafusion_physical_plan::limit::GlobalLimitExec;
 use futures::future::BoxFuture;
 use futures::{Stream, StreamExt, TryStreamExt, future, stream};
 use object_store::ObjectStore;
@@ -79,7 +80,7 @@ pub struct ListFilesResult {
 /// * Reading multiple files as a single table
 /// * Hive style partitioning (e.g., directories named `date=2024-06-01`)
 /// * Merges schemas from files with compatible but not identical schemas (see [`ListingTableConfig::file_schema`])
-/// * `limit`, `filter` and `projection` pushdown for formats that support it (e.g.,
+/// * `limit`, `offset`, `filter` and `projection` pushdown for formats that support it (e.g.,
 ///   Parquet)
 /// * Statistics collection and pruning based on file metadata
 /// * Pre-existing sort order (see [`ListingOptions::file_sort_order`])
@@ -552,6 +553,10 @@ impl TableProvider for ListingTable {
             .collect()
     }
 
+    fn supports_offset_pushdown(&self) -> bool {
+        true
+    }
+
     fn get_table_definition(&self) -> Option<&str> {
         self.definition.as_deref()
     }
@@ -594,6 +599,10 @@ impl ListingTable {
         let projection = args.projection().map(|p| p.to_vec());
         let filters = args.filters().map(|f| f.to_vec()).unwrap_or_default();
         let limit = args.limit();
+        let offset = args.offset();
+        // The scan must read enough rows to satisfy `offset + limit`, not
+        // just `limit`, before any rows are skipped below.
+        let inflated_limit = limit.map(|l| l.saturating_add(offset.unwrap_or(0)));
 
         // extract types of partition columns
         let table_partition_cols = self
@@ -621,7 +630,7 @@ impl ListingTable {
         // or before applying non-partition filters.
         let statistic_file_limit =
             if filters.is_empty() && declared_output_partitioning.is_none() {
-                limit
+                inflated_limit
             } else {
                 None
             };
@@ -737,7 +746,7 @@ impl ListingTable {
             .with_constraints(self.constraints.clone())
             .with_statistics(statistics)
             .with_projection_indices(projection)?
-            .with_limit(limit)
+            .with_limit(inflated_limit)
             .with_output_ordering(output_ordering)
             .with_output_partitioning(output_partitioning)
             .with_expr_adapter(self.expr_adapter_factory.clone())
@@ -750,6 +759,11 @@ impl ListingTable {
             .create_physical_plan(state, scan_config)
             .await?;
 
+        let plan: Arc<dyn ExecutionPlan> = match offset {
+            Some(skip) => Arc::new(GlobalLimitExec::new(plan, skip, limit)),
+            None => plan,
+        };
+
         Ok(ScanResult::new(plan))
     }
 

datafusion/catalog/src/default_table_source.rs

@@ -70,6 +70,10 @@ impl TableSource for DefaultTableSource {
         self.table_provider.supports_filters_pushdown(filter)
     }
 
+    fn supports_offset_pushdown(&self) -> bool {
+        self.table_provider.supports_offset_pushdown()
+    }
+
     fn get_logical_plan(&'_ self) -> Option<Cow<'_, datafusion_expr::LogicalPlan>> {
         self.table_provider.get_logical_plan()
     }

datafusion/core/src/datasource/listing/table.rs

@@ -158,7 +158,7 @@ mod tests {
     };
     use arrow::{compute::SortOptions, record_batch::RecordBatch};
     use arrow_schema::{DataType, Field, Schema, SchemaRef, TimeUnit};
-    use datafusion_catalog::TableProvider;
+    use datafusion_catalog::{ScanArgs, TableProvider};
     use datafusion_catalog_listing::{
         ListingOptions, ListingTable, ListingTableConfig, SchemaSource,
     };
@@ -315,6 +315,48 @@ mod tests {
         Ok(())
     }
 
+    #[cfg(feature = "parquet")]
+    #[tokio::test]
+    async fn scan_with_args_offset_skips_correct_rows() -> Result<()> {
+        for target_partition in [1_usize, 8, 16] {
+            let ctx = SessionContext::new_with_config(
+                SessionConfig::new().with_target_partitions(target_partition),
+            );
+
+            let table = load_table(&ctx, "alltypes_plain.parquet").await?;
+
+            // Full scan (no offset) establishes the expected row order.
+            let full_exec = table
+                .scan_with_args(&ctx.state(), ScanArgs::default())
+                .await?
+                .into_inner();
+            let full_batches = collect(full_exec, ctx.task_ctx()).await?;
+            let full =
+                arrow::compute::concat_batches(&full_batches[0].schema(), &full_batches)?;
+            assert_eq!(full.num_rows(), 8);
+            let expected = full.slice(2, 3);
+
+            // `LIMIT 3 OFFSET 2` should return exactly rows [2, 5), in order —
+            // not just the first 3 rows.
+            let offset_exec = table
+                .scan_with_args(
+                    &ctx.state(),
+                    ScanArgs::default().with_limit(Some(3)).with_offset(Some(2)),
+                )
+                .await?
+                .into_inner();
+            let offset_batches = collect(offset_exec, ctx.task_ctx()).await?;
+            let actual = arrow::compute::concat_batches(
+                &offset_batches[0].schema(),
+                &offset_batches,
+            )?;
+
+            assert_eq!(actual.num_rows(), 3);
+            assert_eq!(batches_to_string(&[expected]), batches_to_string(&[actual]));
+        }
+        Ok(())
+    }
+
     #[cfg(feature = "parquet")]
     #[tokio::test]
     async fn test_try_create_output_ordering() {

datafusion/core/src/physical_planner.rs

@@ -589,6 +589,7 @@ impl DefaultPhysicalPlanner {
                     projection,
                     filters,
                     fetch,
+                    offset,
                     projected_schema,
                     statistics_requests,
                     ..
@@ -606,6 +607,7 @@ impl DefaultPhysicalPlanner {
                         .with_projection(projection.as_deref())
                         .with_filters(Some(&filters_vec))
                         .with_limit(*fetch)
+                        .with_offset(*offset)
                         .with_statistics_requests(&stats_requests);
                     let res = source.scan_with_args(session_state, opts).await?;
                     Arc::clone(res.plan())

datafusion/expr/src/logical_plan/builder.rs

@@ -491,7 +491,14 @@ impl LogicalPlanBuilder {
         projection: Option<Vec<usize>>,
         filters: Vec<Expr>,
     ) -> Result<Self> {
-        Self::scan_with_filters_inner(table_name, table_source, projection, filters, None)
+        Self::scan_with_filters_inner(
+            table_name,
+            table_source,
+            projection,
+            filters,
+            None,
+            None,
+        )
     }
 
     /// Convert a table provider into a builder with a TableScan with filter and fetch
@@ -508,6 +515,26 @@ impl LogicalPlanBuilder {
             projection,
             filters,
             fetch,
+            None,
+        )
+    }
+
+    /// Convert a table provider into a builder with a TableScan with filter, fetch and offset
+    pub fn scan_with_filters_fetch_offset(
+        table_name: impl Into<TableReference>,
+        table_source: Arc<dyn TableSource>,
+        projection: Option<Vec<usize>>,
+        filters: Vec<Expr>,
+        fetch: Option<usize>,
+        offset: Option<usize>,
+    ) -> Result<Self> {
+        Self::scan_with_filters_inner(
+            table_name,
+            table_source,
+            projection,
+            filters,
+            fetch,
+            offset,
         )
     }
 
@@ -517,11 +544,13 @@ impl LogicalPlanBuilder {
         projection: Option<Vec<usize>>,
         filters: Vec<Expr>,
         fetch: Option<usize>,
+        offset: Option<usize>,
     ) -> Result<Self> {
         let table_scan = TableScanBuilder::new(table_name, table_source)
             .with_projection(projection)
             .with_filters(filters)
             .with_fetch(fetch)
+            .with_offset(offset)
             .build()?;
 
         // Inline TableScan
@@ -2229,17 +2258,39 @@ pub fn table_scan_with_filter_and_fetch(
     projection: Option<Vec<usize>>,
     filters: Vec<Expr>,
     fetch: Option<usize>,
+) -> Result<LogicalPlanBuilder> {
+    table_scan_with_filter_and_fetch_and_offset(
+        name,
+        table_schema,
+        projection,
+        filters,
+        fetch,
+        None,
+    )
+}
+
+/// Create a LogicalPlanBuilder representing a scan of a table with the provided name and schema,
+/// filters, inlined fetch and offset.
+/// This is mostly used for testing and documentation.
+pub fn table_scan_with_filter_and_fetch_and_offset(
+    name: Option<impl Into<TableReference>>,
+    table_schema: &Schema,
+    projection: Option<Vec<usize>>,
+    filters: Vec<Expr>,
+    fetch: Option<usize>,
+    offset: Option<usize>,
 ) -> Result<LogicalPlanBuilder> {
     let table_source = table_source(table_schema);
     let name = name
         .map(|n| n.into())
         .unwrap_or_else(|| TableReference::bare(UNNAMED_TABLE));
-    LogicalPlanBuilder::scan_with_filters_fetch(
+    LogicalPlanBuilder::scan_with_filters_fetch_offset(
         name,
         table_source,
         projection,
         filters,
         fetch,
+        offset,
     )
 }
 

datafusion/expr/src/logical_plan/display.rs

@@ -342,6 +342,7 @@ impl<'a, 'b> PgJsonVisitor<'a, 'b> {
                 table_name,
                 filters,
                 fetch,
+                offset,
                 ..
             }) => {
                 let mut object = json!({
@@ -395,6 +396,10 @@ impl<'a, 'b> PgJsonVisitor<'a, 'b> {
                     object["Fetch"] = serde_json::Value::Number((*f).into());
                 }
 
+                if let Some(o) = offset {
+                    object["Offset"] = serde_json::Value::Number((*o).into());
+                }
+
                 object
             }
             LogicalPlan::Projection(Projection { expr, .. }) => {