DataFusion 52 migration

Author: comphead

native/core/src/execution/planner.rs

@@ -4042,14 +4042,13 @@ mod tests {
 
         let source = Arc::new(
             ParquetSource::new(Arc::new(read_schema.clone()))
-                .with_table_parquet_options(TableParquetOptions::new())
+                .with_table_parquet_options(TableParquetOptions::new()),
         ) as Arc<dyn FileSource>;
 
         let object_store_url = ObjectStoreUrl::local_filesystem();
-        let file_scan_config =
-            FileScanConfigBuilder::new(object_store_url, source)
-                .with_file_groups(file_groups)
-                .build();
+        let file_scan_config = FileScanConfigBuilder::new(object_store_url, source)
+            .with_file_groups(file_groups)
+            .build();
 
         // Run native read
         let scan = Arc::new(DataSourceExec::new(Arc::new(file_scan_config.clone())));

native/core/src/parquet/parquet_exec.rs

@@ -85,7 +85,9 @@ pub(crate) fn init_datasource_exec(
             let partition_fields: Vec<_> = partition_schema
                 .fields()
                 .iter()
-                .map(|f| Arc::new(Field::new(f.name(), f.data_type().clone(), f.is_nullable())) as _)
+                .map(|f| {
+                    Arc::new(Field::new(f.name(), f.data_type().clone(), f.is_nullable())) as _
+                })
                 .collect();
             TableSchema::new(Arc::clone(data_schema), partition_fields)
         } else {
@@ -95,8 +97,8 @@ pub(crate) fn init_datasource_exec(
         TableSchema::from_file_schema(Arc::clone(&required_schema))
     };
 
-    let mut parquet_source = ParquetSource::new(table_schema)
-        .with_table_parquet_options(table_parquet_options);
+    let mut parquet_source =
+        ParquetSource::new(table_schema).with_table_parquet_options(table_parquet_options);
 
     // Create a conjunctive form of the vector because ParquetExecBuilder takes
     // a single expression
@@ -129,13 +131,14 @@ pub(crate) fn init_datasource_exec(
         .map(|files| FileGroup::new(files.clone()))
         .collect();
 
-    let mut file_scan_config_builder = FileScanConfigBuilder::new(object_store_url, file_source)
-        .with_file_groups(file_groups);
-    
+    let mut file_scan_config_builder =
+        FileScanConfigBuilder::new(object_store_url, file_source).with_file_groups(file_groups);
+
     if let Some(projection_vector) = projection_vector {
-        file_scan_config_builder = file_scan_config_builder.with_projection_indices(Some(projection_vector))?;
+        file_scan_config_builder =
+            file_scan_config_builder.with_projection_indices(Some(projection_vector))?;
     }
-    
+
     let file_scan_config = file_scan_config_builder.build();
 
     Ok(Arc::new(DataSourceExec::new(Arc::new(file_scan_config))))

native/core/src/parquet/schema_adapter.rs

@@ -346,14 +346,13 @@ mod test {
 
         let parquet_source = Arc::new(
             ParquetSource::new(Arc::clone(&required_schema))
-                .with_table_parquet_options(TableParquetOptions::new())
+                .with_table_parquet_options(TableParquetOptions::new()),
         ) as Arc<dyn FileSource>;
 
         let files = FileGroup::new(vec![PartitionedFile::from_path(filename.to_string())?]);
-        let file_scan_config =
-            FileScanConfigBuilder::new(object_store_url, parquet_source)
-                .with_file_groups(vec![files])
-                .build();
+        let file_scan_config = FileScanConfigBuilder::new(object_store_url, parquet_source)
+            .with_file_groups(vec![files])
+            .build();
 
         let parquet_exec = DataSourceExec::new(Arc::new(file_scan_config));
 

native/spark-expr/src/agg_funcs/covariance.rs

@@ -23,9 +23,7 @@ use arrow::{
     compute::cast,
     datatypes::{DataType, Field},
 };
-use datafusion::common::{
-    downcast_value, unwrap_or_internal_err, Result, ScalarValue,
-};
+use datafusion::common::{downcast_value, unwrap_or_internal_err, Result, ScalarValue};
 use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs};
 use datafusion::logical_expr::type_coercion::aggregates::NUMERICS;
 use datafusion::logical_expr::{Accumulator, AggregateUDFImpl, Signature, Volatility};

native/spark-expr/src/math_funcs/round.rs

@@ -21,11 +21,11 @@ use arrow::array::{Array, ArrowNativeTypeOp};
 use arrow::array::{Int16Array, Int32Array, Int64Array, Int8Array};
 use arrow::datatypes::{DataType, Field};
 use arrow::error::ArrowError;
-use datafusion::common::{exec_err, internal_err, DataFusionError, ScalarValue};
 use datafusion::common::config::ConfigOptions;
-use datafusion::physical_plan::ColumnarValue;
+use datafusion::common::{exec_err, internal_err, DataFusionError, ScalarValue};
 use datafusion::functions::math::round::RoundFunc;
-use datafusion::logical_expr::{ScalarUDFImpl, ScalarFunctionArgs};
+use datafusion::logical_expr::{ScalarFunctionArgs, ScalarUDFImpl};
+use datafusion::physical_plan::ColumnarValue;
 use std::{cmp::min, sync::Arc};
 
 macro_rules! integer_round {
@@ -133,10 +133,7 @@ pub fn spark_round(
                 let round_udf = RoundFunc::new();
                 let return_field = Arc::new(Field::new("round", array.data_type().clone(), true));
                 let args_for_round = ScalarFunctionArgs {
-                    args: vec![
-                        ColumnarValue::Array(Arc::clone(array)),
-                        args[1].clone(),
-                    ],
+                    args: vec![ColumnarValue::Array(Arc::clone(array)), args[1].clone()],
                     number_rows: array.len(),
                     return_field,
                     arg_fields: vec![],
@@ -169,10 +166,7 @@ pub fn spark_round(
                 let data_type = a.data_type();
                 let return_field = Arc::new(Field::new("round", data_type, true));
                 let args_for_round = ScalarFunctionArgs {
-                    args: vec![
-                        ColumnarValue::Scalar(a.clone()),
-                        args[1].clone(),
-                    ],
+                    args: vec![ColumnarValue::Scalar(a.clone()), args[1].clone()],
                     number_rows: 1,
                     return_field,
                     arg_fields: vec![],