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/21 05:40:35 UTC
(datafusion) 01/01: fix: infer missing nested Parquet fields as nullable (#26195)
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-26195-f9bc5a3fc60cc7f4319cd2c156cb3d024a8c1b21
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 99193cbd0ec4f7bc46b46c0fd2efc35014515b3d
Author: Rishabh Gupta <ri...@gmail.com>
AuthorDate: Wed Oct 21 05:40:12 2026 +0000
fix: infer missing nested Parquet fields as nullable (#26195)
## Which issue does this PR close?
- Closes #26172.
## Rationale for this change
When a required child is missing from one Parquet file but its
containing struct exists in both files, schema inference leaves the
child non-nullable. Selecting that child then fails, and an `IS NULL`
predicate can incorrectly return no rows. With this change, selecting
`s.y` returns `10` and `NULL`, and `WHERE s.y IS NULL` returns the row
from the file without `y`.
## What changes are included in this PR?
### Implementation
- Extend partial-field nullability correction recursively through
structs, lists, and large lists.
- Compare children only across schemas containing their parent. A
missing parent becomes nullable without weakening required children or
list elements; a missing struct child within a present parent becomes
nullable.
- Preserve existing field/schema metadata and existing nullability.
Other container types retain their current merge behavior.
- Keep the identical-schema path inexpensive by returning early when the
recursive field trees are equal, both at the schema root and within
unchanged nested levels.
- Treat a missing occurrence-list entry defensively as an absent field
instead of indexing the map directly.
### Performance
- A fresh optimized synthetic rerun compared local PR head `c14317f1f`
with current `upstream/main` at `cdc2fc041`. Identical flat schemas were
faster in the PR build: 32 files x 128 fields took 1.737 -> 1.650 ms
(-4.99%), and 256 files x 1,024 fields took 317.490 -> 308.674 ms
(-2.78%).
- Evolving nested schemas retained a small inference cost: 32 files x
128 children took 1.925 -> 1.966 ms (+2.11%, approximately 0.041 ms),
and 256 files x 128 children took 15.610 -> 16.557 ms (+6.07%,
approximately 0.947 ms).
- An isolated pre-review/post-review comparison on the same base
confirmed that the recursive-equality fast path improved the two
identical-schema cases by 2.78% and 1.18%. Evolving nested cases cannot
take that shortcut and were 5.19% and 7.75% slower in this rerun.
- Each reported value is the median of three process medians; each
process used 5 warm-ups followed by 40 measured schema-inference calls.
Measurements used an Apple M4 Mac Mini, `release-nonlto`, in-memory
one-row Parquet files, and warmed metadata.
- The benchmark isolates schema inference. It does not measure
production query throughput or establish statistical significance. The
change adds no footer reads and no scan-time correction pass.
- [Benchmark methodology and detailed
results](https://drive.google.com/file/d/1Vx-ya0ZjglcwX1VzI6YutT403DyZqvuR/view?usp=sharing)
— synthetic data, experiment setup, timings, and limitations. (Benchmark
preparation and report drafting assisted by Codex)
## What is the testing strategy for this PR?
- Unit tests cover missing nested children, absent parents, independent
nested paths, metadata preservation, required list elements, and the
defensive missing-map-entry path.
- A Parquet integration regression checks struct/list/large-list
projection and `IS NULL` across 12 configurations (metadata and filter
pushdown on/off). It fails without the fix.
- A SQL logic regression verifies that a missing child projects as
`NULL` and that `IS NULL` returns the corresponding row.
- Focused unit, Parquet integration, and SQL logic tests passed.
- All 534 SQL logic test files passed.
- Extended workspace Rust tests passed with the repository's required
feature set. Four path-sensitive snapshot cases also passed after
rerunning with canonical workspace environment paths.
- `cargo fmt --all` passed. The repository lint workflow completed
through the available checks; documentation HTML generation was blocked
locally by a missing `cargo-depgraph` integration. The literal
all-target/all-feature Clippy command reports three pre-existing
`mutable_key_type` errors in unchanged upstream optimizer code.
## Are there any user-facing changes?
Inferred nested fields that are absent from some containing Parquet
schemas now become nullable. This allows missing values to be read as
`NULL` and makes `IS NULL` queries return the correct rows. Explicit
schemas and public APIs are unchanged.
---
datafusion/core/tests/parquet/schema.rs | 131 +++++++++++-
datafusion/datasource-parquet/src/file_format.rs | 238 +++++++++++++++++----
.../test_files/schema_evolution_nested.slt | 29 +++
3 files changed, 361 insertions(+), 37 deletions(-)
diff --git a/datafusion/core/tests/parquet/schema.rs b/datafusion/core/tests/parquet/schema.rs
index 58f3d1c087..0ec969d5b6 100644
--- a/datafusion/core/tests/parquet/schema.rs
+++ b/datafusion/core/tests/parquet/schema.rs
@@ -18,7 +18,9 @@
//! Tests for parquet schema handling
use std::{collections::HashMap, fs, path::Path};
-use arrow::array::ArrayRef;
+use arrow::array::{Array, ArrayRef, LargeListArray, ListArray, StructArray};
+use arrow::buffer::OffsetBuffer;
+use parquet::file::properties::{EnabledStatistics, WriterProperties};
use tempfile::TempDir;
@@ -325,3 +327,130 @@ fn assert_metadata(batches: &[RecordBatch], expected_metadata: &HashMap<String,
assert_eq!(batch.schema().metadata(), expected_metadata,);
}
}
+
+#[tokio::test]
+async fn schema_merge_marks_missing_nested_children_nullable() {
+ for kind in ["struct", "list", "large_list"] {
+ let dir = TempDir::new().unwrap();
+ for (name, id, with_y) in [("a.parquet", 1, true), ("b.parquet", 2, false)] {
+ let mut fields = vec![Arc::new(Field::new("x", DataType::Int32, false))];
+ let mut arrays: Vec<ArrayRef> = vec![Arc::new(Int32Array::from(vec![id]))];
+ if with_y {
+ fields.push(Arc::new(Field::new("y", DataType::Int32, false)));
+ arrays.push(Arc::new(Int32Array::from(vec![10])));
+ }
+ let s = StructArray::new(fields.into(), arrays, None);
+ let s: ArrayRef = match kind {
+ "struct" => Arc::new(s),
+ "list" => Arc::new(ListArray::new(
+ Arc::new(Field::new("item", s.data_type().clone(), false)),
+ OffsetBuffer::new(vec![0_i32, 1].into()),
+ Arc::new(s),
+ None,
+ )),
+ "large_list" => Arc::new(LargeListArray::new(
+ Arc::new(Field::new("item", s.data_type().clone(), false)),
+ OffsetBuffer::new(vec![0_i64, 1].into()),
+ Arc::new(s),
+ None,
+ )),
+ _ => unreachable!(),
+ };
+ write_nested_file(
+ &dir,
+ name,
+ vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("s", s.data_type().clone(), false),
+ ],
+ vec![Arc::new(Int32Array::from(vec![id])), s],
+ EnabledStatistics::Page,
+ );
+ }
+ for skip_metadata in [true, false] {
+ for pushdown in [true, false] {
+ let config = SessionConfig::new()
+ .set_bool("datafusion.execution.parquet.pushdown_filters", pushdown);
+ let ctx = SessionContext::new_with_config(config);
+ ctx.register_parquet(
+ "t",
+ dir.path().to_str().unwrap(),
+ ParquetReadOptions::default().skip_metadata(skip_metadata),
+ )
+ .await
+ .unwrap();
+ let df = ctx.table("t").await.unwrap();
+ let mut data_type = df
+ .schema()
+ .field_with_unqualified_name("s")
+ .unwrap()
+ .data_type();
+ if let DataType::List(item) | DataType::LargeList(item) = data_type {
+ assert!(!item.is_nullable());
+ data_type = item.data_type();
+ }
+ let DataType::Struct(fields) = data_type else {
+ panic!("expected struct")
+ };
+ let child = if kind == "struct" {
+ "s.y"
+ } else {
+ "get_field(s[1], 'y')"
+ };
+ assert!(!fields[0].is_nullable());
+ assert!(fields[1].is_nullable());
+ let batches = ctx
+ .sql(&format!("SELECT id, {child} AS y FROM t ORDER BY id"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ batches_to_sort_string(&batches).trim(),
+ "+----+----+\n| id | y |\n+----+----+\n| 1 | 10 |\n| 2 | |\n+----+----+"
+ );
+ let batches = ctx
+ .sql(&format!("SELECT id FROM t WHERE {child} IS NULL"))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
+ assert_eq!(
+ batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 2
+ );
+ }
+ }
+ }
+}
+
+fn write_nested_file(
+ dir: &TempDir,
+ name: &str,
+ fields: Vec<Field>,
+ arrays: Vec<ArrayRef>,
+ statistics: EnabledStatistics,
+) {
+ let schema = Arc::new(Schema::new(fields));
+ let batch = RecordBatch::try_new(schema.clone(), arrays).unwrap();
+ let props = WriterProperties::builder()
+ .set_max_row_group_row_count(Some(1))
+ .set_statistics_enabled(statistics)
+ .build();
+ let mut writer = ArrowWriter::try_new(
+ fs::File::create(dir.path().join(name)).unwrap(),
+ schema,
+ Some(props),
+ )
+ .unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+}
diff --git a/datafusion/datasource-parquet/src/file_format.rs b/datafusion/datasource-parquet/src/file_format.rs
index 76d0e83134..aa8e82c970 100644
--- a/datafusion/datasource-parquet/src/file_format.rs
+++ b/datafusion/datasource-parquet/src/file_format.rs
@@ -33,7 +33,7 @@ pub use crate::schema_coercion::{
pub use crate::sink::ParquetSink;
-use arrow::datatypes::{Fields, Metadata, Schema, SchemaRef};
+use arrow::datatypes::{DataType, FieldRef, Fields, Metadata, Schema, SchemaRef};
use datafusion_datasource::TableSchema;
use datafusion_datasource::file_compression_type::FileCompressionType;
use datafusion_datasource::file_sink_config::FileSinkConfig;
@@ -271,32 +271,100 @@ 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) {
+/// Widen fields missing from some of their containing schemas. An absent parent
+/// produces a null parent, not a null value for each required child, so children
+/// are compared only across files in which their parent exists.
+fn mark_partially_present_fields_nullable(schema: Schema, sources: &[Schema]) -> Schema {
+ if sources.len() <= 1
+ || sources[1..]
+ .iter()
+ .all(|source| source.fields() == sources[0].fields())
+ {
return schema;
}
- let metadata = schema.metadata().clone();
- let fields = schema
- .fields()
+ let fields = widen_fields(
+ schema.fields(),
+ &sources.iter().map(Schema::fields).collect::<Vec<_>>(),
+ );
+ Schema::new_with_metadata(fields, schema.metadata().clone())
+}
+
+fn widen_fields(fields: &Fields, sources: &[&Fields]) -> Fields {
+ if !sources.is_empty()
+ && fields == sources[0]
+ && sources[1..].iter().all(|source| *source == sources[0])
+ {
+ return fields.clone();
+ }
+
+ let mut present: HashMap<&str, Vec<&FieldRef>> = HashMap::new();
+ for source in sources {
+ for field in *source {
+ present.entry(field.name()).or_default().push(field);
+ }
+ }
+ 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)
+ let occurrences = present
+ .get(field.name().as_str())
+ .map(Vec::as_slice)
+ .unwrap_or_default();
+ let missing = occurrences.is_empty() || occurrences.len() < sources.len();
+ widen_field(field, occurrences, missing)
+ })
+ .collect()
+}
+
+fn widen_field(field: &FieldRef, sources: &[&FieldRef], missing: bool) -> FieldRef {
+ let source_types = sources.iter().map(|f| f.data_type()).collect::<Vec<_>>();
+ let data_type = widen_nested_type(field.data_type(), &source_types);
+ let nullable = field.is_nullable() || missing;
+ if &data_type == field.data_type() && nullable == field.is_nullable() {
+ Arc::clone(field)
+ } else {
+ Arc::new(
+ field
+ .as_ref()
+ .clone()
+ .with_data_type(data_type)
+ .with_nullable(nullable),
+ )
+ }
+}
+
+fn widen_nested_type(data_type: &DataType, sources: &[&DataType]) -> DataType {
+ match data_type {
+ DataType::Struct(fields) => {
+ let sources = sources
+ .iter()
+ .filter_map(|t| match t {
+ DataType::Struct(fields) => Some(fields),
+ _ => None,
+ })
+ .collect::<Vec<_>>();
+ DataType::Struct(widen_fields(fields, &sources))
+ }
+ DataType::List(field) | DataType::LargeList(field) => {
+ let sources = sources
+ .iter()
+ .filter_map(|t| match t {
+ DataType::List(f) | DataType::LargeList(f) => Some(f),
+ _ => None,
+ })
+ .collect::<Vec<_>>();
+ let field = widen_field(field, &sources, false);
+ if matches!(data_type, DataType::List(_)) {
+ DataType::List(field)
} else {
- Arc::new(field.as_ref().clone().with_nullable(true))
+ DataType::LargeList(field)
}
- })
- .collect::<Vec<_>>();
- Schema::new_with_metadata(fields, metadata)
+ }
+ // Parquet cannot represent unions. Arrow requires exact type equality
+ // for ListView, LargeListView, FixedSizeList, Map, Dictionary, and
+ // RunEndEncoded, so nested evolution in those types is rejected earlier.
+ _ => data_type.clone(),
+ }
}
#[cfg(feature = "parquet_encryption")]
@@ -423,19 +491,6 @@ 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();
@@ -443,14 +498,14 @@ impl FileFormat for ParquetFormat {
schemas = crate::schema_coercion::uniform_dict_schemas(schemas);
}
+ let source_schemas = schemas.clone();
let schema = if self.skip_metadata() {
Schema::try_merge(clear_metadata(schemas))
} else {
Schema::try_merge(schemas)
}?;
- let schema =
- mark_partially_present_fields_nullable(schema, &presence, file_count);
+ let schema = mark_partially_present_fields_nullable(schema, &source_schemas);
let schema = if self.binary_as_string() {
transform_binary_to_string(&schema)
@@ -813,3 +868,114 @@ impl From<&ParquetFormatFactory> for protobuf::TableParquetOptions {
}
}
}
+
+#[cfg(test)]
+mod schema_presence_tests {
+ use super::*;
+ use arrow::datatypes::Field;
+
+ fn structure(include_y: bool) -> DataType {
+ let mut fields = vec![Field::new("x", DataType::Int32, false)];
+ if include_y {
+ fields.push(
+ Field::new("y", DataType::Int32, false)
+ .with_metadata(HashMap::from([("tag".into(), "retained".into())])),
+ );
+ }
+ DataType::Struct(fields.into())
+ }
+
+ #[test]
+ fn nested_presence_respects_parent_and_container_constraints() {
+ for wrapper in ["struct", "list", "large_list"] {
+ let make = |include_y| {
+ let nested = structure(include_y);
+ let item = Arc::new(Field::new("item", nested.clone(), false));
+ let dt = match wrapper {
+ "struct" => nested,
+ "list" => DataType::List(item),
+ "large_list" => DataType::LargeList(item),
+ _ => unreachable!(),
+ };
+ Schema::new(vec![Field::new("s", dt, false)])
+ };
+ let sources = vec![make(true), make(false), Schema::empty()];
+ let merged = Schema::try_merge(sources.clone())
+ .unwrap()
+ .with_metadata(HashMap::from([("schema".into(), "retained".into())]));
+ let result = mark_partially_present_fields_nullable(merged, &sources);
+ assert_eq!(result.metadata()["schema"], "retained");
+ let parent = result.field(0);
+ assert!(parent.is_nullable(), "{wrapper}");
+ let nested = match parent.data_type() {
+ DataType::Struct(_) => parent.data_type(),
+ DataType::List(f) | DataType::LargeList(f) => {
+ assert!(!f.is_nullable());
+ f.data_type()
+ }
+ _ => panic!(),
+ };
+ let DataType::Struct(children) = nested else {
+ panic!()
+ };
+ assert!(!children[0].is_nullable(), "{wrapper}");
+ assert!(children[1].is_nullable(), "{wrapper}");
+ assert_eq!(children[1].metadata()["tag"], "retained");
+ }
+ }
+
+ #[test]
+ fn nested_paths_are_independent() {
+ let sources = [true, false].map(|include_y| {
+ Schema::new(vec![
+ Field::new(
+ "a",
+ DataType::Struct(
+ vec![Field::new("s", structure(include_y), false)].into(),
+ ),
+ false,
+ ),
+ Field::new("b", structure(true), false),
+ Field::new("a.s.y", DataType::Int32, false),
+ ])
+ });
+ let merged = Schema::try_merge(sources.clone()).unwrap();
+ let result = mark_partially_present_fields_nullable(merged, &sources);
+ let DataType::Struct(a) = result.field(0).data_type() else {
+ panic!()
+ };
+ let DataType::Struct(children) = a[0].data_type() else {
+ panic!()
+ };
+ assert!(children[1].is_nullable());
+ let DataType::Struct(b) = result.field(1).data_type() else {
+ panic!()
+ };
+ assert!(!b[1].is_nullable());
+ assert!(!result.field(2).is_nullable());
+ }
+
+ #[test]
+ fn absent_parent_does_not_widen_required_children() {
+ let sources = vec![
+ Schema::new(vec![Field::new("s", structure(true), false)]),
+ Schema::empty(),
+ ];
+ let merged = Schema::try_merge(sources.clone()).unwrap();
+ let result = mark_partially_present_fields_nullable(merged, &sources);
+ assert!(result.field(0).is_nullable());
+ let DataType::Struct(children) = result.field(0).data_type() else {
+ panic!()
+ };
+ assert!(children.iter().all(|f| !f.is_nullable()));
+ }
+
+ #[test]
+ fn missing_merged_field_is_defensively_nullable() {
+ let fields = Fields::from(vec![Field::new("missing", DataType::Int32, false)]);
+ let source = Fields::from(vec![Field::new("present", DataType::Int32, false)]);
+ let result = widen_fields(&fields, &[&source]);
+
+ assert!(result[0].is_nullable());
+ }
+}
diff --git a/datafusion/sqllogictest/test_files/schema_evolution_nested.slt b/datafusion/sqllogictest/test_files/schema_evolution_nested.slt
index ce6b50d5d1..efed9f44c2 100644
--- a/datafusion/sqllogictest/test_files/schema_evolution_nested.slt
+++ b/datafusion/sqllogictest/test_files/schema_evolution_nested.slt
@@ -247,3 +247,32 @@ RESET datafusion.execution.parquet.enable_page_index;
statement ok
RESET datafusion.execution.parquet.pushdown_filters;
+
+# Inferred nested fields absent from a file must allow NULL, including predicates.
+statement ok
+COPY (SELECT 1 AS id, named_struct('x', 1, 'y', 10) AS s)
+TO 'test_files/scratch/schema_evolution_nested/inferred_child/a.parquet'
+STORED AS PARQUET;
+
+statement ok
+COPY (SELECT 2 AS id, named_struct('x', 2) AS s)
+TO 'test_files/scratch/schema_evolution_nested/inferred_child/b.parquet'
+STORED AS PARQUET;
+
+statement ok
+CREATE EXTERNAL TABLE inferred_child STORED AS PARQUET
+LOCATION 'test_files/scratch/schema_evolution_nested/inferred_child/';
+
+query II rowsort
+SELECT id, s.y FROM inferred_child;
+----
+1 10
+2 NULL
+
+query I
+SELECT id FROM inferred_child WHERE s.y IS NULL;
+----
+2
+
+statement ok
+DROP TABLE inferred_child;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org