Fix parquet statistics for ListingTable and Utf8View with schema_force_string_view, rename config option to schema_force_view_types

datafusion/common/src/config.rs

@@ -380,6 +380,10 @@ config_namespace! {
         /// the filters are applied in the same order as written in the query
         pub reorder_filters: bool, default = false
 
+        /// (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`,
+        /// and `Binary/BinaryLarge` with `BinaryView`.
+        pub schema_force_string_view: bool, default = false
+
         // The following options affect writing to parquet files
         // and map to parquet::file::properties::WriterProperties
 
@@ -483,10 +487,6 @@ config_namespace! {
         /// writing out already in-memory data, such as from a cached
         /// data frame.
         pub maximum_buffered_record_batches_per_stream: usize, default = 2
-
-        /// (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`,
-        /// and `Binary/BinaryLarge` with `BinaryView`.
-        pub schema_force_string_view: bool, default = false
     }
 }
 

datafusion/core/src/datasource/file_format/mod.rs

@@ -229,6 +229,53 @@ pub fn transform_schema_to_view(schema: &Schema) -> Schema {
     Schema::new_with_metadata(transformed_fields, schema.metadata.clone())
 }
 
+/// Merge table schema into the parquet file schema.
+///
+/// This coerces the file schema if the table schema uses a view type.
+pub(crate) fn merge_file_schema_on_view_type(
+    table_schema: &Schema,
+    file_schema: &Schema,
+) -> Option<Schema> {
+    let mut transform = false;
+    let table_fields: HashMap<_, _> = table_schema
+        .fields
+        .iter()
+        .map(|f| {
+            let dt = f.data_type();
+            if dt.equals_datatype(&DataType::Utf8View) {
+                transform = true;
+            }
+            (f.name(), dt)
+        })
+        .collect();
+    if !transform {
+        return None;
+    }
+
+    let transformed_fields: Vec<Arc<Field>> = file_schema
+        .fields
+        .iter()
+        .map(
+            |field| match (table_fields.get(field.name()), field.data_type()) {
+                (Some(DataType::Utf8View), DataType::Utf8)
+                | (Some(DataType::Utf8View), DataType::LargeUtf8) => Arc::new(
+                    Field::new(field.name(), DataType::Utf8View, field.is_nullable()),
+                ),
+                (Some(DataType::BinaryView), DataType::Binary)
+                | (Some(DataType::BinaryView), DataType::LargeBinary) => Arc::new(
+                    Field::new(field.name(), DataType::BinaryView, field.is_nullable()),
+                ),
+                _ => field.clone(),
+            },
+        )
+        .collect();
+
+    Some(Schema::new_with_metadata(
+        transformed_fields,
+        file_schema.metadata.clone(),
+    ))
+}
+
 #[cfg(test)]
 pub(crate) mod test_util {
     use std::ops::Range;