Fix: ListingTableFactory hive column detection

 - Fixes an issue in the ListingTableFactory where hive columns are not
   detected and incorporated into the table schema when an explicit
   schema has not been set by the user
 - Fixes an issue where subdirectories that do not follow Hive
   formatting (e.g. key=value) could be erroneously interpreted as
   contributing to the table schema

Author: BlakeOrth

datafusion/core/src/datasource/listing/table.rs

@@ -802,6 +802,9 @@ impl ListingOptions {
                     .rev()
                     .skip(1) // get parents only; skip the file itself
                     .rev()
+                    // Partitions are expected to follow the format "column_name=value", so we
+                    // should ignore any path part that cannot be parsed into the expected format
+                    .filter(|s| s.contains('='))
                     .map(|s| s.split('=').take(1).collect())
                     .collect_vec()
             })

datafusion/core/src/datasource/listing_table_factory.rs

@@ -64,15 +64,25 @@ impl TableProviderFactory for ListingTableFactory {
             .create(session_state, &cmd.options)?;
 
         let file_extension = get_extension(cmd.location.as_str());
+        let mut table_path = ListingTableUrl::parse(&cmd.location)?;
+        let mut options = ListingOptions::new(file_format)
+            .with_session_config_options(session_state.config())
+            .with_file_extension(file_extension);
 
         let (provided_schema, table_partition_cols) = if cmd.schema.fields().is_empty() {
+            let part_cols = match cmd.table_partition_cols.is_empty() {
+                true => options
+                    .infer_partitions(session_state, &table_path)
+                    .await?
+                    .into_iter(),
+                false => cmd.table_partition_cols.clone().into_iter(),
+            };
             (
                 None,
-                cmd.table_partition_cols
-                    .iter()
-                    .map(|x| {
+                part_cols
+                    .map(|p| {
                         (
-                            x.clone(),
+                            p,
                             DataType::Dictionary(
                                 Box::new(DataType::UInt16),
                                 Box::new(DataType::Utf8),
@@ -108,13 +118,7 @@ impl TableProviderFactory for ListingTableFactory {
             (Some(schema), table_partition_cols)
         };
 
-        let mut table_path = ListingTableUrl::parse(&cmd.location)?;
-
-        let options = ListingOptions::new(file_format)
-            .with_file_extension(&file_extension)
-            .with_session_config_options(session_state.config())
-            .with_table_partition_cols(table_partition_cols);
-
+        options = options.with_table_partition_cols(table_partition_cols);
         options
             .validate_partitions(session_state, &table_path)
             .await?;
@@ -189,6 +193,8 @@ fn get_extension(path: &str) -> String {
 mod tests {
     use glob::Pattern;
     use std::collections::HashMap;
+    use std::fs;
+    use std::path::PathBuf;
 
     use super::*;
     use crate::{
@@ -375,4 +381,49 @@ mod tests {
             Pattern::new("*.csv").unwrap()
         );
     }
+
+    #[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");
+        fs::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 {
+            name,
+            location: dir.path().to_str().unwrap().to_string(),
+            file_type: "parquet".to_string(),
+            schema: Arc::new(DFSchema::empty()),
+            table_partition_cols: vec![],
+            if_not_exists: false,
+            temporary: false,
+            definition: None,
+            order_exprs: vec![],
+            unbounded: false,
+            options: HashMap::new(),
+            constraints: Constraints::default(),
+            column_defaults: HashMap::new(),
+        };
+        let table_provider = factory.create(&state, &cmd).await.unwrap();
+        let listing_table = table_provider
+            .as_any()
+            .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);
+    }
 }

datafusion/sqllogictest/test_files/insert_to_external.slt

@@ -175,6 +175,34 @@ select * from partitioned_insert_test order by a,b,c
 1 20 200
 2 20 200
 
+statement count 0
+CREATE EXTERNAL TABLE
+partitioned_insert_test_readback
+STORED AS csv
+LOCATION 'test_files/scratch/insert_to_external/insert_to_partitioned/';
+
+query TTT
+describe partitioned_insert_test_readback;
+----
+c Int64 YES
+a Dictionary(UInt16, Utf8) NO
+b Dictionary(UInt16, Utf8) NO
+
+query ITT
+select * from partitioned_insert_test_readback order by a,b,c;
+----
+1 10 100
+1 10 200
+1 20 100
+2 20 100
+1 20 200
+2 20 200
+
+query I
+select count(*) from partitioned_insert_test_readback where b=100;
+----
+3
+
 statement ok
 CREATE EXTERNAL TABLE
 partitioned_insert_test_verify(c bigint)