Merge remote-tracking branch 'apache/main' into alamb/features2

Author: alamb

benchmarks/src/imdb/convert.rs

@@ -16,6 +16,7 @@
 // under the License.
 
 use datafusion::dataframe::DataFrameWriteOptions;
+use datafusion::logical_expr::select_expr::SelectExpr;
 use datafusion_common::instant::Instant;
 use std::path::PathBuf;
 
@@ -74,7 +75,8 @@ impl ConvertOpt {
                 .iter()
                 .take(schema.fields.len())
                 .map(Expr::from)
-                .collect();
+                .map(SelectExpr::from)
+                .collect::<Vec<_>>();
 
             csv = csv.select(selection)?;
 

benchmarks/src/tpch/convert.rs

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use datafusion::logical_expr::select_expr::SelectExpr;
 use datafusion_common::instant::Instant;
 use std::fs;
 use std::path::{Path, PathBuf};
@@ -89,7 +90,8 @@ impl ConvertOpt {
                 .iter()
                 .take(schema.fields.len() - 1)
                 .map(Expr::from)
-                .collect();
+                .map(SelectExpr::from)
+                .collect::<Vec<_>>();
 
             csv = csv.select(selection)?;
             // optionally, repartition the file

datafusion/core/src/dataframe/mod.rs

@@ -54,6 +54,7 @@ use datafusion_common::{
     exec_err, not_impl_err, plan_datafusion_err, plan_err, Column, DFSchema,
     DataFusionError, ParamValues, ScalarValue, SchemaError, UnnestOptions,
 };
+use datafusion_expr::select_expr::SelectExpr;
 use datafusion_expr::{
     case,
     dml::InsertOp,
@@ -342,13 +343,28 @@ impl DataFrame {
     /// # Ok(())
     /// # }
     /// ```
-    pub fn select(self, expr_list: Vec<Expr>) -> Result<DataFrame> {
-        let window_func_exprs = find_window_exprs(&expr_list);
+    pub fn select(
+        self,
+        expr_list: impl IntoIterator<Item = impl Into<SelectExpr>>,
+    ) -> Result<DataFrame> {
+        let expr_list: Vec<SelectExpr> =
+            expr_list.into_iter().map(|e| e.into()).collect::<Vec<_>>();
+
+        let expressions = expr_list
+            .iter()
+            .filter_map(|e| match e {
+                SelectExpr::Expression(expr) => Some(expr.clone()),
+                _ => None,
+            })
+            .collect::<Vec<_>>();
+
+        let window_func_exprs = find_window_exprs(&expressions);
         let plan = if window_func_exprs.is_empty() {
             self.plan
         } else {
             LogicalPlanBuilder::window_plan(self.plan, window_func_exprs)?
         };
+
         let project_plan = LogicalPlanBuilder::from(plan).project(expr_list)?.build()?;
 
         Ok(DataFrame {

datafusion/core/tests/dataframe/mod.rs

@@ -73,7 +73,7 @@ use datafusion_expr::expr::{GroupingSet, Sort, WindowFunction};
 use datafusion_expr::var_provider::{VarProvider, VarType};
 use datafusion_expr::{
     cast, col, create_udf, exists, in_subquery, lit, out_ref_col, placeholder,
-    scalar_subquery, when, Expr, ExprFunctionExt, ExprSchemable, LogicalPlan,
+    scalar_subquery, when, wildcard, Expr, ExprFunctionExt, ExprSchemable, LogicalPlan,
     ScalarFunctionImplementation, WindowFrame, WindowFrameBound, WindowFrameUnits,
     WindowFunctionDefinition,
 };
@@ -269,6 +269,16 @@ async fn select_expr() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn select_all() -> Result<()> {
+    let t = test_table().await?;
+    let plan = t.select([wildcard()])?.logical_plan().clone();
+    let sql_plan = create_plan("SELECT * FROM aggregate_test_100").await?;
+    assert_same_plan(&plan, &sql_plan);
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn select_exprs() -> Result<()> {
     // build plan using `select_expr``

datafusion/core/tests/physical_optimizer/projection_pushdown.rs

@@ -29,7 +29,9 @@ use datafusion_common::Result;
 use datafusion_common::{JoinSide, JoinType, ScalarValue};
 use datafusion_execution::object_store::ObjectStoreUrl;
 use datafusion_execution::{SendableRecordBatchStream, TaskContext};
-use datafusion_expr::{Operator, ScalarUDF, ScalarUDFImpl, Signature, Volatility};
+use datafusion_expr::{
+    Operator, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, Volatility,
+};
 use datafusion_physical_expr::expressions::{
     binary, cast, col, BinaryExpr, CaseExpr, CastExpr, Column, Literal, NegativeExpr,
 };
@@ -57,6 +59,7 @@ use datafusion_physical_plan::streaming::StreamingTableExec;
 use datafusion_physical_plan::union::UnionExec;
 use datafusion_physical_plan::{get_plan_string, ExecutionPlan};
 
+use datafusion_expr_common::columnar_value::ColumnarValue;
 use itertools::Itertools;
 
 /// Mocked UDF
@@ -89,6 +92,10 @@ impl ScalarUDFImpl for DummyUDF {
     fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
         Ok(DataType::Int32)
     }
+
+    fn invoke_with_args(&self, _args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+        panic!("dummy - not implemented")
+    }
 }
 
 #[test]

datafusion/core/tests/user_defined/user_defined_scalar_functions.rs

@@ -679,6 +679,10 @@ impl ScalarUDFImpl for CastToI64UDF {
         Ok(DataType::Int64)
     }
 
+    fn invoke_with_args(&self, _args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+        panic!("dummy - not implemented")
+    }
+
     // Demonstrate simplifying a UDF
     fn simplify(
         &self,
@@ -946,6 +950,10 @@ impl ScalarUDFImpl for ScalarFunctionWrapper {
         Ok(self.return_type.clone())
     }
 
+    fn invoke_with_args(&self, _args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+        panic!("dummy - not implemented")
+    }
+
     fn simplify(
         &self,
         args: Vec<Expr>,

datafusion/datasource-parquet/src/source.rs

@@ -34,6 +34,7 @@ use datafusion_common::config::TableParquetOptions;
 use datafusion_common::Statistics;
 use datafusion_datasource::file::FileSource;
 use datafusion_datasource::file_scan_config::FileScanConfig;
+use datafusion_physical_expr_common::physical_expr::fmt_sql;
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
 use datafusion_physical_optimizer::pruning::PruningPredicate;
 use datafusion_physical_plan::metrics::{ExecutionPlanMetricsSet, MetricBuilder};
@@ -580,7 +581,7 @@ impl FileSource for ParquetSource {
             }
             DisplayFormatType::TreeRender => {
                 if let Some(predicate) = self.predicate() {
-                    writeln!(f, "predicate={predicate}")?;
+                    writeln!(f, "predicate={}", fmt_sql(predicate.as_ref()))?;
                 }
                 Ok(())
             }

datafusion/expr/src/expr_fn.rs

@@ -25,6 +25,7 @@ use crate::function::{
     AccumulatorArgs, AccumulatorFactoryFunction, PartitionEvaluatorFactory,
     StateFieldsArgs,
 };
+use crate::select_expr::SelectExpr;
 use crate::{
     conditional_expressions::CaseBuilder, expr::Sort, logical_plan::Subquery,
     AggregateUDF, Expr, LogicalPlan, Operator, PartitionEvaluator, ScalarFunctionArgs,
@@ -120,21 +121,13 @@ pub fn placeholder(id: impl Into<String>) -> Expr {
 /// let p = wildcard();
 /// assert_eq!(p.to_string(), "*")
 /// ```
-pub fn wildcard() -> Expr {
-    #[expect(deprecated)]
-    Expr::Wildcard {
-        qualifier: None,
-        options: Box::new(WildcardOptions::default()),
-    }
+pub fn wildcard() -> SelectExpr {
+    SelectExpr::Wildcard(WildcardOptions::default())
 }
 
 /// Create an '*' [`Expr::Wildcard`] expression with the wildcard options
-pub fn wildcard_with_options(options: WildcardOptions) -> Expr {
-    #[expect(deprecated)]
-    Expr::Wildcard {
-        qualifier: None,
-        options: Box::new(options),
-    }
+pub fn wildcard_with_options(options: WildcardOptions) -> SelectExpr {
+    SelectExpr::Wildcard(options)
 }
 
 /// Create an 't.*' [`Expr::Wildcard`] expression that matches all columns from a specific table
@@ -147,24 +140,16 @@ pub fn wildcard_with_options(options: WildcardOptions) -> Expr {
 /// let p = qualified_wildcard(TableReference::bare("t"));
 /// assert_eq!(p.to_string(), "t.*")
 /// ```
-pub fn qualified_wildcard(qualifier: impl Into<TableReference>) -> Expr {
-    #[expect(deprecated)]
-    Expr::Wildcard {
-        qualifier: Some(qualifier.into()),
-        options: Box::new(WildcardOptions::default()),
-    }
+pub fn qualified_wildcard(qualifier: impl Into<TableReference>) -> SelectExpr {
+    SelectExpr::QualifiedWildcard(qualifier.into(), WildcardOptions::default())
 }
 
 /// Create an 't.*' [`Expr::Wildcard`] expression with the wildcard options
 pub fn qualified_wildcard_with_options(
     qualifier: impl Into<TableReference>,
     options: WildcardOptions,
-) -> Expr {
-    #[expect(deprecated)]
-    Expr::Wildcard {
-        qualifier: Some(qualifier.into()),
-        options: Box::new(options),
-    }
+) -> SelectExpr {
+    SelectExpr::QualifiedWildcard(qualifier.into(), options)
 }
 
 /// Return a new expression `left <op> right`

datafusion/expr/src/lib.rs

@@ -48,6 +48,7 @@ pub mod expr_fn;
 pub mod expr_rewriter;
 pub mod expr_schema;
 pub mod function;
+pub mod select_expr;
 pub mod groups_accumulator {
     pub use datafusion_expr_common::groups_accumulator::*;
 }

datafusion/expr/src/logical_plan/builder.rs

@@ -24,7 +24,7 @@ use std::iter::once;
 use std::sync::Arc;
 
 use crate::dml::CopyTo;
-use crate::expr::{Alias, Sort as SortExpr};
+use crate::expr::{Alias, PlannedReplaceSelectItem, Sort as SortExpr};
 use crate::expr_rewriter::{
     coerce_plan_expr_for_schema, normalize_col,
     normalize_col_with_schemas_and_ambiguity_check, normalize_cols, normalize_sorts,
@@ -36,9 +36,11 @@ use crate::logical_plan::{
     Projection, Repartition, Sort, SubqueryAlias, TableScan, Union, Unnest, Values,
     Window,
 };
+use crate::select_expr::SelectExpr;
 use crate::utils::{
-    can_hash, columnize_expr, compare_sort_expr, expr_to_columns,
-    find_valid_equijoin_key_pair, group_window_expr_by_sort_keys,
+    can_hash, columnize_expr, compare_sort_expr, expand_qualified_wildcard,
+    expand_wildcard, expr_to_columns, find_valid_equijoin_key_pair,
+    group_window_expr_by_sort_keys,
 };
 use crate::{
     and, binary_expr, lit, DmlStatement, Expr, ExprSchemable, Operator, RecursiveQuery,
@@ -520,10 +522,11 @@ impl LogicalPlanBuilder {
         }
         Ok(plan)
     }
+
     /// Apply a projection without alias.
     pub fn project(
         self,
-        expr: impl IntoIterator<Item = impl Into<Expr>>,
+        expr: impl IntoIterator<Item = impl Into<SelectExpr>>,
     ) -> Result<Self> {
         project(Arc::unwrap_or_clone(self.plan), expr).map(Self::new)
     }
@@ -532,7 +535,7 @@ impl LogicalPlanBuilder {
     /// (true to validate, false to not validate)
     pub fn project_with_validation(
         self,
-        expr: Vec<(impl Into<Expr>, bool)>,
+        expr: Vec<(impl Into<SelectExpr>, bool)>,
     ) -> Result<Self> {
         project_with_validation(Arc::unwrap_or_clone(self.plan), expr).map(Self::new)
     }
@@ -1657,7 +1660,7 @@ pub fn union_by_name(
 /// * An invalid expression is used (e.g. a `sort` expression)
 pub fn project(
     plan: LogicalPlan,
-    expr: impl IntoIterator<Item = impl Into<Expr>>,
+    expr: impl IntoIterator<Item = impl Into<SelectExpr>>,
 ) -> Result<LogicalPlan> {
     project_with_validation(plan, expr.into_iter().map(|e| (e, true)))
 }
@@ -1671,15 +1674,54 @@ pub fn project(
 /// * An invalid expression is used (e.g. a `sort` expression)
 fn project_with_validation(
     plan: LogicalPlan,
-    expr: impl IntoIterator<Item = (impl Into<Expr>, bool)>,
+    expr: impl IntoIterator<Item = (impl Into<SelectExpr>, bool)>,
 ) -> Result<LogicalPlan> {
     let mut projected_expr = vec![];
     for (e, validate) in expr {
         let e = e.into();
         match e {
-            #[expect(deprecated)]
-            Expr::Wildcard { .. } => projected_expr.push(e),
-            _ => {
+            SelectExpr::Wildcard(opt) => {
+                let expanded = expand_wildcard(plan.schema(), &plan, Some(&opt))?;
+
+                // If there is a REPLACE statement, replace that column with the given
+                // replace expression. Column name remains the same.
+                let expanded = if let Some(replace) = opt.replace {
+                    replace_columns(expanded, &replace)?
+                } else {
+                    expanded
+                };
+
+                for e in expanded {
+                    if validate {
+                        projected_expr
+                            .push(columnize_expr(normalize_col(e, &plan)?, &plan)?)
+                    } else {
+                        projected_expr.push(e)
+                    }
+                }
+            }
+            SelectExpr::QualifiedWildcard(table_ref, opt) => {
+                let expanded =
+                    expand_qualified_wildcard(&table_ref, plan.schema(), Some(&opt))?;
+
+                // If there is a REPLACE statement, replace that column with the given
+                // replace expression. Column name remains the same.
+                let expanded = if let Some(replace) = opt.replace {
+                    replace_columns(expanded, &replace)?
+                } else {
+                    expanded
+                };
+
+                for e in expanded {
+                    if validate {
+                        projected_expr
+                            .push(columnize_expr(normalize_col(e, &plan)?, &plan)?)
+                    } else {
+                        projected_expr.push(e)
+                    }
+                }
+            }
+            SelectExpr::Expression(e) => {
                 if validate {
                     projected_expr.push(columnize_expr(normalize_col(e, &plan)?, &plan)?)
                 } else {
@@ -1693,6 +1735,29 @@ fn project_with_validation(
     Projection::try_new(projected_expr, Arc::new(plan)).map(LogicalPlan::Projection)
 }
 
+/// If there is a REPLACE statement in the projected expression in the form of
+/// "REPLACE (some_column_within_an_expr AS some_column)", this function replaces
+/// that column with the given replace expression. Column name remains the same.
+/// Multiple REPLACEs are also possible with comma separations.
+fn replace_columns(
+    mut exprs: Vec<Expr>,
+    replace: &PlannedReplaceSelectItem,
+) -> Result<Vec<Expr>> {
+    for expr in exprs.iter_mut() {
+        if let Expr::Column(Column { name, .. }) = expr {
+            if let Some((_, new_expr)) = replace
+                .items()
+                .iter()
+                .zip(replace.expressions().iter())
+                .find(|(item, _)| item.column_name.value == *name)
+            {
+                *expr = new_expr.clone().alias(name.clone())
+            }
+        }
+    }
+    Ok(exprs)
+}
+
 /// Create a SubqueryAlias to wrap a LogicalPlan.
 pub fn subquery_alias(
     plan: LogicalPlan,
@@ -1811,7 +1876,7 @@ pub fn wrap_projection_for_join_if_necessary(
         projection.extend(join_key_items);
 
         LogicalPlanBuilder::from(input)
-            .project(projection)?
+            .project(projection.into_iter().map(SelectExpr::from))?
             .build()?
     } else {
         input