Merge remote-tracking branch 'apache/df52' into fix/schema-adapter-nested-types

Author: andygrove

native/core/src/execution/planner.rs

@@ -730,6 +730,41 @@ impl PhysicalPlanner {
                     Arc::new(ConfigOptions::default()),
                 )))
             }
+            // Date +/- Int8/Int16/Int32: DataFusion 52's arrow-arith kernels only
+            // support Date32 +/- Interval types, not raw integers. Use the Spark
+            // date_add / date_sub UDFs which handle Int8/Int16/Int32 directly.
+            (
+                DataFusionOperator::Plus,
+                Ok(DataType::Date32),
+                Ok(DataType::Int8 | DataType::Int16 | DataType::Int32),
+            ) => {
+                let udf = Arc::new(ScalarUDF::new_from_impl(
+                    datafusion_spark::function::datetime::date_add::SparkDateAdd::new(),
+                ));
+                Ok(Arc::new(ScalarFunctionExpr::new(
+                    "date_add",
+                    udf,
+                    vec![left, right],
+                    Arc::new(Field::new("date_add", DataType::Date32, true)),
+                    Arc::new(ConfigOptions::default()),
+                )))
+            }
+            (
+                DataFusionOperator::Minus,
+                Ok(DataType::Date32),
+                Ok(DataType::Int8 | DataType::Int16 | DataType::Int32),
+            ) => {
+                let udf = Arc::new(ScalarUDF::new_from_impl(
+                    datafusion_spark::function::datetime::date_sub::SparkDateSub::new(),
+                ));
+                Ok(Arc::new(ScalarFunctionExpr::new(
+                    "date_sub",
+                    udf,
+                    vec![left, right],
+                    Arc::new(Field::new("date_sub", DataType::Date32, true)),
+                    Arc::new(ConfigOptions::default()),
+                )))
+            }
             _ => {
                 let data_type = return_type.map(to_arrow_datatype).unwrap();
                 if [EvalMode::Try, EvalMode::Ansi].contains(&eval_mode)

native/core/src/parquet/parquet_exec.rs

@@ -83,11 +83,15 @@ pub(crate) fn init_datasource_exec(
     // dbg!(&required_schema, &data_schema);
 
     // Determine the schema to use for ParquetSource
-    // Use data_schema only if both data_schema and data_filters are set
-    let base_schema = match (&data_schema, &data_filters) {
+    // // Use data_schema only if both data_schema and data_filters are set
+    let base_schema = match (&data_schema, &projection_vector) {
         (Some(schema), Some(_)) => Arc::clone(schema),
         _ => Arc::clone(&required_schema),
     };
+    //let base_schema = required_schema;
+    // dbg!(&base_schema);
+    // dbg!(&data_schema);
+    // dbg!(&data_filters);
     let partition_fields: Vec<_> = partition_schema
         .iter()
         .flat_map(|s| s.fields().iter())

native/spark-expr/src/conversion_funcs/cast.rs

@@ -906,10 +906,10 @@ pub fn spark_cast(
     data_type: &DataType,
     cast_options: &SparkCastOptions,
 ) -> DataFusionResult<ColumnarValue> {
-    let input_type = match &arg {
-        ColumnarValue::Array(array) => array.data_type().clone(),
-        ColumnarValue::Scalar(scalar) => scalar.data_type(),
-    };
+    // let input_type = match &arg {
+    //     ColumnarValue::Array(array) => array.data_type().clone(),
+    //     ColumnarValue::Scalar(scalar) => scalar.data_type(),
+    // };
 
     let result = match arg {
         ColumnarValue::Array(array) => {
@@ -927,10 +927,10 @@ pub fn spark_cast(
         }
     };
 
-    let result_type = match &result {
-        ColumnarValue::Array(array) => array.data_type().clone(),
-        ColumnarValue::Scalar(scalar) => scalar.data_type(),
-    };
+    // let result_type = match &result {
+    //     ColumnarValue::Array(array) => array.data_type().clone(),
+    //     ColumnarValue::Scalar(scalar) => scalar.data_type(),
+    // };
 
     // println!(
     //     "spark_cast: {} -> {} (requested: {})",

native/spark-expr/src/csv_funcs/to_csv.rs

@@ -115,8 +115,8 @@ impl PhysicalExpr for ToCsv {
         )))
     }
 
-    fn fmt_sql(&self, _: &mut Formatter<'_>) -> std::fmt::Result {
-        unimplemented!()
+    fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        Display::fmt(self, f)
     }
 }
 

native/spark-expr/src/unbound.rs

@@ -20,7 +20,7 @@ use arrow::datatypes::{DataType, Schema};
 use datafusion::common::{internal_err, Result};
 use datafusion::physical_expr::PhysicalExpr;
 use datafusion::physical_plan::ColumnarValue;
-use std::fmt::Formatter;
+use std::fmt::{Display, Formatter};
 use std::{hash::Hash, sync::Arc};
 
 /// This is similar to `UnKnownColumn` in DataFusion, but it has data type.
@@ -64,8 +64,8 @@ impl PhysicalExpr for UnboundColumn {
         self
     }
 
-    fn fmt_sql(&self, _: &mut Formatter<'_>) -> std::fmt::Result {
-        unimplemented!()
+    fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        Display::fmt(self, f)
     }
 
     /// Get the data type of this expression, given the schema of the input