From c4fdbdd475b0addaa55570bc9027301ed3f347ba Mon Sep 17 00:00:00 2001 From: Shayan Gh <98089795+ShayanGho@users.noreply.github.com> Date: Fri, 11 Sep 2026 21:38:16 -0500 Subject: [PATCH] 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 --- datafusion/catalog-listing/src/table.rs | 26 +++++++++++++ .../core/src/datasource/listing/table.rs | 37 +++++++++++++++++++ datafusion/core/tests/sql/path_partition.rs | 34 ++++++++++++++--- datafusion/sqllogictest/test_files/copy.slt | 27 ++++++++++++++ 4 files changed, 118 insertions(+), 6 deletions(-) diff --git a/datafusion/catalog-listing/src/table.rs b/datafusion/catalog-listing/src/table.rs index 6c294fe077db4..b143e3eeb5d29 100644 --- a/datafusion/catalog-listing/src/table.rs +++ b/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 = 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 { diff --git a/datafusion/core/src/datasource/listing/table.rs b/datafusion/core/src/datasource/listing/table.rs index 982766dc88519..3c8a086bda424 100644 --- a/datafusion/core/src/datasource/listing/table.rs +++ b/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 + #[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::>(); + 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<()> { diff --git a/datafusion/core/tests/sql/path_partition.rs b/datafusion/core/tests/sql/path_partition.rs index f8701ccc568a3..137192e83d4d8 100644 --- a/datafusion/core/tests/sql/path_partition.rs +++ b/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::>(); + 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(()) } diff --git a/datafusion/sqllogictest/test_files/copy.slt b/datafusion/sqllogictest/test_files/copy.slt index 49a3c964be3d3..7c65c99731f67 100644 --- a/datafusion/sqllogictest/test_files/copy.slt +++ b/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;