Refactor file schema type coercions (apache#15268)

* Refactor file schema type coercions

* resolve comments

* keep old api and add deprecated

* resolve comments

Author: xudong963

datafusion/datasource-parquet/src/file_format.rs

@@ -465,7 +465,114 @@ impl FileFormat for ParquetFormat {
     }
 }
 
+/// Apply necessary schema type coercions to make file schema match table schema.
+///
+/// This function performs two main types of transformations in a single pass:
+/// 1. Binary types to string types conversion - Converts binary data types to their
+///    corresponding string types when the table schema expects string data
+/// 2. Regular to view types conversion - Converts standard string/binary types to
+///    view types when the table schema uses view types
+///
+/// # Arguments
+/// * `table_schema` - The table schema containing the desired types
+/// * `file_schema` - The file schema to be transformed
+///
+/// # Returns
+/// * `Some(Schema)` - If any transformations were applied, returns the transformed schema
+/// * `None` - If no transformations were needed
+pub fn apply_file_schema_type_coercions(
+    table_schema: &Schema,
+    file_schema: &Schema,
+) -> Option<Schema> {
+    let mut needs_view_transform = false;
+    let mut needs_string_transform = false;
+
+    // Create a mapping of table field names to their data types for fast lookup
+    // and simultaneously check if we need any transformations
+    let table_fields: HashMap<_, _> = table_schema
+        .fields()
+        .iter()
+        .map(|f| {
+            let dt = f.data_type();
+            // Check if we need view type transformation
+            if matches!(dt, &DataType::Utf8View | &DataType::BinaryView) {
+                needs_view_transform = true;
+            }
+            // Check if we need string type transformation
+            if matches!(
+                dt,
+                &DataType::Utf8 | &DataType::LargeUtf8 | &DataType::Utf8View
+            ) {
+                needs_string_transform = true;
+            }
+
+            (f.name(), dt)
+        })
+        .collect();
+
+    // Early return if no transformation needed
+    if !needs_view_transform && !needs_string_transform {
+        return None;
+    }
+
+    let transformed_fields: Vec<Arc<Field>> = file_schema
+        .fields()
+        .iter()
+        .map(|field| {
+            let field_name = field.name();
+            let field_type = field.data_type();
+
+            // Look up the corresponding field type in the table schema
+            if let Some(table_type) = table_fields.get(field_name) {
+                match (table_type, field_type) {
+                    // table schema uses string type, coerce the file schema to use string type
+                    (
+                        &DataType::Utf8,
+                        DataType::Binary | DataType::LargeBinary | DataType::BinaryView,
+                    ) => {
+                        return field_with_new_type(field, DataType::Utf8);
+                    }
+                    // table schema uses large string type, coerce the file schema to use large string type
+                    (
+                        &DataType::LargeUtf8,
+                        DataType::Binary | DataType::LargeBinary | DataType::BinaryView,
+                    ) => {
+                        return field_with_new_type(field, DataType::LargeUtf8);
+                    }
+                    // table schema uses string view type, coerce the file schema to use view type
+                    (
+                        &DataType::Utf8View,
+                        DataType::Binary | DataType::LargeBinary | DataType::BinaryView,
+                    ) => {
+                        return field_with_new_type(field, DataType::Utf8View);
+                    }
+                    // Handle view type conversions
+                    (&DataType::Utf8View, DataType::Utf8 | DataType::LargeUtf8) => {
+                        return field_with_new_type(field, DataType::Utf8View);
+                    }
+                    (&DataType::BinaryView, DataType::Binary | DataType::LargeBinary) => {
+                        return field_with_new_type(field, DataType::BinaryView);
+                    }
+                    _ => {}
+                }
+            }
+
+            // If no transformation is needed, keep the original field
+            Arc::clone(field)
+        })
+        .collect();
+
+    Some(Schema::new_with_metadata(
+        transformed_fields,
+        file_schema.metadata.clone(),
+    ))
+}
+
 /// Coerces the file schema if the table schema uses a view type.
+#[deprecated(
+    since = "47.0.0",
+    note = "Use `apply_file_schema_type_coercions` instead"
+)]
 pub fn coerce_file_schema_to_view_type(
     table_schema: &Schema,
     file_schema: &Schema,
@@ -515,6 +622,10 @@ pub fn coerce_file_schema_to_view_type(
 /// If the table schema uses a string type, coerce the file schema to use a string type.
 ///
 /// See [ParquetFormat::binary_as_string] for details
+#[deprecated(
+    since = "47.0.0",
+    note = "Use `apply_file_schema_type_coercions` instead"
+)]
 pub fn coerce_file_schema_to_string_type(
     table_schema: &Schema,
     file_schema: &Schema,
@@ -718,11 +829,8 @@ pub fn statistics_from_parquet_meta_calc(
         file_metadata.schema_descr(),
         file_metadata.key_value_metadata(),
     )?;
-    if let Some(merged) = coerce_file_schema_to_string_type(&table_schema, &file_schema) {
-        file_schema = merged;
-    }
 
-    if let Some(merged) = coerce_file_schema_to_view_type(&table_schema, &file_schema) {
+    if let Some(merged) = apply_file_schema_type_coercions(&table_schema, &file_schema) {
         file_schema = merged;
     }
 

datafusion/datasource-parquet/src/opener.rs

@@ -19,14 +19,11 @@
 
 use std::sync::Arc;
 
-use crate::file_format::{
-    coerce_file_schema_to_string_type, coerce_file_schema_to_view_type,
-};
 use crate::page_filter::PagePruningAccessPlanFilter;
 use crate::row_group_filter::RowGroupAccessPlanFilter;
 use crate::{
-    row_filter, should_enable_page_index, ParquetAccessPlan, ParquetFileMetrics,
-    ParquetFileReaderFactory,
+    apply_file_schema_type_coercions, row_filter, should_enable_page_index,
+    ParquetAccessPlan, ParquetFileMetrics, ParquetFileReaderFactory,
 };
 use datafusion_datasource::file_meta::FileMeta;
 use datafusion_datasource::file_stream::{FileOpenFuture, FileOpener};
@@ -131,14 +128,8 @@ impl FileOpener for ParquetOpener {
                 ArrowReaderMetadata::load_async(&mut reader, options.clone()).await?;
             let mut schema = Arc::clone(metadata.schema());
 
-            if let Some(merged) =
-                coerce_file_schema_to_string_type(&table_schema, &schema)
-            {
-                schema = Arc::new(merged);
-            }
-
             // read with view types
-            if let Some(merged) = coerce_file_schema_to_view_type(&table_schema, &schema)
+            if let Some(merged) = apply_file_schema_type_coercions(&table_schema, &schema)
             {
                 schema = Arc::new(merged);
             }