Mismatch in MemTable of Select Into when projecting on aggregate window functions

datafusion/core/tests/sqllogictests/test_files/ddl.slt

@@ -348,6 +348,9 @@ SELECT * FROM new_table;
 statement ok
 DROP TABLE new_table
 
+statement ok
+DROP TABLE my_table;
+
 # create_table_with_schema_as_multiple_values
 statement ok
 CREATE TABLE test_table(c1 int, c2 float, c3 varchar) AS VALUES(1, 2, 'hello'),(2, 1, 'there'),(3, 0, '!');
@@ -362,7 +365,18 @@ SELECT * FROM new_table
 2 1 there
 
 statement ok
-DROP TABLE my_table;
+DROP TABLE new_table;
+
+# Select into without alias names of window aggregates
+statement ok
+SELECT SUM(c1) OVER(ORDER BY c2), c2, c3 INTO new_table FROM test_table
+
+query IRT
+SELECT * FROM new_table
+----
+3 0 !
+5 1 there
+6 2 hello
 
 statement ok
 DROP TABLE new_table;

datafusion/sql/src/select.rs

@@ -15,12 +15,15 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::fmt::Write;
+
 use crate::planner::{ContextProvider, PlannerContext, SqlToRel};
 use crate::utils::{
     check_columns_satisfy_exprs, extract_aliases, rebase_expr, resolve_aliases_to_exprs,
     resolve_columns, resolve_positions_to_exprs,
 };
 use datafusion_common::{DataFusionError, Result};
+use datafusion_expr::expr::*;
 use datafusion_expr::expr_rewriter::{
     normalize_col, normalize_col_with_schemas_and_ambiguity_check,
 };
@@ -31,7 +34,8 @@ use datafusion_expr::utils::{
 };
 use datafusion_expr::Expr::Alias;
 use datafusion_expr::{
-    Expr, Filter, GroupingSet, LogicalPlan, LogicalPlanBuilder, Partitioning,
+    BinaryExpr, Cast, Expr, Filter, GroupingSet, LogicalPlan, LogicalPlanBuilder,
+    Partitioning, TryCast,
 };
 
 use sqlparser::ast::{Distinct, Expr as SQLExpr, WildcardAdditionalOptions, WindowType};
@@ -194,7 +198,13 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
                 .iter()
                 .map(|expr| rebase_expr(expr, &window_func_exprs, &plan))
                 .collect::<Result<Vec<Expr>>>()?;
-
+            if select.into.is_some() {
+                for expr in select_exprs_post_aggr.iter_mut() {
+                    if let Expr::Column(_) = expr.clone() {
+                        *expr = expr.clone().alias(physical_name(expr)?);
+                    }
+                }
+            }
             plan
         };
 
@@ -555,3 +565,288 @@ fn match_window_definitions(
     }
     Ok(())
 }
+
+fn create_function_physical_name(
+    fun: &str,
+    distinct: bool,
+    args: &[Expr],
+) -> Result<String> {
+    let names: Vec<String> = args
+        .iter()
+        .map(|e| create_physical_name(e, false))
+        .collect::<Result<_>>()?;
+
+    let distinct_str = match distinct {
+        true => "DISTINCT ",
+        false => "",
+    };
+    Ok(format!("{}({}{})", fun, distinct_str, names.join(",")))
+}
+
+fn physical_name(e: &Expr) -> Result<String> {
+    create_physical_name(e, true)
+}
+
+fn create_physical_name(e: &Expr, is_first_expr: bool) -> Result<String> {
+    match e {
+        Expr::Column(c) => {
+            if is_first_expr {
+                Ok(c.name.clone())
+            } else {
+                Ok(c.flat_name())
+            }
+        }
+        Expr::Alias(_, name) => Ok(name.clone()),
+        Expr::ScalarVariable(_, variable_names) => Ok(variable_names.join(".")),
+        Expr::Literal(value) => Ok(format!("{value:?}")),
+        Expr::BinaryExpr(BinaryExpr { left, op, right }) => {
+            let left = create_physical_name(left, false)?;
+            let right = create_physical_name(right, false)?;
+            Ok(format!("{left} {op} {right}"))
+        }
+        Expr::Case(case) => {
+            let mut name = "CASE ".to_string();
+            if let Some(e) = &case.expr {
+                let _ = write!(name, "{e:?} ");
+            }
+            for (w, t) in &case.when_then_expr {
+                let _ = write!(name, "WHEN {w:?} THEN {t:?} ");
+            }
+            if let Some(e) = &case.else_expr {
+                let _ = write!(name, "ELSE {e:?} ");
+            }
+            name += "END";
+            Ok(name)
+        }
+        Expr::Cast(Cast { expr, .. }) => {
+            // CAST does not change the expression name
+            create_physical_name(expr, false)
+        }
+        Expr::TryCast(TryCast { expr, .. }) => {
+            // CAST does not change the expression name
+            create_physical_name(expr, false)
+        }
+        Expr::Not(expr) => {
+            let expr = create_physical_name(expr, false)?;
+            Ok(format!("NOT {expr}"))
+        }
+        Expr::Negative(expr) => {
+            let expr = create_physical_name(expr, false)?;
+            Ok(format!("(- {expr})"))
+        }
+        Expr::IsNull(expr) => {
+            let expr = create_physical_name(expr, false)?;
+            Ok(format!("{expr} IS NULL"))
+        }
+        Expr::IsNotNull(expr) => {
+            let expr = create_physical_name(expr, false)?;
+            Ok(format!("{expr} IS NOT NULL"))
+        }
+        Expr::IsTrue(expr) => {
+            let expr = create_physical_name(expr, false)?;
+            Ok(format!("{expr} IS TRUE"))
+        }
+        Expr::IsFalse(expr) => {
+            let expr = create_physical_name(expr, false)?;
+            Ok(format!("{expr} IS FALSE"))
+        }
+        Expr::IsUnknown(expr) => {
+            let expr = create_physical_name(expr, false)?;
+            Ok(format!("{expr} IS UNKNOWN"))
+        }
+        Expr::IsNotTrue(expr) => {
+            let expr = create_physical_name(expr, false)?;
+            Ok(format!("{expr} IS NOT TRUE"))
+        }
+        Expr::IsNotFalse(expr) => {
+            let expr = create_physical_name(expr, false)?;
+            Ok(format!("{expr} IS NOT FALSE"))
+        }
+        Expr::IsNotUnknown(expr) => {
+            let expr = create_physical_name(expr, false)?;
+            Ok(format!("{expr} IS NOT UNKNOWN"))
+        }
+        Expr::GetIndexedField(GetIndexedField { key, expr }) => {
+            let expr = create_physical_name(expr, false)?;
+            Ok(format!("{expr}[{key}]"))
+        }
+        Expr::ScalarFunction(func) => {
+            create_function_physical_name(&func.fun.to_string(), false, &func.args)
+        }
+        Expr::ScalarUDF(ScalarUDF { fun, args }) => {
+            create_function_physical_name(&fun.name, false, args)
+        }
+        Expr::WindowFunction(WindowFunction { fun, args, .. }) => {
+            create_function_physical_name(&fun.to_string(), false, args)
+        }
+        Expr::AggregateFunction(AggregateFunction {
+            fun,
+            distinct,
+            args,
+            ..
+        }) => create_function_physical_name(&fun.to_string(), *distinct, args),
+        Expr::AggregateUDF(AggregateUDF {
+            fun,
+            args,
+            filter,
+            order_by,
+        }) => {
+            // TODO: Add support for filter and order by in AggregateUDF
+            if filter.is_some() {
+                return Err(DataFusionError::Execution(
+                    "aggregate expression with filter is not supported".to_string(),
+                ));
+            }
+            if order_by.is_some() {
+                return Err(DataFusionError::Execution(
+                    "aggregate expression with order_by is not supported".to_string(),
+                ));
+            }
+            let mut names = Vec::with_capacity(args.len());
+            for e in args {
+                names.push(create_physical_name(e, false)?);
+            }
+            Ok(format!("{}({})", fun.name, names.join(",")))
+        }
+        Expr::GroupingSet(grouping_set) => match grouping_set {
+            GroupingSet::Rollup(exprs) => Ok(format!(
+                "ROLLUP ({})",
+                exprs
+                    .iter()
+                    .map(|e| create_physical_name(e, false))
+                    .collect::<Result<Vec<_>>>()?
+                    .join(", ")
+            )),
+            GroupingSet::Cube(exprs) => Ok(format!(
+                "CUBE ({})",
+                exprs
+                    .iter()
+                    .map(|e| create_physical_name(e, false))
+                    .collect::<Result<Vec<_>>>()?
+                    .join(", ")
+            )),
+            GroupingSet::GroupingSets(lists_of_exprs) => {
+                let mut strings = vec![];
+                for exprs in lists_of_exprs {
+                    let exprs_str = exprs
+                        .iter()
+                        .map(|e| create_physical_name(e, false))
+                        .collect::<Result<Vec<_>>>()?
+                        .join(", ");
+                    strings.push(format!("({exprs_str})"));
+                }
+                Ok(format!("GROUPING SETS ({})", strings.join(", ")))
+            }
+        },
+
+        Expr::InList(InList {
+            expr,
+            list,
+            negated,
+        }) => {
+            let expr = create_physical_name(expr, false)?;
+            let list = list.iter().map(|expr| create_physical_name(expr, false));
+            if *negated {
+                Ok(format!("{expr} NOT IN ({list:?})"))
+            } else {
+                Ok(format!("{expr} IN ({list:?})"))
+            }
+        }
+        Expr::Exists { .. } => Err(DataFusionError::NotImplemented(
+            "EXISTS is not yet supported in the physical plan".to_string(),
+        )),
+        Expr::InSubquery(_) => Err(DataFusionError::NotImplemented(
+            "IN subquery is not yet supported in the physical plan".to_string(),
+        )),
+        Expr::ScalarSubquery(_) => Err(DataFusionError::NotImplemented(
+            "Scalar subqueries are not yet supported in the physical plan".to_string(),
+        )),
+        Expr::Between(Between {
+            expr,
+            negated,
+            low,
+            high,
+        }) => {
+            let expr = create_physical_name(expr, false)?;
+            let low = create_physical_name(low, false)?;
+            let high = create_physical_name(high, false)?;
+            if *negated {
+                Ok(format!("{expr} NOT BETWEEN {low} AND {high}"))
+            } else {
+                Ok(format!("{expr} BETWEEN {low} AND {high}"))
+            }
+        }
+        Expr::Like(Like {
+            negated,
+            expr,
+            pattern,
+            escape_char,
+        }) => {
+            let expr = create_physical_name(expr, false)?;
+            let pattern = create_physical_name(pattern, false)?;
+            let escape = if let Some(char) = escape_char {
+                format!("CHAR '{char}'")
+            } else {
+                "".to_string()
+            };
+            if *negated {
+                Ok(format!("{expr} NOT LIKE {pattern}{escape}"))
+            } else {
+                Ok(format!("{expr} LIKE {pattern}{escape}"))
+            }
+        }
+        Expr::ILike(Like {
+            negated,
+            expr,
+            pattern,
+            escape_char,
+        }) => {
+            let expr = create_physical_name(expr, false)?;
+            let pattern = create_physical_name(pattern, false)?;
+            let escape = if let Some(char) = escape_char {
+                format!("CHAR '{char}'")
+            } else {
+                "".to_string()
+            };
+            if *negated {
+                Ok(format!("{expr} NOT ILIKE {pattern}{escape}"))
+            } else {
+                Ok(format!("{expr} ILIKE {pattern}{escape}"))
+            }
+        }
+        Expr::SimilarTo(Like {
+            negated,
+            expr,
+            pattern,
+            escape_char,
+        }) => {
+            let expr = create_physical_name(expr, false)?;
+            let pattern = create_physical_name(pattern, false)?;
+            let escape = if let Some(char) = escape_char {
+                format!("CHAR '{char}'")
+            } else {
+                "".to_string()
+            };
+            if *negated {
+                Ok(format!("{expr} NOT SIMILAR TO {pattern}{escape}"))
+            } else {
+                Ok(format!("{expr} SIMILAR TO {pattern}{escape}"))
+            }
+        }
+        Expr::Sort { .. } => Err(DataFusionError::Internal(
+            "Create physical name does not support sort expression".to_string(),
+        )),
+        Expr::Wildcard => Err(DataFusionError::Internal(
+            "Create physical name does not support wildcard".to_string(),
+        )),
+        Expr::QualifiedWildcard { .. } => Err(DataFusionError::Internal(
+            "Create physical name does not support qualified wildcard".to_string(),
+        )),
+        Expr::Placeholder(_) => Err(DataFusionError::Internal(
+            "Create physical name does not support placeholder".to_string(),
+        )),
+        Expr::OuterReferenceColumn(_, _) => Err(DataFusionError::Internal(
+            "Create physical name does not support OuterReferenceColumn".to_string(),
+        )),
+    }
+}