You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/12 01:51:16 UTC
(datafusion) 01/01: fix: infer a Parquet column as nullable when some files lack it (#26104)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-26104-b40ef06a7aaacd7adec278eb8130202d2304326f
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit b295a672ce6fcef216c94aaca88987c98449803c
Author: Qi Zhu <zh...@gmail.com>
AuthorDate: Mon Oct 12 01:50:51 2026 +0000
fix: infer a Parquet column as nullable when some files lack it (#26104)
## Which issue does this PR close?
- Closes #26103
## Rationale for this change
`ListingTable` schema inference over Parquet merges the per-file schemas
with `Schema::try_merge`, which widens nullability only for fields that
appear in more than one schema. A column that is `required` in some
files and absent from the others therefore comes out `NOT NULL`, but
reading the files without it fills it with nulls, and the scan fails:
```
Arrow error: Invalid argument error: Column 'c' is declared as non-nullable but contains null values
```
Pure SQL reproducer (on `main`):
```sql
COPY (SELECT 1 AS id, 10 AS c) TO '/tmp/evo/a.parquet';
COPY (SELECT 2 AS id) TO '/tmp/evo/b.parquet';
CREATE EXTERNAL TABLE t STORED AS PARQUET LOCATION '/tmp/evo/';
SELECT * FROM t; -- fails; DESCRIBE shows c Int64 NO
```
Before DataFusion 52 this was masked by `SchemaAdapter::map_batch`,
which rebuilt batches under the table schema without validating
nullability (removed in #18998). The strict check is right; the inferred
schema is what is wrong.
## What changes are included in this PR?
In `ParquetFormat::infer_schema`, count in how many files each top-level
field appears and, after the merge, mark every field that is missing
from at least one file as nullable. Fields present in every file keep
whatever `try_merge` produced, so tables whose files all share a schema
are unchanged.
## Are these changes tested?
- New `schema_evolution.slt` case: two `COPY TO` files, `CREATE EXTERNAL
TABLE` without a schema, `DESCRIBE` shows the partial column as
nullable, `SELECT` returns the null. Fails on `main` with the error
above.
- New `schema_merge_marks_partially_present_columns_nullable` in
`core/tests/parquet/schema.rs`: three files, column in a different
position, an already-nullable partial column, and both `skip_metadata`
paths.
## Are there any user-facing changes?
An inferred Parquet table schema now reports a column as nullable when
some files do not contain it. Tables where every file has every column
are unaffected. Explicitly declared schemas are not touched.
---
datafusion/core/tests/parquet/schema.rs | 94 ++++++++++++++++++++++
datafusion/datasource-parquet/src/file_format.rs | 48 ++++++++++-
.../sqllogictest/test_files/schema_evolution.slt | 43 ++++++++++
3 files changed, 183 insertions(+), 2 deletions(-)
diff --git a/datafusion/core/tests/parquet/schema.rs b/datafusion/core/tests/parquet/schema.rs
index 29afd39704..58f3d1c087 100644
--- a/datafusion/core/tests/parquet/schema.rs
+++ b/datafusion/core/tests/parquet/schema.rs
@@ -18,6 +18,8 @@
//! Tests for parquet schema handling
use std::{collections::HashMap, fs, path::Path};
+use arrow::array::ArrayRef;
+
use tempfile::TempDir;
use super::*;
@@ -185,6 +187,71 @@ async fn schema_merge_can_preserve_metadata() {
assert_metadata(&actual, &expected_metadata);
}
+/// A column that only some files carry must be inferred as nullable, because
+/// reading the files without it yields nulls, even when every file that has
+/// the column declares it required.
+#[tokio::test]
+async fn schema_merge_marks_partially_present_columns_nullable() {
+ let tmp_dir = TempDir::new().unwrap();
+ let table_dir = tmp_dir.path().join("parquet_test");
+
+ let id = Field::new("id", DataType::Int32, false);
+ let name = Field::new("name", DataType::Utf8, false);
+ let extra = Field::new("extra", DataType::Int32, false);
+ let opt = Field::new("opt", DataType::Int32, true);
+
+ let schemas = vec![
+ // required `extra` and nullable `opt` in the first file only
+ Schema::new(vec![id.clone(), name.clone(), extra.clone(), opt]),
+ // neither column
+ Schema::new(vec![id.clone(), name.clone()]),
+ // required `extra` again, in a different position
+ Schema::new(vec![extra, id, name]),
+ ];
+ write_files_with_columns(table_dir.as_path(), schemas);
+ let table_path = table_dir.to_str().unwrap().to_string();
+
+ for skip_metadata in [true, false] {
+ let options = ParquetReadOptions::default().skip_metadata(skip_metadata);
+ let ctx = SessionContext::new();
+ let df = ctx.read_parquet(&table_path, options).await.unwrap();
+
+ let nullability: Vec<(&str, bool)> = df
+ .schema()
+ .fields()
+ .iter()
+ .map(|f| (f.name().as_str(), f.is_nullable()))
+ .collect();
+ assert_eq!(
+ nullability,
+ vec![
+ ("id", false),
+ ("name", false),
+ ("extra", true),
+ ("opt", true)
+ ],
+ "skip_metadata={skip_metadata}"
+ );
+
+ let actual = df.collect().await.unwrap();
+ let expected = [
+ "+----+------+-------+-----+",
+ "| id | name | extra | opt |",
+ "+----+------+-------+-----+",
+ "| 0 | test | 0 | |",
+ "| 1 | test | | |",
+ "| 2 | test | 2 | |",
+ "+----+------+-------+-----+",
+ ]
+ .join("\n");
+ assert_eq!(
+ batches_to_sort_string(&actual).trim(),
+ expected,
+ "skip_metadata={skip_metadata}"
+ );
+ }
+}
+
fn make_meta(k: impl Into<String>, v: impl Into<String>) -> HashMap<String, String> {
let mut meta = HashMap::new();
meta.insert(k.into(), v.into());
@@ -214,6 +281,33 @@ fn write_files(table_path: &Path, schemas: Vec<Schema>) {
}
}
+/// Writes one file per schema, filling each column from its name: `id` and
+/// `extra` get the file index, `name` gets "test", anything else is null.
+fn write_files_with_columns(table_path: &Path, schemas: Vec<Schema>) {
+ fs::create_dir(table_path).expect("Error creating temp dir");
+
+ for (i, schema) in schemas.into_iter().enumerate() {
+ let schema = Arc::new(schema);
+ let path = table_path.join(format!("part-{i}.parquet"));
+ let file = fs::File::create(path).unwrap();
+ let mut writer = ArrowWriter::try_new(file, schema.clone(), None).unwrap();
+
+ let columns: Vec<ArrayRef> = schema
+ .fields()
+ .iter()
+ .map(|field| match field.name().as_str() {
+ "id" | "extra" => Arc::new(Int32Array::from(vec![i as i32])) as ArrayRef,
+ "name" => Arc::new(StringArray::from(vec!["test"])),
+ _ => Arc::new(Int32Array::from(vec![None::<i32>])),
+ })
+ .collect();
+ let rec_batch = RecordBatch::try_new(schema, columns).unwrap();
+
+ writer.write(&rec_batch).unwrap();
+ writer.close().unwrap();
+ }
+}
+
fn assert_no_metadata(batches: &[RecordBatch]) {
// all batches should have no metadata
for batch in batches {
diff --git a/datafusion/datasource-parquet/src/file_format.rs b/datafusion/datasource-parquet/src/file_format.rs
index e4954a138e..95ba19a975 100644
--- a/datafusion/datasource-parquet/src/file_format.rs
+++ b/datafusion/datasource-parquet/src/file_format.rs
@@ -17,7 +17,7 @@
//! [`ParquetFormat`]: Parquet [`FileFormat`] abstractions
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
use std::fmt;
use std::fmt::Debug;
use std::ops::Range;
@@ -99,7 +99,7 @@ impl FileFormatFactory for ParquetFormatFactory {
fn create(
&self,
state: &dyn Session,
- format_options: &std::collections::HashMap<String, String>,
+ format_options: &HashMap<String, String>,
) -> Result<Arc<dyn FileFormat>> {
let parquet_options = match &self.options {
None => {
@@ -272,6 +272,34 @@ fn clear_metadata(
})
}
+/// Marks every field that is missing from at least one of the `file_count`
+/// files as nullable: reading such a file yields nulls for that column.
+fn mark_partially_present_fields_nullable(
+ schema: Schema,
+ presence: &HashMap<String, usize>,
+ file_count: usize,
+) -> Schema {
+ if presence.values().all(|&count| count == file_count) {
+ return schema;
+ }
+ let metadata = schema.metadata().clone();
+ let fields = schema
+ .fields()
+ .iter()
+ .map(|field| {
+ let present_everywhere = presence
+ .get(field.name())
+ .is_some_and(|&count| count == file_count);
+ if present_everywhere || field.is_nullable() {
+ Arc::clone(field)
+ } else {
+ Arc::new(field.as_ref().clone().with_nullable(true))
+ }
+ })
+ .collect::<Vec<_>>();
+ Schema::new_with_metadata(fields, metadata)
+}
+
#[cfg(feature = "parquet_encryption")]
async fn get_file_decryption_properties(
state: &dyn Session,
@@ -400,6 +428,19 @@ impl FileFormat for ParquetFormat {
}
drop(seen);
+ // A column that some files do not have is read as all-null from those
+ // files, so the merged table schema must declare it nullable even
+ // when every file that has it declares it required. `Schema::try_merge`
+ // only widens nullability for fields that appear in more than one
+ // schema, so track presence separately.
+ let file_count = schemas.len();
+ let mut presence: HashMap<String, usize> = HashMap::new();
+ for (_, schema) in &schemas {
+ for field in schema.fields() {
+ *presence.entry(field.name().clone()).or_insert(0) += 1;
+ }
+ }
+
// Normalize dict-promoted schemas before merging so mixed dict/plain files merge cleanly.
let mut schemas: Vec<Schema> =
schemas.into_iter().map(|(_, schema)| schema).collect();
@@ -413,6 +454,9 @@ impl FileFormat for ParquetFormat {
Schema::try_merge(schemas)
}?;
+ let schema =
+ mark_partially_present_fields_nullable(schema, &presence, file_count);
+
let schema = if self.binary_as_string() {
transform_binary_to_string(&schema)
} else {
diff --git a/datafusion/sqllogictest/test_files/schema_evolution.slt b/datafusion/sqllogictest/test_files/schema_evolution.slt
index 6b13d877b1..dae147091f 100644
--- a/datafusion/sqllogictest/test_files/schema_evolution.slt
+++ b/datafusion/sqllogictest/test_files/schema_evolution.slt
@@ -282,3 +282,46 @@ query TIR rowsort
select * from parquet_table where b = 100 AND c = 10.5;
----
foo 100 10.5
+
+##########
+# Inferred schema: a column that some files lack must be nullable, because
+# reading those files yields nulls for it, even when every file that has the
+# column declares it required.
+##########
+
+statement ok
+COPY (SELECT 1 AS id, 10 AS c)
+TO 'test_files/scratch/schema_evolution/inferred_partial/a.parquet'
+STORED AS PARQUET;
+
+statement ok
+COPY (SELECT 2 AS id)
+TO 'test_files/scratch/schema_evolution/inferred_partial/b.parquet'
+STORED AS PARQUET;
+
+statement ok
+CREATE EXTERNAL TABLE inferred_partial STORED AS PARQUET
+LOCATION 'test_files/scratch/schema_evolution/inferred_partial/';
+
+# `id` is in every file and stays required; `c` is missing from b.parquet.
+query TTT
+DESCRIBE inferred_partial;
+----
+id Int64 NO
+c Int64 YES
+
+query II rowsort
+SELECT id, c FROM inferred_partial;
+----
+1 10
+2 NULL
+
+# With `c` wrongly inferred NOT NULL, the optimizer folds `c IS NULL` to false
+# and this returns no rows.
+query I
+SELECT id FROM inferred_partial WHERE c IS NULL;
+----
+2
+
+statement ok
+DROP TABLE inferred_partial;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org