From b1a51f2a734f90302622d77def790d0e48221b18 Mon Sep 17 00:00:00 2001 From: Martin Tzvetanov Grigorov Date: Thu, 17 Sep 2026 10:19:00 +0300 Subject: [PATCH 01/10] Deprecate TableProvider::scan() in favor of ::scan_with_args() Pass `offset` in addition to limit/fetch --- datafusion/expr/src/logical_plan/builder.rs | 48 ++++++++++++++++++++- datafusion/expr/src/logical_plan/plan.rs | 15 +++++++ datafusion/session/src/table.rs | 26 +++++++++++ 3 files changed, 87 insertions(+), 2 deletions(-) diff --git a/datafusion/expr/src/logical_plan/builder.rs b/datafusion/expr/src/logical_plan/builder.rs index 36aa67bbe7e3..bf363de08933 100644 --- a/datafusion/expr/src/logical_plan/builder.rs +++ b/datafusion/expr/src/logical_plan/builder.rs @@ -491,7 +491,7 @@ impl LogicalPlanBuilder { projection: Option>, filters: Vec, ) -> Result { - 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 +508,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, + table_source: Arc, + projection: Option>, + filters: Vec, + fetch: Option, + offset: Option, + ) -> Result { + Self::scan_with_filters_inner( + table_name, + table_source, + projection, + filters, + fetch, + offset, ) } @@ -517,11 +537,13 @@ impl LogicalPlanBuilder { projection: Option>, filters: Vec, fetch: Option, + offset: Option, ) -> Result { 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 +2251,39 @@ pub fn table_scan_with_filter_and_fetch( projection: Option>, filters: Vec, fetch: Option, +) -> Result { + 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>, + table_schema: &Schema, + projection: Option>, + filters: Vec, + fetch: Option, + offset: Option, ) -> Result { 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, ) } diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs index de6d1667c75b..466d2ec4168c 100644 --- a/datafusion/expr/src/logical_plan/plan.rs +++ b/datafusion/expr/src/logical_plan/plan.rs @@ -3129,6 +3129,8 @@ pub struct TableScan { pub filters: Vec, /// Optional number of rows to read pub fetch: Option, + /// Optional number of rows to skip + pub offset: Option, /// Statistics the planner would like the provider to answer for this /// scan, typically attached by a custom optimizer rule from the /// surrounding plan (e.g. Min/Max for sort keys). @@ -3238,6 +3240,7 @@ pub struct TableScanBuilder { projection: Option>, filters: Vec, fetch: Option, + offset: Option, statistics_requests: BTreeSet, } @@ -3253,6 +3256,7 @@ impl TableScanBuilder { projection: None, filters: vec![], fetch: None, + offset: None, statistics_requests: BTreeSet::new(), } } @@ -3275,6 +3279,12 @@ impl TableScanBuilder { self } + /// Set the number of rows to skip. + pub fn with_offset(mut self, offset: Option) -> Self { + self.offset = offset; + self + } + /// Set the statistics requests for the scan. See /// [`TableScan::statistics_requests`]. pub fn with_statistics_requests( @@ -3294,6 +3304,7 @@ impl TableScanBuilder { projection, filters, fetch, + offset, statistics_requests, } = self; @@ -3335,6 +3346,7 @@ impl TableScanBuilder { projected_schema, filters, fetch, + offset, statistics_requests, }) } @@ -3348,6 +3360,7 @@ impl From for TableScanBuilder { projection: scan.projection, filters: scan.filters, fetch: scan.fetch, + offset: scan.offset, statistics_requests: scan.statistics_requests, } } @@ -6410,6 +6423,7 @@ mod tests { projected_schema: Arc::clone(&schema), filters: vec![], fetch: None, + offset: None, statistics_requests: BTreeSet::new(), })); let col = schema.field_names()[0].clone(); @@ -6441,6 +6455,7 @@ mod tests { projected_schema: Arc::clone(&unique_schema), filters: vec![], fetch: None, + offset: None, statistics_requests: BTreeSet::new(), })); let col = schema.field_names()[0].clone(); diff --git a/datafusion/session/src/table.rs b/datafusion/session/src/table.rs index a6cddd10ce80..28e539e36880 100644 --- a/datafusion/session/src/table.rs +++ b/datafusion/session/src/table.rs @@ -184,6 +184,12 @@ pub trait TableProvider: Any + Debug + Sync + Send { /// /// As noted above, columns referenced only by pushed-down filters may be /// absent from `projection`. + /// + /// # Deprecation + /// + /// Deprecated in favour of [`TableProvider::scan_with_args`] that brings more arguments, + /// e.g. [`ScanArgs::offset`] + #[deprecated(since = "56.0.0", note = "Please use [`TableProvider::scan_with_args`] instead")] async fn scan( &self, state: &dyn Session, @@ -469,6 +475,7 @@ pub struct ScanArgs<'a> { filters: Option<&'a [Expr]>, projection: Option<&'a [usize]>, limit: Option, + offset: Option, statistics_requests: &'a [StatisticsRequest], } @@ -532,6 +539,25 @@ impl<'a> ScanArgs<'a> { self.limit } + /// Set the number of rows to skip from the scan. + /// + /// If specified, the scan should skip this many rows. This is typically + /// used to optimize queries with `OFFSET` clauses. + /// + /// # Arguments + /// * `offset` - Optional number of rows to skip + pub fn with_offset(mut self, offset: Option) -> Self { + self.offset = offset; + self + } + + /// Get the number of rows to skip from the scan. + /// + /// Returns the row offset, or `None` if no offset was specified. + pub fn offset(&self) -> Option { + self.offset + } + /// Specifies the statistics the caller may use when optimizing the query. /// /// This is intended to allow the `TableProvider` to cheaply provide From 9a48bf3d2432c84bd005980e3984c81f7ef1d5f7 Mon Sep 17 00:00:00 2001 From: Martin Tzvetanov Grigorov Date: Thu, 17 Sep 2026 11:47:24 +0300 Subject: [PATCH 02/10] Update all callers of TableProvider::scan() with ::scan_with_args() --- datafusion/catalog-listing/src/table.rs | 20 +++- .../catalog/src/default_table_source.rs | 4 + datafusion/catalog/src/memory/table.rs | 7 +- .../core/src/datasource/listing/table.rs | 106 +++++++++++++++--- datafusion/core/src/datasource/memory_test.rs | 25 ++++- datafusion/core/src/physical_planner.rs | 2 + .../core/tests/parquet/file_statistics.rs | 51 +++++++-- .../partition_statistics.rs | 5 +- datafusion/expr/src/logical_plan/builder.rs | 9 +- datafusion/expr/src/logical_plan/display.rs | 5 + datafusion/expr/src/logical_plan/plan.rs | 22 +++- datafusion/expr/src/logical_plan/tree_node.rs | 2 + datafusion/expr/src/table_source.rs | 12 ++ datafusion/optimizer/src/push_down_filter.rs | 3 +- datafusion/session/src/table.rs | 22 +++- .../library-user-guide/upgrading/56.0.0.md | 53 +++++++++ 16 files changed, 302 insertions(+), 46 deletions(-) diff --git a/datafusion/catalog-listing/src/table.rs b/datafusion/catalog-listing/src/table.rs index 6c294fe077db..4e91d2761de0 100644 --- a/datafusion/catalog-listing/src/table.rs +++ b/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; @@ -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 + 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,13 @@ impl ListingTable { .create_physical_plan(state, scan_config) .await?; + // `supports_offset_pushdown` returns `true`, so we must actually + // skip the first `offset` rows here rather than merely hinting. + let plan: Arc = match offset { + Some(skip) => Arc::new(GlobalLimitExec::new(plan, skip, limit)), + None => plan, + }; + Ok(ScanResult::new(plan)) } diff --git a/datafusion/catalog/src/default_table_source.rs b/datafusion/catalog/src/default_table_source.rs index 3342db54de92..0343242ddf0b 100644 --- a/datafusion/catalog/src/default_table_source.rs +++ b/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> { self.table_provider.get_logical_plan() } diff --git a/datafusion/catalog/src/memory/table.rs b/datafusion/catalog/src/memory/table.rs index 1cc7287c32cf..777119762d73 100644 --- a/datafusion/catalog/src/memory/table.rs +++ b/datafusion/catalog/src/memory/table.rs @@ -49,7 +49,7 @@ use datafusion_physical_plan::{ ChildrenPropertiesMode, DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, PhysicalExpr, PlanProperties, ReplaceChildrenOptions, collect_partitioned, }; -use datafusion_session::Session; +use datafusion_session::{ScanArgs, Session}; use async_trait::async_trait; use futures::future::BoxFuture; @@ -149,7 +149,10 @@ impl MemTable { let schema = t.schema(); let constraints = t.constraints().cloned().unwrap_or_default(); - let exec = t.scan(state, None, &[], None).await?; + let exec = t + .scan_with_args(state, ScanArgs::default()) + .await? + .into_inner(); let data = collect_partitioned(exec, state.task_ctx()).await?; // Optionally repartition the collected batches. diff --git a/datafusion/core/src/datasource/listing/table.rs b/datafusion/core/src/datasource/listing/table.rs index 982766dc8851..f09c01142503 100644 --- a/datafusion/core/src/datasource/listing/table.rs +++ b/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, }; @@ -291,9 +291,13 @@ mod tests { let table = load_table(&ctx, "alltypes_plain.parquet").await?; let projection = None; let exec = table - .scan(&ctx.state(), projection, &[], None) + .scan_with_args( + &ctx.state(), + ScanArgs::default().with_projection(projection), + ) .await - .expect("Scan table"); + .expect("Scan table") + .into_inner(); assert_eq!(exec.children().len(), 0); assert_eq!(exec.output_partitioning().partition_count(), 1); @@ -315,6 +319,47 @@ mod tests { Ok(()) } + #[cfg(feature = "parquet")] + #[tokio::test] + async fn scan_with_args_offset_skips_correct_rows() -> Result<()> { + let ctx = SessionContext::new_with_config( + SessionConfig::new() + .with_collect_statistics(true) + .with_target_partitions(1), + ); + + 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() { @@ -459,9 +504,13 @@ mod tests { let filter = Expr::not_eq(col("p1"), lit("v1")); let scan = table - .scan(&ctx.state(), None, &[filter], None) + .scan_with_args( + &ctx.state(), + ScanArgs::default().with_filters(Some(&[filter])), + ) .await - .expect("Empty execution plan"); + .expect("Empty execution plan") + .into_inner(); assert!(scan.is::()); assert_eq!( @@ -1427,7 +1476,10 @@ mod tests { )]), )?; - let scan = table.scan(&ctx.state(), None, &[], None).await?; + let scan = table + .scan_with_args(&ctx.state(), ScanArgs::default()) + .await? + .into_inner(); assert_eq!(scan.output_partitioning(), &expected_output_partitioning); Ok(()) @@ -1458,7 +1510,10 @@ mod tests { Schema::new(vec![Field::new("a", DataType::Int32, false)]), )?; - let err = table.scan(&ctx.state(), None, &[], None).await.unwrap_err(); + let err = table + .scan_with_args(&ctx.state(), ScanArgs::default()) + .await + .unwrap_err(); assert_contains!( err.to_string(), "Range output partitioning split point 0 value 0 with type Utf8 cannot be represented exactly as ordering expression type Int32" @@ -1497,7 +1552,10 @@ mod tests { )]), )?; - let err = table.scan(&ctx.state(), None, &[], None).await.unwrap_err(); + let err = table + .scan_with_args(&ctx.state(), ScanArgs::default()) + .await + .unwrap_err(); assert_contains!( err.to_string(), "Range output partitioning split point 0 value 0 with type Timestamp(ns) cannot be represented exactly as ordering expression type Timestamp(s)" @@ -1540,7 +1598,10 @@ mod tests { Schema::new(vec![Field::new("a", DataType::Boolean, false)]), )?; - let unfiltered = table.scan(&ctx.state(), None, &[], None).await?; + let unfiltered = table + .scan_with_args(&ctx.state(), ScanArgs::default()) + .await? + .into_inner(); assert_eq!( unfiltered.output_partitioning(), &expected_output_partitioning @@ -1568,7 +1629,13 @@ mod tests { ] ); - let filtered = table.scan(&ctx.state(), None, &[filter], None).await?; + let filtered = table + .scan_with_args( + &ctx.state(), + ScanArgs::default().with_filters(Some(&[filter])), + ) + .await? + .into_inner(); assert_eq!( filtered.output_partitioning(), &expected_output_partitioning @@ -1648,7 +1715,10 @@ mod tests { let table_default = ListingTable::try_new(config_default)?; - let exec_default = table_default.scan(&state, None, &[], None).await?; + let exec_default = table_default + .scan_with_args(&state, ScanArgs::default()) + .await? + .into_inner(); assert_eq!( StatisticsContext::new() .compute(exec_default.as_ref(), &StatisticsArgs::new())? @@ -1674,7 +1744,10 @@ mod tests { .with_schema(schema_disabled); let table_disabled = ListingTable::try_new(config_disabled)?; - let exec_disabled = table_disabled.scan(&state, None, &[], None).await?; + let exec_disabled = table_disabled + .scan_with_args(&state, ScanArgs::default()) + .await? + .into_inner(); assert_eq!( StatisticsContext::new() .compute(exec_disabled.as_ref(), &StatisticsArgs::new())? @@ -1698,7 +1771,10 @@ mod tests { .with_schema(schema_enabled); let table_enabled = ListingTable::try_new(config_enabled)?; - let exec_enabled = table_enabled.scan(&state, None, &[], None).await?; + let exec_enabled = table_enabled + .scan_with_args(&state, ScanArgs::default()) + .await? + .into_inner(); assert_eq!( StatisticsContext::new() .compute(exec_enabled.as_ref(), &StatisticsArgs::new())? @@ -1785,7 +1861,9 @@ mod tests { let table = ListingTable::try_new(config)?; // The scan should work correctly - let scan_result = table.scan(&ctx.state(), None, &[], None).await; + let scan_result = table + .scan_with_args(&ctx.state(), ScanArgs::default()) + .await; assert!(scan_result.is_ok(), "Scan should succeed"); // Verify file listing works diff --git a/datafusion/core/src/datasource/memory_test.rs b/datafusion/core/src/datasource/memory_test.rs index 033a8036f5ec..4336e00d569c 100644 --- a/datafusion/core/src/datasource/memory_test.rs +++ b/datafusion/core/src/datasource/memory_test.rs @@ -27,7 +27,7 @@ mod tests { use arrow::error::ArrowError; use arrow::record_batch::RecordBatch; use arrow_schema::SchemaRef; - use datafusion_catalog::TableProvider; + use datafusion_catalog::{ScanArgs, TableProvider}; use datafusion_common::{Constraint, Constraints, DataFusionError, Result}; use datafusion_expr::LogicalPlanBuilder; use datafusion_expr::dml::InsertOp; @@ -60,8 +60,12 @@ mod tests { // scan with projection let exec = provider - .scan(&session_ctx.state(), Some(&[2, 1]), &[], None) - .await?; + .scan_with_args( + &session_ctx.state(), + ScanArgs::default().with_projection(Some(&[2, 1])), + ) + .await? + .into_inner(); let mut it = exec.execute(0, task_ctx)?; let batch2 = it.next().await.unwrap()?; @@ -94,7 +98,10 @@ mod tests { let provider = MemTable::try_new(schema, vec![vec![batch]])?; - let exec = provider.scan(&session_ctx.state(), None, &[], None).await?; + let exec = provider + .scan_with_args(&session_ctx.state(), ScanArgs::default()) + .await? + .into_inner(); let mut it = exec.execute(0, task_ctx)?; let batch1 = it.next().await.unwrap()?; assert_eq!(3, batch1.schema().fields().len()); @@ -178,7 +185,10 @@ mod tests { let projection: Vec = vec![0, 4]; match provider - .scan(&session_ctx.state(), Some(&projection), &[], None) + .scan_with_args( + &session_ctx.state(), + ScanArgs::default().with_projection(Some(&projection)), + ) .await { Err(DataFusionError::ArrowError(err, _)) => match err.as_ref() { @@ -305,7 +315,10 @@ mod tests { let provider = MemTable::try_new(Arc::new(merged_schema), vec![vec![batch1, batch2]])?; - let exec = provider.scan(&session_ctx.state(), None, &[], None).await?; + let exec = provider + .scan_with_args(&session_ctx.state(), ScanArgs::default()) + .await? + .into_inner(); let mut it = exec.execute(0, task_ctx)?; let batch1 = it.next().await.unwrap()?; assert_eq!(3, batch1.schema().fields().len()); diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs index 77997c619e5c..5a4af25f1900 100644 --- a/datafusion/core/src/physical_planner.rs +++ b/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()) diff --git a/datafusion/core/tests/parquet/file_statistics.rs b/datafusion/core/tests/parquet/file_statistics.rs index f6d733ec6972..5f4efb795344 100644 --- a/datafusion/core/tests/parquet/file_statistics.rs +++ b/datafusion/core/tests/parquet/file_statistics.rs @@ -28,6 +28,7 @@ use datafusion::datasource::source::DataSourceExec; use datafusion::execution::context::SessionState; use datafusion::execution::session_state::SessionStateBuilder; use datafusion::prelude::{ParquetReadOptions, SessionContext}; +use datafusion_catalog::ScanArgs; use datafusion_common::stats::Precision; use datafusion_common::{DFSchema, TableReference}; use datafusion_execution::cache::cache_manager::{ @@ -63,7 +64,11 @@ async fn check_stats_precision_with_filter_pushdown() { options.execution.collect_statistics = true; // Scan without filter, stats are exact - let exec = table.scan(&state, None, &[], None).await.unwrap(); + let exec = table + .scan_with_args(&state, ScanArgs::default()) + .await + .unwrap() + .into_inner(); assert_eq!( StatisticsContext::new() .compute(exec.as_ref(), &StatisticsArgs::new()) @@ -78,9 +83,13 @@ async fn check_stats_precision_with_filter_pushdown() { // source operator after the appropriate optimizer pass. let filter_expr = Expr::gt(col("id"), lit(1)); let exec_with_filter = table - .scan(&state, None, std::slice::from_ref(&filter_expr), None) + .scan_with_args( + &state, + ScanArgs::default().with_filters(Some(std::slice::from_ref(&filter_expr))), + ) .await - .unwrap(); + .unwrap() + .into_inner(); let ctx = SessionContext::new(); let df_schema = DFSchema::try_from(table.schema()).unwrap(); @@ -133,7 +142,11 @@ async fn load_table_stats_with_session_level_cache() { //Session 1 first time list files assert_eq!(get_static_cache_size(&state1), 0); - let exec1 = table1.scan(&state1, None, &[], None).await.unwrap(); + let exec1 = table1 + .scan_with_args(&state1, ScanArgs::default()) + .await + .unwrap() + .into_inner(); assert_eq!( StatisticsContext::new() @@ -156,7 +169,11 @@ async fn load_table_stats_with_session_level_cache() { //Session 2 first time list files //check session 1 cache result not show in session 2 assert_eq!(get_static_cache_size(&state2), 0); - let exec2 = table2.scan(&state2, None, &[], None).await.unwrap(); + let exec2 = table2 + .scan_with_args(&state2, ScanArgs::default()) + .await + .unwrap() + .into_inner(); assert_eq!( StatisticsContext::new() .compute(exec2.as_ref(), &StatisticsArgs::new()) @@ -177,7 +194,11 @@ async fn load_table_stats_with_session_level_cache() { //Session 1 second time list files //check session 1 cache result not show in session 2 assert_eq!(get_static_cache_size(&state1), 1); - let exec3 = table1.scan(&state1, None, &[], None).await.unwrap(); + let exec3 = table1 + .scan_with_args(&state1, ScanArgs::default()) + .await + .unwrap() + .into_inner(); assert_eq!( StatisticsContext::new() .compute(exec3.as_ref(), &StatisticsArgs::new()) @@ -314,7 +335,11 @@ async fn list_files_with_session_level_cache() { //Session 1 first time list files assert_eq!(get_list_file_cache_size(&state1), 0); - let exec1 = table1.scan(&state1, None, &[], None).await.unwrap(); + let exec1 = table1 + .scan_with_args(&state1, ScanArgs::default()) + .await + .unwrap() + .into_inner(); let data_source_exec = exec1.downcast_ref::().unwrap(); let data_source = data_source_exec.data_source(); let parquet1 = data_source.downcast_ref::().unwrap(); @@ -327,7 +352,11 @@ async fn list_files_with_session_level_cache() { //Session 2 first time list files //check session 1 cache result not show in session 2 assert_eq!(get_list_file_cache_size(&state2), 0); - let exec2 = table2.scan(&state2, None, &[], None).await.unwrap(); + let exec2 = table2 + .scan_with_args(&state2, ScanArgs::default()) + .await + .unwrap() + .into_inner(); let data_source_exec = exec2.downcast_ref::().unwrap(); let data_source = data_source_exec.data_source(); let parquet2 = data_source.downcast_ref::().unwrap(); @@ -340,7 +369,11 @@ async fn list_files_with_session_level_cache() { //Session 1 second time list files //check session 1 cache result not show in session 2 assert_eq!(get_list_file_cache_size(&state1), 1); - let exec3 = table1.scan(&state1, None, &[], None).await.unwrap(); + let exec3 = table1 + .scan_with_args(&state1, ScanArgs::default()) + .await + .unwrap() + .into_inner(); let data_source_exec = exec3.downcast_ref::().unwrap(); let data_source = data_source_exec.data_source(); let parquet3 = data_source.downcast_ref::().unwrap(); diff --git a/datafusion/core/tests/physical_optimizer/partition_statistics.rs b/datafusion/core/tests/physical_optimizer/partition_statistics.rs index 6cabcdb71039..e1c66bd65aea 100644 --- a/datafusion/core/tests/physical_optimizer/partition_statistics.rs +++ b/datafusion/core/tests/physical_optimizer/partition_statistics.rs @@ -24,7 +24,7 @@ mod test { use arrow_schema::{DataType, Field, Schema, SortOptions}; use datafusion::datasource::listing::ListingTable; use datafusion::prelude::SessionContext; - use datafusion_catalog::TableProvider; + use datafusion_catalog::{ScanArgs, TableProvider}; use datafusion_common::Result; use datafusion_common::stats::Precision; use datafusion_common::{ @@ -109,9 +109,10 @@ mod test { let table = ctx.table_provider(table_name.as_str()).await.unwrap(); let listing_table = table.downcast_ref::().unwrap().clone(); listing_table - .scan(&ctx.state(), None, &[], None) + .scan_with_args(&ctx.state(), ScanArgs::default()) .await .unwrap() + .into_inner() } // Date32 values for test data (days since 1970-01-01): diff --git a/datafusion/expr/src/logical_plan/builder.rs b/datafusion/expr/src/logical_plan/builder.rs index bf363de08933..935a3a378515 100644 --- a/datafusion/expr/src/logical_plan/builder.rs +++ b/datafusion/expr/src/logical_plan/builder.rs @@ -491,7 +491,14 @@ impl LogicalPlanBuilder { projection: Option>, filters: Vec, ) -> Result { - Self::scan_with_filters_inner(table_name, table_source, projection, filters, None, 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 diff --git a/datafusion/expr/src/logical_plan/display.rs b/datafusion/expr/src/logical_plan/display.rs index 1d95bf894bfa..67e9454972f4 100644 --- a/datafusion/expr/src/logical_plan/display.rs +++ b/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, .. }) => { diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs index 466d2ec4168c..2553edcf666c 100644 --- a/datafusion/expr/src/logical_plan/plan.rs +++ b/datafusion/expr/src/logical_plan/plan.rs @@ -2109,6 +2109,7 @@ impl LogicalPlan { projection, filters, fetch, + offset, .. }) => { let projected_fields = match projection { @@ -2176,6 +2177,10 @@ impl LogicalPlan { write!(f, ", fetch={n}")?; } + if let Some(n) = offset { + write!(f, ", offset={n}")?; + } + Ok(()) } LogicalPlan::Projection(Projection { expr, .. }) => { @@ -3136,7 +3141,11 @@ pub struct TableScan { /// surrounding plan (e.g. Min/Max for sort keys). /// /// A [`BTreeSet`], not a `Vec` to keep the resulting plan deterministic. - pub statistics_requests: BTreeSet, + /// + /// Boxed to keep this rarely-populated field from growing every + /// `TableScan` (and thus `LogicalPlan`) by its own size; see + /// `test_size_of_logical_plan`. + pub statistics_requests: Box>, } impl Debug for TableScan { @@ -3241,7 +3250,8 @@ pub struct TableScanBuilder { filters: Vec, fetch: Option, offset: Option, - statistics_requests: BTreeSet, + #[expect(clippy::box_collection)] + statistics_requests: Box>, } impl TableScanBuilder { @@ -3257,7 +3267,7 @@ impl TableScanBuilder { filters: vec![], fetch: None, offset: None, - statistics_requests: BTreeSet::new(), + statistics_requests: Box::default(), } } @@ -3291,7 +3301,7 @@ impl TableScanBuilder { mut self, statistics_requests: BTreeSet, ) -> Self { - self.statistics_requests = statistics_requests; + self.statistics_requests = Box::new(statistics_requests); self } @@ -6424,7 +6434,7 @@ mod tests { filters: vec![], fetch: None, offset: None, - statistics_requests: BTreeSet::new(), + statistics_requests: Box::default(), })); let col = schema.field_names()[0].clone(); @@ -6456,7 +6466,7 @@ mod tests { filters: vec![], fetch: None, offset: None, - statistics_requests: BTreeSet::new(), + statistics_requests: Box::default(), })); let col = schema.field_names()[0].clone(); diff --git a/datafusion/expr/src/logical_plan/tree_node.rs b/datafusion/expr/src/logical_plan/tree_node.rs index ee43666736fe..8a539e19b3c0 100644 --- a/datafusion/expr/src/logical_plan/tree_node.rs +++ b/datafusion/expr/src/logical_plan/tree_node.rs @@ -697,6 +697,7 @@ impl LogicalPlan { projected_schema, filters, fetch, + offset, statistics_requests, }) => filters.map_elements(f)?.update_data(|filters| { LogicalPlan::TableScan(TableScan { @@ -706,6 +707,7 @@ impl LogicalPlan { projected_schema, filters, fetch, + offset, statistics_requests, }) }), diff --git a/datafusion/expr/src/table_source.rs b/datafusion/expr/src/table_source.rs index 65dce8f3c8b0..8d720e3af55b 100644 --- a/datafusion/expr/src/table_source.rs +++ b/datafusion/expr/src/table_source.rs @@ -116,6 +116,18 @@ pub trait TableSource: Any + Sync + Send { .collect()) } + /// Tests whether the underlying table provider can guarantee that a scan + /// omits *exactly* the first `offset` rows it would otherwise have + /// produced. + /// + /// Returning `true` is a firm guarantee, not a hint: the optimizer may + /// rely on it to push a `LIMIT ... OFFSET ...` skip into the scan and + /// avoid re-applying it above. Returning `false` (the default) keeps the + /// skip always enforced above the scan. + fn supports_offset_pushdown(&self) -> bool { + false + } + /// Get the Logical plan of this table provider, if available. /// /// For example, a view may have a logical plan, but a CSV file does not. diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs index 8a1dcc12ef87..d6bfa5abe110 100644 --- a/datafusion/optimizer/src/push_down_filter.rs +++ b/datafusion/optimizer/src/push_down_filter.rs @@ -3138,7 +3138,8 @@ mod tests { projection, source: Arc::new(test_provider), fetch: None, - statistics_requests: std::collections::BTreeSet::new(), + offset: None, + statistics_requests: Box::default(), }); Ok(LogicalPlanBuilder::from(table_scan)) diff --git a/datafusion/session/src/table.rs b/datafusion/session/src/table.rs index 28e539e36880..c448a7527d11 100644 --- a/datafusion/session/src/table.rs +++ b/datafusion/session/src/table.rs @@ -185,11 +185,10 @@ pub trait TableProvider: Any + Debug + Sync + Send { /// As noted above, columns referenced only by pushed-down filters may be /// absent from `projection`. /// - /// # Deprecation + /// # Note /// - /// Deprecated in favour of [`TableProvider::scan_with_args`] that brings more arguments, + /// Overriding [`TableProvider::scan_with_args`] will give you access to more arguments, /// e.g. [`ScanArgs::offset`] - #[deprecated(since = "56.0.0", note = "Please use [`TableProvider::scan_with_args`] instead")] async fn scan( &self, state: &dyn Session, @@ -228,6 +227,10 @@ pub trait TableProvider: Any + Debug + Sync + Send { 'life1: 'async_trait, Self: 'async_trait, { + // Note: this bridge cannot honor `args.offset()`, since `scan` has no + // offset parameter. This is exactly why `supports_offset_pushdown` + // defaults to `false`: only providers that override `scan_with_args` + // directly can honor an offset. let plan = self.scan( state, args.projection(), @@ -237,6 +240,19 @@ pub trait TableProvider: Any + Debug + Sync + Send { Box::pin(async move { Ok(plan.await?.into()) }) } + /// Tests whether this table provider can guarantee that a scan built via + /// [`Self::scan_with_args`] omits *exactly* the first [`ScanArgs::offset`] + /// rows it would otherwise have produced. + /// + /// Returning `true` is a firm guarantee, not a hint: a caller that + /// pushes a skip into `ScanArgs::offset` may rely on it to avoid + /// re-applying the same skip above the scan. Returning `false` (the + /// default) means `ScanArgs::offset` is ignored or only partially + /// honored, so callers must not assume rows were skipped. + fn supports_offset_pushdown(&self) -> bool { + false + } + /// Specify if DataFusion should provide filter expressions to the /// TableProvider to apply *during* the scan. /// diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index 0a9828ff7fba..f8a1f4f47af5 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -427,3 +427,56 @@ let plan = DistributionContext::new_default(plan) Callers that only need the previous behaviour can keep using the deprecated form, or pass a freshly constructed `StatisticsContext` per call. + +### `TableProvider::scan_with_args` supports offset pushdown + +`TableProvider` gained `scan_with_args(state, args: ScanArgs)`, an +alternative to `scan(state, projection, filters, limit)` that passes the same +information through a structured, extensible `ScanArgs` value and also +carries a new `offset` (number of rows to skip) alongside `limit`. +`scan_with_args` has a default implementation that forwards to `scan`, so +existing `TableProvider` implementations keep compiling and working +unchanged; implementations that want to take advantage of `offset` (or future +`ScanArgs` fields) can override `scan_with_args` instead of, or in addition +to, `scan`. + +A new `TableProvider::supports_offset_pushdown() -> bool` (default `false`) +lets a provider guarantee it honors `ScanArgs::offset` exactly — that a scan +built with `offset` omits _exactly_ the first `offset` rows it would +otherwise have produced, not just approximately. `ListingTable` overrides this +to `true` and honors `offset`. + +Note that DataFusion's SQL/DataFrame optimizer does not yet populate +`LogicalPlan::TableScan::offset` from a query's `LIMIT ... OFFSET ...` clause +(that skip is still always enforced above the scan, as before); `offset` is +reachable today by constructing a `TableScan` directly (for example via +`LogicalPlanBuilder::scan_with_filters_fetch_offset`) or by calling +`scan_with_args` directly with `ScanArgs::with_offset`. Wiring the optimizer +to push a SQL `OFFSET` into the scan is tracked as follow-up work, since doing +so safely requires the pushdown to not be silently lost when the resulting +logical plan is serialized (e.g. via `datafusion-substrait` or +`datafusion-proto`) by a consumer unaware of the new field. + +**Who is affected:** + +- Custom `TableProvider` implementations that want offset pushdown: override + `scan_with_args` to read `ScanArgs::offset()` and honor it exactly, and + override `supports_offset_pushdown` to return `true`. +- Callers that already build a scan through `ScanArgs`/`scan_with_args` (for + example custom query planners) can now also set `with_offset`. + +**Example:** + +```rust,ignore +let plan = provider + .scan_with_args( + &state, + ScanArgs::default() + .with_projection(projection) + .with_filters(Some(&filters)) + .with_limit(limit) + .with_offset(offset), + ) + .await? + .into_inner(); +``` From 5003d141acbcd27fffc7b7f2072430177660a7aa Mon Sep 17 00:00:00 2001 From: Martin Tzvetanov Grigorov Date: Thu, 17 Sep 2026 14:49:20 +0300 Subject: [PATCH 03/10] Revert unnecessary changes from scan() to scan_with_args() --- datafusion/catalog-listing/src/table.rs | 4 +- datafusion/catalog/src/memory/table.rs | 7 +-- .../core/src/datasource/listing/table.rs | 63 ++++--------------- datafusion/core/src/datasource/memory_test.rs | 23 ++----- .../core/tests/parquet/file_statistics.rs | 50 +++------------ .../partition_statistics.rs | 5 +- datafusion/expr/src/logical_plan/plan.rs | 18 +++--- datafusion/expr/src/table_source.rs | 5 -- datafusion/optimizer/src/push_down_filter.rs | 3 +- datafusion/session/src/table.rs | 15 +---- .../library-user-guide/upgrading/56.0.0.md | 10 +-- 11 files changed, 46 insertions(+), 157 deletions(-) diff --git a/datafusion/catalog-listing/src/table.rs b/datafusion/catalog-listing/src/table.rs index 4e91d2761de0..086a3e30430e 100644 --- a/datafusion/catalog-listing/src/table.rs +++ b/datafusion/catalog-listing/src/table.rs @@ -80,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`]) @@ -759,8 +759,6 @@ impl ListingTable { .create_physical_plan(state, scan_config) .await?; - // `supports_offset_pushdown` returns `true`, so we must actually - // skip the first `offset` rows here rather than merely hinting. let plan: Arc = match offset { Some(skip) => Arc::new(GlobalLimitExec::new(plan, skip, limit)), None => plan, diff --git a/datafusion/catalog/src/memory/table.rs b/datafusion/catalog/src/memory/table.rs index 777119762d73..1cc7287c32cf 100644 --- a/datafusion/catalog/src/memory/table.rs +++ b/datafusion/catalog/src/memory/table.rs @@ -49,7 +49,7 @@ use datafusion_physical_plan::{ ChildrenPropertiesMode, DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, PhysicalExpr, PlanProperties, ReplaceChildrenOptions, collect_partitioned, }; -use datafusion_session::{ScanArgs, Session}; +use datafusion_session::Session; use async_trait::async_trait; use futures::future::BoxFuture; @@ -149,10 +149,7 @@ impl MemTable { let schema = t.schema(); let constraints = t.constraints().cloned().unwrap_or_default(); - let exec = t - .scan_with_args(state, ScanArgs::default()) - .await? - .into_inner(); + let exec = t.scan(state, None, &[], None).await?; let data = collect_partitioned(exec, state.task_ctx()).await?; // Optionally repartition the collected batches. diff --git a/datafusion/core/src/datasource/listing/table.rs b/datafusion/core/src/datasource/listing/table.rs index f09c01142503..5d70a4d1de09 100644 --- a/datafusion/core/src/datasource/listing/table.rs +++ b/datafusion/core/src/datasource/listing/table.rs @@ -291,13 +291,9 @@ mod tests { let table = load_table(&ctx, "alltypes_plain.parquet").await?; let projection = None; let exec = table - .scan_with_args( - &ctx.state(), - ScanArgs::default().with_projection(projection), - ) + .scan(&ctx.state(), projection, &[], None) .await - .expect("Scan table") - .into_inner(); + .expect("Scan table"); assert_eq!(exec.children().len(), 0); assert_eq!(exec.output_partitioning().partition_count(), 1); @@ -504,13 +500,9 @@ mod tests { let filter = Expr::not_eq(col("p1"), lit("v1")); let scan = table - .scan_with_args( - &ctx.state(), - ScanArgs::default().with_filters(Some(&[filter])), - ) + .scan(&ctx.state(), None, &[filter], None) .await - .expect("Empty execution plan") - .into_inner(); + .expect("Empty execution plan"); assert!(scan.is::()); assert_eq!( @@ -1476,10 +1468,7 @@ mod tests { )]), )?; - let scan = table - .scan_with_args(&ctx.state(), ScanArgs::default()) - .await? - .into_inner(); + let scan = table.scan(&ctx.state(), None, &[], None).await?; assert_eq!(scan.output_partitioning(), &expected_output_partitioning); Ok(()) @@ -1510,10 +1499,7 @@ mod tests { Schema::new(vec![Field::new("a", DataType::Int32, false)]), )?; - let err = table - .scan_with_args(&ctx.state(), ScanArgs::default()) - .await - .unwrap_err(); + let err = table.scan(&ctx.state(), None, &[], None).await.unwrap_err(); assert_contains!( err.to_string(), "Range output partitioning split point 0 value 0 with type Utf8 cannot be represented exactly as ordering expression type Int32" @@ -1552,10 +1538,7 @@ mod tests { )]), )?; - let err = table - .scan_with_args(&ctx.state(), ScanArgs::default()) - .await - .unwrap_err(); + let err = table.scan(&ctx.state(), None, &[], None).await.unwrap_err(); assert_contains!( err.to_string(), "Range output partitioning split point 0 value 0 with type Timestamp(ns) cannot be represented exactly as ordering expression type Timestamp(s)" @@ -1598,10 +1581,7 @@ mod tests { Schema::new(vec![Field::new("a", DataType::Boolean, false)]), )?; - let unfiltered = table - .scan_with_args(&ctx.state(), ScanArgs::default()) - .await? - .into_inner(); + let unfiltered = table.scan(&ctx.state(), None, &[], None).await?; assert_eq!( unfiltered.output_partitioning(), &expected_output_partitioning @@ -1629,13 +1609,7 @@ mod tests { ] ); - let filtered = table - .scan_with_args( - &ctx.state(), - ScanArgs::default().with_filters(Some(&[filter])), - ) - .await? - .into_inner(); + let filtered = table.scan(&ctx.state(), None, &[filter], None).await?; assert_eq!( filtered.output_partitioning(), &expected_output_partitioning @@ -1715,10 +1689,7 @@ mod tests { let table_default = ListingTable::try_new(config_default)?; - let exec_default = table_default - .scan_with_args(&state, ScanArgs::default()) - .await? - .into_inner(); + let exec_default = table_default.scan(&state, None, &[], None).await?; assert_eq!( StatisticsContext::new() .compute(exec_default.as_ref(), &StatisticsArgs::new())? @@ -1744,10 +1715,7 @@ mod tests { .with_schema(schema_disabled); let table_disabled = ListingTable::try_new(config_disabled)?; - let exec_disabled = table_disabled - .scan_with_args(&state, ScanArgs::default()) - .await? - .into_inner(); + let exec_disabled = table_disabled.scan(&state, None, &[], None).await?; assert_eq!( StatisticsContext::new() .compute(exec_disabled.as_ref(), &StatisticsArgs::new())? @@ -1771,10 +1739,7 @@ mod tests { .with_schema(schema_enabled); let table_enabled = ListingTable::try_new(config_enabled)?; - let exec_enabled = table_enabled - .scan_with_args(&state, ScanArgs::default()) - .await? - .into_inner(); + let exec_enabled = table_enabled.scan(&state, None, &[], None).await?; assert_eq!( StatisticsContext::new() .compute(exec_enabled.as_ref(), &StatisticsArgs::new())? @@ -1861,9 +1826,7 @@ mod tests { let table = ListingTable::try_new(config)?; // The scan should work correctly - let scan_result = table - .scan_with_args(&ctx.state(), ScanArgs::default()) - .await; + let scan_result = table.scan(&ctx.state(), None, &[], None).await; assert!(scan_result.is_ok(), "Scan should succeed"); // Verify file listing works diff --git a/datafusion/core/src/datasource/memory_test.rs b/datafusion/core/src/datasource/memory_test.rs index 4336e00d569c..2ff61027d3e1 100644 --- a/datafusion/core/src/datasource/memory_test.rs +++ b/datafusion/core/src/datasource/memory_test.rs @@ -60,12 +60,8 @@ mod tests { // scan with projection let exec = provider - .scan_with_args( - &session_ctx.state(), - ScanArgs::default().with_projection(Some(&[2, 1])), - ) - .await? - .into_inner(); + .scan(&session_ctx.state(), Some(&[2, 1]), &[], None) + .await?; let mut it = exec.execute(0, task_ctx)?; let batch2 = it.next().await.unwrap()?; @@ -98,10 +94,7 @@ mod tests { let provider = MemTable::try_new(schema, vec![vec![batch]])?; - let exec = provider - .scan_with_args(&session_ctx.state(), ScanArgs::default()) - .await? - .into_inner(); + let exec = provider.scan(&session_ctx.state(), None, &[], None).await?; let mut it = exec.execute(0, task_ctx)?; let batch1 = it.next().await.unwrap()?; assert_eq!(3, batch1.schema().fields().len()); @@ -185,10 +178,7 @@ mod tests { let projection: Vec = vec![0, 4]; match provider - .scan_with_args( - &session_ctx.state(), - ScanArgs::default().with_projection(Some(&projection)), - ) + .scan(&session_ctx.state(), Some(&projection), &[], None) .await { Err(DataFusionError::ArrowError(err, _)) => match err.as_ref() { @@ -315,10 +305,7 @@ mod tests { let provider = MemTable::try_new(Arc::new(merged_schema), vec![vec![batch1, batch2]])?; - let exec = provider - .scan_with_args(&session_ctx.state(), ScanArgs::default()) - .await? - .into_inner(); + let exec = provider.scan(&session_ctx.state(), None, &[], None).await?; let mut it = exec.execute(0, task_ctx)?; let batch1 = it.next().await.unwrap()?; assert_eq!(3, batch1.schema().fields().len()); diff --git a/datafusion/core/tests/parquet/file_statistics.rs b/datafusion/core/tests/parquet/file_statistics.rs index 5f4efb795344..6346ea562e39 100644 --- a/datafusion/core/tests/parquet/file_statistics.rs +++ b/datafusion/core/tests/parquet/file_statistics.rs @@ -64,11 +64,7 @@ async fn check_stats_precision_with_filter_pushdown() { options.execution.collect_statistics = true; // Scan without filter, stats are exact - let exec = table - .scan_with_args(&state, ScanArgs::default()) - .await - .unwrap() - .into_inner(); + let exec = table.scan(&state, None, &[], None).await.unwrap(); assert_eq!( StatisticsContext::new() .compute(exec.as_ref(), &StatisticsArgs::new()) @@ -83,13 +79,9 @@ async fn check_stats_precision_with_filter_pushdown() { // source operator after the appropriate optimizer pass. let filter_expr = Expr::gt(col("id"), lit(1)); let exec_with_filter = table - .scan_with_args( - &state, - ScanArgs::default().with_filters(Some(std::slice::from_ref(&filter_expr))), - ) + .scan(&state, None, std::slice::from_ref(&filter_expr), None) .await - .unwrap() - .into_inner(); + .unwrap(); let ctx = SessionContext::new(); let df_schema = DFSchema::try_from(table.schema()).unwrap(); @@ -142,11 +134,7 @@ async fn load_table_stats_with_session_level_cache() { //Session 1 first time list files assert_eq!(get_static_cache_size(&state1), 0); - let exec1 = table1 - .scan_with_args(&state1, ScanArgs::default()) - .await - .unwrap() - .into_inner(); + let exec1 = table1.scan(&state1, None, &[], None).await.unwrap(); assert_eq!( StatisticsContext::new() @@ -169,11 +157,7 @@ async fn load_table_stats_with_session_level_cache() { //Session 2 first time list files //check session 1 cache result not show in session 2 assert_eq!(get_static_cache_size(&state2), 0); - let exec2 = table2 - .scan_with_args(&state2, ScanArgs::default()) - .await - .unwrap() - .into_inner(); + let exec2 = table2.scan(&state2, None, &[], None).await.unwrap(); assert_eq!( StatisticsContext::new() .compute(exec2.as_ref(), &StatisticsArgs::new()) @@ -194,11 +178,7 @@ async fn load_table_stats_with_session_level_cache() { //Session 1 second time list files //check session 1 cache result not show in session 2 assert_eq!(get_static_cache_size(&state1), 1); - let exec3 = table1 - .scan_with_args(&state1, ScanArgs::default()) - .await - .unwrap() - .into_inner(); + let exec3 = table1.scan(&state1, None, &[], None).await.unwrap(); assert_eq!( StatisticsContext::new() .compute(exec3.as_ref(), &StatisticsArgs::new()) @@ -335,11 +315,7 @@ async fn list_files_with_session_level_cache() { //Session 1 first time list files assert_eq!(get_list_file_cache_size(&state1), 0); - let exec1 = table1 - .scan_with_args(&state1, ScanArgs::default()) - .await - .unwrap() - .into_inner(); + let exec1 = table1.scan(&state1, None, &[], None).await.unwrap(); let data_source_exec = exec1.downcast_ref::().unwrap(); let data_source = data_source_exec.data_source(); let parquet1 = data_source.downcast_ref::().unwrap(); @@ -352,11 +328,7 @@ async fn list_files_with_session_level_cache() { //Session 2 first time list files //check session 1 cache result not show in session 2 assert_eq!(get_list_file_cache_size(&state2), 0); - let exec2 = table2 - .scan_with_args(&state2, ScanArgs::default()) - .await - .unwrap() - .into_inner(); + let exec2 = table2.scan(&state2, None, &[], None).await.unwrap(); let data_source_exec = exec2.downcast_ref::().unwrap(); let data_source = data_source_exec.data_source(); let parquet2 = data_source.downcast_ref::().unwrap(); @@ -369,11 +341,7 @@ async fn list_files_with_session_level_cache() { //Session 1 second time list files //check session 1 cache result not show in session 2 assert_eq!(get_list_file_cache_size(&state1), 1); - let exec3 = table1 - .scan_with_args(&state1, ScanArgs::default()) - .await - .unwrap() - .into_inner(); + let exec3 = table1.scan(&state1, None, &[], None).await.unwrap(); let data_source_exec = exec3.downcast_ref::().unwrap(); let data_source = data_source_exec.data_source(); let parquet3 = data_source.downcast_ref::().unwrap(); diff --git a/datafusion/core/tests/physical_optimizer/partition_statistics.rs b/datafusion/core/tests/physical_optimizer/partition_statistics.rs index e1c66bd65aea..6cabcdb71039 100644 --- a/datafusion/core/tests/physical_optimizer/partition_statistics.rs +++ b/datafusion/core/tests/physical_optimizer/partition_statistics.rs @@ -24,7 +24,7 @@ mod test { use arrow_schema::{DataType, Field, Schema, SortOptions}; use datafusion::datasource::listing::ListingTable; use datafusion::prelude::SessionContext; - use datafusion_catalog::{ScanArgs, TableProvider}; + use datafusion_catalog::TableProvider; use datafusion_common::Result; use datafusion_common::stats::Precision; use datafusion_common::{ @@ -109,10 +109,9 @@ mod test { let table = ctx.table_provider(table_name.as_str()).await.unwrap(); let listing_table = table.downcast_ref::().unwrap().clone(); listing_table - .scan_with_args(&ctx.state(), ScanArgs::default()) + .scan(&ctx.state(), None, &[], None) .await .unwrap() - .into_inner() } // Date32 values for test data (days since 1970-01-01): diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs index 2553edcf666c..c7ce897a8aee 100644 --- a/datafusion/expr/src/logical_plan/plan.rs +++ b/datafusion/expr/src/logical_plan/plan.rs @@ -3141,11 +3141,7 @@ pub struct TableScan { /// surrounding plan (e.g. Min/Max for sort keys). /// /// A [`BTreeSet`], not a `Vec` to keep the resulting plan deterministic. - /// - /// Boxed to keep this rarely-populated field from growing every - /// `TableScan` (and thus `LogicalPlan`) by its own size; see - /// `test_size_of_logical_plan`. - pub statistics_requests: Box>, + pub statistics_requests: BTreeSet, } impl Debug for TableScan { @@ -3157,6 +3153,7 @@ impl Debug for TableScan { .field("projected_schema", &self.projected_schema) .field("filters", &self.filters) .field("fetch", &self.fetch) + .field("offset", &self.offset) .finish_non_exhaustive() } } @@ -3250,8 +3247,7 @@ pub struct TableScanBuilder { filters: Vec, fetch: Option, offset: Option, - #[expect(clippy::box_collection)] - statistics_requests: Box>, + statistics_requests: BTreeSet, } impl TableScanBuilder { @@ -3267,7 +3263,7 @@ impl TableScanBuilder { filters: vec![], fetch: None, offset: None, - statistics_requests: Box::default(), + statistics_requests: BTreeSet::default(), } } @@ -3301,7 +3297,7 @@ impl TableScanBuilder { mut self, statistics_requests: BTreeSet, ) -> Self { - self.statistics_requests = Box::new(statistics_requests); + self.statistics_requests = statistics_requests; self } @@ -6434,7 +6430,7 @@ mod tests { filters: vec![], fetch: None, offset: None, - statistics_requests: Box::default(), + statistics_requests: BTreeSet::default(), })); let col = schema.field_names()[0].clone(); @@ -6466,7 +6462,7 @@ mod tests { filters: vec![], fetch: None, offset: None, - statistics_requests: Box::default(), + statistics_requests: BTreeSet::default(), })); let col = schema.field_names()[0].clone(); diff --git a/datafusion/expr/src/table_source.rs b/datafusion/expr/src/table_source.rs index 8d720e3af55b..3dea20c7f619 100644 --- a/datafusion/expr/src/table_source.rs +++ b/datafusion/expr/src/table_source.rs @@ -119,11 +119,6 @@ pub trait TableSource: Any + Sync + Send { /// Tests whether the underlying table provider can guarantee that a scan /// omits *exactly* the first `offset` rows it would otherwise have /// produced. - /// - /// Returning `true` is a firm guarantee, not a hint: the optimizer may - /// rely on it to push a `LIMIT ... OFFSET ...` skip into the scan and - /// avoid re-applying it above. Returning `false` (the default) keeps the - /// skip always enforced above the scan. fn supports_offset_pushdown(&self) -> bool { false } diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs index d6bfa5abe110..243f317aab7a 100644 --- a/datafusion/optimizer/src/push_down_filter.rs +++ b/datafusion/optimizer/src/push_down_filter.rs @@ -1435,6 +1435,7 @@ fn expr_columns(exprs: &[Expr]) -> HashSet { #[cfg(test)] mod tests { use std::cmp::Ordering; + use std::collections::BTreeSet; use std::fmt::{Debug, Formatter}; use arrow::datatypes::{Field, Schema, SchemaRef}; @@ -3139,7 +3140,7 @@ mod tests { source: Arc::new(test_provider), fetch: None, offset: None, - statistics_requests: Box::default(), + statistics_requests: std::collections::BTreeSet::new(), }); Ok(LogicalPlanBuilder::from(table_scan)) diff --git a/datafusion/session/src/table.rs b/datafusion/session/src/table.rs index c448a7527d11..a92732e5de7a 100644 --- a/datafusion/session/src/table.rs +++ b/datafusion/session/src/table.rs @@ -227,10 +227,6 @@ pub trait TableProvider: Any + Debug + Sync + Send { 'life1: 'async_trait, Self: 'async_trait, { - // Note: this bridge cannot honor `args.offset()`, since `scan` has no - // offset parameter. This is exactly why `supports_offset_pushdown` - // defaults to `false`: only providers that override `scan_with_args` - // directly can honor an offset. let plan = self.scan( state, args.projection(), @@ -240,15 +236,8 @@ pub trait TableProvider: Any + Debug + Sync + Send { Box::pin(async move { Ok(plan.await?.into()) }) } - /// Tests whether this table provider can guarantee that a scan built via - /// [`Self::scan_with_args`] omits *exactly* the first [`ScanArgs::offset`] - /// rows it would otherwise have produced. - /// - /// Returning `true` is a firm guarantee, not a hint: a caller that - /// pushes a skip into `ScanArgs::offset` may rely on it to avoid - /// re-applying the same skip above the scan. Returning `false` (the - /// default) means `ScanArgs::offset` is ignored or only partially - /// honored, so callers must not assume rows were skipped. + /// Specify if DataFusion should provide the offset to the + /// TableProvider to apply *during* the scan. fn supports_offset_pushdown(&self) -> bool { false } diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index f8a1f4f47af5..f09846f54e90 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -430,15 +430,11 @@ form, or pass a freshly constructed `StatisticsContext` per call. ### `TableProvider::scan_with_args` supports offset pushdown -`TableProvider` gained `scan_with_args(state, args: ScanArgs)`, an -alternative to `scan(state, projection, filters, limit)` that passes the same -information through a structured, extensible `ScanArgs` value and also -carries a new `offset` (number of rows to skip) alongside `limit`. -`scan_with_args` has a default implementation that forwards to `scan`, so +The `ScanArgs` struct now carries a new `offset` (number of rows to skip) field +alongside `limit`. `TableProvider::scan_with_args()` by default forwards to `scan()`, so existing `TableProvider` implementations keep compiling and working unchanged; implementations that want to take advantage of `offset` (or future -`ScanArgs` fields) can override `scan_with_args` instead of, or in addition -to, `scan`. +`ScanArgs` fields) can override `scan_with_args` in addition to `scan`. A new `TableProvider::supports_offset_pushdown() -> bool` (default `false`) lets a provider guarantee it honors `ScanArgs::offset` exactly — that a scan From 1ffbcd90ba4a28c285569045abaf6f1783fbc643 Mon Sep 17 00:00:00 2001 From: Martin Tzvetanov Grigorov Date: Thu, 17 Sep 2026 14:52:24 +0300 Subject: [PATCH 04/10] Optimize imports --- datafusion/core/src/datasource/memory_test.rs | 2 +- datafusion/core/tests/parquet/file_statistics.rs | 1 - datafusion/expr/src/logical_plan/plan.rs | 6 +++--- datafusion/optimizer/src/push_down_filter.rs | 1 - 4 files changed, 4 insertions(+), 6 deletions(-) diff --git a/datafusion/core/src/datasource/memory_test.rs b/datafusion/core/src/datasource/memory_test.rs index 2ff61027d3e1..033a8036f5ec 100644 --- a/datafusion/core/src/datasource/memory_test.rs +++ b/datafusion/core/src/datasource/memory_test.rs @@ -27,7 +27,7 @@ mod tests { use arrow::error::ArrowError; use arrow::record_batch::RecordBatch; use arrow_schema::SchemaRef; - use datafusion_catalog::{ScanArgs, TableProvider}; + use datafusion_catalog::TableProvider; use datafusion_common::{Constraint, Constraints, DataFusionError, Result}; use datafusion_expr::LogicalPlanBuilder; use datafusion_expr::dml::InsertOp; diff --git a/datafusion/core/tests/parquet/file_statistics.rs b/datafusion/core/tests/parquet/file_statistics.rs index 6346ea562e39..f6d733ec6972 100644 --- a/datafusion/core/tests/parquet/file_statistics.rs +++ b/datafusion/core/tests/parquet/file_statistics.rs @@ -28,7 +28,6 @@ use datafusion::datasource::source::DataSourceExec; use datafusion::execution::context::SessionState; use datafusion::execution::session_state::SessionStateBuilder; use datafusion::prelude::{ParquetReadOptions, SessionContext}; -use datafusion_catalog::ScanArgs; use datafusion_common::stats::Precision; use datafusion_common::{DFSchema, TableReference}; use datafusion_execution::cache::cache_manager::{ diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs index c7ce897a8aee..f3f2f26910c3 100644 --- a/datafusion/expr/src/logical_plan/plan.rs +++ b/datafusion/expr/src/logical_plan/plan.rs @@ -3263,7 +3263,7 @@ impl TableScanBuilder { filters: vec![], fetch: None, offset: None, - statistics_requests: BTreeSet::default(), + statistics_requests: BTreeSet::new(), } } @@ -6430,7 +6430,7 @@ mod tests { filters: vec![], fetch: None, offset: None, - statistics_requests: BTreeSet::default(), + statistics_requests: BTreeSet::new(), })); let col = schema.field_names()[0].clone(); @@ -6462,7 +6462,7 @@ mod tests { filters: vec![], fetch: None, offset: None, - statistics_requests: BTreeSet::default(), + statistics_requests: BTreeSet::new(), })); let col = schema.field_names()[0].clone(); diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs index 243f317aab7a..3a4340c81e97 100644 --- a/datafusion/optimizer/src/push_down_filter.rs +++ b/datafusion/optimizer/src/push_down_filter.rs @@ -1435,7 +1435,6 @@ fn expr_columns(exprs: &[Expr]) -> HashSet { #[cfg(test)] mod tests { use std::cmp::Ordering; - use std::collections::BTreeSet; use std::fmt::{Debug, Formatter}; use arrow::datatypes::{Field, Schema, SchemaRef}; From 741424d9f79f1ed9d6d78285f01086eb9e09644b Mon Sep 17 00:00:00 2001 From: Martin Tzvetanov Grigorov Date: Thu, 17 Sep 2026 15:47:05 +0300 Subject: [PATCH 05/10] Add support for pushing down the offset to push_down_limit Box TableScan's statistics_requests field because otherwise it becomes bigger than 176 bytes and `test_size_of_logical_plan()` fails --- datafusion/expr/src/logical_plan/plan.rs | 17 ++- datafusion/optimizer/src/push_down_filter.rs | 2 +- datafusion/optimizer/src/push_down_limit.rs | 63 ++++++++- .../proto-models/proto/datafusion.proto | 12 ++ .../proto-models/src/generated/pbjson.rs | 126 ++++++++++++++++++ .../proto-models/src/generated/prost.rs | 18 +++ datafusion/proto/src/logical_plan/mod.rs | 21 ++- .../tests/cases/roundtrip_logical_plan.rs | 40 ++++++ .../logical_plan/producer/rel/fetch_rel.rs | 25 +++- .../library-user-guide/upgrading/56.0.0.md | 35 +++-- 10 files changed, 331 insertions(+), 28 deletions(-) diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs index f3f2f26910c3..e18c32910d7b 100644 --- a/datafusion/expr/src/logical_plan/plan.rs +++ b/datafusion/expr/src/logical_plan/plan.rs @@ -3141,7 +3141,11 @@ pub struct TableScan { /// surrounding plan (e.g. Min/Max for sort keys). /// /// A [`BTreeSet`], not a `Vec` to keep the resulting plan deterministic. - pub statistics_requests: BTreeSet, + /// + /// Boxed to keep this rarely-populated field from growing every + /// `TableScan` (and thus `LogicalPlan`) by its own size; see + /// `test_size_of_logical_plan`. + pub statistics_requests: Box>, } impl Debug for TableScan { @@ -3247,7 +3251,8 @@ pub struct TableScanBuilder { filters: Vec, fetch: Option, offset: Option, - statistics_requests: BTreeSet, + #[expect(clippy::box_collection)] + statistics_requests: Box>, } impl TableScanBuilder { @@ -3263,7 +3268,7 @@ impl TableScanBuilder { filters: vec![], fetch: None, offset: None, - statistics_requests: BTreeSet::new(), + statistics_requests: Box::default(), } } @@ -3297,7 +3302,7 @@ impl TableScanBuilder { mut self, statistics_requests: BTreeSet, ) -> Self { - self.statistics_requests = statistics_requests; + self.statistics_requests = Box::new(statistics_requests); self } @@ -6430,7 +6435,7 @@ mod tests { filters: vec![], fetch: None, offset: None, - statistics_requests: BTreeSet::new(), + statistics_requests: Box::default(), })); let col = schema.field_names()[0].clone(); @@ -6462,7 +6467,7 @@ mod tests { filters: vec![], fetch: None, offset: None, - statistics_requests: BTreeSet::new(), + statistics_requests: Box::default(), })); let col = schema.field_names()[0].clone(); diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs index 3a4340c81e97..d6bfa5abe110 100644 --- a/datafusion/optimizer/src/push_down_filter.rs +++ b/datafusion/optimizer/src/push_down_filter.rs @@ -3139,7 +3139,7 @@ mod tests { source: Arc::new(test_provider), fetch: None, offset: None, - statistics_requests: std::collections::BTreeSet::new(), + statistics_requests: Box::default(), }); Ok(LogicalPlanBuilder::from(table_scan)) diff --git a/datafusion/optimizer/src/push_down_limit.rs b/datafusion/optimizer/src/push_down_limit.rs index 79c18fbdeb4e..c160140815a0 100644 --- a/datafusion/optimizer/src/push_down_limit.rs +++ b/datafusion/optimizer/src/push_down_limit.rs @@ -125,6 +125,21 @@ fn rewrite_limit(mut limit: Limit) -> Result> { }; match Arc::unwrap_or_clone(limit.input) { + LogicalPlan::TableScan(mut scan) + if skip > 0 && scan.source.supports_offset_pushdown() => + { + // The source guarantees it will omit exactly the first `skip` + // rows itself, so the remaining `Limit` only needs to trim to + // `fetch` — its skip becomes 0. + scan.offset = Some(scan.offset.unwrap_or(0) + skip); + let new_fetch = if fetch != 0 { + scan.fetch.map(|x| min(x, fetch)).or(Some(fetch)) + } else { + Some(0) + }; + scan.fetch = new_fetch; + transformed_limit(0, fetch, LogicalPlan::TableScan(scan)) + } LogicalPlan::TableScan(mut scan) => { let rows_needed = if fetch != 0 { fetch + skip } else { 0 }; let new_fetch = scan @@ -305,10 +320,11 @@ mod test { use crate::test::*; use crate::OptimizerContext; + use arrow::datatypes::{Schema, SchemaRef}; use datafusion_common::DFSchemaRef; use datafusion_expr::{ - Expr, Extension, UserDefinedLogicalNodeCore, col, exists, - logical_plan::builder::LogicalPlanBuilder, + Expr, Extension, TableScanBuilder, TableSource, UserDefinedLogicalNodeCore, col, + exists, logical_plan::builder::LogicalPlanBuilder, }; use datafusion_functions_aggregate::expr_fn::max; @@ -489,6 +505,49 @@ mod test { ) } + /// A `TableSource` that declares it will honor `offset` exactly, so + /// `push_down_limit` is allowed to push `skip` into `TableScan::offset` + /// and elide the outer `Limit`'s skip. + #[derive(Debug)] + struct OffsetPushdownTableSource { + schema: SchemaRef, + } + + impl TableSource for OffsetPushdownTableSource { + fn schema(&self) -> SchemaRef { + Arc::clone(&self.schema) + } + + fn supports_offset_pushdown(&self) -> bool { + true + } + } + + fn offset_pushdown_table_scan() -> Result { + let schema = Arc::new(Schema::new(test_table_scan_fields())); + let source = Arc::new(OffsetPushdownTableSource { schema }); + Ok(LogicalPlan::TableScan( + TableScanBuilder::new("test", source).build()?, + )) + } + + #[test] + fn limit_pushdown_offset_supported() -> Result<()> { + let table_scan = offset_pushdown_table_scan()?; + + let plan = LogicalPlanBuilder::from(table_scan) + .limit(10, Some(1000))? + .build()?; + + assert_optimized_plan_equal!( + plan, + @r" + Limit: skip=0, fetch=1000 + TableScan: test, fetch=1000, offset=10 + " + ) + } + #[test] fn limit_pushdown_multiple_limits() -> Result<()> { let table_scan = test_table_scan()?; diff --git a/datafusion/proto-models/proto/datafusion.proto b/datafusion/proto-models/proto/datafusion.proto index 0f9386969ab3..4a960bb33fb8 100644 --- a/datafusion/proto-models/proto/datafusion.proto +++ b/datafusion/proto-models/proto/datafusion.proto @@ -103,6 +103,10 @@ message ListingTableScanNode { datafusion_common.ArrowFormat arrow = 16; } repeated SortExprNodeCollection file_sort_order = 13; + // Optional number of rows to read. + optional uint64 fetch = 17; + // Optional number of rows to skip. + optional uint64 offset = 18; } message ViewTableScanNode { @@ -112,6 +116,10 @@ message ViewTableScanNode { datafusion_common.Schema schema = 3; ProjectionColumns projection = 4; string definition = 5; + // Optional number of rows to read. + optional uint64 fetch = 7; + // Optional number of rows to skip. + optional uint64 offset = 8; } // Logical Plan to Scan a CustomTableProvider registered at runtime @@ -122,6 +130,10 @@ message CustomTableScanNode { datafusion_common.Schema schema = 3; repeated LogicalExprNode filters = 4; bytes custom_table_data = 5; + // Optional number of rows to read. + optional uint64 fetch = 7; + // Optional number of rows to skip. + optional uint64 offset = 8; } message ProjectionNode { diff --git a/datafusion/proto-models/src/generated/pbjson.rs b/datafusion/proto-models/src/generated/pbjson.rs index d97912d2654d..48552ff0cd84 100644 --- a/datafusion/proto-models/src/generated/pbjson.rs +++ b/datafusion/proto-models/src/generated/pbjson.rs @@ -5704,6 +5704,12 @@ impl serde::Serialize for CustomTableScanNode { if !self.custom_table_data.is_empty() { len += 1; } + if self.fetch.is_some() { + len += 1; + } + if self.offset.is_some() { + len += 1; + } let mut struct_ser = serializer.serialize_struct("datafusion.CustomTableScanNode", len)?; if let Some(v) = self.table_name.as_ref() { struct_ser.serialize_field("tableName", v)?; @@ -5722,6 +5728,16 @@ impl serde::Serialize for CustomTableScanNode { #[allow(clippy::needless_borrows_for_generic_args)] struct_ser.serialize_field("customTableData", pbjson::private::base64::encode(&self.custom_table_data).as_str())?; } + if let Some(v) = self.fetch.as_ref() { + #[allow(clippy::needless_borrow)] + #[allow(clippy::needless_borrows_for_generic_args)] + struct_ser.serialize_field("fetch", ToString::to_string(&v).as_str())?; + } + if let Some(v) = self.offset.as_ref() { + #[allow(clippy::needless_borrow)] + #[allow(clippy::needless_borrows_for_generic_args)] + struct_ser.serialize_field("offset", ToString::to_string(&v).as_str())?; + } struct_ser.end() } } @@ -5739,6 +5755,8 @@ impl<'de> serde::Deserialize<'de> for CustomTableScanNode { "filters", "custom_table_data", "customTableData", + "fetch", + "offset", ]; #[allow(clippy::enum_variant_names)] @@ -5748,6 +5766,8 @@ impl<'de> serde::Deserialize<'de> for CustomTableScanNode { Schema, Filters, CustomTableData, + Fetch, + Offset, } impl<'de> serde::Deserialize<'de> for GeneratedField { fn deserialize(deserializer: D) -> std::result::Result @@ -5774,6 +5794,8 @@ impl<'de> serde::Deserialize<'de> for CustomTableScanNode { "schema" => Ok(GeneratedField::Schema), "filters" => Ok(GeneratedField::Filters), "customTableData" | "custom_table_data" => Ok(GeneratedField::CustomTableData), + "fetch" => Ok(GeneratedField::Fetch), + "offset" => Ok(GeneratedField::Offset), _ => Err(serde::de::Error::unknown_field(value, FIELDS)), } } @@ -5798,6 +5820,8 @@ impl<'de> serde::Deserialize<'de> for CustomTableScanNode { let mut schema__ = None; let mut filters__ = None; let mut custom_table_data__ = None; + let mut fetch__ = None; + let mut offset__ = None; while let Some(k) = map_.next_key()? { match k { GeneratedField::TableName => { @@ -5832,6 +5856,22 @@ impl<'de> serde::Deserialize<'de> for CustomTableScanNode { Some(map_.next_value::<::pbjson::private::BytesDeserialize<_>>()?.0) ; } + GeneratedField::Fetch => { + if fetch__.is_some() { + return Err(serde::de::Error::duplicate_field("fetch")); + } + fetch__ = + map_.next_value::<::std::option::Option<::pbjson::private::NumberDeserialize<_>>>()?.map(|x| x.0) + ; + } + GeneratedField::Offset => { + if offset__.is_some() { + return Err(serde::de::Error::duplicate_field("offset")); + } + offset__ = + map_.next_value::<::std::option::Option<::pbjson::private::NumberDeserialize<_>>>()?.map(|x| x.0) + ; + } } } Ok(CustomTableScanNode { @@ -5840,6 +5880,8 @@ impl<'de> serde::Deserialize<'de> for CustomTableScanNode { schema: schema__, filters: filters__.unwrap_or_default(), custom_table_data: custom_table_data__.unwrap_or_default(), + fetch: fetch__, + offset: offset__, }) } } @@ -13087,6 +13129,12 @@ impl serde::Serialize for ListingTableScanNode { if !self.file_sort_order.is_empty() { len += 1; } + if self.fetch.is_some() { + len += 1; + } + if self.offset.is_some() { + len += 1; + } if self.file_format_type.is_some() { len += 1; } @@ -13115,6 +13163,16 @@ impl serde::Serialize for ListingTableScanNode { if !self.file_sort_order.is_empty() { struct_ser.serialize_field("fileSortOrder", &self.file_sort_order)?; } + if let Some(v) = self.fetch.as_ref() { + #[allow(clippy::needless_borrow)] + #[allow(clippy::needless_borrows_for_generic_args)] + struct_ser.serialize_field("fetch", ToString::to_string(&v).as_str())?; + } + if let Some(v) = self.offset.as_ref() { + #[allow(clippy::needless_borrow)] + #[allow(clippy::needless_borrows_for_generic_args)] + struct_ser.serialize_field("offset", ToString::to_string(&v).as_str())?; + } if let Some(v) = self.file_format_type.as_ref() { match v { listing_table_scan_node::FileFormatType::Csv(v) => { @@ -13156,6 +13214,8 @@ impl<'de> serde::Deserialize<'de> for ListingTableScanNode { "tablePartitionCols", "file_sort_order", "fileSortOrder", + "fetch", + "offset", "csv", "parquet", "avro", @@ -13173,6 +13233,8 @@ impl<'de> serde::Deserialize<'de> for ListingTableScanNode { Filters, TablePartitionCols, FileSortOrder, + Fetch, + Offset, Csv, Parquet, Avro, @@ -13207,6 +13269,8 @@ impl<'de> serde::Deserialize<'de> for ListingTableScanNode { "filters" => Ok(GeneratedField::Filters), "tablePartitionCols" | "table_partition_cols" => Ok(GeneratedField::TablePartitionCols), "fileSortOrder" | "file_sort_order" => Ok(GeneratedField::FileSortOrder), + "fetch" => Ok(GeneratedField::Fetch), + "offset" => Ok(GeneratedField::Offset), "csv" => Ok(GeneratedField::Csv), "parquet" => Ok(GeneratedField::Parquet), "avro" => Ok(GeneratedField::Avro), @@ -13239,6 +13303,8 @@ impl<'de> serde::Deserialize<'de> for ListingTableScanNode { let mut filters__ = None; let mut table_partition_cols__ = None; let mut file_sort_order__ = None; + let mut fetch__ = None; + let mut offset__ = None; let mut file_format_type__ = None; while let Some(k) = map_.next_key()? { match k { @@ -13290,6 +13356,22 @@ impl<'de> serde::Deserialize<'de> for ListingTableScanNode { } file_sort_order__ = Some(map_.next_value()?); } + GeneratedField::Fetch => { + if fetch__.is_some() { + return Err(serde::de::Error::duplicate_field("fetch")); + } + fetch__ = + map_.next_value::<::std::option::Option<::pbjson::private::NumberDeserialize<_>>>()?.map(|x| x.0) + ; + } + GeneratedField::Offset => { + if offset__.is_some() { + return Err(serde::de::Error::duplicate_field("offset")); + } + offset__ = + map_.next_value::<::std::option::Option<::pbjson::private::NumberDeserialize<_>>>()?.map(|x| x.0) + ; + } GeneratedField::Csv => { if file_format_type__.is_some() { return Err(serde::de::Error::duplicate_field("csv")); @@ -13336,6 +13418,8 @@ impl<'de> serde::Deserialize<'de> for ListingTableScanNode { filters: filters__.unwrap_or_default(), table_partition_cols: table_partition_cols__.unwrap_or_default(), file_sort_order: file_sort_order__.unwrap_or_default(), + fetch: fetch__, + offset: offset__, file_format_type: file_format_type__, }) } @@ -29202,6 +29286,12 @@ impl serde::Serialize for ViewTableScanNode { if !self.definition.is_empty() { len += 1; } + if self.fetch.is_some() { + len += 1; + } + if self.offset.is_some() { + len += 1; + } let mut struct_ser = serializer.serialize_struct("datafusion.ViewTableScanNode", len)?; if let Some(v) = self.table_name.as_ref() { struct_ser.serialize_field("tableName", v)?; @@ -29218,6 +29308,16 @@ impl serde::Serialize for ViewTableScanNode { if !self.definition.is_empty() { struct_ser.serialize_field("definition", &self.definition)?; } + if let Some(v) = self.fetch.as_ref() { + #[allow(clippy::needless_borrow)] + #[allow(clippy::needless_borrows_for_generic_args)] + struct_ser.serialize_field("fetch", ToString::to_string(&v).as_str())?; + } + if let Some(v) = self.offset.as_ref() { + #[allow(clippy::needless_borrow)] + #[allow(clippy::needless_borrows_for_generic_args)] + struct_ser.serialize_field("offset", ToString::to_string(&v).as_str())?; + } struct_ser.end() } } @@ -29234,6 +29334,8 @@ impl<'de> serde::Deserialize<'de> for ViewTableScanNode { "schema", "projection", "definition", + "fetch", + "offset", ]; #[allow(clippy::enum_variant_names)] @@ -29243,6 +29345,8 @@ impl<'de> serde::Deserialize<'de> for ViewTableScanNode { Schema, Projection, Definition, + Fetch, + Offset, } impl<'de> serde::Deserialize<'de> for GeneratedField { fn deserialize(deserializer: D) -> std::result::Result @@ -29269,6 +29373,8 @@ impl<'de> serde::Deserialize<'de> for ViewTableScanNode { "schema" => Ok(GeneratedField::Schema), "projection" => Ok(GeneratedField::Projection), "definition" => Ok(GeneratedField::Definition), + "fetch" => Ok(GeneratedField::Fetch), + "offset" => Ok(GeneratedField::Offset), _ => Err(serde::de::Error::unknown_field(value, FIELDS)), } } @@ -29293,6 +29399,8 @@ impl<'de> serde::Deserialize<'de> for ViewTableScanNode { let mut schema__ = None; let mut projection__ = None; let mut definition__ = None; + let mut fetch__ = None; + let mut offset__ = None; while let Some(k) = map_.next_key()? { match k { GeneratedField::TableName => { @@ -29325,6 +29433,22 @@ impl<'de> serde::Deserialize<'de> for ViewTableScanNode { } definition__ = Some(map_.next_value()?); } + GeneratedField::Fetch => { + if fetch__.is_some() { + return Err(serde::de::Error::duplicate_field("fetch")); + } + fetch__ = + map_.next_value::<::std::option::Option<::pbjson::private::NumberDeserialize<_>>>()?.map(|x| x.0) + ; + } + GeneratedField::Offset => { + if offset__.is_some() { + return Err(serde::de::Error::duplicate_field("offset")); + } + offset__ = + map_.next_value::<::std::option::Option<::pbjson::private::NumberDeserialize<_>>>()?.map(|x| x.0) + ; + } } } Ok(ViewTableScanNode { @@ -29333,6 +29457,8 @@ impl<'de> serde::Deserialize<'de> for ViewTableScanNode { schema: schema__, projection: projection__, definition: definition__.unwrap_or_default(), + fetch: fetch__, + offset: offset__, }) } } diff --git a/datafusion/proto-models/src/generated/prost.rs b/datafusion/proto-models/src/generated/prost.rs index f8cf7f82f425..ffdf3df5c57f 100644 --- a/datafusion/proto-models/src/generated/prost.rs +++ b/datafusion/proto-models/src/generated/prost.rs @@ -123,6 +123,12 @@ pub struct ListingTableScanNode { pub table_partition_cols: ::prost::alloc::vec::Vec, #[prost(message, repeated, tag = "13")] pub file_sort_order: ::prost::alloc::vec::Vec, + /// Optional number of rows to read. + #[prost(uint64, optional, tag = "17")] + pub fetch: ::core::option::Option, + /// Optional number of rows to skip. + #[prost(uint64, optional, tag = "18")] + pub offset: ::core::option::Option, #[prost( oneof = "listing_table_scan_node::FileFormatType", tags = "10, 11, 12, 15, 16" @@ -159,6 +165,12 @@ pub struct ViewTableScanNode { pub projection: ::core::option::Option, #[prost(string, tag = "5")] pub definition: ::prost::alloc::string::String, + /// Optional number of rows to read. + #[prost(uint64, optional, tag = "7")] + pub fetch: ::core::option::Option, + /// Optional number of rows to skip. + #[prost(uint64, optional, tag = "8")] + pub offset: ::core::option::Option, } /// Logical Plan to Scan a CustomTableProvider registered at runtime #[derive(Clone, PartialEq, ::prost::Message)] @@ -173,6 +185,12 @@ pub struct CustomTableScanNode { pub filters: ::prost::alloc::vec::Vec, #[prost(bytes = "vec", tag = "5")] pub custom_table_data: ::prost::alloc::vec::Vec, + /// Optional number of rows to read. + #[prost(uint64, optional, tag = "7")] + pub fetch: ::core::option::Option, + /// Optional number of rows to skip. + #[prost(uint64, optional, tag = "8")] + pub offset: ::core::option::Option, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct ProjectionNode { diff --git a/datafusion/proto/src/logical_plan/mod.rs b/datafusion/proto/src/logical_plan/mod.rs index 37b42fc9a3ba..9db9e8c9525a 100644 --- a/datafusion/proto/src/logical_plan/mod.rs +++ b/datafusion/proto/src/logical_plan/mod.rs @@ -696,11 +696,13 @@ impl AsLogicalPlan for LogicalPlanNode { projection = Some(column_indices); } - LogicalPlanBuilder::scan_with_filters( + LogicalPlanBuilder::scan_with_filters_fetch_offset( table_name, provider_as_source(Arc::new(provider)), projection, filters, + scan.fetch.map(|f| f as usize), + scan.offset.map(|o| o as usize), )? .build() } @@ -730,11 +732,13 @@ impl AsLogicalPlan for LogicalPlanNode { ctx, )?; - LogicalPlanBuilder::scan_with_filters( + LogicalPlanBuilder::scan_with_filters_fetch_offset( table_name, provider_as_source(provider), projection, filters, + scan.fetch.map(|f| f as usize), + scan.offset.map(|o| o as usize), )? .build() } @@ -1210,10 +1214,13 @@ impl AsLogicalPlan for LogicalPlanNode { let table_name = from_table_reference(scan.table_name.as_ref(), "ViewScan")?; - LogicalPlanBuilder::scan( + LogicalPlanBuilder::scan_with_filters_fetch_offset( table_name, provider_as_source(Arc::new(provider)), projection, + vec![], + scan.fetch.map(|f| f as usize), + scan.offset.map(|o| o as usize), )? .build() } @@ -1408,6 +1415,8 @@ impl AsLogicalPlan for LogicalPlanNode { source, filters, projection, + fetch, + offset, .. }) => { let provider = source_as_provider(source)?; @@ -1540,6 +1549,8 @@ impl AsLogicalPlan for LogicalPlanNode { projection, filters, file_sort_order: exprs_vec, + fetch: fetch.map(|f| f as u64), + offset: offset.map(|o| o as u64), }, )), }) @@ -1563,6 +1574,8 @@ impl AsLogicalPlan for LogicalPlanNode { .definition() .map(|s| s.to_string()) .unwrap_or_default(), + fetch: fetch.map(|f| f as u64), + offset: offset.map(|o| o as u64), }, ))), }) @@ -1610,6 +1623,8 @@ impl AsLogicalPlan for LogicalPlanNode { schema: Some(schema), filters, custom_table_data: bytes, + fetch: fetch.map(|f| f as u64), + offset: offset.map(|o| o as u64), }); let node = LogicalPlanNode { logical_plan_type: Some(scan), diff --git a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs index b4c121ef3071..37735e779e1f 100644 --- a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs @@ -617,6 +617,46 @@ async fn roundtrip_logical_plan_sort() -> Result<()> { Ok(()) } +#[tokio::test] +async fn roundtrip_logical_plan_limit_offset() -> Result<()> { + let ctx = SessionContext::new(); + + let schema = Schema::new(vec![ + Field::new("a", DataType::Int64, true), + Field::new("b", DataType::Decimal128(15, 2), true), + ]); + + ctx.register_csv( + "t1", + "tests/testdata/test.csv", + CsvReadOptions::default().schema(&schema), + ) + .await?; + + let query = "SELECT a, b FROM t1 LIMIT 5 OFFSET 3"; + let plan = ctx.sql(query).await?.into_optimized_plan()?; + + // Sanity check that the offset was actually pushed into the `TableScan` + // (ListingTable opts into `supports_offset_pushdown`), so this test + // exercises the new `ListingTableScanNode.offset` wire field rather than + // trivially passing because nothing needed to round-trip. + let plan_str = plan.to_string(); + assert!( + plan_str.contains("offset=3"), + "expected offset to be pushed into the scan, got: {plan_str}" + ); + assert!( + plan_str.contains("limit=5"), + "expected limit to be pushed into the scan, got: {plan_str}" + ); + + let bytes = logical_plan_to_bytes(&plan)?; + let logical_round_trip = logical_plan_from_bytes(&bytes, &ctx.task_ctx())?; + assert_eq!(plan_str, logical_round_trip.to_string()); + + Ok(()) +} + #[tokio::test] async fn roundtrip_logical_plan_dml() -> Result<()> { let ctx = SessionContext::new(); diff --git a/datafusion/substrait/src/logical_plan/producer/rel/fetch_rel.rs b/datafusion/substrait/src/logical_plan/producer/rel/fetch_rel.rs index e878b3816ff4..5d7b12855dd8 100644 --- a/datafusion/substrait/src/logical_plan/producer/rel/fetch_rel.rs +++ b/datafusion/substrait/src/logical_plan/producer/rel/fetch_rel.rs @@ -17,7 +17,7 @@ use crate::logical_plan::producer::SubstraitProducer; use datafusion::common::DFSchema; -use datafusion::logical_expr::Limit; +use datafusion::logical_expr::{Limit, LogicalPlan, lit}; use std::sync::Arc; use substrait::proto::rel::RelType; use substrait::proto::{FetchRel, Rel, fetch_rel}; @@ -28,10 +28,25 @@ pub fn from_limit( ) -> datafusion::common::Result> { let input = producer.handle_plan(limit.input.as_ref())?; let empty_schema = Arc::new(DFSchema::empty()); - let offset_mode = limit - .skip - .as_ref() - .map(|expr| producer.handle_expr(expr.as_ref(), &empty_schema)) + + // A provider-level offset pushdown (see `push_down_limit`) may have moved + // some or all of this `Limit`'s skip into the child `TableScan::offset`, + // reducing `limit.skip` accordingly (down to `None`/0 when the scan + // handles the whole skip itself). Substrait's `ReadRel` has no field of + // its own for it, so it must be folded back into this `FetchRel`'s + // offset — otherwise it is silently lost when the plan is serialized. + let scan_offset = match limit.input.as_ref() { + LogicalPlan::TableScan(scan) => scan.offset, + _ => None, + }; + let skip_expr = match (limit.skip.as_deref(), scan_offset) { + (Some(skip), Some(offset)) => Some(skip.clone() + lit(offset as i64)), + (Some(skip), None) => Some(skip.clone()), + (None, Some(offset)) => Some(lit(offset as i64)), + (None, None) => None, + }; + let offset_mode = skip_expr + .map(|expr| producer.handle_expr(&expr, &empty_schema)) .transpose()? .map(Box::new) .map(fetch_rel::OffsetMode::OffsetExpr); diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index ac023dd57c90..d401c4fd6a37 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -463,24 +463,37 @@ built with `offset` omits _exactly_ the first `offset` rows it would otherwise have produced, not just approximately. `ListingTable` overrides this to `true` and honors `offset`. -Note that DataFusion's SQL/DataFrame optimizer does not yet populate -`LogicalPlan::TableScan::offset` from a query's `LIMIT ... OFFSET ...` clause -(that skip is still always enforced above the scan, as before); `offset` is -reachable today by constructing a `TableScan` directly (for example via -`LogicalPlanBuilder::scan_with_filters_fetch_offset`) or by calling -`scan_with_args` directly with `ScanArgs::with_offset`. Wiring the optimizer -to push a SQL `OFFSET` into the scan is tracked as follow-up work, since doing -so safely requires the pushdown to not be silently lost when the resulting -logical plan is serialized (e.g. via `datafusion-substrait` or -`datafusion-proto`) by a consumer unaware of the new field. +DataFusion's SQL/DataFrame optimizer (`push_down_limit`) now populates +`LogicalPlan::TableScan::offset` from a query's `LIMIT ... OFFSET ...` clause, +but only when the scan's source reports `supports_offset_pushdown() == true`. +In that case the skip is moved entirely into the scan and the remaining +`Limit` above it no longer re-applies it; every other provider keeps today's +behavior unchanged, with the skip always enforced above the scan. `offset` is +also reachable directly, independent of the optimizer, by constructing a +`TableScan` (for example via `LogicalPlanBuilder::scan_with_filters_fetch_offset`) +or by calling `scan_with_args` with `ScanArgs::with_offset`. + +This pushdown is also preserved end-to-end through plan serialization: +`datafusion-substrait`'s producer folds a scan's `offset` into the `FetchRel` +it emits for the wrapping `Limit` (Substrait's `ReadRel` has no field of its +own for it), and `datafusion-proto`'s `ListingTableScanNode`, `ViewTableScanNode`, +and `CustomTableScanNode` messages gained optional `fetch`/`offset` fields so a +`TableScan`'s own `fetch`/`offset` — previously silently dropped — now +round-trips exactly. **Who is affected:** - Custom `TableProvider` implementations that want offset pushdown: override `scan_with_args` to read `ScanArgs::offset()` and honor it exactly, and - override `supports_offset_pushdown` to return `true`. + override `supports_offset_pushdown` to return `true`. Once you do, + `LIMIT ... OFFSET ...` queries against that provider will have their skip pushed + into the scan automatically. - Callers that already build a scan through `ScanArgs`/`scan_with_args` (for example custom query planners) can now also set `with_offset`. +- Anything reading `datafusion-proto`'s `ListingTableScanNode`, + `ViewTableScanNode`, or `CustomTableScanNode` messages directly (rather than + through the `datafusion-proto` Rust API) should be aware of the new + `fetch`/`offset` fields. **Example:** From 513630dacbcdba041881c4ea33564afcbbd103a4 Mon Sep 17 00:00:00 2001 From: Martin Tzvetanov Grigorov Date: Thu, 17 Sep 2026 15:54:27 +0300 Subject: [PATCH 06/10] prettier --- docs/source/library-user-guide/upgrading/56.0.0.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index d401c4fd6a37..2c1f88aa9dc2 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -485,7 +485,7 @@ round-trips exactly. - Custom `TableProvider` implementations that want offset pushdown: override `scan_with_args` to read `ScanArgs::offset()` and honor it exactly, and - override `supports_offset_pushdown` to return `true`. Once you do, + override `supports_offset_pushdown` to return `true`. Once you do, `LIMIT ... OFFSET ...` queries against that provider will have their skip pushed into the scan automatically. - Callers that already build a scan through `ScanArgs`/`scan_with_args` (for From 861a55f1f81211171a967d6533a428327d4b1d6b Mon Sep 17 00:00:00 2001 From: Martin Tzvetanov Grigorov Date: Thu, 17 Sep 2026 16:32:00 +0300 Subject: [PATCH 07/10] Prevent usize overflows, fix test assertion, fix example in a comment --- datafusion/catalog-listing/src/table.rs | 2 +- datafusion/optimizer/src/push_down_limit.rs | 6 ++++-- datafusion/proto/src/lib.rs | 2 +- datafusion/proto/tests/cases/roundtrip_logical_plan.rs | 2 +- datafusion/session/src/table.rs | 5 +++++ datafusion/sql/src/unparser/plan.rs | 9 +++++++-- docs/source/library-user-guide/upgrading/56.0.0.md | 6 +----- 7 files changed, 20 insertions(+), 12 deletions(-) diff --git a/datafusion/catalog-listing/src/table.rs b/datafusion/catalog-listing/src/table.rs index 086a3e30430e..7aa68801901b 100644 --- a/datafusion/catalog-listing/src/table.rs +++ b/datafusion/catalog-listing/src/table.rs @@ -602,7 +602,7 @@ impl ListingTable { 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 + offset.unwrap_or(0)); + let inflated_limit = limit.map(|l| l.saturating_add(offset.unwrap_or(0))); // extract types of partition columns let table_partition_cols = self diff --git a/datafusion/optimizer/src/push_down_limit.rs b/datafusion/optimizer/src/push_down_limit.rs index c160140815a0..be1ae5edf886 100644 --- a/datafusion/optimizer/src/push_down_limit.rs +++ b/datafusion/optimizer/src/push_down_limit.rs @@ -131,9 +131,11 @@ fn rewrite_limit(mut limit: Limit) -> Result> { // The source guarantees it will omit exactly the first `skip` // rows itself, so the remaining `Limit` only needs to trim to // `fetch` — its skip becomes 0. - scan.offset = Some(scan.offset.unwrap_or(0) + skip); + scan.offset = Some(scan.offset.unwrap_or(0).saturating_add(skip)); let new_fetch = if fetch != 0 { - scan.fetch.map(|x| min(x, fetch)).or(Some(fetch)) + scan.fetch + .map(|existing_fetch| min(existing_fetch.saturating_sub(skip), fetch)) + .or(Some(fetch)) } else { Some(0) }; diff --git a/datafusion/proto/src/lib.rs b/datafusion/proto/src/lib.rs index 71feae506dc6..7a965fcebeb4 100644 --- a/datafusion/proto/src/lib.rs +++ b/datafusion/proto/src/lib.rs @@ -65,7 +65,7 @@ //! # use datafusion_expr::{col, lit, Expr}; //! # use datafusion_proto::bytes::Serializeable; //! # fn main() -> Result<()>{ -//! // Create a new `Expr` a < 32 +//! // Create a new `Expr` a < 5 //! let expr = col("a").lt(lit(5i32)); //! //! // Convert it to bytes (for sending over the network, etc.) diff --git a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs index 37735e779e1f..5e5a7e455610 100644 --- a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs @@ -646,7 +646,7 @@ async fn roundtrip_logical_plan_limit_offset() -> Result<()> { "expected offset to be pushed into the scan, got: {plan_str}" ); assert!( - plan_str.contains("limit=5"), + plan_str.contains("fetch=5"), "expected limit to be pushed into the scan, got: {plan_str}" ); diff --git a/datafusion/session/src/table.rs b/datafusion/session/src/table.rs index d138ac61157a..02c3ab763c04 100644 --- a/datafusion/session/src/table.rs +++ b/datafusion/session/src/table.rs @@ -238,6 +238,11 @@ pub trait TableProvider: Any + Debug + Sync + Send { /// Specify if DataFusion should provide the offset to the /// TableProvider to apply *during* the scan. + /// + /// # Note + /// + /// Make sure [`TableProvider::scan_with_args`] is overridden too + /// and [`ScanArgs::offset`] is used! fn supports_offset_pushdown(&self) -> bool { false } diff --git a/datafusion/sql/src/unparser/plan.rs b/datafusion/sql/src/unparser/plan.rs index 18af08fc1836..a961bfa46c1c 100644 --- a/datafusion/sql/src/unparser/plan.rs +++ b/datafusion/sql/src/unparser/plan.rs @@ -2585,8 +2585,13 @@ impl Unparser<'_> { builder = builder.filter(filter)?; } - if let Some(fetch) = table_scan.fetch { - builder = builder.limit(0, Some(fetch))?; + match (table_scan.offset, table_scan.fetch) { + (Some(offset), Some(fetch)) => { + builder = builder.limit(offset, Some(fetch))? + } + (Some(offset), None) => builder = builder.limit(offset, None)?, + (None, Some(fetch)) => builder = builder.limit(0, Some(fetch))?, + (None, None) => (), } // If the table scan has an alias but no projection or filters, it means no column references are rebased. diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index 2c1f88aa9dc2..fcd9f4e9b33f 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -302,7 +302,7 @@ part of `StatisticsRegistry::default_with_builtin_providers()`. Use the walk instead: -```rust +```rust,ignore StatisticsContext::new_with_registry(registry) .compute_extended(plan, &StatisticsArgs::new())?; // or .compute(...) for core Statistics ``` @@ -490,10 +490,6 @@ round-trips exactly. into the scan automatically. - Callers that already build a scan through `ScanArgs`/`scan_with_args` (for example custom query planners) can now also set `with_offset`. -- Anything reading `datafusion-proto`'s `ListingTableScanNode`, - `ViewTableScanNode`, or `CustomTableScanNode` messages directly (rather than - through the `datafusion-proto` Rust API) should be aware of the new - `fetch`/`offset` fields. **Example:** From 16d6669782c5485bfec01fb7629377637269585a Mon Sep 17 00:00:00 2001 From: Martin Tzvetanov Grigorov Date: Fri, 18 Sep 2026 14:29:39 +0300 Subject: [PATCH 08/10] Minor improvements and adding comments --- .../core/src/datasource/listing/table.rs | 68 +++++++++---------- datafusion/expr/src/logical_plan/plan.rs | 8 +-- .../tests/cases/roundtrip_logical_plan.rs | 10 +-- .../library-user-guide/upgrading/56.0.0.md | 16 ++++- 4 files changed, 55 insertions(+), 47 deletions(-) diff --git a/datafusion/core/src/datasource/listing/table.rs b/datafusion/core/src/datasource/listing/table.rs index 5d70a4d1de09..341c27ff1392 100644 --- a/datafusion/core/src/datasource/listing/table.rs +++ b/datafusion/core/src/datasource/listing/table.rs @@ -318,41 +318,41 @@ mod tests { #[cfg(feature = "parquet")] #[tokio::test] async fn scan_with_args_offset_skips_correct_rows() -> Result<()> { - let ctx = SessionContext::new_with_config( - SessionConfig::new() - .with_collect_statistics(true) - .with_target_partitions(1), - ); - - 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])); + 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(()) } diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs index e18c32910d7b..48ea12f567d8 100644 --- a/datafusion/expr/src/logical_plan/plan.rs +++ b/datafusion/expr/src/logical_plan/plan.rs @@ -3142,9 +3142,9 @@ pub struct TableScan { /// /// A [`BTreeSet`], not a `Vec` to keep the resulting plan deterministic. /// - /// Boxed to keep this rarely-populated field from growing every - /// `TableScan` (and thus `LogicalPlan`) by its own size; see - /// `test_size_of_logical_plan`. + // Boxed to keep this rarely-populated field from growing every + // `TableScan` (and thus `LogicalPlan`) by its own size; + // see `test_size_of_logical_plan`. pub statistics_requests: Box>, } @@ -3251,7 +3251,7 @@ pub struct TableScanBuilder { filters: Vec, fetch: Option, offset: Option, - #[expect(clippy::box_collection)] + #[expect(clippy::box_collection)] // additional indirection for smaller size_of() statistics_requests: Box>, } diff --git a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs index 8f8c6f0f79ec..e68bde0b34bf 100644 --- a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs @@ -641,13 +641,9 @@ async fn roundtrip_logical_plan_limit_offset() -> Result<()> { // exercises the new `ListingTableScanNode.offset` wire field rather than // trivially passing because nothing needed to round-trip. let plan_str = plan.to_string(); - assert!( - plan_str.contains("offset=3"), - "expected offset to be pushed into the scan, got: {plan_str}" - ); - assert!( - plan_str.contains("fetch=5"), - "expected limit to be pushed into the scan, got: {plan_str}" + assert_eq!( + plan_str, "Limit: skip=0, fetch=5\n TableScan: t1 projection=[a, b], fetch=5, offset=3", + "expected 'fetch=5' and 'offset=3' to be pushed into the scan, got: {plan_str}" ); let bytes = logical_plan_to_bytes(&plan)?; diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index 45e5cb8f7eba..266abb9ccddd 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -545,7 +545,7 @@ and `CustomTableScanNode` messages gained optional `fetch`/`offset` fields so a `TableScan`'s own `fetch`/`offset` — previously silently dropped — now round-trips exactly. -**Who is affected:** +#### Who is affected - Custom `TableProvider` implementations that want offset pushdown: override `scan_with_args` to read `ScanArgs::offset()` and honor it exactly, and @@ -555,7 +555,19 @@ round-trips exactly. - Callers that already build a scan through `ScanArgs`/`scan_with_args` (for example custom query planners) can now also set `with_offset`. -**Example:** +#### API breaks + +The _public_ `TableScan` struct has one new _public_ field - `offset` and +one field whose type is changed - `statistics_requests`. The addition of `offset` +field increases the size of `TableScan` and thus the size of `LogicalPlan` enum. To keep +the size of `LogicalPlan` reasonable the type of the `TableScan::statistics_requests` +field is changed from `BTreeSet` (24 bytes) to +`Box>` (8 bytes). +The users' applications are recommended to use `TableScanBuilder` to construct new +instances of `TableScan` instead of instantiating directly `TableScan { ... }` to avoid +the API breaks. + +#### Example ```rust,ignore let plan = provider From 4b67a6eddf02b91a3924e9b6e37775449be62557 Mon Sep 17 00:00:00 2001 From: Martin Tzvetanov Grigorov Date: Fri, 18 Sep 2026 14:59:26 +0300 Subject: [PATCH 09/10] Add SqlLogicTests for the offset pushdown functionality --- .../test_files/offset_pushdown.slt | 752 ++++++++++++++++++ 1 file changed, 752 insertions(+) create mode 100644 datafusion/sqllogictest/test_files/offset_pushdown.slt diff --git a/datafusion/sqllogictest/test_files/offset_pushdown.slt b/datafusion/sqllogictest/test_files/offset_pushdown.slt new file mode 100644 index 000000000000..67820eed9744 --- /dev/null +++ b/datafusion/sqllogictest/test_files/offset_pushdown.slt @@ -0,0 +1,752 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +########## +## OFFSET pushdown tests +## +## `push_down_limit` pushes a `LIMIT ... OFFSET ...` skip into +## `TableScan::offset` whenever the underlying `TableSource` reports +## `supports_offset_pushdown() == true` (today: `ListingTable`, i.e. +## CSV/JSON/Parquet/Arrow files). Providers that do not opt in (e.g. the +## in-memory `MemTable` created by `CREATE TABLE ... AS VALUES`) keep the +## skip in a `Limit` node above the scan instead. +########## + +# Source data: 20 rows, split into 5 files of 4 rows each so that OFFSET +# values can be chosen to land mid-file and to span a file boundary. +statement ok +CREATE TABLE offset_src AS +SELECT i AS id, i * 10 AS val FROM generate_series(0, 19) t(i); + +query II +SELECT id, val FROM offset_src ORDER BY id +---- +0 0 +1 10 +2 20 +3 30 +4 40 +5 50 +6 60 +7 70 +8 80 +9 90 +10 100 +11 110 +12 120 +13 130 +14 140 +15 150 +16 160 +17 170 +18 180 +19 190 + +# File 0: ids 0-3 +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 0 AND 3) +TO 'test_files/scratch/offset_pushdown/csv/part-0.csv' +STORED AS CSV OPTIONS ('format.has_header' 'true'); +---- +4 + +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 0 AND 3) +TO 'test_files/scratch/offset_pushdown/json/part-0.json' +STORED AS JSON; +---- +4 + +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 0 AND 3) +TO 'test_files/scratch/offset_pushdown/parquet/part-0.parquet' +STORED AS PARQUET; +---- +4 + +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 0 AND 3) +TO 'test_files/scratch/offset_pushdown/arrow/part-0.arrow' +STORED AS ARROW; +---- +4 + +# File 1: ids 4-7 +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 4 AND 7) +TO 'test_files/scratch/offset_pushdown/csv/part-1.csv' +STORED AS CSV OPTIONS ('format.has_header' 'true'); +---- +4 + +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 4 AND 7) +TO 'test_files/scratch/offset_pushdown/json/part-1.json' +STORED AS JSON; +---- +4 + +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 4 AND 7) +TO 'test_files/scratch/offset_pushdown/parquet/part-1.parquet' +STORED AS PARQUET; +---- +4 + +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 4 AND 7) +TO 'test_files/scratch/offset_pushdown/arrow/part-1.arrow' +STORED AS ARROW; +---- +4 + +# File 2: ids 8-11 +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 8 AND 11) +TO 'test_files/scratch/offset_pushdown/csv/part-2.csv' +STORED AS CSV OPTIONS ('format.has_header' 'true'); +---- +4 + +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 8 AND 11) +TO 'test_files/scratch/offset_pushdown/json/part-2.json' +STORED AS JSON; +---- +4 + +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 8 AND 11) +TO 'test_files/scratch/offset_pushdown/parquet/part-2.parquet' +STORED AS PARQUET; +---- +4 + +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 8 AND 11) +TO 'test_files/scratch/offset_pushdown/arrow/part-2.arrow' +STORED AS ARROW; +---- +4 + +# File 3: ids 12-15 +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 12 AND 15) +TO 'test_files/scratch/offset_pushdown/csv/part-3.csv' +STORED AS CSV OPTIONS ('format.has_header' 'true'); +---- +4 + +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 12 AND 15) +TO 'test_files/scratch/offset_pushdown/json/part-3.json' +STORED AS JSON; +---- +4 + +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 12 AND 15) +TO 'test_files/scratch/offset_pushdown/parquet/part-3.parquet' +STORED AS PARQUET; +---- +4 + +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 12 AND 15) +TO 'test_files/scratch/offset_pushdown/arrow/part-3.arrow' +STORED AS ARROW; +---- +4 + +# File 4: ids 16-19 +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 16 AND 19) +TO 'test_files/scratch/offset_pushdown/csv/part-4.csv' +STORED AS CSV OPTIONS ('format.has_header' 'true'); +---- +4 + +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 16 AND 19) +TO 'test_files/scratch/offset_pushdown/json/part-4.json' +STORED AS JSON; +---- +4 + +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 16 AND 19) +TO 'test_files/scratch/offset_pushdown/parquet/part-4.parquet' +STORED AS PARQUET; +---- +4 + +query I +COPY (SELECT * FROM offset_src WHERE id BETWEEN 16 AND 19) +TO 'test_files/scratch/offset_pushdown/arrow/part-4.arrow' +STORED AS ARROW; +---- +4 + +statement ok +CREATE EXTERNAL TABLE offset_csv (id BIGINT, val BIGINT) +STORED AS CSV +LOCATION 'test_files/scratch/offset_pushdown/csv/' +OPTIONS ('format.has_header' 'true'); + +statement ok +CREATE EXTERNAL TABLE offset_json (id BIGINT, val BIGINT) +STORED AS JSON +LOCATION 'test_files/scratch/offset_pushdown/json/'; + +statement ok +CREATE EXTERNAL TABLE offset_parquet (id BIGINT, val BIGINT) +STORED AS PARQUET +LOCATION 'test_files/scratch/offset_pushdown/parquet/'; + +statement ok +CREATE EXTERNAL TABLE offset_arrow (id BIGINT, val BIGINT) +STORED AS ARROW +LOCATION 'test_files/scratch/offset_pushdown/arrow/'; + +#################### +# supports_offset_pushdown(): MemTable does not opt in, so the skip stays +# on the `Limit` node and `TableScan` gets no `offset` attribute. +#################### + +query TT +EXPLAIN SELECT id FROM offset_src OFFSET 5 LIMIT 3 +---- +logical_plan +01)Limit: skip=5, fetch=3 +02)--TableScan: offset_src projection=[id], fetch=8 +physical_plan +01)GlobalLimitExec: skip=5, fetch=3 +02)--CoalescePartitionsExec: fetch=8 +03)----DataSourceExec: partitions=4, partition_sizes=[1, 0, 0, 0], fetch=8 + +#################### +# supports_offset_pushdown(): ListingTable (CSV/JSON/Parquet/Arrow) opts +# in, so the skip is folded into `TableScan::offset` and the outer `Limit` +# no longer needs to skip anything. +#################### + +statement ok +set datafusion.explain.logical_plan_only = true; + +query TT +EXPLAIN SELECT id FROM offset_csv OFFSET 5 LIMIT 3 +---- +logical_plan +01)Limit: skip=0, fetch=3 +02)--TableScan: offset_csv projection=[id], fetch=3, offset=5 + +query TT +EXPLAIN SELECT id FROM offset_json OFFSET 5 LIMIT 3 +---- +logical_plan +01)Limit: skip=0, fetch=3 +02)--TableScan: offset_json projection=[id], fetch=3, offset=5 + +query TT +EXPLAIN SELECT id FROM offset_parquet OFFSET 5 LIMIT 3 +---- +logical_plan +01)Limit: skip=0, fetch=3 +02)--TableScan: offset_parquet projection=[id], fetch=3, offset=5 + +query TT +EXPLAIN SELECT id FROM offset_arrow OFFSET 5 LIMIT 3 +---- +logical_plan +01)Limit: skip=0, fetch=3 +02)--TableScan: offset_arrow projection=[id], fetch=3, offset=5 + +statement ok +reset datafusion.explain.logical_plan_only; + +# Physical plan with a single partition: the offset is applied by a +# `GlobalLimitExec` wrapped directly around the file scan (built inside +# `ListingTable::scan_with_args`). +# +# NB: `OFFSET 15 LIMIT 10` is chosen so that `offset + limit` (25) exceeds +# the table's 20 rows. `ListingTable` stops *listing* files once file-level +# statistics show enough rows to satisfy a smaller limit (see +# `get_files_with_limit`), and which files that leaves out is not +# deterministic (files are discovered concurrently). Requiring all rows +# keeps the file set -- and thus this plan -- deterministic. +statement ok +set datafusion.execution.target_partitions = 1; + +query TT +EXPLAIN SELECT id FROM offset_parquet OFFSET 15 LIMIT 10 +---- +logical_plan +01)Limit: skip=0, fetch=10 +02)--TableScan: offset_parquet projection=[id], fetch=10, offset=15 +physical_plan +01)GlobalLimitExec: skip=15, fetch=10 +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/offset_pushdown/parquet/part-0.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/offset_pushdown/parquet/part-1.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/offset_pushdown/parquet/part-2.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/offset_pushdown/parquet/part-3.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/offset_pushdown/parquet/part-4.parquet]]}, projection=[id], limit=25, file_type=parquet + +# Physical plan with multiple partitions: `GlobalLimitExec` requires a +# single input partition, so a `CoalescePartitionsExec` is inserted below +# it to merge the per-file partitions before the skip is applied. +statement ok +set datafusion.execution.target_partitions = 4; + +query TT +EXPLAIN SELECT id FROM offset_parquet OFFSET 15 LIMIT 10 +---- +logical_plan +01)Limit: skip=0, fetch=10 +02)--TableScan: offset_parquet projection=[id], fetch=10, offset=15 +physical_plan +01)GlobalLimitExec: skip=15, fetch=10 +02)--CoalescePartitionsExec: fetch=25 +03)----DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/offset_pushdown/parquet/part-0.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/offset_pushdown/parquet/part-1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/offset_pushdown/parquet/part-2.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/offset_pushdown/parquet/part-3.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/offset_pushdown/parquet/part-4.parquet]]}, projection=[id], limit=25, file_type=parquet + +statement ok +reset datafusion.execution.target_partitions; + +#################### +# Correctness: OFFSET must skip exactly the right number of rows, for +# every source format and across a range of `target_partitions` settings +# (1 = no parallelism, 8 = more partitions than files so some partitions +# read no file at all). +# +# The `COUNT`-based checks below intentionally avoid asserting exact row +# *values* without an `ORDER BY`: which of the 5 underlying files +# `ListingTable` decides to read for a given `LIMIT` is not deterministic +# (see the note above `get_files_with_limit`), so only counts, distinct +# counts and set-membership are checked. The final `ORDER BY` query per +# table cross-checks the actual values deterministically -- note that an +# `ORDER BY` blocks offset pushdown into `TableScan` (a `Sort` sits +# between the `Limit` and the scan), so it exercises the plain +# `GlobalLimitExec` path rather than the pushdown path exercised above. +#################### + +# offset_csv + +statement ok +set datafusion.execution.target_partitions = 1; + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_csv OFFSET 0 LIMIT 3) +---- +3 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_csv OFFSET 7 LIMIT 6) +---- +6 + +query I +SELECT COUNT(DISTINCT id) FROM (SELECT id FROM offset_csv OFFSET 7 LIMIT 6) +---- +6 + +query I +SELECT COUNT(*) FROM ((SELECT id FROM offset_csv OFFSET 7 LIMIT 6) EXCEPT (SELECT id FROM offset_src)) +---- +0 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_csv OFFSET 19 LIMIT 5) +---- +1 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_csv OFFSET 20) +---- +0 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_csv OFFSET 25 LIMIT 3) +---- +0 + +query II +SELECT id, val FROM offset_csv ORDER BY id OFFSET 7 LIMIT 6 +---- +7 70 +8 80 +9 90 +10 100 +11 110 +12 120 + +statement ok +set datafusion.execution.target_partitions = 8; + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_csv OFFSET 0 LIMIT 3) +---- +3 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_csv OFFSET 7 LIMIT 6) +---- +6 + +query I +SELECT COUNT(DISTINCT id) FROM (SELECT id FROM offset_csv OFFSET 7 LIMIT 6) +---- +6 + +query I +SELECT COUNT(*) FROM ((SELECT id FROM offset_csv OFFSET 7 LIMIT 6) EXCEPT (SELECT id FROM offset_src)) +---- +0 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_csv OFFSET 19 LIMIT 5) +---- +1 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_csv OFFSET 20) +---- +0 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_csv OFFSET 25 LIMIT 3) +---- +0 + +query II +SELECT id, val FROM offset_csv ORDER BY id OFFSET 7 LIMIT 6 +---- +7 70 +8 80 +9 90 +10 100 +11 110 +12 120 + +# offset_json + +statement ok +set datafusion.execution.target_partitions = 1; + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_json OFFSET 0 LIMIT 3) +---- +3 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_json OFFSET 7 LIMIT 6) +---- +6 + +query I +SELECT COUNT(DISTINCT id) FROM (SELECT id FROM offset_json OFFSET 7 LIMIT 6) +---- +6 + +query I +SELECT COUNT(*) FROM ((SELECT id FROM offset_json OFFSET 7 LIMIT 6) EXCEPT (SELECT id FROM offset_src)) +---- +0 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_json OFFSET 19 LIMIT 5) +---- +1 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_json OFFSET 20) +---- +0 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_json OFFSET 25 LIMIT 3) +---- +0 + +query II +SELECT id, val FROM offset_json ORDER BY id OFFSET 7 LIMIT 6 +---- +7 70 +8 80 +9 90 +10 100 +11 110 +12 120 + +statement ok +set datafusion.execution.target_partitions = 8; + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_json OFFSET 0 LIMIT 3) +---- +3 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_json OFFSET 7 LIMIT 6) +---- +6 + +query I +SELECT COUNT(DISTINCT id) FROM (SELECT id FROM offset_json OFFSET 7 LIMIT 6) +---- +6 + +query I +SELECT COUNT(*) FROM ((SELECT id FROM offset_json OFFSET 7 LIMIT 6) EXCEPT (SELECT id FROM offset_src)) +---- +0 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_json OFFSET 19 LIMIT 5) +---- +1 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_json OFFSET 20) +---- +0 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_json OFFSET 25 LIMIT 3) +---- +0 + +query II +SELECT id, val FROM offset_json ORDER BY id OFFSET 7 LIMIT 6 +---- +7 70 +8 80 +9 90 +10 100 +11 110 +12 120 + +# offset_parquet + +statement ok +set datafusion.execution.target_partitions = 1; + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_parquet OFFSET 0 LIMIT 3) +---- +3 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_parquet OFFSET 7 LIMIT 6) +---- +6 + +query I +SELECT COUNT(DISTINCT id) FROM (SELECT id FROM offset_parquet OFFSET 7 LIMIT 6) +---- +6 + +query I +SELECT COUNT(*) FROM ((SELECT id FROM offset_parquet OFFSET 7 LIMIT 6) EXCEPT (SELECT id FROM offset_src)) +---- +0 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_parquet OFFSET 19 LIMIT 5) +---- +1 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_parquet OFFSET 20) +---- +0 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_parquet OFFSET 25 LIMIT 3) +---- +0 + +query II +SELECT id, val FROM offset_parquet ORDER BY id OFFSET 7 LIMIT 6 +---- +7 70 +8 80 +9 90 +10 100 +11 110 +12 120 + +statement ok +set datafusion.execution.target_partitions = 8; + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_parquet OFFSET 0 LIMIT 3) +---- +3 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_parquet OFFSET 7 LIMIT 6) +---- +6 + +query I +SELECT COUNT(DISTINCT id) FROM (SELECT id FROM offset_parquet OFFSET 7 LIMIT 6) +---- +6 + +query I +SELECT COUNT(*) FROM ((SELECT id FROM offset_parquet OFFSET 7 LIMIT 6) EXCEPT (SELECT id FROM offset_src)) +---- +0 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_parquet OFFSET 19 LIMIT 5) +---- +1 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_parquet OFFSET 20) +---- +0 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_parquet OFFSET 25 LIMIT 3) +---- +0 + +query II +SELECT id, val FROM offset_parquet ORDER BY id OFFSET 7 LIMIT 6 +---- +7 70 +8 80 +9 90 +10 100 +11 110 +12 120 + +# offset_arrow + +statement ok +set datafusion.execution.target_partitions = 1; + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_arrow OFFSET 0 LIMIT 3) +---- +3 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_arrow OFFSET 7 LIMIT 6) +---- +6 + +query I +SELECT COUNT(DISTINCT id) FROM (SELECT id FROM offset_arrow OFFSET 7 LIMIT 6) +---- +6 + +query I +SELECT COUNT(*) FROM ((SELECT id FROM offset_arrow OFFSET 7 LIMIT 6) EXCEPT (SELECT id FROM offset_src)) +---- +0 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_arrow OFFSET 19 LIMIT 5) +---- +1 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_arrow OFFSET 20) +---- +0 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_arrow OFFSET 25 LIMIT 3) +---- +0 + +query II +SELECT id, val FROM offset_arrow ORDER BY id OFFSET 7 LIMIT 6 +---- +7 70 +8 80 +9 90 +10 100 +11 110 +12 120 + +statement ok +set datafusion.execution.target_partitions = 8; + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_arrow OFFSET 0 LIMIT 3) +---- +3 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_arrow OFFSET 7 LIMIT 6) +---- +6 + +query I +SELECT COUNT(DISTINCT id) FROM (SELECT id FROM offset_arrow OFFSET 7 LIMIT 6) +---- +6 + +query I +SELECT COUNT(*) FROM ((SELECT id FROM offset_arrow OFFSET 7 LIMIT 6) EXCEPT (SELECT id FROM offset_src)) +---- +0 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_arrow OFFSET 19 LIMIT 5) +---- +1 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_arrow OFFSET 20) +---- +0 + +query I +SELECT COUNT(*) FROM (SELECT id FROM offset_arrow OFFSET 25 LIMIT 3) +---- +0 + +query II +SELECT id, val FROM offset_arrow ORDER BY id OFFSET 7 LIMIT 6 +---- +7 70 +8 80 +9 90 +10 100 +11 110 +12 120 + +# Config reset +# The SLT runner sets `target_partitions` to 4 instead of using the +# default, so reset it explicitly. +statement ok +set datafusion.execution.target_partitions = 4; + +statement ok +DROP TABLE offset_src; + +statement ok +DROP TABLE offset_csv; + +statement ok +DROP TABLE offset_json; + +statement ok +DROP TABLE offset_parquet; + +statement ok +DROP TABLE offset_arrow; From ef43a124dac20db0cd97057792b8fb2a321c307f Mon Sep 17 00:00:00 2001 From: Martin Tzvetanov Grigorov Date: Fri, 18 Sep 2026 15:40:06 +0300 Subject: [PATCH 10/10] fmt && prettier --- datafusion/core/src/datasource/listing/table.rs | 9 +++++---- datafusion/proto/tests/cases/roundtrip_logical_plan.rs | 3 ++- docs/source/library-user-guide/upgrading/56.0.0.md | 4 ++-- 3 files changed, 9 insertions(+), 7 deletions(-) diff --git a/datafusion/core/src/datasource/listing/table.rs b/datafusion/core/src/datasource/listing/table.rs index 341c27ff1392..194de0a89aaa 100644 --- a/datafusion/core/src/datasource/listing/table.rs +++ b/datafusion/core/src/datasource/listing/table.rs @@ -320,8 +320,7 @@ mod tests { 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), + SessionConfig::new().with_target_partitions(target_partition), ); let table = load_table(&ctx, "alltypes_plain.parquet").await?; @@ -347,8 +346,10 @@ mod tests { .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)?; + 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])); diff --git a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs index e68bde0b34bf..f4f95d9b2d09 100644 --- a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs @@ -642,7 +642,8 @@ async fn roundtrip_logical_plan_limit_offset() -> Result<()> { // trivially passing because nothing needed to round-trip. let plan_str = plan.to_string(); assert_eq!( - plan_str, "Limit: skip=0, fetch=5\n TableScan: t1 projection=[a, b], fetch=5, offset=3", + plan_str, + "Limit: skip=0, fetch=5\n TableScan: t1 projection=[a, b], fetch=5, offset=3", "expected 'fetch=5' and 'offset=3' to be pushed into the scan, got: {plan_str}" ); diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index 266abb9ccddd..d63ffc7537f5 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -557,11 +557,11 @@ round-trips exactly. #### API breaks -The _public_ `TableScan` struct has one new _public_ field - `offset` and +The _public_ `TableScan` struct has one new _public_ field - `offset` and one field whose type is changed - `statistics_requests`. The addition of `offset` field increases the size of `TableScan` and thus the size of `LogicalPlan` enum. To keep the size of `LogicalPlan` reasonable the type of the `TableScan::statistics_requests` -field is changed from `BTreeSet` (24 bytes) to +field is changed from `BTreeSet` (24 bytes) to `Box>` (8 bytes). The users' applications are recommended to use `TableScanBuilder` to construct new instances of `TableScan` instead of instantiating directly `TableScan { ... }` to avoid