rgbuilds commented on code in PR #26195:
URL: https://github.com/apache/datafusion/pull/26195#discussion_r4291740783
##########
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:
Good point. Addressed in `5bacdb13b`.
I took the second option and made the invariant explicit: unions are not
representable in Parquet, while the other child-bearing Arrow types must match
exactly during schema merging and therefore cannot reach this pass with nested
evolution.
--
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]
