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..f8bfe59b9222e 100644 --- a/datafusion/core/src/datasource/listing/table.rs +++ b/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 + #[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(()) + } + + /// 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::>(); + assert_eq!(names, vec!["a", "b", "c", "month", "year"]); + + 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;