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

Files written with `keep_partition_by_columns = true` physically contain
the partition column. `ListingTable::try_new` appended the configured
partition columns to the inferred file schema unconditionally, so the
table schema listed the column twice and any query failed with
"Schema contains duplicate qualified field name".

The partition column now appears once, with the declared partition type,
and its values come from the path, matching what CREATE EXTERNAL TABLE
with an explicit column list already did.

Closes #17420

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Author: ShayanGho

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,43 @@ 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(())
+    }
+
     #[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;