datafusion 55.0.0

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

//! Factory for creating ListingTables with default options

use std::collections::HashSet;
use std::path::Path;
use std::sync::Arc;

use crate::catalog::{TableProvider, TableProviderFactory};
use crate::datasource::listing::{
    ListingOptions, ListingTable, ListingTableConfig, ListingTableUrl,
};
use crate::execution::context::SessionState;

use arrow::datatypes::{DataType, SchemaRef};
use datafusion_common::{Result, config_datafusion_err};
use datafusion_common::{
    ToDFSchema, arrow_datafusion_err, internal_datafusion_err, plan_err,
};
use datafusion_expr::CreateExternalTable;

use async_trait::async_trait;
use datafusion_catalog::Session;

/// A `TableProviderFactory` capable of creating new `ListingTable`s
#[derive(Debug, Default)]
pub struct ListingTableFactory {}

impl ListingTableFactory {
    /// Creates a new `ListingTableFactory`
    pub fn new() -> Self {
        Self::default()
    }
}

#[async_trait]
impl TableProviderFactory for ListingTableFactory {
    async fn create(
        &self,
        state: &dyn Session,
        cmd: &CreateExternalTable,
    ) -> Result<Arc<dyn TableProvider>> {
        // TODO (https://github.com/apache/datafusion/issues/11600) remove downcast_ref from here. Should file format factory be an extension to session state?
        let session_state =
            state
                .as_any()
                .downcast_ref::<SessionState>()
                .ok_or_else(|| {
                    datafusion_common::internal_datafusion_err!(
                        "ListingTableFactory requires SessionState"
                    )
                })?;
        let file_format = session_state
            .get_file_format_factory(cmd.file_type.as_str())
            .ok_or(config_datafusion_err!(
                "Unable to create table with format {}! Could not find FileFormat.",
                cmd.file_type
            ))?
            .create(session_state, &cmd.options)?;

        let table_paths = cmd
            .locations
            .iter()
            .map(|location| {
                Ok(ListingTableUrl::parse(location)?.with_table_ref(cmd.name.clone()))
            })
            .collect::<Result<Vec<_>>>()?;
        let Some(first_path) = table_paths.first() else {
            return plan_err!("CREATE EXTERNAL TABLE requires at least one location");
        };

        let mut seen_paths = HashSet::with_capacity(table_paths.len());
        if let Some(duplicate) = table_paths.iter().find(|path| !seen_paths.insert(*path))
        {
            return plan_err!(
                "Duplicate location '{}' in CREATE EXTERNAL TABLE",
                duplicate.as_str()
            );
        }

        // `ListingTable` resolves a single object store (from the first location)
        // and scans every location with it, so locations spanning different
        // object stores would silently read the wrong data. Reading across
        // object stores is intentionally not supported (see
        // https://github.com/apache/datafusion/issues/16303); reject it here with
        // a clear error rather than producing incorrect results at scan time.
        let object_store_url = first_path.object_store();
        if let Some(other) = table_paths
            .iter()
            .find(|path| path.object_store() != object_store_url)
        {
            return plan_err!(
                "All locations of a CREATE EXTERNAL TABLE must be on the same \
                 object store, but found '{}' and '{}'",
                object_store_url.as_str(),
                other.object_store().as_str()
            );
        }

        // With a single location the historical extension handling is kept. With
        // more than one location the files may have different extensions, so the
        // extension filter is left empty and the explicit paths/globs are used
        // as provided.
        let file_extension = if table_paths.len() == 1 {
            match first_path.is_collection() {
                // Setting the extension to be empty instead of allowing the default extension seems
                // odd, but was done to ensure existing behavior isn't modified. It seems like this
                // could be refactored to either use the default extension or set the fully expected
                // extension when compression is included (e.g. ".csv.gz")
                true => String::new(),
                false => get_extension(&cmd.locations[0]),
            }
        } else {
            String::new()
        };
        let mut options =
            ListingOptions::new(file_format).with_file_extension(file_extension);

        // Partition columns are derived from the first location; all locations
        // are expected to share the same partitioning.
        let (provided_schema, table_partition_cols) = if cmd.schema.fields().is_empty() {
            let infer_parts = session_state
                .config_options()
                .execution
                .listing_table_factory_infer_partitions;
            let part_cols = if cmd.table_partition_cols.is_empty() && infer_parts {
                options
                    .infer_partitions(session_state, first_path)
                    .await?
                    .into_iter()
            } else {
                cmd.table_partition_cols.clone().into_iter()
            };

            (
                None,
                part_cols
                    .map(|p| {
                        (
                            p,
                            DataType::Dictionary(
                                Box::new(DataType::UInt16),
                                Box::new(DataType::Utf8),
                            ),
                        )
                    })
                    .collect::<Vec<_>>(),
            )
        } else {
            let schema = Arc::clone(cmd.schema.inner());
            let table_partition_cols = cmd
                .table_partition_cols
                .iter()
                .map(|col| {
                    schema
                        .field_with_name(col)
                        .map_err(|e| arrow_datafusion_err!(e))
                })
                .collect::<Result<Vec<_>>>()?
                .into_iter()
                .map(|f| (f.name().to_owned(), f.data_type().to_owned()))
                .collect();
            // exclude partition columns to support creating partitioned external table
            // with a specified column definition like
            // `create external table a(c0 int, c1 int) stored as csv partitioned by (c1)...`
            let mut project_idx = Vec::new();
            for i in 0..schema.fields().len() {
                if !cmd.table_partition_cols.contains(schema.field(i).name()) {
                    project_idx.push(i);
                }
            }
            let schema = Arc::new(schema.project(&project_idx)?);
            (Some(schema), table_partition_cols)
        };

        options = options.with_table_partition_cols(table_partition_cols);

        // Validate partitions against every location before any glob rewriting.
        for table_path in &table_paths {
            options
                .validate_partitions(session_state, table_path)
                .await?;
        }

        let (resolved_table_paths, resolved_schema) = match provided_schema {
            // We will need to check the table columns against the schema
            // this is done so that we can do an ORDER BY for external table creation
            // specifically for parquet file format.
            // See: https://github.com/apache/datafusion/issues/7317
            None => {
                let mut resolved_paths = Vec::with_capacity(table_paths.len());
                let mut inferred_schema: Option<(String, SchemaRef)> = None;
                for mut table_path in table_paths {
                    // if the folder then rewrite a file path as 'path/*.parquet'
                    // to only read the files the reader can understand
                    if table_path.is_folder() && table_path.get_glob().is_none() {
                        // Since there are no files yet to infer an actual extension,
                        // derive the pattern based on compression type.
                        // So for gzipped CSV the pattern is `*.csv.gz`
                        let glob = match options.format.compression_type() {
                            Some(compression) => {
                                match options
                                    .format
                                    .get_ext_with_compression(&compression)
                                {
                                    // Use glob based on `FileFormat` extension
                                    Ok(ext) => format!("*.{ext}"),
                                    // Fallback to `file_type`, if not supported by `FileFormat`
                                    Err(_) => {
                                        format!("*.{}", cmd.file_type.to_lowercase())
                                    }
                                }
                            }
                            None => format!("*.{}", cmd.file_type.to_lowercase()),
                        };
                        table_path = table_path.with_glob(glob.as_ref())?;
                    }
                    let schema = options.infer_schema(session_state, &table_path).await?;
                    // All locations must resolve to the same fields. Schema
                    // and field metadata may differ between files without
                    // changing the fields read by the table.
                    let location = table_path.to_string();
                    match &inferred_schema {
                        None => inferred_schema = Some((location, schema)),
                        Some((existing_location, existing))
                            if !schemas_have_same_fields(existing, &schema) =>
                        {
                            return plan_err!(
                                "All locations of a CREATE EXTERNAL TABLE must have the \
                                 same schema, but schema inferred from '{}' differs from \
                                 schema inferred from '{}'",
                                location,
                                existing_location
                            );
                        }
                        Some(_) => {}
                    }
                    resolved_paths.push(table_path);
                }
                // `table_paths` was guaranteed non-empty above, so the loop ran
                // at least once and `inferred_schema` is always `Some` here.
                let (_, schema) = inferred_schema.ok_or_else(|| {
                    internal_datafusion_err!(
                        "no schema could be inferred from the provided locations"
                    )
                })?;
                let df_schema = Arc::clone(&schema).to_dfschema()?;
                let column_refs: HashSet<_> = cmd
                    .order_exprs
                    .iter()
                    .flat_map(|sort| sort.iter())
                    .flat_map(|s| s.expr.column_refs())
                    .collect();

                for column in &column_refs {
                    if !df_schema.has_column(column) {
                        return plan_err!("Column {column} is not in schema");
                    }
                }

                (resolved_paths, schema)
            }
            Some(s) => (table_paths, s),
        };
        let config = ListingTableConfig::new_with_multi_paths(resolved_table_paths)
            .with_listing_options(options.with_file_sort_order(cmd.order_exprs.clone()))
            .with_schema(resolved_schema);
        let provider = ListingTable::try_new(config)?
            .with_cache(state.runtime_env().cache_manager.get_file_statistic_cache());
        let table = provider
            .with_definition(cmd.definition.clone())
            .with_constraints(cmd.constraints.clone())
            .with_column_defaults(cmd.column_defaults.clone());

        // Pre-warm statistics cache if collect_statistics is enabled
        if session_state.config().collect_statistics() {
            let filters = &[];
            let limit = None;
            if let Err(e) = table.list_files_for_scan(state, filters, limit).await {
                log::warn!("Failed to pre-warm statistics cache: {e}");
            }
        }

        Ok(Arc::new(table))
    }
}

// Get file extension from path
fn get_extension(path: &str) -> String {
    let res = Path::new(path).extension().and_then(|ext| ext.to_str());
    match res {
        Some(ext) => format!(".{ext}"),
        None => "".to_string(),
    }
}

fn schemas_have_same_fields(left: &SchemaRef, right: &SchemaRef) -> bool {
    left.fields().len() == right.fields().len()
        && left
            .fields()
            .iter()
            .zip(right.fields())
            .all(|(left, right)| {
                left.name() == right.name()
                    && left.data_type() == right.data_type()
                    && left.is_nullable() == right.is_nullable()
            })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        datasource::file_format::csv::CsvFormat, execution::context::SessionContext,
        test_util::parquet_test_data,
    };
    use arrow::datatypes::{Field, Schema};
    use datafusion_execution::cache::cache_manager::{
        CacheManagerConfig, DEFAULT_FILE_STATISTICS_MEMORY_LIMIT,
    };
    use datafusion_execution::config::SessionConfig;
    use datafusion_execution::runtime_env::RuntimeEnvBuilder;
    use glob::Pattern;
    use std::collections::HashMap;
    use std::fs;
    use std::fs::File;
    use std::path::{Path, PathBuf};

    use datafusion_common::parsers::CompressionTypeVariant;
    use datafusion_common::{DFSchema, TableReference};
    use datafusion_execution::cache::Cache;
    use datafusion_execution::cache::default_cache::DefaultCache;
    use datafusion_expr::registry::ExtensionTypeRegistryRef;

    fn factory_and_state() -> (ListingTableFactory, SessionState) {
        let factory = ListingTableFactory::new();
        let context = SessionContext::new();
        let state = context.state();
        (factory, state)
    }

    fn write_csv(path: &Path, contents: &str) {
        fs::write(path, contents).unwrap();
    }

    fn csv_cmd_with_locations(paths: &[&Path]) -> CreateExternalTable {
        let locations = paths
            .iter()
            .map(|path| path.to_str().unwrap().to_string())
            .collect::<Vec<_>>();

        CreateExternalTable::builder(
            TableReference::bare("foo"),
            locations[0].clone(),
            "csv",
            Arc::new(DFSchema::empty()),
        )
        .with_locations(locations)
        .with_options(HashMap::from([("format.has_header".into(), "true".into())]))
        .build()
    }

    fn assert_error_contains(error: impl std::fmt::Display, expected: &str) {
        let error = error.to_string();
        assert!(
            error.contains(expected),
            "expected error to contain '{expected}', got: {error}"
        );
    }

    #[tokio::test]
    async fn test_create_using_non_std_file_ext() {
        let csv_file = tempfile::Builder::new()
            .prefix("foo")
            .suffix(".tbl")
            .tempfile()
            .unwrap();

        let factory = ListingTableFactory::new();
        let context = SessionContext::new();
        let state = context.state();
        let name = TableReference::bare("foo");
        let cmd = CreateExternalTable::builder(
            name,
            csv_file.path().to_str().unwrap().to_string(),
            "csv",
            Arc::new(DFSchema::empty()),
        )
        .with_options(HashMap::from([("format.has_header".into(), "true".into())]))
        .build();
        let table_provider = factory.create(&state, &cmd).await.unwrap();
        let listing_table = table_provider.downcast_ref::<ListingTable>().unwrap();
        let listing_options = listing_table.options();
        assert_eq!(".tbl", listing_options.file_extension);
    }

    #[tokio::test]
    async fn test_create_using_non_std_file_ext_csv_options() {
        let csv_file = tempfile::Builder::new()
            .prefix("foo")
            .suffix(".tbl")
            .tempfile()
            .unwrap();

        let factory = ListingTableFactory::new();
        let context = SessionContext::new();
        let state = context.state();
        let name = TableReference::bare("foo");

        let mut options = HashMap::new();
        options.insert("format.schema_infer_max_rec".to_owned(), "1000".to_owned());
        options.insert("format.has_header".into(), "true".into());
        let cmd = CreateExternalTable::builder(
            name,
            csv_file.path().to_str().unwrap().to_string(),
            "csv",
            Arc::new(DFSchema::empty()),
        )
        .with_options(options)
        .build();
        let table_provider = factory.create(&state, &cmd).await.unwrap();
        let listing_table = table_provider.downcast_ref::<ListingTable>().unwrap();

        let format = listing_table.options().format.clone();
        let csv_format = format.downcast_ref::<CsvFormat>().unwrap();
        let csv_options = csv_format.options().clone();
        assert_eq!(csv_options.schema_infer_max_rec, Some(1000));
        let listing_options = listing_table.options();
        assert_eq!(".tbl", listing_options.file_extension);
    }

    /// Validates that CreateExternalTable with compression
    /// searches for gzipped files in a directory location
    #[tokio::test]
    async fn test_create_using_folder_with_compression() {
        let dir = tempfile::tempdir().unwrap();
        // Schema inference now requires at least one file at the location.
        // The file itself can be 0-byte — it will be filtered out before the
        // format-specific inference runs, leaving an empty inferred schema.
        File::create_new(dir.path().join("placeholder.csv.gz")).unwrap();

        let factory = ListingTableFactory::new();
        let context = SessionContext::new();
        let state = context.state();
        let name = TableReference::bare("foo");

        let mut options = HashMap::new();
        options.insert("format.schema_infer_max_rec".to_owned(), "1000".to_owned());
        options.insert("format.has_header".into(), "true".into());
        options.insert("format.compression".into(), "gzip".into());
        let cmd = CreateExternalTable::builder(
            name,
            dir.path().to_str().unwrap().to_string(),
            "csv",
            Arc::new(DFSchema::empty()),
        )
        .with_options(options)
        .build();
        let table_provider = factory.create(&state, &cmd).await.unwrap();
        let listing_table = table_provider.downcast_ref::<ListingTable>().unwrap();

        // Verify compression is used
        let format = listing_table.options().format.clone();
        let csv_format = format.downcast_ref::<CsvFormat>().unwrap();
        let csv_options = csv_format.options().clone();
        assert_eq!(csv_options.compression, CompressionTypeVariant::GZIP);

        let listing_options = listing_table.options();
        assert_eq!("", listing_options.file_extension);
        // Glob pattern is set to search for gzipped files
        let table_path = listing_table.table_paths().first().unwrap();
        assert_eq!(
            table_path.get_glob().clone().unwrap(),
            Pattern::new("*.csv.gz").unwrap()
        );
    }

    /// Validates that CreateExternalTable without compression
    /// searches for normal files in a directory location
    #[tokio::test]
    async fn test_create_using_folder_without_compression() {
        let dir = tempfile::tempdir().unwrap();
        // See `test_create_using_folder_with_compression` — a placeholder file
        // is required so schema inference does not error on an empty location.
        File::create_new(dir.path().join("placeholder.csv")).unwrap();

        let factory = ListingTableFactory::new();
        let context = SessionContext::new();
        let state = context.state();
        let name = TableReference::bare("foo");

        let mut options = HashMap::new();
        options.insert("format.schema_infer_max_rec".to_owned(), "1000".to_owned());
        options.insert("format.has_header".into(), "true".into());
        let cmd = CreateExternalTable::builder(
            name,
            dir.path().to_str().unwrap().to_string(),
            "csv",
            Arc::new(DFSchema::empty()),
        )
        .with_options(options)
        .build();
        let table_provider = factory.create(&state, &cmd).await.unwrap();
        let listing_table = table_provider.downcast_ref::<ListingTable>().unwrap();

        let listing_options = listing_table.options();
        assert_eq!("", listing_options.file_extension);
        // Glob pattern is set to search for gzipped files
        let table_path = listing_table.table_paths().first().unwrap();
        assert_eq!(
            table_path.get_glob().clone().unwrap(),
            Pattern::new("*.csv").unwrap()
        );
    }

    #[tokio::test]
    async fn test_odd_directory_names() {
        let dir = tempfile::tempdir().unwrap();
        let mut path = PathBuf::from(dir.path());
        path.extend(["odd.v1", "odd.v2"]);
        fs::create_dir_all(&path).unwrap();
        // Placeholder so schema inference does not error on an empty location.
        File::create_new(path.join("placeholder.parquet")).unwrap();

        let factory = ListingTableFactory::new();
        let context = SessionContext::new();
        let state = context.state();
        let name = TableReference::bare("foo");

        let cmd = CreateExternalTable::builder(
            name,
            String::from(path.to_str().unwrap()),
            "parquet",
            Arc::new(DFSchema::empty()),
        )
        .build();
        let table_provider = factory.create(&state, &cmd).await.unwrap();
        let listing_table = table_provider.downcast_ref::<ListingTable>().unwrap();

        let listing_options = listing_table.options();
        assert_eq!("", listing_options.file_extension);
    }

    #[tokio::test]
    async fn test_create_with_hive_partitions() {
        let dir = tempfile::tempdir().unwrap();
        let mut path = PathBuf::from(dir.path());
        path.extend(["key1=value1", "key2=value2"]);
        fs::create_dir_all(&path).unwrap();
        path.push("data.parquet");
        File::create_new(&path).unwrap();

        let factory = ListingTableFactory::new();
        let context = SessionContext::new();
        let state = context.state();
        let name = TableReference::bare("foo");

        let cmd = CreateExternalTable::builder(
            name,
            dir.path().to_str().unwrap(),
            "parquet",
            Arc::new(DFSchema::empty()),
        )
        .build();
        let table_provider = factory.create(&state, &cmd).await.unwrap();
        let listing_table = table_provider.downcast_ref::<ListingTable>().unwrap();

        let listing_options = listing_table.options();
        let dtype =
            DataType::Dictionary(Box::new(DataType::UInt16), Box::new(DataType::Utf8));
        let expected_cols = vec![
            (String::from("key1"), dtype.clone()),
            (String::from("key2"), dtype.clone()),
        ];
        assert_eq!(expected_cols, listing_options.table_partition_cols);

        // Ensure partition detection can be disabled via config
        let factory = ListingTableFactory::new();
        let mut cfg = SessionConfig::new();
        cfg.options_mut()
            .execution
            .listing_table_factory_infer_partitions = false;
        let context = SessionContext::new_with_config(cfg);
        let state = context.state();
        let name = TableReference::bare("foo");

        let cmd = CreateExternalTable::builder(
            name,
            dir.path().to_str().unwrap().to_string(),
            "parquet",
            Arc::new(DFSchema::empty()),
        )
        .build();
        let table_provider = factory.create(&state, &cmd).await.unwrap();
        let listing_table = table_provider.downcast_ref::<ListingTable>().unwrap();

        let listing_options = listing_table.options();
        assert!(listing_options.table_partition_cols.is_empty());
    }

    #[tokio::test]
    async fn test_create_with_multiple_locations() {
        let dir = tempfile::tempdir().unwrap();
        let file_a = dir.path().join("file_a.csv");
        let file_b = dir.path().join("file_b.csv");
        write_csv(&file_a, "c1,c2\n1,a\n2,b\n");
        write_csv(&file_b, "c1,c2\n3,c\n");

        let (factory, state) = factory_and_state();
        let cmd = csv_cmd_with_locations(&[&file_a, &file_b]);

        let table_provider = factory.create(&state, &cmd).await.unwrap();
        let listing_table = table_provider.downcast_ref::<ListingTable>().unwrap();

        // Both locations are registered as table paths
        assert_eq!(2, listing_table.table_paths().len());

        // Schema is inferred from the files and shared across both locations
        let field_names: Vec<_> = listing_table
            .schema()
            .fields()
            .iter()
            .map(|f| f.name().clone())
            .collect();
        assert_eq!(field_names, vec!["c1".to_string(), "c2".to_string()]);
    }

    #[tokio::test]
    async fn test_create_with_duplicate_locations_errors() {
        let dir = tempfile::tempdir().unwrap();
        let file = dir.path().join("file.csv");
        write_csv(&file, "c1,c2\n1,a\n");

        let (factory, state) = factory_and_state();
        let cmd = csv_cmd_with_locations(&[&file, &file]);
        let err = factory.create(&state, &cmd).await.unwrap_err();
        assert_error_contains(err, "Duplicate location");
    }

    #[tokio::test]
    async fn test_create_with_overlapping_locations_reads_each_file_once() {
        let dir = tempfile::tempdir().unwrap();
        let file_a = dir.path().join("file_a.csv");
        let file_b = dir.path().join("file_b.csv");
        write_csv(&file_a, "c1,c2\n1,a\n");
        write_csv(&file_b, "c1,c2\n2,b\n");

        let (factory, state) = factory_and_state();
        let cmd = csv_cmd_with_locations(&[dir.path(), file_a.as_path()]);
        let table_provider = factory.create(&state, &cmd).await.unwrap();
        let listing_table = table_provider.downcast_ref::<ListingTable>().unwrap();

        let listed_files = listing_table
            .list_files_for_scan(&state, &[], None)
            .await
            .unwrap()
            .file_groups
            .iter()
            .map(|group| group.len())
            .sum::<usize>();
        assert_eq!(listed_files, 2);
    }

    #[tokio::test]
    async fn test_create_with_multiple_locations_mismatched_schema_errors() {
        let dir = tempfile::tempdir().unwrap();
        let file_a = dir.path().join("file_a.csv");
        let file_b = dir.path().join("file_b.csv");
        write_csv(&file_a, "c1,c2\n1,a\n");
        // Different column names -> different inferred schema
        write_csv(&file_b, "x1,x2\n1,a\n");

        let (factory, state) = factory_and_state();
        let cmd = csv_cmd_with_locations(&[&file_a, &file_b]);
        let err = factory.create(&state, &cmd).await.unwrap_err();
        assert_error_contains(err, "same schema");
    }

    #[test]
    fn test_schema_comparison_ignores_schema_metadata() {
        let fields =
            vec![
                Field::new("c1", DataType::Int32, true).with_metadata(HashMap::from([(
                    "field_source".to_string(),
                    "a".to_string(),
                )])),
            ];
        let schema_a = Arc::new(Schema::new_with_metadata(
            fields.clone(),
            HashMap::from([("source".to_string(), "a".to_string())]),
        ));
        let schema_b =
            Arc::new(Schema::new_with_metadata(
                vec![Field::new("c1", DataType::Int32, true).with_metadata(
                    HashMap::from([("field_source".to_string(), "b".to_string())]),
                )],
                HashMap::from([("source".to_string(), "b".to_string())]),
            ));
        let schema_c =
            Arc::new(Schema::new(vec![Field::new("c2", DataType::Int32, true)]));

        assert_ne!(schema_a, schema_b);
        assert!(schemas_have_same_fields(&schema_a, &schema_b));
        assert!(!schemas_have_same_fields(&schema_a, &schema_c));
    }

    #[tokio::test]
    async fn test_create_with_no_locations_errors() {
        let (factory, state) = factory_and_state();

        let cmd = CreateExternalTable::builder(
            TableReference::bare("foo"),
            "unused",
            "csv",
            Arc::new(DFSchema::empty()),
        )
        .with_locations(vec![])
        .build();

        let err = factory.create(&state, &cmd).await.unwrap_err();
        assert_error_contains(err, "at least one location");
    }

    #[tokio::test]
    async fn test_create_with_locations_on_different_stores_errors() {
        let (factory, state) = factory_and_state();

        // Two locations on different object stores (different buckets) are not
        // supported: ListingTable would scan both against the first store.
        let cmd = CreateExternalTable::builder(
            TableReference::bare("foo"),
            "s3://bucket_a/file.parquet",
            "parquet",
            Arc::new(DFSchema::empty()),
        )
        .with_locations(vec![
            "s3://bucket_a/file.parquet".to_string(),
            "s3://bucket_b/file.parquet".to_string(),
        ])
        .build();

        let err = factory.create(&state, &cmd).await.unwrap_err();
        assert_error_contains(err, "same object store");
    }

    #[tokio::test]
    async fn test_statistics_cache_prewarming() {
        let factory = ListingTableFactory::new();

        let location = PathBuf::from(parquet_test_data())
            .join("alltypes_tiny_pages_plain.parquet")
            .to_string_lossy()
            .to_string();

        // Test with collect_statistics enabled
        let file_statistics_cache =
            Arc::new(DefaultCache::new(DEFAULT_FILE_STATISTICS_MEMORY_LIMIT));
        let cache_config = CacheManagerConfig::default()
            .with_file_statistics_cache(Some(file_statistics_cache.clone()));
        let runtime = RuntimeEnvBuilder::new()
            .with_cache_manager(cache_config)
            .build_arc()
            .unwrap();

        let mut config = SessionConfig::new();
        config.options_mut().execution.collect_statistics = true;
        let context = SessionContext::new_with_config_rt(config, runtime);
        let state = context.state();
        let name = TableReference::bare("test");

        let cmd = CreateExternalTable::builder(
            name,
            location.clone(),
            "parquet",
            Arc::new(DFSchema::empty()),
        )
        .build();

        let _table_provider = factory.create(&state, &cmd).await.unwrap();

        assert!(
            file_statistics_cache.len() > 0,
            "Statistics cache should be pre-warmed when collect_statistics is enabled"
        );

        // Test with collect_statistics disabled
        let file_statistics_cache =
            Arc::new(DefaultCache::new(DEFAULT_FILE_STATISTICS_MEMORY_LIMIT));
        let cache_config = CacheManagerConfig::default()
            .with_file_statistics_cache(Some(file_statistics_cache.clone()));
        let runtime = RuntimeEnvBuilder::new()
            .with_cache_manager(cache_config)
            .build_arc()
            .unwrap();

        let mut config = SessionConfig::new();
        config.options_mut().execution.collect_statistics = false;
        let context = SessionContext::new_with_config_rt(config, runtime);
        let state = context.state();
        let name = TableReference::bare("test");

        let cmd = CreateExternalTable::builder(
            name,
            location,
            "parquet",
            Arc::new(DFSchema::empty()),
        )
        .build();

        let _table_provider = factory.create(&state, &cmd).await.unwrap();

        assert_eq!(
            file_statistics_cache.len(),
            0,
            "Statistics cache should not be pre-warmed when collect_statistics is disabled"
        );
    }

    #[tokio::test]
    async fn test_create_with_invalid_session() {
        use datafusion_common::config::TableOptions;
        use datafusion_execution::TaskContext;
        use datafusion_execution::config::SessionConfig;
        use datafusion_physical_expr::PhysicalExpr;
        use datafusion_physical_plan::ExecutionPlan;
        use datafusion_session::{CatalogProviderList, EmptyCatalogProviderList};
        use std::any::Any;
        use std::collections::HashMap;

        // A mock Session that is NOT SessionState
        #[derive(Debug)]
        struct MockSession;

        #[async_trait]
        impl Session for MockSession {
            fn session_id(&self) -> &str {
                "mock_session"
            }
            fn config(&self) -> &SessionConfig {
                unimplemented!()
            }
            fn catalog_list(&self) -> Arc<dyn CatalogProviderList> {
                Arc::new(EmptyCatalogProviderList)
            }
            async fn create_physical_plan(
                &self,
                _logical_plan: &datafusion_expr::LogicalPlan,
            ) -> Result<Arc<dyn ExecutionPlan>> {
                unimplemented!()
            }
            fn create_physical_expr(
                &self,
                _expr: datafusion_expr::Expr,
                _df_schema: &DFSchema,
            ) -> Result<Arc<dyn PhysicalExpr>> {
                unimplemented!()
            }
            fn scalar_functions(
                &self,
            ) -> &HashMap<String, Arc<datafusion_expr::ScalarUDF>> {
                unimplemented!()
            }
            fn higher_order_functions(
                &self,
            ) -> &HashMap<String, Arc<datafusion_expr::HigherOrderUDF>> {
                unimplemented!()
            }
            fn aggregate_functions(
                &self,
            ) -> &HashMap<String, Arc<datafusion_expr::AggregateUDF>> {
                unimplemented!()
            }
            fn window_functions(
                &self,
            ) -> &HashMap<String, Arc<datafusion_expr::WindowUDF>> {
                unimplemented!()
            }

            fn extension_type_registry(&self) -> &ExtensionTypeRegistryRef {
                unreachable!()
            }

            fn runtime_env(&self) -> &Arc<datafusion_execution::runtime_env::RuntimeEnv> {
                unimplemented!()
            }
            fn execution_props(
                &self,
            ) -> &datafusion_expr::execution_props::ExecutionProps {
                unimplemented!()
            }
            fn as_any(&self) -> &dyn Any {
                self
            }
            fn table_options(&self) -> &TableOptions {
                unimplemented!()
            }
            fn table_options_mut(&mut self) -> &mut TableOptions {
                unimplemented!()
            }
            fn task_ctx(&self) -> Arc<TaskContext> {
                unimplemented!()
            }
        }

        let factory = ListingTableFactory::new();
        let mock_session = MockSession;

        let name = TableReference::bare("foo");
        let cmd = CreateExternalTable::builder(
            name,
            "foo.csv".to_string(),
            "csv",
            Arc::new(DFSchema::empty()),
        )
        .build();

        // This should return an error, not panic
        let result = factory.create(&mock_session, &cmd).await;
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .strip_backtrace()
                .contains("Internal error: ListingTableFactory requires SessionState")
        );
    }
}