Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions datafusion/catalog-listing/src/table.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
37 changes: 37 additions & 0 deletions datafusion/core/src/datasource/listing/table.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<()> {
Expand Down
34 changes: 28 additions & 6 deletions datafusion/core/tests/sql/path_partition.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
&[
Expand All @@ -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(())
}

Expand Down
27 changes: 27 additions & 0 deletions datafusion/sqllogictest/test_files/copy.slt
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down