Infer a Parquet column as nullable when some files lack it

ListingTable schema inference merges per-file schemas with
Schema::try_merge, which only widens nullability for fields present in more
than one file. A column that exists, as required, in some files and not at
all in others therefore came out NOT NULL, and reading the files without it
failed with "declared as non-nullable but contains null values". Track
which files carry each field and mark fields that are missing from any file
nullable after the merge.

Author: zhuqi-lucas

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 {

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 {

datafusion/sqllogictest/test_files/schema_evolution.slt

@@ -282,3 +282,39 @@ 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
+
+statement ok
+DROP TABLE inferred_partial;