This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 12ba4b74d8 fix: reject duplicate field names during CSV and Parquet
schema inference (#24387)
12ba4b74d8 is described below
commit 12ba4b74d8383ffaecca7a51da7cfec324ec8f26
Author: batu <[email protected]>
AuthorDate: Thu Aug 20 18:27:49 2026 +0000
fix: reject duplicate field names during CSV and Parquet schema inference
(#24387)
## Which issue does this PR close?
- Closes #24381.
- Closes #12852.
## Rationale for this change
When a Parquet file has multiple columns with the same name and
compatible types, DataFusion merges them during schema inference. The
scan succeeds without warning but returns only one of those columns.
PyArrow can produce such a file via its public API:
```python
left = pa.table({"id": [1, 2, 3], "value": [10, 20, 30]})
right = pa.table({"id": [1, 2, 3], "value": [100, 200, 300]})
joined = left.join(right, keys="id") # left_suffix/right_suffix default
to None
pq.write_table(joined, path) # writes id, value, value
```
DataFusion 54.0.0 returns ['id', 'value'] for this file; the [100, 200,
300] column is missing.
Other readers either preserve the data or reject the file:
| Reader | Result |
| --- | --- |
| `pyarrow` `ParquetFile.read()` | keeps all three columns |
| `pyarrow` `pq.read_table()` | `ArrowInvalid: Multiple matches for
FieldRef.Name(value)` |
| `pyarrow` dataset | `ArrowInvalid: Can't unify schema with duplicate
field names` |
| `duckdb` | keeps all three, renames the second to `value_1` |
| `polars` | `DuplicateError` |
| `datafusion` | **succeeds, missing column(s)** |
The same bug is behind two issues #24381 and #12852. CSV and Parquet
schema inference both pass each file's schema to `Schema::try_merge`,
which matches fields by name. Duplicate names in one file are merged
before the scan. Thus, column(s) disappear silently.
## What changes are included in this PR?
This PR adds `ensure_unique_field_names` to `datafusion-datasource`. It
is called for each inferred CSV and Parquet schema before the merge. It
uses the existing `SchemaError::DuplicateUnqualifiedField` error and
adds the file location to the message.
Duplicate names are rejected. The PR does not rename columns or change
how fields with unique names are merged across files.
## Are these changes tested?
Yes. Two regression tests in order to cover the CSV and Parquet paths:
- `datafusion/core/src/datasource/file_format/parquet.rs` —
`infer_schema_rejects_duplicate_field_names`
- `datafusion/core/src/datasource/file_format/csv.rs` —
`infer_schema_rejects_duplicate_header_names`
Both tests create their inputs in a temporary directory; no fixtures are
added. Both fail on `main` and pass with this change. On `main`, the CSV
test infers `Schema { fields: [id, value] }` from a three-column header,
matching the behavior reported in #12852.
## Are there any user-facing changes?
Yes. CSV and Parquet schema inference now returns a clear error when an
inferred
file schema repeats a field name. The error names the file and the
repeated
column instead of returning an incomplete schema.
---------
Co-authored-by: Andrew Lamb <[email protected]>
---
datafusion/core/src/datasource/file_format/csv.rs | 27 +++++++++++++++++
.../core/src/datasource/file_format/parquet.rs | 34 ++++++++++++++++++++++
datafusion/core/tests/data/duplicate_header.csv | 2 ++
datafusion/datasource-csv/src/file_format.rs | 15 +++++++++-
datafusion/datasource-parquet/src/file_format.rs | 16 +++++++++-
datafusion/datasource/src/file_format.rs | 29 ++++++++++++++++--
datafusion/sqllogictest/test_files/csv_files.slt | 7 +++++
7 files changed, 125 insertions(+), 5 deletions(-)
diff --git a/datafusion/core/src/datasource/file_format/csv.rs
b/datafusion/core/src/datasource/file_format/csv.rs
index 90d7eb3b41..447d5244f9 100644
--- a/datafusion/core/src/datasource/file_format/csv.rs
+++ b/datafusion/core/src/datasource/file_format/csv.rs
@@ -1628,4 +1628,31 @@ mod tests {
Ok(())
}
+
+ #[tokio::test]
+ async fn infer_schema_rejects_duplicate_header_names() -> Result<()> {
+ let directory = tempfile::tempdir()?;
+ let path = directory.path().join("duplicate_header.csv");
+ std::fs::write(&path, "id,value,value\n1,10,100\n")?;
+
+ let store = Arc::new(LocalFileSystem::new()) as _;
+ let meta = crate::test::object_store::local_unpartitioned_file(&path);
+
+ let ctx = SessionContext::new().state();
+ let error = CsvFormat::default()
+ .with_has_header(true)
+ .infer_schema(&ctx, &store, std::slice::from_ref(&meta))
+ .await
+ .expect_err("duplicate header names must not infer a schema")
+ .to_string();
+
+ assert!(
+ error.contains("duplicate unqualified field name")
+ && error.contains("value")
+ && error.contains("duplicate_header.csv"),
+ "unexpected error: {error}"
+ );
+
+ Ok(())
+ }
}
diff --git a/datafusion/core/src/datasource/file_format/parquet.rs
b/datafusion/core/src/datasource/file_format/parquet.rs
index 0f5db4a057..bbd9d0937a 100644
--- a/datafusion/core/src/datasource/file_format/parquet.rs
+++ b/datafusion/core/src/datasource/file_format/parquet.rs
@@ -1822,4 +1822,38 @@ mod tests {
Ok(())
}
+
+ #[tokio::test]
+ async fn infer_schema_rejects_duplicate_field_names() -> Result<()> {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int64, false),
+ Field::new("value", DataType::Int64, false),
+ Field::new("value", DataType::Int64, false),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int64Array::from(vec![1, 2, 3])) as ArrayRef,
+ Arc::new(Int64Array::from(vec![10, 20, 30])) as ArrayRef,
+ Arc::new(Int64Array::from(vec![100, 200, 300])) as ArrayRef,
+ ],
+ )?;
+
+ let store = Arc::new(LocalFileSystem::new()) as _;
+ let (meta, _files) = store_parquet(vec![batch], false).await?;
+
+ let ctx = SessionContext::new().state();
+ let error = ParquetFormat::default()
+ .infer_schema(&ctx, &store, &meta)
+ .await
+ .expect_err("duplicate field names must not infer a schema")
+ .to_string();
+
+ assert!(
+ error.contains("duplicate unqualified field name") &&
error.contains("value"),
+ "unexpected error: {error}"
+ );
+
+ Ok(())
+ }
}
diff --git a/datafusion/core/tests/data/duplicate_header.csv
b/datafusion/core/tests/data/duplicate_header.csv
new file mode 100644
index 0000000000..0b2719a8ec
--- /dev/null
+++ b/datafusion/core/tests/data/duplicate_header.csv
@@ -0,0 +1,2 @@
+id,value,value
+1,10,100
diff --git a/datafusion/datasource-csv/src/file_format.rs
b/datafusion/datasource-csv/src/file_format.rs
index c0d22b80f0..a094fdc3bf 100644
--- a/datafusion/datasource-csv/src/file_format.rs
+++ b/datafusion/datasource-csv/src/file_format.rs
@@ -41,6 +41,7 @@ use datafusion_datasource::file::FileSource;
use datafusion_datasource::file_compression_type::FileCompressionType;
use datafusion_datasource::file_format::{
DEFAULT_SCHEMA_INFER_MAX_RECORD, FileFormat, FileFormatFactory,
+ ensure_unique_field_names,
};
use datafusion_datasource::file_scan_config::{FileScanConfig,
FileScanConfigBuilder};
use datafusion_datasource::file_sink_config::{FileSink, FileSinkConfig};
@@ -397,12 +398,24 @@ impl FileFormat for CsvFormat {
)
})?;
records_to_read -= records_read;
- schemas.push(schema);
+ schemas.push((&object.location, schema));
if records_to_read == 0 {
break;
}
}
+ let mut seen = HashSet::new();
+ for (location, schema) in &schemas {
+ ensure_unique_field_names(schema, &mut seen).map_err(|err| {
+ DataFusionError::Context(
+ format!("Error when processing CSV file {location}"),
+ Box::new(err),
+ )
+ })?;
+ }
+ drop(seen);
+
+ let schemas = schemas.into_iter().map(|(_, schema)| schema);
let merged_schema = Schema::try_merge(schemas)?;
Ok(Arc::new(merged_schema))
}
diff --git a/datafusion/datasource-parquet/src/file_format.rs
b/datafusion/datasource-parquet/src/file_format.rs
index 6358201c06..1b25a2c632 100644
--- a/datafusion/datasource-parquet/src/file_format.rs
+++ b/datafusion/datasource-parquet/src/file_format.rs
@@ -17,6 +17,7 @@
//! [`ParquetFormat`]: Parquet [`FileFormat`] abstractions
+use std::collections::HashSet;
use std::fmt;
use std::fmt::Debug;
use std::ops::Range;
@@ -37,7 +38,9 @@ use datafusion_datasource::TableSchema;
use datafusion_datasource::file_compression_type::FileCompressionType;
use datafusion_datasource::file_sink_config::FileSinkConfig;
-use datafusion_datasource::file_format::{FileFormat, FileFormatFactory};
+use datafusion_datasource::file_format::{
+ FileFormat, FileFormatFactory, ensure_unique_field_names,
+};
use datafusion_common::Statistics;
use datafusion_common::config::{ConfigField, ConfigFileType,
TableParquetOptions};
@@ -388,6 +391,17 @@ impl FileFormat for ParquetFormat {
schemas
.sort_unstable_by(|(location1, _), (location2, _)|
location1.cmp(location2));
+ let mut seen = HashSet::new();
+ for (location, schema) in &schemas {
+ ensure_unique_field_names(schema, &mut seen).map_err(|err| {
+ DataFusionError::Context(
+ format!("Error when processing Parquet file {location}"),
+ Box::new(err),
+ )
+ })?;
+ }
+ drop(seen);
+
let schemas = schemas.into_iter().map(|(_, schema)| schema);
let schema = if self.skip_metadata() {
diff --git a/datafusion/datasource/src/file_format.rs
b/datafusion/datasource/src/file_format.rs
index dd30881610..2c2a94e8cc 100644
--- a/datafusion/datasource/src/file_format.rs
+++ b/datafusion/datasource/src/file_format.rs
@@ -19,7 +19,7 @@
//! See write.rs for write related helper methods
use std::any::Any;
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
use std::fmt;
use std::sync::Arc;
@@ -28,9 +28,11 @@ use crate::file_compression_type::FileCompressionType;
use crate::file_scan_config::FileScanConfig;
use crate::file_sink_config::FileSinkConfig;
-use arrow::datatypes::SchemaRef;
+use arrow::datatypes::{Schema, SchemaRef};
use datafusion_common::file_options::file_type::FileType;
-use datafusion_common::{GetExt, Result, Statistics, internal_err,
not_impl_err};
+use datafusion_common::{
+ GetExt, Result, SchemaError, Statistics, internal_err, not_impl_err,
schema_err,
+};
use datafusion_physical_expr::LexRequirement;
use datafusion_physical_expr_common::sort_expr::LexOrdering;
use datafusion_physical_plan::ExecutionPlan;
@@ -42,6 +44,27 @@ use object_store::{ObjectMeta, ObjectStore};
/// Default max records to scan to infer the schema
pub const DEFAULT_SCHEMA_INFER_MAX_RECORD: usize = 1000;
+/// Rejects an inferred schema that names the same field more than once.
+///
+/// [`Schema::try_merge`] coalesces fields by name, so callers validate each
+/// inferred file schema before merging. `seen` is cleared before use so
callers
+/// can reuse its allocation across schemas.
+pub fn ensure_unique_field_names<'a>(
+ schema: &'a Schema,
+ seen: &mut HashSet<&'a str>,
+) -> Result<()> {
+ seen.clear();
+ seen.reserve(schema.fields().len());
+ for field in schema.fields() {
+ if !seen.insert(field.name()) {
+ return schema_err!(SchemaError::DuplicateUnqualifiedField {
+ name: field.name().clone(),
+ });
+ }
+ }
+ Ok(())
+}
+
/// Metadata fetched from a file, including statistics and ordering.
///
/// This struct is returned by [`FileFormat::infer_stats_and_ordering`] to
diff --git a/datafusion/sqllogictest/test_files/csv_files.slt
b/datafusion/sqllogictest/test_files/csv_files.slt
index af2c6d41af..a64f1df6e4 100644
--- a/datafusion/sqllogictest/test_files/csv_files.slt
+++ b/datafusion/sqllogictest/test_files/csv_files.slt
@@ -15,6 +15,13 @@
# specific language governing permissions and limitations
# under the License.
+# duplicate_header_is_rejected
+statement error Schema contains duplicate unqualified field name value
+CREATE EXTERNAL TABLE csv_with_duplicate_header
+STORED AS CSV
+LOCATION '../core/tests/data/duplicate_header.csv'
+OPTIONS ('format.has_header' 'true');
+
# create_external_table_with_quote_escape
statement ok
CREATE EXTERNAL TABLE csv_with_quote (
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
