This is an automated email from the ASF dual-hosted git repository.
jayzhan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new bf51babc9c Optionally create name of aggregate expression from
expressions (#11776)
bf51babc9c is described below
commit bf51babc9c7dc3e95476f84169646b4452fb9799
Author: 张林伟 <[email protected]>
AuthorDate: Tue Aug 6 13:05:55 2024 +0800
Optionally create name of aggregate expression from expressions (#11776)
* Optionally create name of aggregate expression from expressions
* Fix test
* name -> alias
* keep trying to create aggr expr's display name first
---
.../combine_partial_final_agg.rs | 4 +-
datafusion/core/src/physical_planner.rs | 272 +--------------------
datafusion/core/src/test_util/mod.rs | 2 +-
datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs | 2 +-
datafusion/expr/src/expr.rs | 262 +++++++++++++++++++-
.../physical-expr-common/src/aggregate/mod.rs | 40 ++-
datafusion/physical-plan/src/aggregates/mod.rs | 52 ++--
datafusion/physical-plan/src/windows/mod.rs | 2 +-
datafusion/proto/src/physical_plan/mod.rs | 2 +-
.../proto/tests/cases/roundtrip_physical_plan.rs | 20 +-
10 files changed, 337 insertions(+), 321 deletions(-)
diff --git
a/datafusion/core/src/physical_optimizer/combine_partial_final_agg.rs
b/datafusion/core/src/physical_optimizer/combine_partial_final_agg.rs
index 6f3274820c..843efcc7b0 100644
--- a/datafusion/core/src/physical_optimizer/combine_partial_final_agg.rs
+++ b/datafusion/core/src/physical_optimizer/combine_partial_final_agg.rs
@@ -280,7 +280,7 @@ mod tests {
) -> Arc<dyn AggregateExpr> {
AggregateExprBuilder::new(count_udaf(), vec![expr])
.schema(Arc::new(schema.clone()))
- .name(name)
+ .alias(name)
.build()
.unwrap()
}
@@ -364,7 +364,7 @@ mod tests {
vec![
AggregateExprBuilder::new(sum_udaf(), vec![col("b", &schema)?])
.schema(Arc::clone(&schema))
- .name("Sum(b)")
+ .alias("Sum(b)")
.build()
.unwrap(),
];
diff --git a/datafusion/core/src/physical_planner.rs
b/datafusion/core/src/physical_planner.rs
index 378a892111..65cdbf9fe6 100644
--- a/datafusion/core/src/physical_planner.rs
+++ b/datafusion/core/src/physical_planner.rs
@@ -19,7 +19,6 @@
use std::borrow::Cow;
use std::collections::HashMap;
-use std::fmt::Write;
use std::sync::Arc;
use crate::datasource::file_format::file_type_to_format;
@@ -74,11 +73,9 @@ use datafusion_common::{
};
use datafusion_expr::dml::CopyTo;
use datafusion_expr::expr::{
- self, AggregateFunction, Alias, Between, BinaryExpr, Cast, GroupingSet,
InList, Like,
- TryCast, WindowFunction,
+ self, physical_name, AggregateFunction, Alias, GroupingSet, WindowFunction,
};
use datafusion_expr::expr_rewriter::unnormalize_cols;
-use datafusion_expr::expr_vec_fmt;
use
datafusion_expr::logical_plan::builder::wrap_projection_for_join_if_necessary;
use datafusion_expr::{
DescribeTable, DmlStatement, Extension, Filter, RecursiveQuery,
StringifiedPlan,
@@ -97,265 +94,6 @@ use log::{debug, trace};
use sqlparser::ast::NullTreatment;
use tokio::sync::Mutex;
-fn create_function_physical_name(
- fun: &str,
- distinct: bool,
- args: &[Expr],
- order_by: Option<&Vec<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 => "",
- };
-
- let phys_name = format!("{}({}{})", fun, distinct_str, names.join(","));
-
- Ok(order_by
- .map(|order_by| format!("{} ORDER BY [{}]", phys_name,
expr_vec_fmt!(order_by)))
- .unwrap_or(phys_name))
-}
-
-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::Unnest(_) => {
- internal_err!(
- "Expr::Unnest should have been converted to
LogicalPlan::Unnest"
- )
- }
- Expr::Column(c) => {
- if is_first_expr {
- Ok(c.name.clone())
- } else {
- Ok(c.flat_name())
- }
- }
- Expr::Alias(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, "{} ", create_physical_name(e, false)?);
- }
- for (w, t) in &case.when_then_expr {
- let _ = write!(
- name,
- "WHEN {} THEN {} ",
- create_physical_name(w, false)?,
- create_physical_name(t, false)?
- );
- }
- if let Some(e) = &case.else_expr {
- let _ = write!(name, "ELSE {} ", create_physical_name(e,
false)?);
- }
- 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::ScalarFunction(fun) => fun.func.display_name(&fun.args),
- Expr::WindowFunction(WindowFunction {
- fun,
- args,
- order_by,
- ..
- }) => {
- create_function_physical_name(&fun.to_string(), false, args,
Some(order_by))
- }
- Expr::AggregateFunction(AggregateFunction {
- func,
- distinct,
- args,
- filter: _,
- order_by,
- null_treatment: _,
- }) => {
- create_function_physical_name(func.name(), *distinct, args,
order_by.as_ref())
- }
- 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 { .. } => {
- not_impl_err!("EXISTS is not yet supported in the physical plan")
- }
- Expr::InSubquery(_) => {
- not_impl_err!("IN subquery is not yet supported in the physical
plan")
- }
- Expr::ScalarSubquery(_) => {
- not_impl_err!("Scalar subqueries are not yet supported in the
physical plan")
- }
- 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,
- case_insensitive,
- }) => {
- let expr = create_physical_name(expr, false)?;
- let pattern = create_physical_name(pattern, false)?;
- let op_name = if *case_insensitive { "ILIKE" } else { "LIKE" };
- let escape = if let Some(char) = escape_char {
- format!("CHAR '{char}'")
- } else {
- "".to_string()
- };
- if *negated {
- Ok(format!("{expr} NOT {op_name} {pattern}{escape}"))
- } else {
- Ok(format!("{expr} {op_name} {pattern}{escape}"))
- }
- }
- Expr::SimilarTo(Like {
- negated,
- expr,
- pattern,
- escape_char,
- case_insensitive: _,
- }) => {
- 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 { .. } => {
- internal_err!("Create physical name does not support sort
expression")
- }
- Expr::Wildcard { .. } => {
- internal_err!("Create physical name does not support wildcard")
- }
- Expr::Placeholder(_) => {
- internal_err!("Create physical name does not support placeholder")
- }
- Expr::OuterReferenceColumn(_, _) => {
- internal_err!("Create physical name does not support
OuterReferenceColumn")
- }
- }
-}
-
/// Physical query planner that converts a `LogicalPlan` to an
/// `ExecutionPlan` suitable for execution.
#[async_trait]
@@ -1807,7 +1545,7 @@ type AggregateExprWithOptionalArgs = (
/// Create an aggregate expression with a name from a logical expression
pub fn create_aggregate_expr_with_name_and_maybe_filter(
e: &Expr,
- name: impl Into<String>,
+ name: Option<String>,
logical_input_schema: &DFSchema,
_physical_input_schema: &Schema,
execution_props: &ExecutionProps,
@@ -1881,9 +1619,9 @@ pub fn create_aggregate_expr_and_maybe_filter(
) -> Result<AggregateExprWithOptionalArgs> {
// unpack (nested) aliased logical expressions, e.g. "sum(col) as total"
let (name, e) = match e {
- Expr::Alias(Alias { expr, name, .. }) => (name.clone(), expr.as_ref()),
- Expr::AggregateFunction(_) =>
(e.display_name().unwrap_or(physical_name(e)?), e),
- _ => (physical_name(e)?, e),
+ Expr::Alias(Alias { expr, name, .. }) => (Some(name.clone()),
expr.as_ref()),
+ Expr::AggregateFunction(_) => (e.display_name().ok(), e),
+ _ => (None, e),
};
create_aggregate_expr_with_name_and_maybe_filter(
diff --git a/datafusion/core/src/test_util/mod.rs
b/datafusion/core/src/test_util/mod.rs
index 6eb82dece3..9610a7f203 100644
--- a/datafusion/core/src/test_util/mod.rs
+++ b/datafusion/core/src/test_util/mod.rs
@@ -432,7 +432,7 @@ impl TestAggregate {
pub fn count_expr(&self, schema: &Schema) -> Arc<dyn AggregateExpr> {
AggregateExprBuilder::new(count_udaf(), vec![self.column()])
.schema(Arc::new(schema.clone()))
- .name(self.column_name())
+ .alias(self.column_name())
.build()
.unwrap()
}
diff --git a/datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs
b/datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs
index 6f286c9aeb..4cecb0b693 100644
--- a/datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs
+++ b/datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs
@@ -107,7 +107,7 @@ async fn run_aggregate_test(input1: Vec<RecordBatch>,
group_by_columns: Vec<&str
vec![
AggregateExprBuilder::new(sum_udaf(), vec![col("d",
&schema).unwrap()])
.schema(Arc::clone(&schema))
- .name("sum1")
+ .alias("sum1")
.build()
.unwrap(),
];
diff --git a/datafusion/expr/src/expr.rs b/datafusion/expr/src/expr.rs
index 1a51c181f4..edf45a244e 100644
--- a/datafusion/expr/src/expr.rs
+++ b/datafusion/expr/src/expr.rs
@@ -38,7 +38,8 @@ use datafusion_common::tree_node::{
Transformed, TransformedResult, TreeNode, TreeNodeRecursion,
};
use datafusion_common::{
- internal_err, plan_err, Column, DFSchema, Result, ScalarValue,
TableReference,
+ internal_err, not_impl_err, plan_err, Column, DFSchema, Result,
ScalarValue,
+ TableReference,
};
use sqlparser::ast::NullTreatment;
@@ -2277,6 +2278,265 @@ fn write_names_join<W: Write>(w: &mut W, exprs:
&[Expr], sep: &str) -> Result<()
Ok(())
}
+pub fn create_function_physical_name(
+ fun: &str,
+ distinct: bool,
+ args: &[Expr],
+ order_by: Option<&Vec<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 => "",
+ };
+
+ let phys_name = format!("{}({}{})", fun, distinct_str, names.join(","));
+
+ Ok(order_by
+ .map(|order_by| format!("{} ORDER BY [{}]", phys_name,
expr_vec_fmt!(order_by)))
+ .unwrap_or(phys_name))
+}
+
+pub 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::Unnest(_) => {
+ internal_err!(
+ "Expr::Unnest should have been converted to
LogicalPlan::Unnest"
+ )
+ }
+ Expr::Column(c) => {
+ if is_first_expr {
+ Ok(c.name.clone())
+ } else {
+ Ok(c.flat_name())
+ }
+ }
+ Expr::Alias(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, "{} ", create_physical_name(e, false)?);
+ }
+ for (w, t) in &case.when_then_expr {
+ let _ = write!(
+ name,
+ "WHEN {} THEN {} ",
+ create_physical_name(w, false)?,
+ create_physical_name(t, false)?
+ );
+ }
+ if let Some(e) = &case.else_expr {
+ let _ = write!(name, "ELSE {} ", create_physical_name(e,
false)?);
+ }
+ 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::ScalarFunction(fun) => fun.func.display_name(&fun.args),
+ Expr::WindowFunction(WindowFunction {
+ fun,
+ args,
+ order_by,
+ ..
+ }) => {
+ create_function_physical_name(&fun.to_string(), false, args,
Some(order_by))
+ }
+ Expr::AggregateFunction(AggregateFunction {
+ func,
+ distinct,
+ args,
+ filter: _,
+ order_by,
+ null_treatment: _,
+ }) => {
+ create_function_physical_name(func.name(), *distinct, args,
order_by.as_ref())
+ }
+ 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 { .. } => {
+ not_impl_err!("EXISTS is not yet supported in the physical plan")
+ }
+ Expr::InSubquery(_) => {
+ not_impl_err!("IN subquery is not yet supported in the physical
plan")
+ }
+ Expr::ScalarSubquery(_) => {
+ not_impl_err!("Scalar subqueries are not yet supported in the
physical plan")
+ }
+ 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,
+ case_insensitive,
+ }) => {
+ let expr = create_physical_name(expr, false)?;
+ let pattern = create_physical_name(pattern, false)?;
+ let op_name = if *case_insensitive { "ILIKE" } else { "LIKE" };
+ let escape = if let Some(char) = escape_char {
+ format!("CHAR '{char}'")
+ } else {
+ "".to_string()
+ };
+ if *negated {
+ Ok(format!("{expr} NOT {op_name} {pattern}{escape}"))
+ } else {
+ Ok(format!("{expr} {op_name} {pattern}{escape}"))
+ }
+ }
+ Expr::SimilarTo(Like {
+ negated,
+ expr,
+ pattern,
+ escape_char,
+ case_insensitive: _,
+ }) => {
+ 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 { .. } => {
+ internal_err!("Create physical name does not support sort
expression")
+ }
+ Expr::Wildcard { .. } => {
+ internal_err!("Create physical name does not support wildcard")
+ }
+ Expr::Placeholder(_) => {
+ internal_err!("Create physical name does not support placeholder")
+ }
+ Expr::OuterReferenceColumn(_, _) => {
+ internal_err!("Create physical name does not support
OuterReferenceColumn")
+ }
+ }
+}
+
#[cfg(test)]
mod test {
use crate::expr_fn::col;
diff --git a/datafusion/physical-expr-common/src/aggregate/mod.rs
b/datafusion/physical-expr-common/src/aggregate/mod.rs
index 665cdd7083..350023352b 100644
--- a/datafusion/physical-expr-common/src/aggregate/mod.rs
+++ b/datafusion/physical-expr-common/src/aggregate/mod.rs
@@ -22,6 +22,7 @@ use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use datafusion_common::exec_err;
use datafusion_common::{internal_err, not_impl_err, DFSchema, Result};
+use datafusion_expr::expr::create_function_physical_name;
use datafusion_expr::function::StateFieldsArgs;
use datafusion_expr::type_coercion::aggregates::check_arg_count;
use datafusion_expr::utils::AggregateOrderSensitivity;
@@ -67,7 +68,7 @@ pub fn create_aggregate_expr(
sort_exprs: &[Expr],
ordering_req: &[PhysicalSortExpr],
schema: &Schema,
- name: impl Into<String>,
+ name: Option<String>,
ignore_nulls: bool,
is_distinct: bool,
) -> Result<Arc<dyn AggregateExpr>> {
@@ -77,7 +78,9 @@ pub fn create_aggregate_expr(
builder = builder.order_by(ordering_req.to_vec());
builder = builder.logical_exprs(input_exprs.to_vec());
builder = builder.schema(Arc::new(schema.clone()));
- builder = builder.name(name);
+ if let Some(name) = name {
+ builder = builder.alias(name);
+ }
if ignore_nulls {
builder = builder.ignore_nulls();
@@ -98,7 +101,7 @@ pub fn create_aggregate_expr_with_dfschema(
sort_exprs: &[Expr],
ordering_req: &[PhysicalSortExpr],
dfschema: &DFSchema,
- name: impl Into<String>,
+ alias: Option<String>,
ignore_nulls: bool,
is_distinct: bool,
is_reversed: bool,
@@ -111,7 +114,9 @@ pub fn create_aggregate_expr_with_dfschema(
builder = builder.dfschema(dfschema.clone());
let schema: Schema = dfschema.into();
builder = builder.schema(Arc::new(schema));
- builder = builder.name(name);
+ if let Some(alias) = alias {
+ builder = builder.alias(alias);
+ }
if ignore_nulls {
builder = builder.ignore_nulls();
@@ -137,7 +142,7 @@ pub struct AggregateExprBuilder {
args: Vec<Arc<dyn PhysicalExpr>>,
/// Logical expressions of the aggregate function, it will be deprecated
in <https://github.com/apache/datafusion/issues/11359>
logical_args: Vec<Expr>,
- name: String,
+ alias: Option<String>,
/// Arrow Schema for the aggregate function
schema: SchemaRef,
/// Datafusion Schema for the aggregate function
@@ -160,7 +165,7 @@ impl AggregateExprBuilder {
fun,
args,
logical_args: vec![],
- name: String::new(),
+ alias: None,
schema: Arc::new(Schema::empty()),
dfschema: DFSchema::empty(),
sort_exprs: vec![],
@@ -176,7 +181,7 @@ impl AggregateExprBuilder {
fun,
args,
logical_args,
- name,
+ alias,
schema,
dfschema,
sort_exprs,
@@ -213,6 +218,19 @@ impl AggregateExprBuilder {
)?;
let data_type = fun.return_type(&input_exprs_types)?;
+ let name = match alias {
+ None => create_function_physical_name(
+ fun.name(),
+ is_distinct,
+ &logical_args,
+ if sort_exprs.is_empty() {
+ None
+ } else {
+ Some(&sort_exprs)
+ },
+ )?,
+ Some(alias) => alias,
+ };
Ok(Arc::new(AggregateFunctionExpr {
fun: Arc::unwrap_or_clone(fun),
@@ -232,8 +250,8 @@ impl AggregateExprBuilder {
}))
}
- pub fn name(mut self, name: impl Into<String>) -> Self {
- self.name = name.into();
+ pub fn alias(mut self, alias: impl Into<String>) -> Self {
+ self.alias = Some(alias.into());
self
}
@@ -680,7 +698,7 @@ impl AggregateExpr for AggregateFunctionExpr {
&self.sort_exprs,
&self.ordering_req,
&self.dfschema,
- self.name(),
+ Some(self.name().to_string()),
self.ignore_nulls,
self.is_distinct,
self.is_reversed,
@@ -721,7 +739,7 @@ impl AggregateExpr for AggregateFunctionExpr {
&reverse_sort_exprs,
&reverse_ordering_req,
&self.dfschema,
- name,
+ Some(name),
self.ignore_nulls,
self.is_distinct,
!self.is_reversed,
diff --git a/datafusion/physical-plan/src/aggregates/mod.rs
b/datafusion/physical-plan/src/aggregates/mod.rs
index 8941418c12..d72da9b300 100644
--- a/datafusion/physical-plan/src/aggregates/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/mod.rs
@@ -1362,7 +1362,7 @@ mod tests {
let aggregates = vec![AggregateExprBuilder::new(count_udaf(),
vec![lit(1i8)])
.schema(Arc::clone(&input_schema))
- .name("COUNT(1)")
+ .alias("COUNT(1)")
.logical_exprs(vec![datafusion_expr::lit(1i8)])
.build()?];
@@ -1507,7 +1507,7 @@ mod tests {
vec![
AggregateExprBuilder::new(avg_udaf(), vec![col("b",
&input_schema)?])
.schema(Arc::clone(&input_schema))
- .name("AVG(b)")
+ .alias("AVG(b)")
.build()?,
];
@@ -1803,7 +1803,7 @@ mod tests {
fn test_median_agg_expr(schema: SchemaRef) -> Result<Arc<dyn
AggregateExpr>> {
AggregateExprBuilder::new(median_udaf(), vec![col("a", &schema)?])
.schema(schema)
- .name("MEDIAN(a)")
+ .alias("MEDIAN(a)")
.build()
}
@@ -1834,7 +1834,7 @@ mod tests {
vec![
AggregateExprBuilder::new(avg_udaf(), vec![col("b",
&input_schema)?])
.schema(Arc::clone(&input_schema))
- .name("AVG(b)")
+ .alias("AVG(b)")
.build()?,
];
@@ -1894,7 +1894,7 @@ mod tests {
vec![
AggregateExprBuilder::new(avg_udaf(), vec![col("a", &schema)?])
.schema(Arc::clone(&schema))
- .name("AVG(a)")
+ .alias("AVG(a)")
.build()?,
];
@@ -1934,7 +1934,7 @@ mod tests {
vec![
AggregateExprBuilder::new(avg_udaf(), vec![col("b", &schema)?])
.schema(Arc::clone(&schema))
- .name("AVG(b)")
+ .alias("AVG(b)")
.build()?,
];
@@ -2002,7 +2002,7 @@ mod tests {
&sort_exprs,
&ordering_req,
dfschema,
- "FIRST_VALUE(b)",
+ None,
false,
false,
false,
@@ -2034,7 +2034,7 @@ mod tests {
&sort_exprs,
&ordering_req,
dfschema,
- "LAST_VALUE(b)",
+ None,
false,
false,
false,
@@ -2130,24 +2130,24 @@ mod tests {
let result = crate::collect(aggregate_final, task_ctx).await?;
if is_first_acc {
let expected = [
- "+---+----------------+",
- "| a | FIRST_VALUE(b) |",
- "+---+----------------+",
- "| 2 | 0.0 |",
- "| 3 | 1.0 |",
- "| 4 | 3.0 |",
- "+---+----------------+",
+ "+---+--------------------------------------------+",
+ "| a | first_value(b) ORDER BY [b ASC NULLS LAST] |",
+ "+---+--------------------------------------------+",
+ "| 2 | 0.0 |",
+ "| 3 | 1.0 |",
+ "| 4 | 3.0 |",
+ "+---+--------------------------------------------+",
];
assert_batches_eq!(expected, &result);
} else {
let expected = [
- "+---+---------------+",
- "| a | LAST_VALUE(b) |",
- "+---+---------------+",
- "| 2 | 3.0 |",
- "| 3 | 5.0 |",
- "| 4 | 6.0 |",
- "+---+---------------+",
+ "+---+-------------------------------------------+",
+ "| a | last_value(b) ORDER BY [b ASC NULLS LAST] |",
+ "+---+-------------------------------------------+",
+ "| 2 | 3.0 |",
+ "| 3 | 5.0 |",
+ "| 4 | 6.0 |",
+ "+---+-------------------------------------------+",
];
assert_batches_eq!(expected, &result);
};
@@ -2267,7 +2267,7 @@ mod tests {
&sort_exprs,
&ordering_req,
&test_df_schema,
- "array_agg",
+ None,
false,
false,
false,
@@ -2363,7 +2363,7 @@ mod tests {
let aggregates: Vec<Arc<dyn AggregateExpr>> =
vec![AggregateExprBuilder::new(count_udaf(), vec![lit(1)])
.schema(Arc::clone(&schema))
- .name("1")
+ .alias("1")
.build()?];
let input_batches = (0..4)
@@ -2427,7 +2427,7 @@ mod tests {
&[],
&[],
&df_schema,
- "COUNT(val)",
+ Some("COUNT(val)".to_string()),
false,
false,
false,
@@ -2515,7 +2515,7 @@ mod tests {
&[],
&[],
&df_schema,
- "COUNT(val)",
+ Some("COUNT(val)".to_string()),
false,
false,
false,
diff --git a/datafusion/physical-plan/src/windows/mod.rs
b/datafusion/physical-plan/src/windows/mod.rs
index 65cef28efc..b41f3ad71b 100644
--- a/datafusion/physical-plan/src/windows/mod.rs
+++ b/datafusion/physical-plan/src/windows/mod.rs
@@ -128,7 +128,7 @@ pub fn create_window_expr(
let aggregate = AggregateExprBuilder::new(Arc::clone(fun),
args.to_vec())
.schema(Arc::new(input_schema.clone()))
- .name(name)
+ .alias(name)
.order_by(order_by.to_vec())
.sort_exprs(sort_exprs)
.with_ignore_nulls(ignore_nulls)
diff --git a/datafusion/proto/src/physical_plan/mod.rs
b/datafusion/proto/src/physical_plan/mod.rs
index aefa1d87a2..59db791c75 100644
--- a/datafusion/proto/src/physical_plan/mod.rs
+++ b/datafusion/proto/src/physical_plan/mod.rs
@@ -492,7 +492,7 @@ impl AsExecutionPlan for protobuf::PhysicalPlanNode {
//
https://github.com/apache/datafusion/issues/11804
AggregateExprBuilder::new(agg_udf,
input_phy_expr)
.schema(Arc::clone(&physical_schema))
- .name(name)
+ .alias(name)
.with_ignore_nulls(agg_node.ignore_nulls)
.with_distinct(agg_node.distinct)
.build()
diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
index 0e2bc9cbb3..712182791b 100644
--- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
@@ -296,7 +296,7 @@ fn roundtrip_window() -> Result<()> {
vec![cast(col("b", &schema)?, &schema, DataType::Float64)?],
)
.schema(Arc::clone(&schema))
- .name("avg(b)")
+ .alias("avg(b)")
.build()?,
&[],
&[],
@@ -312,7 +312,7 @@ fn roundtrip_window() -> Result<()> {
let args = vec![cast(col("a", &schema)?, &schema, DataType::Float64)?];
let sum_expr = AggregateExprBuilder::new(sum_udaf(), args)
.schema(Arc::clone(&schema))
- .name("SUM(a) RANGE BETWEEN CURRENT ROW AND UNBOUNDED PRECEEDING")
+ .alias("SUM(a) RANGE BETWEEN CURRENT ROW AND UNBOUNDED PRECEEDING")
.build()?;
let sliding_aggr_window_expr = Arc::new(SlidingAggregateWindowExpr::new(
@@ -346,17 +346,17 @@ fn rountrip_aggregate() -> Result<()> {
let avg_expr = AggregateExprBuilder::new(avg_udaf(), vec![col("b",
&schema)?])
.schema(Arc::clone(&schema))
- .name("AVG(b)")
+ .alias("AVG(b)")
.build()?;
let nth_expr =
AggregateExprBuilder::new(nth_value_udaf(), vec![col("b", &schema)?,
lit(1u64)])
.schema(Arc::clone(&schema))
- .name("NTH_VALUE(b, 1)")
+ .alias("NTH_VALUE(b, 1)")
.build()?;
let str_agg_expr =
AggregateExprBuilder::new(string_agg_udaf(), vec![col("b", &schema)?,
lit(1u64)])
.schema(Arc::clone(&schema))
- .name("NTH_VALUE(b, 1)")
+ .alias("NTH_VALUE(b, 1)")
.build()?;
let test_cases: Vec<Vec<Arc<dyn AggregateExpr>>> = vec![
@@ -396,7 +396,7 @@ fn rountrip_aggregate_with_limit() -> Result<()> {
vec![
AggregateExprBuilder::new(avg_udaf(), vec![col("b", &schema)?])
.schema(Arc::clone(&schema))
- .name("AVG(b)")
+ .alias("AVG(b)")
.build()?,
];
@@ -463,7 +463,7 @@ fn roundtrip_aggregate_udaf() -> Result<()> {
vec![
AggregateExprBuilder::new(Arc::new(udaf), vec![col("b", &schema)?])
.schema(Arc::clone(&schema))
- .name("example_agg")
+ .alias("example_agg")
.build()?,
];
@@ -914,7 +914,7 @@ fn roundtrip_scalar_udf_extension_codec() -> Result<()> {
vec![udf_expr.clone() as Arc<dyn PhysicalExpr>],
)
.schema(schema.clone())
- .name("max")
+ .alias("max")
.build()?;
let window = Arc::new(WindowAggExec::try_new(
@@ -965,7 +965,7 @@ fn roundtrip_aggregate_udf_extension_codec() -> Result<()> {
let aggr_expr = AggregateExprBuilder::new(Arc::clone(&udaf),
aggr_args.clone())
.schema(Arc::clone(&schema))
- .name("aggregate_udf")
+ .alias("aggregate_udf")
.build()?;
let filter = Arc::new(FilterExec::try_new(
@@ -990,7 +990,7 @@ fn roundtrip_aggregate_udf_extension_codec() -> Result<()> {
let aggr_expr = AggregateExprBuilder::new(udaf, aggr_args.clone())
.schema(Arc::clone(&schema))
- .name("aggregate_udf")
+ .alias("aggregate_udf")
.distinct()
.ignore_nulls()
.build()?;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
