fix: drop partition columns already present in the file schema of a ListingTable

datafusion/catalog-listing/src/table.rs

@@ -225,6 +225,32 @@ impl ListingTable {
             .options
             .ok_or_else(|| internal_datafusion_err!("No ListingOptions provided"))?;
 
+        // Files may physically contain the partition columns, for example when
+        // they were written with `keep_partition_by_columns = true`. Partition
+        // column values are always taken from the path, so drop such columns
+        // from the file schema to avoid duplicated fields in the table schema.
+        let file_schema = if options
+            .table_partition_cols
+            .iter()
+            .any(|(name, _)| file_schema.field_with_name(name).is_ok())
+        {
+            let indices: Vec<usize> = file_schema
+                .fields()
+                .iter()
+                .enumerate()
+                .filter(|(_, field)| {
+                    !options
+                        .table_partition_cols
+                        .iter()
+                        .any(|(name, _)| name == field.name())
+                })
+                .map(|(idx, _)| idx)
+                .collect();
+            Arc::new(file_schema.project(&indices)?)
+        } else {
+            file_schema
+        };
+
         // 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 {

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

@@ -1625,6 +1625,77 @@ mod tests {
         Ok(())
     }
 
+    /// Files written with `keep_partition_by_columns = true` physically contain
+    /// the partition column, so the (inferred) file schema and the declared
+    /// partition columns overlap. The table schema must list the column once.
+    /// See <https://github.com/apache/datafusion/issues/17420>
+    #[test]
+    fn test_partition_column_present_in_file_schema_is_not_duplicated() -> Result<()> {
+        let partition_type =
+            DataType::Dictionary(Box::new(DataType::UInt16), Box::new(DataType::Utf8));
+        let opt = ListingOptions::new(Arc::new(JsonFormat::default()))
+            .with_file_extension_opt(Some(""))
+            .with_table_partition_cols(vec![("pid".to_string(), partition_type.clone())]);
+
+        let table_path = ListingTableUrl::parse("test:///bucket/test/")?;
+        let file_schema = Schema::new(vec![
+            Field::new("a", DataType::Boolean, false),
+            Field::new("pid", DataType::Utf8, false),
+        ]);
+        let config = ListingTableConfig::new(table_path)
+            .with_listing_options(opt)
+            .with_schema(Arc::new(file_schema));
+
+        let table = ListingTable::try_new(config)?;
+        let schema = table.schema();
+
+        let names = schema
+            .fields()
+            .iter()
+            .map(|f| f.name().as_str())
+            .collect::<Vec<_>>();
+        assert_eq!(names, vec!["a", "pid"]);
+        // The partition column keeps the declared partition type, as it does
+        // when the file schema is given explicitly in CREATE EXTERNAL TABLE.
+        assert_eq!(schema.field_with_name("pid")?.data_type(), &partition_type);
+
+        Ok(())
+    }
+
+    /// Two partition columns present in the file, one of them mid-schema. The
+    /// remaining file fields keep their order and the partition columns follow
+    /// in their configured order, not their order in the file.
+    #[test]
+    fn test_overlapping_partition_columns_order() -> Result<()> {
+        let opt = ListingOptions::new(Arc::new(JsonFormat::default()))
+            .with_file_extension_opt(Some(""))
+            .with_table_partition_cols(vec![
+                ("month".to_string(), DataType::Utf8),
+                ("year".to_string(), DataType::Utf8),
+            ]);
+        let file_schema = Schema::new(vec![
+            Field::new("a", DataType::Boolean, false),
+            Field::new("year", DataType::Utf8, true),
+            Field::new("b", DataType::Int32, false),
+            Field::new("month", DataType::Utf8, true),
+            Field::new("c", DataType::Float64, true),
+        ]);
+        let config =
+            ListingTableConfig::new(ListingTableUrl::parse("test:///bucket/test/")?)
+                .with_listing_options(opt)
+                .with_schema(Arc::new(file_schema));
+
+        let schema = ListingTable::try_new(config)?.schema();
+        let names = schema
+            .fields()
+            .iter()
+            .map(|f| f.name().as_str())
+            .collect::<Vec<_>>();
+        assert_eq!(names, vec!["a", "b", "c", "month", "year"]);
+
+        Ok(())
+    }
+
     #[cfg(feature = "parquet")]
     #[tokio::test]
     async fn test_table_stats_behaviors() -> Result<()> {

datafusion/core/tests/sql/path_partition.rs

@@ -514,7 +514,11 @@ async fn parquet_statistics() -> Result<()> {
 async fn parquet_overlapping_columns() -> Result<()> {
     let ctx = SessionContext::new();
 
-    // `id` is both a column of the file and a partitioning col
+    // `id` is both a column of the file (Int32, values 0..8) and a partition
+    // column (Int64). Files written with `keep_partition_by_columns = true`
+    // look exactly like this. The partition column wins: it appears once in
+    // the schema, with the partition type, and its values come from the path.
+    // See https://github.com/apache/datafusion/issues/17420
     register_partitioned_alltypes_parquet(
         &ctx,
         &[
@@ -528,12 +532,30 @@ async fn parquet_overlapping_columns() -> Result<()> {
     )
     .await;
 
-    let result = ctx.sql("SELECT id FROM t WHERE id=1 ORDER BY id").await;
+    let schema = ctx.table_provider("t").await?.schema();
+    let id_fields = schema
+        .fields()
+        .iter()
+        .filter(|f| f.name() == "id")
+        .collect::<Vec<_>>();
+    assert_eq!(id_fields.len(), 1, "partition column must appear only once");
+    assert_eq!(id_fields[0].data_type(), &DataType::Int64);
+
+    // All 8 rows of the `id=2` file report the path value, not the file value
+    let result = ctx
+        .sql("SELECT id, count(*) AS n FROM t WHERE id = 2 GROUP BY id")
+        .await?
+        .collect()
+        .await?;
+
+    assert_snapshot!(batches_to_sort_string(&result), @r"
+    +----+---+
+    | id | n |
+    +----+---+
+    | 2  | 8 |
+    +----+---+
+    ");
 
-    assert!(
-        result.is_err(),
-        "Duplicate qualified name should raise error"
-    );
     Ok(())
 }
 

datafusion/sqllogictest/test_files/copy.slt

@@ -228,6 +228,33 @@ select column1, column2 from validate_partitioned_parquet4 order by column1,colu
 ----
 1 a
 
+# The files written above physically contain the partition column (because
+# keep_partition_by_columns was enabled), so a hive-partitioned table over the
+# whole directory must not end up with that column twice.
+# See https://github.com/apache/datafusion/issues/17420
+statement ok
+CREATE EXTERNAL TABLE validate_partitioned_parquet4_hive STORED AS PARQUET
+LOCATION 'test_files/scratch/copy/partitioned_table4/' PARTITIONED BY (column1);
+
+query TT
+select column1, column2 from validate_partitioned_parquet4_hive order by column1, column2;
+----
+1 a
+2 b
+3 c
+
+# Same directory, but letting the factory infer the hive partitions from the path
+statement ok
+CREATE EXTERNAL TABLE validate_partitioned_parquet4_inferred STORED AS PARQUET
+LOCATION 'test_files/scratch/copy/partitioned_table4/';
+
+query TT
+select column1, column2 from validate_partitioned_parquet4_inferred order by column1, column2;
+----
+1 a
+2 b
+3 c
+
 # Copy more files to directory via query
 query I
 COPY (select * from source_table UNION ALL select * from source_table) to 'test_files/scratch/copy/table/' STORED AS PARQUET;