From d7c732ed514abb928a03a82b9dd55c5e2aa33d80 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Sat, 10 Oct 2026 15:54:01 +0300 Subject: [PATCH] Allow prelisted file metadata for read-only ListingTable scans --- datafusion/catalog-listing/src/config.rs | 18 ++++- datafusion/catalog-listing/src/table.rs | 72 ++++++++++++++----- .../tests/datasource/object_store_access.rs | 66 ++++++++++++++++- 3 files changed, 136 insertions(+), 20 deletions(-) diff --git a/datafusion/catalog-listing/src/config.rs b/datafusion/catalog-listing/src/config.rs index a3791c23da04..7ccb5ad8bf6e 100644 --- a/datafusion/catalog-listing/src/config.rs +++ b/datafusion/catalog-listing/src/config.rs @@ -19,10 +19,10 @@ use crate::options::ListingOptions; use arrow::datatypes::{DataType, Schema, SchemaRef}; use datafusion_catalog::Session; use datafusion_common::{config_err, internal_err}; -use datafusion_datasource::ListingTableUrl; use datafusion_datasource::file_compression_type::FileCompressionType; #[expect(deprecated)] use datafusion_datasource::schema_adapter::SchemaAdapterFactory; +use datafusion_datasource::{ListingTableUrl, PartitionedFile}; use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory; use std::str::FromStr; use std::sync::Arc; @@ -74,6 +74,9 @@ pub struct ListingTableConfig { /// Paths on the `ObjectStore` for creating [`crate::ListingTable`]. /// They should share the same schema and object store. pub table_paths: Vec, + /// Optional caller-selected files with known object metadata. When present, + /// listing and per-file HEAD requests are skipped during scan planning. + pub(crate) prelisted_files: Option>, /// Optional `SchemaRef` for the to be created [`crate::ListingTable`]. /// /// See details on [`ListingTableConfig::with_schema`] @@ -107,6 +110,16 @@ impl ListingTableConfig { } } + /// Use an already selected set of files for scan planning. The caller is + /// responsible for supplying current metadata from the table's object store. + /// This mode requires an explicit schema and does not support partition columns, + /// declared output partitioning, or inserts into the resulting table. + #[must_use] + pub fn with_prelisted_files(mut self, files: Vec) -> Self { + self.prelisted_files = Some(files.into()); + self + } + /// Returns the source of the schema for this configuration pub fn schema_source(&self) -> SchemaSource { self.schema_source @@ -228,6 +241,7 @@ impl ListingTableConfig { Some(options) => { let ListingTableConfig { table_paths, + prelisted_files, file_schema, options: _, schema_source, @@ -250,6 +264,7 @@ impl ListingTableConfig { Ok(Self { table_paths, + prelisted_files, file_schema: Some(schema), options: Some(options), schema_source: new_schema_source, @@ -290,6 +305,7 @@ impl ListingTableConfig { let options = options.with_table_partition_cols(partitions); Ok(Self { table_paths: self.table_paths, + prelisted_files: self.prelisted_files, file_schema: self.file_schema, options: Some(options), schema_source: self.schema_source, diff --git a/datafusion/catalog-listing/src/table.rs b/datafusion/catalog-listing/src/table.rs index f65b8db1d37d..f1cfad9c2b81 100644 --- a/datafusion/catalog-listing/src/table.rs +++ b/datafusion/catalog-listing/src/table.rs @@ -195,6 +195,7 @@ pub struct ListFilesResult { #[derive(Debug, Clone)] pub struct ListingTable { table_paths: Vec, + prelisted_files: Option>, /// `file_schema` contains only the columns physically stored in the data files themselves. /// - Represents the actual fields found in files like Parquet, CSV, etc. /// - Used when reading the raw data from files @@ -240,6 +241,21 @@ impl ListingTable { .options .ok_or_else(|| internal_datafusion_err!("No ListingOptions provided"))?; + if config.prelisted_files.is_some() + && (schema_source != SchemaSource::Specified + || !options.table_partition_cols.is_empty() + || options.output_partitioning.is_some()) + { + return plan_err!( + "Prelisted files require an explicit schema, no partition columns, and no declared output partitioning" + ); + } + if config.prelisted_files.as_ref().is_some_and(|files| { + files.iter().any(|file| !file.partition_values.is_empty()) + }) { + return plan_err!("Prelisted files cannot contain partition values"); + } + // Add the partition columns to the file schema let mut builder = SchemaBuilder::from(file_schema.as_ref().to_owned()); for (part_col_name, part_col_type) in &options.table_partition_cols { @@ -257,6 +273,7 @@ impl ListingTable { let table = Self { table_paths: config.table_paths, + prelisted_files: config.prelisted_files, file_schema, table_schema, schema_source, @@ -720,6 +737,11 @@ impl TableProvider for ListingTable { input: Arc, insert_op: InsertOp, ) -> datafusion_common::Result> { + if self.prelisted_files.is_some() { + return plan_err!( + "Inserting into a ListingTable with prelisted files is not supported" + ); + } // Check that the schema of the plan matches the schema of this table. self.schema() .logically_equivalent_names_and_types(&input.schema())?; @@ -824,28 +846,42 @@ impl ListingTable { // Bound metadata requests for explicit file lists as well as directories. let meta_fetch_concurrency = ctx.config_options().execution.meta_fetch_concurrency.get(); - let mut requests = Vec::with_capacity(self.table_paths.len()); - for table_path in &self.table_paths { - requests.push(pruned_partition_list( - ctx, - store.as_ref(), - table_path, - listing_time_filters, - &self.options.file_extension, - self.options.auto_file_suffixes.as_deref(), - &self.options.table_partition_cols, - )); - } - let file_list = bounded_try_join_all(requests, meta_fetch_concurrency).await?; + let file_list: stream::BoxStream<'a, datafusion_common::Result> = + if let Some(files) = &self.prelisted_files { + stream::iter( + files + .iter() + .filter(|file| file.object_meta.size > 0) + .cloned() + .map(Ok), + ) + .boxed() + } else { + let mut requests = Vec::with_capacity(self.table_paths.len()); + for table_path in &self.table_paths { + requests.push(pruned_partition_list( + ctx, + store.as_ref(), + table_path, + listing_time_filters, + &self.options.file_extension, + self.options.auto_file_suffixes.as_deref(), + &self.options.table_partition_cols, + )); + } + let file_list = + bounded_try_join_all(requests, meta_fetch_concurrency).await?; + stream::iter(file_list) + .flatten_unordered(meta_fetch_concurrency) + .boxed() + }; // Table paths can overlap, for example when one path is a directory and // another names a file inside it. A ListingTable uses one object store, // so the object path uniquely identifies a file within this scan. let mut seen_files = HashSet::new(); - let file_list = stream::iter(file_list) - .flatten_unordered(meta_fetch_concurrency) - .try_filter(move |file| { - future::ready(seen_files.insert(file.object_meta.location.clone())) - }); + let file_list = file_list.try_filter(move |file| { + future::ready(seen_files.insert(file.object_meta.location.clone())) + }); // collect the statistics and ordering if required by the config let files = file_list .map(|part_file| async { diff --git a/datafusion/core/tests/datasource/object_store_access.rs b/datafusion/core/tests/datasource/object_store_access.rs index 2503de862e06..b902f0851eaa 100644 --- a/datafusion/core/tests/datasource/object_store_access.rs +++ b/datafusion/core/tests/datasource/object_store_access.rs @@ -25,13 +25,14 @@ //! [`ListingTable`]: datafusion::datasource::listing::ListingTable use arrow::array::{ArrayRef, Int32Array, RecordBatch}; +use arrow::datatypes::{DataType, Field, Schema}; use async_trait::async_trait; use bytes::Bytes; use datafusion::prelude::{ CsvReadOptions, JsonReadOptions, ParquetReadOptions, SessionContext, }; use datafusion_catalog_listing::{ListingOptions, ListingTable, ListingTableConfig}; -use datafusion_datasource::ListingTableUrl; +use datafusion_datasource::{ListingTableUrl, PartitionedFile}; use datafusion_datasource_csv::CsvFormat; use datafusion_datasource_json::JsonFormat; use futures::stream::BoxStream; @@ -50,6 +51,69 @@ use std::ops::Range; use std::sync::Arc; use url::Url; +#[tokio::test] +async fn prelisted_files_skip_listing_and_head_requests() { + let test = Test::new() + .with_bytes("data/a.csv", "id\n1\n") + .await + .with_bytes("data/b.csv", "id\n2\n") + .await; + let selected = test + .object_store + .inner + .head(&Path::from("data/a.csv")) + .await + .unwrap(); + let config = ListingTableConfig::new(ListingTableUrl::parse("mem:///data/").unwrap()) + .with_listing_options(ListingOptions::new(Arc::new( + CsvFormat::default().with_has_header(true), + ))) + .with_schema(Arc::new(Schema::new(vec![Field::new( + "id", + DataType::Int32, + true, + )]))) + .with_prelisted_files(vec![PartitionedFile::new_from_meta(selected)]); + test.session_context + .register_table( + "selected_files", + Arc::new(ListingTable::try_new(config).unwrap()), + ) + .unwrap(); + + let batches = test + .session_context + .sql("SELECT id FROM selected_files") + .await + .unwrap() + .collect() + .await + .unwrap(); + assert_eq!(batches.len(), 1); + assert_eq!(batches[0].num_rows(), 1); + assert_eq!( + batches[0] + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 1 + ); + let error = test + .session_context + .sql("INSERT INTO selected_files VALUES (3)") + .await + .unwrap() + .collect() + .await + .unwrap_err(); + assert!(error.to_string().contains("prelisted files"), "{error}"); + let requests = test.requests(); + assert!(!requests.contains("LIST"), "{requests}"); + assert!(!requests.contains("head=true"), "{requests}"); +} + #[tokio::test] async fn create_single_csv_file() { let test = Test::new().with_single_file_csv().await;