zhuqi-lucas commented on code in PR #26195:
URL: https://github.com/apache/datafusion/pull/26195#discussion_r4244372355
##########
datafusion/datasource-parquet/src/file_format.rs:
##########
@@ -272,32 +272,83 @@ 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 {
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 {
+ 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[field.name().as_str()];
+ widen_field(field, occurrences, occurrences.len() < sources.len())
+ })
+ .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)
+ }
+ // Arrow rejects evolution within other containers before this pass.
Review Comment:
This comment is not accurate for `Union`. `Field::try_merge` in
`arrow-schema/src/field.rs` recurses into `Struct`, `Union`, `List` and
`LargeList`; `Map` and `FixedSizeList` fall into the equality arm, so the
comment holds for those. A union whose child is missing from some files merges
fine and then lands here, where it is silently left alone, which is the same
bug one container away.
Low risk in practice since Parquet cannot carry a union, but the comment
states an invariant that does not hold, and the next person will rely on it.
Either widen unions too, or name the types that are actually rejected.
##########
datafusion/datasource-parquet/src/file_format.rs:
##########
@@ -272,32 +272,83 @@ 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 {
Review Comment:
The old fast path is gone. It returned early when every field was present in
every file, using the precomputed counts, which is the common case of all files
sharing a schema. Now only the single-source case returns early, so identical
schemas still walk the whole tree, build a `HashMap` per struct level and
allocate a fresh `Fields` per level. That looks like the 2.4-3.0% the
description measures.
Cheap to restore: compare the field name sets across sources first, or
short-circuit a level when every `widen_field` returned the same `Arc`.
##########
datafusion/datasource-parquet/src/file_format.rs:
##########
@@ -272,32 +272,83 @@ 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 {
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 {
+ 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[field.name().as_str()];
Review Comment:
Unguarded map index, so a missing key panics rather than erroring. I could
not construct a reachable case, since a merged field name always comes from
some source, but that leans on a `try_merge` invariant instead of enforcing it
here. `.get()` with an empty slice treated as missing is both safer and the
semantically right default.
Separately, and this predates the PR so take it or leave it: the loop above
pushes per field rather than per source, so a file carrying two fields of the
same name contributes two occurrences. `occurrences.len() < sources.len()` can
then be false even though another file lacks the field, leaving it required.
Counting distinct sources would fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
