DataFusion 52 migration

Author: comphead

native/core/src/parquet/cast_column.rs

@@ -23,7 +23,7 @@ use arrow::{
 
 use datafusion::common::format::DEFAULT_CAST_OPTIONS;
 use datafusion::common::Result as DataFusionResult;
-use datafusion::common::{cast_column, ScalarValue};
+use datafusion::common::ScalarValue;
 use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::PhysicalExpr;
 use std::{
@@ -70,8 +70,7 @@ fn cast_timestamp_micros_to_millis_scalar(
 }
 
 #[derive(Debug, Clone, Eq)]
-pub struct
-CometCastColumnExpr {
+pub struct CometCastColumnExpr {
     /// The physical expression producing the value to cast.
     expr: Arc<dyn PhysicalExpr>,
     /// The physical field of the input column.
@@ -146,8 +145,11 @@ impl PhysicalExpr for CometCastColumnExpr {
     fn evaluate(&self, batch: &RecordBatch) -> DataFusionResult<ColumnarValue> {
         let value = self.expr.evaluate(batch)?;
 
-        if value.data_type().equals_datatype(self.target_field.data_type()) {
-            return Ok(value)
+        if value
+            .data_type()
+            .equals_datatype(self.target_field.data_type())
+        {
+            return Ok(value);
         }
 
         let input_physical_field = self.input_physical_field.data_type();
@@ -214,10 +216,10 @@ mod tests {
     fn test_cast_timestamp_micros_to_millis_array() {
         // Create a TimestampMicrosecond array with some values
         let micros_array: TimestampMicrosecondArray = vec![
-            Some(1_000_000), // 1 second in micros
-            Some(2_500_000), // 2.5 seconds in micros
-            None,            // null value
-            Some(0),         // zero
+            Some(1_000_000),  // 1 second in micros
+            Some(2_500_000),  // 2.5 seconds in micros
+            None,             // null value
+            Some(0),          // zero
             Some(-1_000_000), // negative value (before epoch)
         ]
         .into();
@@ -263,10 +265,7 @@ mod tests {
     fn test_cast_timestamp_micros_to_millis_scalar() {
         // Test with a value
         let result = cast_timestamp_micros_to_millis_scalar(Some(1_500_000), None);
-        assert_eq!(
-            result,
-            ScalarValue::TimestampMillisecond(Some(1500), None)
-        );
+        assert_eq!(result, ScalarValue::TimestampMillisecond(Some(1500), None));
 
         // Test with null
         let null_result = cast_timestamp_micros_to_millis_scalar(None, None);
@@ -305,7 +304,8 @@ mod tests {
         let cast_expr = CometCastColumnExpr::new(col_expr, input_field, target_field, None);
 
         // Create a record batch with TimestampMicrosecond data
-        let micros_array: TimestampMicrosecondArray = vec![Some(1_000_000), Some(2_000_000), None].into();
+        let micros_array: TimestampMicrosecondArray =
+            vec![Some(1_000_000), Some(2_000_000), None].into();
         let batch = RecordBatch::try_new(Arc::new(schema), vec![Arc::new(micros_array)]).unwrap();
 
         // Evaluate

native/core/src/parquet/schema_adapter.rs

@@ -26,7 +26,7 @@
 use crate::parquet::cast_column::CometCastColumnExpr;
 use crate::parquet::parquet_support::{spark_parquet_convert, SparkParquetOptions};
 use arrow::array::{ArrayRef, RecordBatch, RecordBatchOptions};
-use arrow::datatypes::{Field, Schema, SchemaRef};
+use arrow::datatypes::{Schema, SchemaRef};
 use datafusion::common::tree_node::{Transformed, TransformedResult, TreeNode};
 use datafusion::common::{ColumnStatistics, Result as DataFusionResult};
 use datafusion::datasource::schema_adapter::{SchemaAdapter, SchemaAdapterFactory, SchemaMapper};
@@ -100,6 +100,7 @@ impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory {
 /// 2. Replace standard DataFusion cast expressions with Spark-compatible casts
 /// 3. Handle case-insensitive column matching
 #[derive(Debug)]
+#[allow(dead_code)]
 struct SparkPhysicalExprAdapter {
     /// The logical schema expected by the query
     logical_file_schema: SchemaRef,
@@ -125,6 +126,7 @@ impl PhysicalExprAdapter for SparkPhysicalExprAdapter {
     }
 }
 
+#[allow(dead_code)]
 impl SparkPhysicalExprAdapter {
     /// Replace CastColumnExpr (DataFusion's cast) with Spark's Cast expression.
     fn replace_with_spark_cast(
@@ -179,7 +181,6 @@ impl SparkPhysicalExprAdapter {
                 // Get the physical datatype (actual file schema)
                 let physical_field = self.physical_file_schema.fields().get(col_idx);
 
-
                 // dbg!(&logical_field, &physical_field);
 
                 if let (Some(logical_field), Some(physical_field)) = (logical_field, physical_field)
@@ -191,8 +192,8 @@ impl SparkPhysicalExprAdapter {
                     if logical_type != physical_type {
                         let cast_expr: Arc<dyn PhysicalExpr> = Arc::new(CometCastColumnExpr::new(
                             Arc::clone(&e),
-                            physical_field.clone(),
-                            logical_field.clone(),
+                            Arc::clone(physical_field),
+                            Arc::clone(logical_field),
                             None,
                         ));
                         // dbg!(&cast_expr);