fixes

Author: JanKaul

Cargo.lock

@@ -639,7 +639,7 @@ impl StatisticsCacheFunc {
 }
 
 impl TableFunctionImpl for StatisticsCacheFunc {
-    fn call(&self, exprs: &[Expr]) -> Result<Arc<dyn TableProvider>> {
+    fn call(&self, exprs: &[(Expr, Option<String>)]) -> Result<Arc<dyn TableProvider>> {
         if !exprs.is_empty() {
             return plan_err!("statistics_cache should have no arguments");
         }
@@ -774,7 +774,7 @@ impl ListFilesCacheFunc {
 }
 
 impl TableFunctionImpl for ListFilesCacheFunc {
-    fn call(&self, exprs: &[Expr]) -> Result<Arc<dyn TableProvider>> {
+    fn call(&self, exprs: &[(Expr, Option<String>)]) -> Result<Arc<dyn TableProvider>> {
         if !exprs.is_empty() {
             return plan_err!("list_files_cache should have no arguments");
         }

datafusion-cli/src/functions.rs

@@ -87,7 +87,7 @@ use datafusion_expr::logical_plan::builder::wrap_projection_for_join_if_necessar
 use datafusion_expr::utils::{expr_to_columns, split_conjunction};
 use datafusion_expr::{
     Analyze, BinaryExpr, Cast, DescribeTable, DmlStatement, Explain, ExplainFormat,
-    Extension, FetchType, Filter, JoinType, LogicalPlanBuilder, Operator, RecursiveQuery,
+    Extension, FetchType, Filter, JoinType, LogicalPlanBuilder, RecursiveQuery,
     SkipType, StringifiedPlan, WindowFrame, WindowFrameBound, WriteOp,
 };
 use datafusion_physical_expr::aggregate::{AggregateExprBuilder, AggregateFunctionExpr};
@@ -2212,7 +2212,8 @@ fn extract_dml_filters(
             | LogicalPlan::Repartition(_)
             | LogicalPlan::Aggregate(_)
             | LogicalPlan::Window(_)
-            | LogicalPlan::Subquery(_) => {
+            | LogicalPlan::Subquery(_)
+            | LogicalPlan::Pivot(_) => {
                 // Filter information may appear in child nodes; continue traversal
                 // to extract filters from Filter/TableScan nodes deeper in the plan
             }
@@ -2567,26 +2568,26 @@ pub fn create_aggregate_expr_and_maybe_filter(
 pub fn transform_pivot_to_aggregate(
     input: Arc<LogicalPlan>,
     aggregate_expr: &Expr,
-    pivot_column: &datafusion_common::Column,
+    pivot_column: &Column,
     pivot_values: Vec<ScalarValue>,
     default_on_null_expr: Option<&Expr>,
 ) -> Result<LogicalPlan> {
     let df_schema = input.schema();
 
-    let all_columns: Vec<datafusion_common::Column> = df_schema.columns();
+    let all_columns: Vec<Column> = df_schema.columns();
 
     // Filter to include only columns we want for GROUP BY
     // (exclude pivot column and aggregate expression columns)
     let group_by_columns: Vec<Expr> = all_columns
         .into_iter()
-        .filter(|col: &datafusion_common::Column| {
+        .filter(|col: &Column| {
             col.name != pivot_column.name
                 && !aggregate_expr
                     .column_refs()
                     .iter()
                     .any(|agg_col| agg_col.name == col.name)
         })
-        .map(|col: datafusion_common::Column| Expr::Column(col))
+        .map(|col: Column| Expr::Column(col))
         .collect();
 
     let builder = LogicalPlanBuilder::from(Arc::unwrap_or_clone(Arc::clone(&input)));
@@ -2660,7 +2661,7 @@ pub fn transform_pivot_to_aggregate(
                 .any(|v| field.name() == v.to_string().trim_matches('\''))
             {
                 projection_exprs.push(Expr::Column(
-                    datafusion_common::Column::from_name(field.name()),
+                    Column::from_name(field.name()),
                 ));
             }
         }
@@ -2669,7 +2670,7 @@ pub fn transform_pivot_to_aggregate(
         for value in &pivot_values {
             let field_name = value.to_string().trim_matches('\'').to_string();
             let aggregate_col =
-                Expr::Column(datafusion_common::Column::from_name(&field_name));
+                Expr::Column(Column::from_name(&field_name));
 
             // Create COALESCE expression using CASE: CASE WHEN col IS NULL THEN default_value ELSE col END
             let coalesce_expr = Expr::Case(datafusion_expr::expr::Case {
@@ -3103,7 +3104,7 @@ impl DefaultPhysicalPlanner {
                     {
                         if !physical_exprs.iter().any(|(_, name)| name == field.name()) {
                             all_exprs.push((
-                                Arc::new(Column::new(field.name(), i))
+                                Arc::new(datafusion_physical_expr::expressions::Column::new(field.name(), i))
                                     as Arc<dyn PhysicalExpr>,
                                 field.name().clone(),
                             ));

datafusion/core/src/physical_planner.rs

@@ -174,7 +174,6 @@ impl ScalarUDFImpl for ToDateFunc {
             }
             Float16
             | Float32
-            | Float64
             | Decimal32(_, _)
             | Decimal64(_, _)
             | Decimal128(_, _)

datafusion/functions/src/datetime/to_date.rs

@@ -2016,7 +2016,7 @@ mod test {
         let plan = LogicalPlan::Projection(Projection::try_new(vec![expr], empty)?);
         assert_type_coercion_error(
             plan,
-            "Cannot infer common argument type for comparison operation Int64 IS DISTINCT FROM Boolean",
+            "Cannot infer common argument type for comparison operation Float64 IS DISTINCT FROM Boolean",
         )?;
 
         // is not true

datafusion/optimizer/src/analyzer/type_coercion.rs

@@ -82,12 +82,6 @@ impl OptimizerRule for DecorrelatePredicateSubquery {
                     .into_iter()
                     .partition(has_subquery);
 
-            if with_subqueries.is_empty() {
-                return internal_err!(
-                    "can not find expected subqueries in DecorrelatePredicateSubquery"
-                );
-            }
-
         assert_or_internal_err!(
             !with_subqueries.is_empty(),
             "can not find expected subqueries in DecorrelatePredicateSubquery"
@@ -145,6 +139,7 @@ impl OptimizerRule for DecorrelatePredicateSubquery {
                         join.join_type,
                         join.join_constraint,
                         join.null_equality,
+                        join.null_aware,
                     )?;
                     return Ok(Transformed::yes(LogicalPlan::Join(new_join)));
                 }
@@ -546,8 +541,12 @@ fn build_join(
 
         right_cols_idx_and_col.sort_by_key(|(idx, _)| *idx);
 
+        // Deduplicate by schema index to avoid duplicate projection expressions
+        // when the same column is referenced both qualified and unqualified
+        let mut seen_indices = std::collections::HashSet::new();
         let right_proj_exprs: Vec<Expr> = right_cols_idx_and_col
             .into_iter()
+            .filter(|(idx, _)| seen_indices.insert(*idx))
             .map(|(_, c)| Expr::Column(c))
             .collect();
 
@@ -839,9 +838,10 @@ mod tests {
           LeftMark Join:  Filter: a = __correlated_sq_1.ua [a:Int32;N, mark:Boolean]
             Projection: column1 AS a [a:Int32;N]
               Values: (Int32(1)), (Int32(2)) [column1:Int32;N]
-            SubqueryAlias: __correlated_sq_1 [ua:Int32;N]
-              Projection: column1 AS ua [ua:Int32;N]
-                Values: (Int32(2)) [column1:Int32;N]
+            Projection: __correlated_sq_1.ua [ua:Int32;N]
+              SubqueryAlias: __correlated_sq_1 [ua:Int32;N]
+                Projection: column1 AS ua [ua:Int32;N]
+                  Values: (Int32(2)) [column1:Int32;N]
         "
         )
     }

datafusion/optimizer/src/decorrelate_predicate_subquery.rs

@@ -981,21 +981,22 @@ impl AsLogicalPlan for LogicalPlanNode {
                     dml_node.dml_type().into(),
                     Arc::new(into_logical_plan!(dml_node.input, ctx, extension_codec)?),
                 )))
+            }
             LogicalPlanType::Pivot(pivot) => {
                 let aggregate_expr = pivot
                     .aggregate_expr
                     .as_ref()
                     .map(|expr| from_proto::parse_expr(expr, ctx, extension_codec))
                     .transpose()?
                     .ok_or_else(|| {
-                        DataFusionError::Internal("aggregate_expr required".to_string())
+                        internal_datafusion_err!("aggregate_expr required")
                     })?;
                 let pivot_column = pivot
                     .pivot_column
                     .as_ref()
                     .map(|col| col.clone().into())
                     .ok_or_else(|| {
-                        DataFusionError::Internal("pivot_column required".to_string())
+                        internal_datafusion_err!("pivot_column required")
                     })?;
                 let pivot_values = pivot
                     .pivot_values

datafusion/proto/src/logical_plan/mod.rs

@@ -549,10 +549,15 @@ impl<'a> DFParser<'a> {
             Token::Word(w) => {
                 match w.keyword {
                     Keyword::CREATE => {
-                        if let Token::Word(w) = self.parser.peek_nth_token(2).token {
-                            // use native parser for CREATE EXTERNAL VOLUME
-                            if w.keyword == Keyword::VOLUME {
-                                return self.parse_and_handle_statement();
+                        // use native parser for CREATE EXTERNAL VOLUME
+                        // and CREATE OR REPLACE EXTERNAL VOLUME
+                        for offset in [2, 4] {
+                            if let Token::Word(w) =
+                                self.parser.peek_nth_token(offset).token
+                            {
+                                if w.keyword == Keyword::VOLUME {
+                                    return self.parse_and_handle_statement();
+                                }
                             }
                         }
                         self.parser.next_token(); // CREATE

datafusion/sql/src/parser.rs

@@ -338,7 +338,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
                     return plan_err!("PIVOT value column is required");
                 }
 
-                let column_name = value_column.last().unwrap().value.clone();
+                let column_name = value_column.last().unwrap().to_string();
                 let pivot_column = Column::new(None::<&str>, column_name);
 
                 let default_on_null_expr = default_on_null
@@ -454,7 +454,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
                 let base_plan = self.create_relation(*table, planner_context)?;
                 let base_schema = base_plan.schema();
 
-                let value_column = value.value.clone();
+                let value_column = value.to_string();
                 let name_column = name.value.clone();
 
                 let mut unpivot_column_indices = Vec::new();
@@ -463,7 +463,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
                 let mut common_type = None;
 
                 for column_ident in &columns {
-                    let column_name = column_ident.value.clone();
+                    let column_name = column_ident.expr.to_string();
 
                     let idx = if let Some(i) =
                         base_schema.index_of_column_by_name(None, &column_name)