Support prelisted files in read-only ListingTable scans

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<ListingTableUrl>,
+    /// 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<Arc<[PartitionedFile]>>,
     /// 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<PartitionedFile>) -> 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,

datafusion/catalog-listing/src/table.rs

@@ -195,6 +195,7 @@ pub struct ListFilesResult {
 #[derive(Debug, Clone)]
 pub struct ListingTable {
     table_paths: Vec<ListingTableUrl>,
+    prelisted_files: Option<Arc<[PartitionedFile]>>,
     /// `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<dyn ExecutionPlan>,
         insert_op: InsertOp,
     ) -> datafusion_common::Result<Arc<dyn ExecutionPlan>> {
+        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<PartitionedFile>> =
+            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 {

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::<Int32Array>()
+            .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;