feat: Support multiple external table locations

datafusion-cli/src/exec.rs

@@ -423,16 +423,17 @@ async fn create_plan(
 
         // Expose stdin (e.g. `cat data.csv | datafusion-cli`) as a `stdin://`
         // object store, registered like any other scheme in `get_object_store`.
-        cmd.location = StdinUtils::rewrite_location(&cmd.location, format.as_ref());
-
-        register_object_store_and_config_extensions(
-            ctx,
-            &cmd.location,
-            &cmd.options,
-            format,
-            resolve_region,
-        )
-        .await?;
+        for location in &mut cmd.locations {
+            *location = StdinUtils::rewrite_location(location, format.as_ref());
+            register_object_store_and_config_extensions(
+                ctx,
+                location,
+                &cmd.options,
+                format.clone(),
+                resolve_region,
+            )
+            .await?;
+        }
     }
 
     if let LogicalPlan::Copy(copy_to) = &mut plan {
@@ -535,14 +536,16 @@ mod tests {
 
         if let LogicalPlan::Ddl(DdlStatement::CreateExternalTable(cmd)) = &plan {
             let format = config_file_type_from_str(&cmd.file_type);
-            register_object_store_and_config_extensions(
-                &ctx,
-                &cmd.location,
-                &cmd.options,
-                format,
-                false,
-            )
-            .await?;
+            for location in &cmd.locations {
+                register_object_store_and_config_extensions(
+                    &ctx,
+                    location,
+                    &cmd.options,
+                    format.clone(),
+                    false,
+                )
+                .await?;
+            }
         } else {
             return plan_err!("LogicalPlan is not a CreateExternalTable");
         }

datafusion/catalog-listing/src/table.rs

@@ -54,7 +54,7 @@ use datafusion_physical_plan::ExecutionPlan;
 use datafusion_physical_plan::empty::EmptyExec;
 use futures::{Stream, StreamExt, TryStreamExt, future, stream};
 use object_store::ObjectStore;
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 
 /// Result of a file listing operation from [`ListingTable::list_files_for_scan`].
@@ -816,7 +816,15 @@ impl ListingTable {
         .await?;
         let meta_fetch_concurrency =
             ctx.config_options().execution.meta_fetch_concurrency.get();
-        let file_list = stream::iter(file_list).flatten_unordered(meta_fetch_concurrency);
+        // 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()))
+            });
         // collect the statistics and ordering if required by the config
         let files = file_list
             .map(|part_file| async {

datafusion/catalog/src/stream.rs

@@ -53,7 +53,15 @@ impl TableProviderFactory for StreamTableFactory {
         cmd: &CreateExternalTable,
     ) -> Result<Arc<dyn TableProvider>> {
         let schema: SchemaRef = Arc::clone(cmd.schema.inner());
-        let location = cmd.location.clone();
+        let location = match cmd.locations.as_slice() {
+            [single] => single.clone(),
+            _ => {
+                return config_err!(
+                    "Stream tables support exactly one location; \
+                     use a listing table to read multiple files"
+                );
+            }
+        };
         let encoding = cmd.file_type.parse()?;
         let header = if let Ok(opt) = cmd
             .options