This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 5cf8eef5cf fix(aggregate): show aliased expr in explain (#21739)
5cf8eef5cf is described below
commit 5cf8eef5cfd080e718208f22498e2853adf14433
Author: Kumar Ujjawal <[email protected]>
AuthorDate: Sun May 17 12:56:25 2026 +0530
fix(aggregate): show aliased expr in explain (#21739)
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->
- Closes #19685.
## Rationale for this change
Physical explain output only showed the alias for aliased aggregates.
That made it hard to understand the plan, especially when the aggregate
had a filter, explicit RESPECT NULLS, or a custom UDAF display.
<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
-->
## What changes are included in this PR?
- Show the full aggregate expression in physical explain for
user-written aggregate aliases.
- Keep internal aliases like count(*) compact in physical explain.
- Replace the old hidden metadata approach with an explicit is_internal
flag on Alias.
- Preserve that flag through planner rewrites, tree rewrites, and proto
round-trip.
- Add tests for aliased aggregate explain output, including:
- normal aliased aggregates
- quoted aliases
- explicit RESPECT NULLS
- custom human display
- count(*)
- nested internal alias display
- Add an upgrade note for the public Alias API change.
<!--
There is no need to duplicate the description in the issue here but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->
## Are these changes tested?
Yes
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->
## Are there any user-facing changes?
Yes.
- Physical explain output is clearer for aliased aggregate expressions.
- Alias now has a new is_internal field.
This is a public API change for users who build or pattern match Alias
directly. The upgrade guide has been updated with the needed changes.
<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api
change` label.
-->
---
datafusion/core/src/physical_planner.rs | 365 +++++++++++----
datafusion/core/tests/dataframe/mod.rs | 32 +-
datafusion/expr/src/expr.rs | 52 +++
datafusion/expr/src/expr_rewriter/mod.rs | 12 +-
datafusion/expr/src/expr_schema.rs | 21 +
.../optimizer/src/decorrelate_lateral_join.rs | 7 +-
.../optimizer/src/optimize_projections/mod.rs | 9 +-
datafusion/physical-expr/src/aggregate.rs | 512 +++++++++++++++++++--
datafusion/physical-plan/src/aggregates/mod.rs | 169 ++++++-
.../physical-plan/src/aggregates/row_hash.rs | 8 +-
datafusion/proto/src/physical_plan/mod.rs | 83 +++-
datafusion/proto/src/physical_plan/to_proto.rs | 10 +-
.../proto/tests/cases/roundtrip_physical_plan.rs | 101 +++-
.../test_files/agg_func_substitute.slt | 4 +-
datafusion/sqllogictest/test_files/aggregate.slt | 16 +-
.../test_files/aggregates_simplify.slt | 8 +-
datafusion/sqllogictest/test_files/expr.slt | 2 +-
datafusion/sqllogictest/test_files/group_by.slt | 4 +-
datafusion/sqllogictest/test_files/subquery.slt | 4 +-
.../sqllogictest/test_files/tpch/plans/q1.slt.part | 4 +-
.../test_files/tpch/plans/q10.slt.part | 4 +-
.../test_files/tpch/plans/q12.slt.part | 4 +-
.../test_files/tpch/plans/q14.slt.part | 4 +-
.../test_files/tpch/plans/q15.slt.part | 8 +-
.../test_files/tpch/plans/q19.slt.part | 4 +-
.../sqllogictest/test_files/tpch/plans/q3.slt.part | 2 +-
.../sqllogictest/test_files/tpch/plans/q5.slt.part | 4 +-
.../sqllogictest/test_files/tpch/plans/q8.slt.part | 4 +-
docs/source/library-user-guide/upgrading/54.0.0.md | 67 +++
29 files changed, 1339 insertions(+), 185 deletions(-)
diff --git a/datafusion/core/src/physical_planner.rs
b/datafusion/core/src/physical_planner.rs
index 3b2c7a78e8..d225cff1de 100644
--- a/datafusion/core/src/physical_planner.rs
+++ b/datafusion/core/src/physical_planner.rs
@@ -81,8 +81,8 @@ use datafusion_datasource::memory::MemorySourceConfig;
use datafusion_expr::dml::{CopyTo, InsertOp};
use datafusion_expr::execution_props::{ScalarSubqueryResults, SubqueryIndex};
use datafusion_expr::expr::{
- AggregateFunction, AggregateFunctionParams, Alias, GroupingSet,
NullTreatment,
- WindowFunction, WindowFunctionParams, physical_name,
+ Alias, GroupingSet, NullTreatment, WindowFunction, WindowFunctionParams,
+ physical_name,
};
use datafusion_expr::expr_rewriter::unnormalize_cols;
use datafusion_expr::logical_plan::Subquery;
@@ -93,7 +93,9 @@ use datafusion_expr::{
FetchType, Filter, JoinType, Operator, RecursiveQuery, SkipType,
StringifiedPlan,
WindowFrame, WindowFrameBound, WriteOp,
};
-use datafusion_physical_expr::aggregate::{AggregateExprBuilder,
AggregateFunctionExpr};
+use datafusion_physical_expr::aggregate::{
+ AggregateFunctionExpr, LoweredAggregate, LoweredAggregateBuilder,
+};
use datafusion_physical_expr::expressions::Literal;
use datafusion_physical_expr::{
LexOrdering, PhysicalSortExpr, create_physical_sort_exprs,
@@ -1062,12 +1064,14 @@ impl DefaultPhysicalPlanner {
let agg_filter = aggr_expr
.iter()
.map(|e| {
- create_aggregate_expr_and_maybe_filter(
+ LoweredAggregateBuilder::new(
e,
logical_input_schema,
&physical_input_schema,
execution_props,
)
+ .build()
+ .map(lowered_aggregate_to_tuple)
})
.collect::<Result<Vec<_>>>()?;
@@ -2472,10 +2476,13 @@ pub fn create_window_expr(
) -> Result<Arc<dyn WindowExpr>> {
// unpack aliased logical expressions, e.g. "sum(col) over () as total"
let (name, e) = match e {
- Expr::Alias(Alias { expr, name, .. }) => (name.clone(), expr.as_ref()),
- _ => (e.schema_name().to_string(), e),
+ Expr::Alias(alias) => (
+ alias.name.clone(),
+ alias.expr.as_ref().clone().unalias_nested().data,
+ ),
+ _ => (e.schema_name().to_string(), e.clone()),
};
- create_window_expr_with_name(e, name, logical_schema, execution_props)
+ create_window_expr_with_name(&e, name, logical_schema, execution_props)
}
type AggregateExprWithOptionalArgs = (
@@ -2487,88 +2494,46 @@ type AggregateExprWithOptionalArgs = (
);
/// Create an aggregate expression with a name from a logical expression
+#[deprecated(note = "use LoweredAggregateBuilder")]
pub fn create_aggregate_expr_with_name_and_maybe_filter(
e: &Expr,
name: Option<String>,
- human_displan: String,
+ human_display: String,
logical_input_schema: &DFSchema,
physical_input_schema: &Schema,
execution_props: &ExecutionProps,
) -> Result<AggregateExprWithOptionalArgs> {
- match e {
- Expr::AggregateFunction(AggregateFunction {
- func,
- params:
- AggregateFunctionParams {
- args,
- distinct,
- filter,
- order_by,
- null_treatment,
- },
- }) => {
- let name = if let Some(name) = name {
- name
- } else {
- physical_name(e)?
- };
-
- let physical_args =
- create_physical_exprs(args, logical_input_schema,
execution_props)?;
- let filter = match filter {
- Some(e) => Some(create_physical_expr(
- e,
- logical_input_schema,
- execution_props,
- )?),
- None => None,
- };
-
- let ignore_nulls =
null_treatment.unwrap_or(NullTreatment::RespectNulls)
- == NullTreatment::IgnoreNulls;
-
- let (agg_expr, filter, order_bys) = {
- let order_bys = create_physical_sort_exprs(
- order_by,
- logical_input_schema,
- execution_props,
- )?;
-
- let agg_expr =
- AggregateExprBuilder::new(func.to_owned(),
physical_args.to_vec())
- .order_by(order_bys.clone())
- .schema(Arc::new(physical_input_schema.to_owned()))
- .alias(name)
- .human_display(human_displan)
- .with_ignore_nulls(ignore_nulls)
- .with_distinct(*distinct)
- .build()
- .map(Arc::new)?;
-
- (agg_expr, filter, order_bys)
- };
+ let mut builder = LoweredAggregateBuilder::new(
+ e,
+ logical_input_schema,
+ physical_input_schema,
+ execution_props,
+ )
+ .with_human_display(human_display);
- Ok((agg_expr, filter, order_bys))
- }
- other => internal_err!("Invalid aggregate expression '{other:?}'"),
+ if let Some(name) = name {
+ builder = builder.with_name(name);
}
+
+ builder.build().map(lowered_aggregate_to_tuple)
}
/// Create an aggregate expression from a logical expression or an alias
+#[deprecated(note = "use LoweredAggregateBuilder")]
pub fn create_aggregate_expr_and_maybe_filter(
e: &Expr,
logical_input_schema: &DFSchema,
physical_input_schema: &Schema,
execution_props: &ExecutionProps,
) -> Result<AggregateExprWithOptionalArgs> {
- // Unpack (potentially nested) aliased logical expressions, e.g. "sum(col)
as total"
- // Some functions like `count_all()` create internal aliases,
- // Unwrap all alias layers to get to the underlying aggregate function
+ // Preserve the pre-builder behavior for callers that still use this
helper:
+ // use a single display string and do not attach a separate display alias.
let (name, human_display, e) = match e {
- Expr::Alias(Alias { name, .. }) => {
- let unaliased = e.clone().unalias_nested().data;
- (Some(name.clone()), e.human_display().to_string(), unaliased)
- }
+ Expr::Alias(alias) => (
+ Some(alias.name.clone()),
+ e.human_display().to_string(),
+ e.clone(),
+ ),
Expr::AggregateFunction(_) => (
Some(e.schema_name().to_string()),
e.human_display().to_string(),
@@ -2577,14 +2542,25 @@ pub fn create_aggregate_expr_and_maybe_filter(
_ => (None, String::default(), e.clone()),
};
- create_aggregate_expr_with_name_and_maybe_filter(
+ let mut builder = LoweredAggregateBuilder::new(
&e,
- name,
- human_display,
logical_input_schema,
physical_input_schema,
execution_props,
)
+ .with_human_display(human_display);
+
+ if let Some(name) = name {
+ builder = builder.with_name(name);
+ }
+
+ builder.build().map(lowered_aggregate_to_tuple)
+}
+
+fn lowered_aggregate_to_tuple(
+ lowered: LoweredAggregate,
+) -> AggregateExprWithOptionalArgs {
+ (lowered.aggregate, lowered.filter, lowered.order_bys)
}
impl DefaultPhysicalPlanner {
@@ -3223,6 +3199,7 @@ impl<'n> TreeNodeVisitor<'n> for InvariantChecker {
mod tests {
use std::cmp::Ordering;
use std::fmt::{self, Debug};
+ use std::mem::size_of_val;
use std::ops::{BitAnd, Not};
use super::*;
@@ -3238,18 +3215,23 @@ mod tests {
use crate::execution::session_state::SessionStateBuilder;
use arrow::array::{ArrayRef, DictionaryArray, Int32Array};
use arrow::datatypes::{DataType, Field, Int32Type};
- use arrow_schema::SchemaRef;
+ use arrow_schema::{FieldRef, SchemaRef};
use datafusion_common::config::ConfigOptions;
use datafusion_common::{
- DFSchemaRef, TableReference, ToDFSchema as _, assert_batches_eq,
assert_contains,
+ DFSchemaRef, ScalarValue, TableReference, ToDFSchema as _,
assert_batches_eq,
+ assert_contains,
};
use datafusion_execution::TaskContext;
use datafusion_execution::runtime_env::RuntimeEnv;
use datafusion_expr::builder::subquery_alias;
+ use datafusion_expr::expr::AggregateFunctionParams;
+ use datafusion_expr::function::{AccumulatorArgs, StateFieldsArgs};
use datafusion_expr::{
- LogicalPlanBuilder, TableSource, UserDefinedLogicalNodeCore, col, lit,
+ Accumulator, AggregateUDF, AggregateUDFImpl, ExprFunctionExt,
LogicalPlanBuilder,
+ Signature, TableSource, UserDefinedLogicalNodeCore, Volatility,
+ WindowFunctionDefinition, col, lit,
};
- use datafusion_functions_aggregate::count::count_all;
+ use datafusion_functions_aggregate::count::{count_all, count_udaf};
use datafusion_functions_aggregate::expr_fn::sum;
use datafusion_physical_expr::EquivalenceProperties;
use datafusion_physical_plan::execution_plan::{Boundedness, EmissionType};
@@ -3275,6 +3257,127 @@ mod tests {
.await
}
+ async fn aggregate_explain(logical_plan: &LogicalPlan) -> Result<String> {
+ let physical_plan = plan(logical_plan).await?;
+ Ok(displayable(physical_plan.as_ref()).indent(true).to_string())
+ }
+
+ async fn aggregate_explain_for(
+ schema: Schema,
+ aggr_expr: Vec<Expr>,
+ ) -> Result<String> {
+ let logical_plan = scan_empty(None, &schema, None)?
+ .aggregate(Vec::<Expr>::new(), aggr_expr)?
+ .build()?;
+
+ aggregate_explain(&logical_plan).await
+ }
+
+ fn int64_field(name: &str, nullable: bool) -> Field {
+ Field::new(name, DataType::Int64, nullable)
+ }
+
+ #[test]
+ fn test_create_window_expr_unwraps_alias_with_metadata() -> Result<()> {
+ use std::collections::HashMap;
+
+ use datafusion_common::metadata::FieldMetadata;
+
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "column1",
+ DataType::Int64,
+ true,
+ )]));
+ let logical_schema = schema.as_ref().clone().to_dfschema_ref()?;
+ let metadata = FieldMetadata::from(HashMap::from([(
+ "some_key".to_string(),
+ "some_value".to_string(),
+ )]));
+ let expr = Expr::from(WindowFunction::new(
+ WindowFunctionDefinition::AggregateUDF(count_udaf()),
+ vec![col("column1")],
+ ))
+ .alias_with_metadata("window_alias", Some(metadata));
+
+ let window_expr =
+ create_window_expr(&expr, &logical_schema,
&ExecutionProps::new())?;
+
+ assert_eq!(window_expr.name(), "window_alias");
+ Ok(())
+ }
+
+ #[derive(Debug, Default)]
+ struct NullAccumulator;
+
+ impl Accumulator for NullAccumulator {
+ fn state(&mut self) -> Result<Vec<ScalarValue>> {
+ Ok(vec![self.evaluate()?])
+ }
+
+ fn update_batch(&mut self, _values: &[ArrayRef]) -> Result<()> {
+ Ok(())
+ }
+
+ fn merge_batch(&mut self, _states: &[ArrayRef]) -> Result<()> {
+ Ok(())
+ }
+
+ fn evaluate(&mut self) -> Result<ScalarValue> {
+ Ok(ScalarValue::Int64(None))
+ }
+
+ fn size(&self) -> usize {
+ size_of_val(self)
+ }
+ }
+
+ #[derive(Debug, Clone, PartialEq, Eq, Hash)]
+ struct CustomHumanDisplayUdaf {
+ signature: Signature,
+ }
+
+ impl CustomHumanDisplayUdaf {
+ fn new() -> Self {
+ Self {
+ signature: Signature::exact(vec![DataType::Int64],
Volatility::Immutable),
+ }
+ }
+ }
+
+ impl AggregateUDFImpl for CustomHumanDisplayUdaf {
+ fn name(&self) -> &str {
+ "custom_human_display_udaf"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ Ok(DataType::Int64)
+ }
+
+ fn accumulator(
+ &self,
+ _acc_args: AccumulatorArgs,
+ ) -> Result<Box<dyn Accumulator>> {
+ Ok(Box::new(NullAccumulator))
+ }
+
+ fn state_fields(&self, _args: StateFieldsArgs) ->
Result<Vec<FieldRef>> {
+ Ok(vec![
+ Field::new("custom_state", DataType::Int64, true).into(),
+ ])
+ }
+
+ fn human_display(&self, params: &AggregateFunctionParams) ->
Result<String> {
+ Ok(format!(
+ "custom_display({})",
+ params.args[0].human_display()
+ ))
+ }
+ }
+
async fn plan_sql(query: &str) -> Result<Arc<dyn ExecutionPlan>> {
let ctx = SessionContext::new();
ctx.sql(query).await?.create_physical_plan().await
@@ -4022,6 +4125,114 @@ mod tests {
Ok(())
}
+ #[tokio::test]
+ async fn test_aggregate_explain_shows_quoted_user_alias() -> Result<()> {
+ assert_contains!(
+ aggregate_explain_for(
+ Schema::new(vec![int64_field("column1", false)]),
+ vec![sum(col("column1")).alias("total rows")],
+ )
+ .await?,
+ "AggregateExec: mode=Single, gby=[], aggr=[sum(?table?.column1) as
total rows]"
+ );
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn test_aggregate_explain_shows_aliased_filter_expression() ->
Result<()> {
+ let expr = sum(col("column1"))
+ .filter(col("column2").lt_eq(lit(0_i64)))
+ .build()?
+ .alias("agg");
+
+ assert_contains!(
+ aggregate_explain_for(
+ Schema::new(vec![
+ int64_field("column1", false),
+ int64_field("column2", false),
+ ]),
+ vec![expr],
+ )
+ .await?,
+ "AggregateExec: mode=Single, gby=[], aggr=[sum(?table?.column1)
FILTER (WHERE ?table?.column2 <= Int64(0)) as agg]"
+ );
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn test_aggregate_explain_shows_aliased_respect_nulls() ->
Result<()> {
+ let expr =
datafusion_functions_aggregate::first_last::first_value_udaf()
+ .call(vec![col("column1")])
+ .order_by(vec![col("column2").sort(true, true)])
+ .null_treatment(NullTreatment::RespectNulls)
+ .build()?
+ .alias("agg");
+
+ assert_contains!(
+ aggregate_explain_for(
+ Schema::new(vec![
+ int64_field("column1", true),
+ int64_field("column2", false),
+ ]),
+ vec![expr],
+ )
+ .await?,
+ "AggregateExec: mode=Single, gby=[],
aggr=[first_value(?table?.column1) RESPECT NULLS ORDER BY [?table?.column2 ASC
NULLS FIRST] as agg]"
+ );
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn test_aggregate_explain_shows_count_all() -> Result<()> {
+ let logical_plan = test_csv_scan()
+ .await?
+ .aggregate(Vec::<Expr>::new(), vec![count_all()])?
+ .build()?;
+
+ assert_contains!(
+ aggregate_explain(&logical_plan).await?,
+ "aggr=[count(1) as count(*)]"
+ );
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn test_aggregate_explain_shows_count_all_with_user_alias() ->
Result<()> {
+ let logical_plan = test_csv_scan()
+ .await?
+ .aggregate(Vec::<Expr>::new(),
vec![count_all().alias("total_rows")])?
+ .build()?;
+
+ assert_contains!(
+ aggregate_explain(&logical_plan).await?,
+ "aggr=[count(1) as total_rows]"
+ );
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn test_aggregate_explain_shows_aliased_custom_human_display() ->
Result<()> {
+ assert_contains!(
+ aggregate_explain_for(
+ Schema::new(vec![int64_field("column1", false)]),
+ vec![
+ AggregateUDF::from(CustomHumanDisplayUdaf::new())
+ .call(vec![col("column1")])
+ .alias("agg"),
+ ],
+ )
+ .await?,
+ "AggregateExec: mode=Single, gby=[],
aggr=[custom_display(?table?.column1) as agg]"
+ );
+
+ Ok(())
+ }
+
#[tokio::test]
async fn test_explain() {
let schema = Schema::new(vec![Field::new("id", DataType::Int32,
false)]);
diff --git a/datafusion/core/tests/dataframe/mod.rs
b/datafusion/core/tests/dataframe/mod.rs
index 505ccb3a1b..d1dc964da8 100644
--- a/datafusion/core/tests/dataframe/mod.rs
+++ b/datafusion/core/tests/dataframe/mod.rs
@@ -3022,20 +3022,20 @@ async fn test_count_wildcard_on_sort() -> Result<()> {
assert_snapshot!(
pretty_format_batches(&df_results).unwrap(),
@r"
-
+---------------+----------------------------------------------------------------------------+
- | plan_type | plan
|
-
+---------------+----------------------------------------------------------------------------+
- | logical_plan | Sort: count(*) AS count(*) ASC NULLS LAST
|
- | | Aggregate: groupBy=[[t1.b]], aggr=[[count(Int64(1)) AS
count(*)]] |
- | | TableScan: t1 projection=[b]
|
- | physical_plan | SortPreservingMergeExec: [count(*)@1 ASC NULLS LAST]
|
- | | SortExec: expr=[count(*)@1 ASC NULLS LAST],
preserve_partitioning=[true] |
- | | AggregateExec: mode=FinalPartitioned, gby=[b@0 as
b], aggr=[count(*)] |
- | | RepartitionExec: partitioning=Hash([b@0], 4),
input_partitions=1 |
- | | AggregateExec: mode=Partial, gby=[b@0 as b],
aggr=[count(*)] |
- | | DataSourceExec: partitions=1,
partition_sizes=[1] |
- | |
|
-
+---------------+----------------------------------------------------------------------------+
+
+---------------+---------------------------------------------------------------------------------------+
+ | plan_type | plan
|
+
+---------------+---------------------------------------------------------------------------------------+
+ | logical_plan | Sort: count(*) AS count(*) ASC NULLS LAST
|
+ | | Aggregate: groupBy=[[t1.b]], aggr=[[count(Int64(1)) AS
count(*)]] |
+ | | TableScan: t1 projection=[b]
|
+ | physical_plan | SortPreservingMergeExec: [count(*)@1 ASC NULLS LAST]
|
+ | | SortExec: expr=[count(*)@1 ASC NULLS LAST],
preserve_partitioning=[true] |
+ | | AggregateExec: mode=FinalPartitioned, gby=[b@0 as
b], aggr=[count(1) as count(*)] |
+ | | RepartitionExec: partitioning=Hash([b@0], 4),
input_partitions=1 |
+ | | AggregateExec: mode=Partial, gby=[b@0 as b],
aggr=[count(1) as count(*)] |
+ | | DataSourceExec: partitions=1,
partition_sizes=[1] |
+ | |
|
+
+---------------+---------------------------------------------------------------------------------------+
"
);
Ok(())
@@ -3500,9 +3500,9 @@ async fn test_count_wildcard_on_where_scalar_subquery()
-> Result<()> {
| | HashJoinExec: mode=CollectLeft, join_type=Right,
on=[(a@1, a@0)], projection=[a@3, b@4, count(*)@0, __always_true@2] |
| | CoalescePartitionsExec
|
| | ProjectionExec: expr=[count(*)@1 as count(*),
a@0 as a, true as __always_true] |
- | | AggregateExec: mode=FinalPartitioned, gby=[a@0
as a], aggr=[count(*)] |
+ | | AggregateExec: mode=FinalPartitioned, gby=[a@0
as a], aggr=[count(1) as count(*)] |
| | RepartitionExec: partitioning=Hash([a@0],
4), input_partitions=1 |
- | | AggregateExec: mode=Partial, gby=[a@0 as
a], aggr=[count(*)] |
+ | | AggregateExec: mode=Partial, gby=[a@0 as
a], aggr=[count(1) as count(*)] |
| | DataSourceExec: partitions=1,
partition_sizes=[1] |
| | DataSourceExec: partitions=1, partition_sizes=[1]
|
| |
|
diff --git a/datafusion/expr/src/expr.rs b/datafusion/expr/src/expr.rs
index 6f420a7153..d6276b944c 100644
--- a/datafusion/expr/src/expr.rs
+++ b/datafusion/expr/src/expr.rs
@@ -743,6 +743,28 @@ impl Alias {
self.metadata = metadata;
self
}
+
+ #[doc(hidden)]
+ pub fn with_expr(mut self, expr: Expr) -> Self {
+ self.expr = Box::new(expr);
+ self
+ }
+
+ #[doc(hidden)]
+ pub fn try_map_expr(self, f: impl FnOnce(Expr) -> Result<Expr>) ->
Result<Expr> {
+ let Alias {
+ expr,
+ relation,
+ name,
+ metadata,
+ } = self;
+ Ok(Expr::Alias(Alias {
+ expr: Box::new(f(*expr)?),
+ relation,
+ name,
+ metadata,
+ }))
+ }
}
/// Binary expression for [`Expr::BinaryExpr`]
@@ -4155,6 +4177,36 @@ mod test {
);
}
+ #[test]
+ fn test_unalias_nested_respects_user_metadata() {
+ use std::collections::HashMap;
+
+ let base_expr = col("id");
+
+ let no_metadata = base_expr.clone().alias("alias");
+ assert_eq!(no_metadata.unalias_nested().data, base_expr);
+
+ let Expr::Alias(empty_metadata_alias) =
base_expr.clone().alias("alias") else {
+ unreachable!();
+ };
+ let empty_metadata_alias = Expr::Alias(
+ empty_metadata_alias.with_metadata(Some(FieldMetadata::default())),
+ );
+ assert_eq!(empty_metadata_alias.unalias_nested().data, base_expr);
+
+ let user_metadata = FieldMetadata::from(HashMap::from([(
+ "some_key".to_string(),
+ "some_value".to_string(),
+ )]));
+
+ let Expr::Alias(user_alias) = base_expr.clone().alias("alias") else {
+ unreachable!();
+ };
+ let user_alias =
+ Expr::Alias(user_alias.with_metadata(Some(user_metadata.clone())));
+ assert_eq!(user_alias.clone().unalias_nested().data, user_alias);
+ }
+
fn wildcard_options(
opt_ilike: Option<IlikeSelectItem>,
opt_exclude: Option<ExcludeSelectItem>,
diff --git a/datafusion/expr/src/expr_rewriter/mod.rs
b/datafusion/expr/src/expr_rewriter/mod.rs
index 32a88ab8cf..eab8114d69 100644
--- a/datafusion/expr/src/expr_rewriter/mod.rs
+++ b/datafusion/expr/src/expr_rewriter/mod.rs
@@ -340,8 +340,16 @@ impl NamePreserver {
pub fn save(&self, expr: &Expr) -> SavedName {
if self.use_alias {
- let (relation, name) = expr.qualified_name();
- SavedName::Saved { relation, name }
+ match expr {
+ Expr::Alias(alias) => SavedName::Saved {
+ relation: alias.relation.clone(),
+ name: alias.name.clone(),
+ },
+ _ => {
+ let (relation, name) = expr.qualified_name();
+ SavedName::Saved { relation, name }
+ }
+ }
} else {
SavedName::None
}
diff --git a/datafusion/expr/src/expr_schema.rs
b/datafusion/expr/src/expr_schema.rs
index c989bab304..039bbad65a 100644
--- a/datafusion/expr/src/expr_schema.rs
+++ b/datafusion/expr/src/expr_schema.rs
@@ -1073,6 +1073,27 @@ mod tests {
assert_eq!(meta, outer_ref.metadata(&schema).unwrap());
}
+ #[test]
+ fn test_alias_metadata_is_preserved_in_field_metadata() {
+ let schema = MockExprSchema::new().with_data_type(DataType::Int32);
+ let alias_metadata = FieldMetadata::from(HashMap::from([(
+ "some_key".to_string(),
+ "some_value".to_string(),
+ )]));
+
+ let Expr::Alias(alias) = col("foo").alias("alias") else {
+ unreachable!();
+ };
+ let expr =
Expr::Alias(alias.with_metadata(Some(alias_metadata.clone())));
+
+ let field = expr.to_field(&schema).unwrap().1;
+ assert_eq!(
+ field.metadata().get("some_key"),
+ Some(&"some_value".to_string())
+ );
+ assert_eq!(expr.metadata(&schema).unwrap(), alias_metadata);
+ }
+
#[test]
fn test_expr_placeholder() {
let schema = MockExprSchema::new();
diff --git a/datafusion/optimizer/src/decorrelate_lateral_join.rs
b/datafusion/optimizer/src/decorrelate_lateral_join.rs
index ea25ab479f..a8df5e69e3 100644
--- a/datafusion/optimizer/src/decorrelate_lateral_join.rs
+++ b/datafusion/optimizer/src/decorrelate_lateral_join.rs
@@ -260,7 +260,12 @@ fn rewrite_internal(join: Join) ->
Result<Transformed<LogicalPlan>> {
)],
else_expr: Some(Box::new(col)),
});
- proj_exprs.push(case_expr.alias_qualified(qualifier.cloned(),
name));
+ proj_exprs.push(Expr::Alias(expr::Alias {
+ expr: Box::new(case_expr),
+ relation: qualifier.cloned(),
+ name: name.to_string(),
+ metadata: None,
+ }));
continue;
}
proj_exprs.push(col);
diff --git a/datafusion/optimizer/src/optimize_projections/mod.rs
b/datafusion/optimizer/src/optimize_projections/mod.rs
index af944abc6f..bc923706a4 100644
--- a/datafusion/optimizer/src/optimize_projections/mod.rs
+++ b/datafusion/optimizer/src/optimize_projections/mod.rs
@@ -605,9 +605,12 @@ fn merge_consecutive_projections(proj: Projection) ->
Result<Transformed<Project
if metadata.is_none() && expr.schema_name().to_string() ==
name {
expr
} else {
- Expr::Alias(
- Alias::new(expr, relation,
name).with_metadata(metadata),
- )
+ Expr::Alias(Alias {
+ expr: Box::new(expr),
+ relation,
+ name,
+ metadata,
+ })
}
})
}),
diff --git a/datafusion/physical-expr/src/aggregate.rs
b/datafusion/physical-expr/src/aggregate.rs
index 3fd2b42b2e..e5d55aba4f 100644
--- a/datafusion/physical-expr/src/aggregate.rs
+++ b/datafusion/physical-expr/src/aggregate.rs
@@ -38,13 +38,20 @@ use std::fmt::Debug;
use std::sync::Arc;
use crate::expressions::Column;
+use crate::physical_expr::create_physical_sort_exprs;
+use crate::planner::{create_physical_expr, create_physical_exprs};
use arrow::compute::SortOptions;
use arrow::datatypes::{DataType, FieldRef, Schema, SchemaRef};
+use datafusion_common::metadata::FieldMetadata;
use datafusion_common::{
- Result, ScalarValue, assert_or_internal_err, internal_err, not_impl_err,
+ DFSchema, Result, ScalarValue, assert_or_internal_err, internal_err,
not_impl_err,
};
-use datafusion_expr::{AggregateUDF, ReversedUDAF, SetMonotonicity};
+use datafusion_expr::execution_props::ExecutionProps;
+use datafusion_expr::expr::{
+ AggregateFunction, AggregateFunctionParams, NullTreatment, physical_name,
+};
+use datafusion_expr::{AggregateUDF, Expr, ReversedUDAF, SetMonotonicity};
use datafusion_expr_common::accumulator::Accumulator;
use datafusion_expr_common::groups_accumulator::GroupsAccumulator;
use datafusion_expr_common::type_coercion::aggregates::check_arg_count;
@@ -55,6 +62,57 @@ use
datafusion_functions_aggregate_common::order::AggregateOrderSensitivity;
use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
+#[derive(Debug, Clone)]
+struct AggregateHumanDisplay {
+ expression: String,
+ alias: Option<String>,
+}
+
+impl AggregateHumanDisplay {
+ fn try_new(
+ expression: Option<String>,
+ alias: Option<String>,
+ name: &str,
+ ) -> Result<Option<Self>> {
+ let alias = alias.filter(|alias| !alias.is_empty());
+ let Some(expression) = expression else {
+ if alias.is_some() {
+ return internal_err!(
+ "AggregateExprBuilder::human_display must be provided when
human_display_alias is set"
+ );
+ }
+ return Ok(None);
+ };
+
+ if expression.is_empty() {
+ if alias.is_some() {
+ return internal_err!(
+ "AggregateExprBuilder::human_display must be non-empty
when human_display_alias is set"
+ );
+ }
+ return Ok(None);
+ }
+
+ if let Some(alias) = alias.as_deref()
+ && alias != name
+ {
+ return internal_err!(
+ "aggregate human_display_alias must match aggregate name
`{name}`: {alias}"
+ );
+ }
+
+ Ok(Some(Self { expression, alias }))
+ }
+
+ fn expression(&self) -> &str {
+ &self.expression
+ }
+
+ fn alias(&self) -> Option<&str> {
+ self.alias.as_deref()
+ }
+}
+
/// Builder for physical [`AggregateFunctionExpr`]
///
/// `AggregateFunctionExpr` contains the information necessary to call
@@ -65,8 +123,11 @@ pub struct AggregateExprBuilder {
/// Physical expressions of the aggregate function
args: Vec<Arc<dyn PhysicalExpr>>,
alias: Option<String>,
+ output_metadata: Option<FieldMetadata>,
/// A human readable name
- human_display: String,
+ human_display: Option<String>,
+ /// Optional visible output alias for `human_display`.
+ human_display_alias: Option<String>,
/// Arrow Schema for the aggregate function
schema: SchemaRef,
/// The physical order by expressions
@@ -85,7 +146,9 @@ impl AggregateExprBuilder {
fun,
args,
alias: None,
- human_display: String::default(),
+ output_metadata: None,
+ human_display: None,
+ human_display_alias: None,
schema: Arc::new(Schema::empty()),
order_bys: vec![],
ignore_nulls: false,
@@ -189,7 +252,9 @@ impl AggregateExprBuilder {
fun,
args,
alias,
+ output_metadata,
human_display,
+ human_display_alias,
schema,
order_bys,
ignore_nulls,
@@ -216,7 +281,10 @@ impl AggregateExprBuilder {
&fun.signature().type_signature,
)?;
- let return_field = fun.return_field(&input_exprs_fields)?;
+ let mut return_field = fun.return_field(&input_exprs_fields)?;
+ if let Some(output_metadata) = output_metadata {
+ return_field = output_metadata.add_to_field_ref(return_field);
+ }
let is_nullable = fun.is_nullable();
let name = match alias {
None => {
@@ -227,6 +295,9 @@ impl AggregateExprBuilder {
Some(alias) => alias,
};
+ let human_display =
+ AggregateHumanDisplay::try_new(human_display, human_display_alias,
&name)?;
+
let arg_fields = args
.iter()
.map(|e| e.return_field(schema.as_ref()))
@@ -255,8 +326,24 @@ impl AggregateExprBuilder {
self
}
- pub fn human_display(mut self, name: String) -> Self {
- self.human_display = name;
+ fn output_metadata(mut self, metadata: Option<FieldMetadata>) -> Self {
+ self.output_metadata = metadata;
+ self
+ }
+
+ pub fn human_display(mut self, name: impl Into<String>) -> Self {
+ let name = name.into();
+ self.human_display = (!name.is_empty()).then_some(name);
+ if self.human_display.is_none() {
+ self.human_display_alias = None;
+ }
+ self
+ }
+
+ #[doc(hidden)]
+ pub fn human_display_alias(mut self, alias: impl Into<String>) -> Self {
+ let alias = alias.into();
+ self.human_display_alias = (!alias.is_empty()).then_some(alias);
self
}
@@ -301,6 +388,234 @@ impl AggregateExprBuilder {
}
}
+#[derive(Debug, Clone)]
+struct LoweredAggregateHumanDisplay {
+ expression: String,
+ alias: Option<String>,
+}
+
+/// Result of lowering a logical aggregate expression into physical aggregate
+/// planning pieces.
+#[derive(Debug, Clone)]
+pub struct LoweredAggregate {
+ /// Physical aggregate expression that can be used by an aggregate
execution
+ /// plan.
+ pub aggregate: Arc<AggregateFunctionExpr>,
+ /// Optional physical filter expression for `FILTER (WHERE ...)`.
+ pub filter: Option<Arc<dyn PhysicalExpr>>,
+ /// Physical ordering expressions from aggregate `ORDER BY`.
+ pub order_bys: Vec<PhysicalSortExpr>,
+}
+
+/// Builder for converting a logical aggregate [`Expr`] into physical aggregate
+/// planning pieces.
+///
+/// This builder handles the logical-to-physical work needed for aggregate
+/// planning: unwrapping aggregate aliases, choosing the output name,
preserving
+/// user-facing display text, lowering aggregate arguments, lowering the
optional
+/// filter, and lowering aggregate `ORDER BY` expressions.
+pub struct LoweredAggregateBuilder<'a> {
+ expr: &'a Expr,
+ name: Option<String>,
+ human_display: Option<LoweredAggregateHumanDisplay>,
+ output_metadata: Option<FieldMetadata>,
+ preserve_alias_metadata: bool,
+ logical_input_schema: &'a DFSchema,
+ physical_input_schema: &'a Schema,
+ execution_props: &'a ExecutionProps,
+}
+
+impl<'a> LoweredAggregateBuilder<'a> {
+ /// Create a builder for lowering `expr`.
+ ///
+ /// `logical_input_schema` is used to resolve logical expressions such as
+ /// columns, while `physical_input_schema` is the input schema used by the
+ /// physical aggregate expression.
+ pub fn new(
+ expr: &'a Expr,
+ logical_input_schema: &'a DFSchema,
+ physical_input_schema: &'a Schema,
+ execution_props: &'a ExecutionProps,
+ ) -> Self {
+ Self {
+ expr,
+ name: None,
+ human_display: None,
+ output_metadata: None,
+ preserve_alias_metadata: true,
+ logical_input_schema,
+ physical_input_schema,
+ execution_props,
+ }
+ }
+
+ /// Override the output column name for the aggregate.
+ ///
+ /// If this is not set, the builder uses the alias from `expr` when
present,
+ /// or derives the physical name from the aggregate expression.
+ pub fn with_name(mut self, name: impl Into<String>) -> Self {
+ self.name = Some(name.into());
+ self
+ }
+
+ /// Override the human-readable display text for the aggregate.
+ ///
+ /// This is useful when a caller has already computed the exact display
text
+ /// it wants to preserve. When this override is used, aliases with metadata
+ /// are still unwrapped for planning, but alias metadata is not copied to
the
+ /// aggregate output field.
+ pub fn with_human_display(mut self, human_display: impl Into<String>) ->
Self {
+ self.human_display = Some(LoweredAggregateHumanDisplay {
+ expression: human_display.into(),
+ alias: None,
+ });
+ self.preserve_alias_metadata = false;
+ self
+ }
+
+ /// Lower the logical aggregate expression into physical aggregate pieces.
+ pub fn build(self) -> Result<LoweredAggregate> {
+ let Self {
+ expr,
+ name,
+ human_display,
+ output_metadata,
+ preserve_alias_metadata,
+ logical_input_schema,
+ physical_input_schema,
+ execution_props,
+ } = self;
+
+ let (name, human_display, output_metadata, expr) =
lower_aggregate_display(
+ expr,
+ name,
+ human_display,
+ output_metadata,
+ preserve_alias_metadata,
+ );
+
+ let Expr::AggregateFunction(AggregateFunction {
+ func,
+ params:
+ AggregateFunctionParams {
+ args,
+ distinct,
+ filter,
+ order_by,
+ null_treatment,
+ },
+ }) = &expr
+ else {
+ return internal_err!("Invalid aggregate expression '{expr:?}'");
+ };
+
+ let name = if let Some(name) = name {
+ name
+ } else {
+ physical_name(&expr)?
+ };
+
+ let physical_args =
+ create_physical_exprs(args, logical_input_schema,
execution_props)?;
+ let filter = filter
+ .as_ref()
+ .map(|filter| {
+ create_physical_expr(filter, logical_input_schema,
execution_props)
+ })
+ .transpose()?;
+ let order_bys =
+ create_physical_sort_exprs(order_by, logical_input_schema,
execution_props)?;
+ let ignore_nulls =
null_treatment.unwrap_or(NullTreatment::RespectNulls)
+ == NullTreatment::IgnoreNulls;
+
+ let mut builder = AggregateExprBuilder::new(func.to_owned(),
physical_args)
+ .order_by(order_bys.clone())
+ .schema(Arc::new(physical_input_schema.to_owned()))
+ .alias(name)
+ .output_metadata(output_metadata)
+ .with_ignore_nulls(ignore_nulls)
+ .with_distinct(*distinct);
+
+ if let Some(human_display) = human_display {
+ builder = builder.human_display(human_display.expression);
+ if let Some(alias) = human_display.alias {
+ builder = builder.human_display_alias(alias);
+ }
+ }
+
+ Ok(LoweredAggregate {
+ aggregate: Arc::new(builder.build()?),
+ filter,
+ order_bys,
+ })
+ }
+}
+
+fn lower_aggregate_display(
+ expr: &Expr,
+ name: Option<String>,
+ human_display: Option<LoweredAggregateHumanDisplay>,
+ output_metadata: Option<FieldMetadata>,
+ preserve_alias_metadata: bool,
+) -> (
+ Option<String>,
+ Option<LoweredAggregateHumanDisplay>,
+ Option<FieldMetadata>,
+ Expr,
+) {
+ let mut expr = expr.clone();
+ let mut alias_name = None;
+ let mut alias_metadata = None;
+ while let Expr::Alias(alias) = expr {
+ if alias_name.is_none() {
+ alias_name = Some(alias.name);
+ alias_metadata = alias.metadata;
+ }
+ expr = *alias.expr;
+ }
+
+ let output_metadata = if preserve_alias_metadata {
+ output_metadata.or(alias_metadata)
+ } else {
+ output_metadata
+ };
+
+ if human_display.is_some() {
+ return (name.or(alias_name), human_display, output_metadata, expr);
+ }
+
+ match &expr {
+ Expr::AggregateFunction(_) => {
+ if let Some(alias_name) = alias_name {
+ let name = name.unwrap_or(alias_name);
+ let expression = expr.human_display().to_string();
+ let human_display = if expression.is_empty() || expression ==
name {
+ LoweredAggregateHumanDisplay {
+ expression: name.clone(),
+ alias: None,
+ }
+ } else {
+ LoweredAggregateHumanDisplay {
+ expression,
+ alias: Some(name.clone()),
+ }
+ };
+
+ return (Some(name), Some(human_display), output_metadata,
expr);
+ }
+
+ let name = name.unwrap_or_else(|| expr.schema_name().to_string());
+ let human_display = LoweredAggregateHumanDisplay {
+ expression: expr.human_display().to_string(),
+ alias: None,
+ };
+
+ (Some(name), Some(human_display), output_metadata, expr)
+ }
+ _ => (name.or(alias_name), None, output_metadata, expr),
+ }
+}
+
/// Physical aggregate expression of a UDAF.
///
/// Instances are constructed via [`AggregateExprBuilder`].
@@ -315,7 +630,7 @@ pub struct AggregateFunctionExpr {
/// Output column name that this expression creates
name: String,
/// Simplified name for `tree` explain.
- human_display: String,
+ human_display: Option<AggregateHumanDisplay>,
schema: Schema,
// The physical order by expressions
order_bys: Vec<PhysicalSortExpr>,
@@ -347,8 +662,22 @@ impl AggregateFunctionExpr {
}
/// Simplified name for `tree` explain.
- pub fn human_display(&self) -> &str {
- &self.human_display
+ pub fn human_display(&self) -> Option<&str> {
+ self.human_display
+ .as_ref()
+ .map(AggregateHumanDisplay::expression)
+ }
+
+ #[doc(hidden)]
+ pub fn human_display_alias(&self) -> Option<&str> {
+ self.human_display
+ .as_ref()
+ .and_then(AggregateHumanDisplay::alias)
+ }
+
+ fn return_field_metadata(&self) -> Option<FieldMetadata> {
+ let metadata = FieldMetadata::from(self.return_field.as_ref());
+ (!metadata.is_empty()).then_some(metadata)
}
/// Return if the aggregation is distinct
@@ -456,15 +785,22 @@ impl AggregateFunctionExpr {
return Ok(None);
};
- AggregateExprBuilder::new(Arc::new(updated_fn), self.args.to_vec())
- .order_by(self.order_bys.clone())
- .schema(Arc::new(self.schema.clone()))
- .alias(self.name().to_string())
- .with_ignore_nulls(self.ignore_nulls)
- .with_distinct(self.is_distinct)
- .with_reversed(self.is_reversed)
- .build()
- .map(Some)
+ let mut builder =
+ AggregateExprBuilder::new(Arc::new(updated_fn), self.args.to_vec())
+ .order_by(self.order_bys.clone())
+ .schema(Arc::new(self.schema.clone()))
+ .alias(self.name().to_string())
+ .output_metadata(self.return_field_metadata())
+ .with_ignore_nulls(self.ignore_nulls)
+ .with_distinct(self.is_distinct)
+ .with_reversed(self.is_reversed);
+ if let Some(human_display) = self.human_display() {
+ builder = builder.human_display(human_display);
+ }
+ if let Some(alias) = self.human_display_alias() {
+ builder = builder.human_display_alias(alias);
+ }
+ builder.build().map(Some)
}
/// Creates accumulator implementation that supports retract
@@ -582,23 +918,54 @@ impl AggregateFunctionExpr {
ReversedUDAF::NotSupported => None,
ReversedUDAF::Identical => Some(self.clone()),
ReversedUDAF::Reversed(reverse_udf) => {
+ let was_aliased = self.human_display_alias().is_some();
let mut name = self.name().to_string();
+ let mut human_display = self.human_display.clone();
+ // Reversing display follows two paths:
+ // - aliased display keeps the output `name` unchanged and
rewrites only
+ // the lowered expression in `human_display`.
+ // - non-aliased display rewrites the canonical `name`, and
rewrites
+ // `human_display` only when present.
// If the function is changed, we need to reverse order_by
clause as well
// i.e. First(a order by b asc null first) -> Last(a order by
b desc null last)
- if self.fun().name() != reverse_udf.name() {
+ if !was_aliased && self.fun().name() != reverse_udf.name() {
replace_order_by_clause(&mut name);
}
- replace_fn_name_clause(&mut name, self.fun.name(),
reverse_udf.name());
-
- AggregateExprBuilder::new(reverse_udf, self.args.to_vec())
- .order_by(self.order_bys.iter().map(|e|
e.reverse()).collect())
- .schema(Arc::new(self.schema.clone()))
- .alias(name)
- .with_ignore_nulls(self.ignore_nulls)
- .with_distinct(self.is_distinct)
- .with_reversed(!self.is_reversed)
- .build()
- .ok()
+ if !was_aliased {
+ replace_fn_name_clause(
+ &mut name,
+ self.fun.name(),
+ reverse_udf.name(),
+ );
+ }
+
+ if let Some(human_display) = human_display.as_mut() {
+ if self.fun().name() != reverse_udf.name() {
+ replace_order_by_clause(&mut human_display.expression);
+ }
+ replace_fn_name_clause(
+ &mut human_display.expression,
+ self.fun.name(),
+ reverse_udf.name(),
+ );
+ }
+
+ let mut builder =
+ AggregateExprBuilder::new(reverse_udf, self.args.to_vec())
+ .order_by(self.order_bys.iter().map(|e|
e.reverse()).collect())
+ .schema(Arc::new(self.schema.clone()))
+ .alias(name)
+ .output_metadata(self.return_field_metadata())
+ .with_ignore_nulls(self.ignore_nulls)
+ .with_distinct(self.is_distinct)
+ .with_reversed(!self.is_reversed);
+ if let Some(human_display) = human_display {
+ builder = builder.human_display(human_display.expression);
+ if let Some(alias) = human_display.alias {
+ builder = builder.human_display_alias(alias);
+ }
+ }
+ builder.build().ok()
}
}
}
@@ -753,5 +1120,86 @@ fn replace_order_by_clause(order_by: &mut String) {
}
fn replace_fn_name_clause(aggr_name: &mut String, fn_name_old: &str,
fn_name_new: &str) {
- *aggr_name = aggr_name.replace(fn_name_old, fn_name_new);
+ if let Some(rest) = aggr_name.strip_prefix(fn_name_old) {
+ *aggr_name = format!("{fn_name_new}{rest}");
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ use std::collections::HashMap;
+
+ use arrow::datatypes::Field;
+ use datafusion_common::metadata::FieldMetadata;
+ use datafusion_expr::{col, test::function_stub::sum};
+
+ fn aggregate_test_schema() -> Result<(Schema, DFSchema)> {
+ let schema = Schema::new(vec![Field::new("column1", DataType::Int64,
true)]);
+ let logical_schema = DFSchema::try_from(schema.clone())?;
+ Ok((schema, logical_schema))
+ }
+
+ fn test_metadata() -> FieldMetadata {
+ FieldMetadata::from(HashMap::from([(
+ "some_key".to_string(),
+ "some_value".to_string(),
+ )]))
+ }
+
+ fn aggregate_alias_with_metadata() -> Expr {
+ sum(col("column1")).alias_with_metadata("agg", Some(test_metadata()))
+ }
+
+ #[test]
+ fn lowered_aggregate_builder_unwraps_alias_with_metadata() -> Result<()> {
+ let (schema, logical_schema) = aggregate_test_schema()?;
+ let expr = aggregate_alias_with_metadata();
+
+ let lowered = LoweredAggregateBuilder::new(
+ &expr,
+ &logical_schema,
+ &schema,
+ &ExecutionProps::new(),
+ )
+ .build()?;
+
+ assert_eq!(lowered.aggregate.name(), "agg");
+ assert_eq!(lowered.aggregate.human_display_alias(), Some("agg"));
+ assert_eq!(
+ lowered.aggregate.field().metadata().get("some_key"),
+ Some(&"some_value".to_string())
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn lowered_aggregate_builder_display_override_skips_alias_metadata() ->
Result<()> {
+ let (schema, logical_schema) = aggregate_test_schema()?;
+ let expr = aggregate_alias_with_metadata();
+
+ let lowered = LoweredAggregateBuilder::new(
+ &expr,
+ &logical_schema,
+ &schema,
+ &ExecutionProps::new(),
+ )
+ .with_human_display(expr.human_display().to_string())
+ .build()?;
+
+ assert_eq!(lowered.aggregate.name(), "agg");
+ assert_eq!(lowered.aggregate.human_display_alias(), None);
+ assert!(
+ lowered
+ .aggregate
+ .field()
+ .metadata()
+ .get("some_key")
+ .is_none()
+ );
+
+ Ok(())
+ }
}
diff --git a/datafusion/physical-plan/src/aggregates/mod.rs
b/datafusion/physical-plan/src/aggregates/mod.rs
index 6684622047..9953ce3f84 100644
--- a/datafusion/physical-plan/src/aggregates/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/mod.rs
@@ -17,6 +17,7 @@
//! Aggregates functionalities
+use std::borrow::Cow;
use std::sync::Arc;
use super::{DisplayAs, ExecutionPlanProperties, PlanProperties};
@@ -1374,7 +1375,7 @@ impl DisplayAs for AggregateExec {
let a: Vec<String> = self
.aggr_expr
.iter()
- .map(|agg| agg.name().to_string())
+ .map(|agg| format_aggregate_exec_expr(agg).to_string())
.collect();
write!(f, ", aggr=[{}]", a.join(", "))?;
if let Some(config) = self.limit_options {
@@ -1428,7 +1429,7 @@ impl DisplayAs for AggregateExec {
let a: Vec<String> = self
.aggr_expr
.iter()
- .map(|agg| agg.human_display().to_string())
+ .map(|agg| format_tree_aggregate_expr(agg).to_string())
.collect();
writeln!(f, "mode={:?}", self.mode)?;
if !g.is_empty() {
@@ -1446,6 +1447,29 @@ impl DisplayAs for AggregateExec {
}
}
+fn format_aggregate_exec_expr(agg: &AggregateFunctionExpr) -> Cow<'_, str> {
+ match agg.human_display_alias() {
+ Some(_) => format_human_display(agg.human_display(),
agg.human_display_alias())
+ .unwrap_or_else(|| Cow::Borrowed(agg.name())),
+ None => Cow::Borrowed(agg.name()),
+ }
+}
+
+fn format_tree_aggregate_expr(agg: &AggregateFunctionExpr) -> Cow<'_, str> {
+ format_human_display(agg.human_display(), agg.human_display_alias())
+ .unwrap_or_else(|| Cow::Borrowed(agg.name()))
+}
+
+fn format_human_display<'a>(
+ human_display: Option<&'a str>,
+ alias: Option<&'a str>,
+) -> Option<Cow<'a, str>> {
+ human_display.map(|human_display| match alias {
+ Some(alias) => Cow::Owned(format!("{human_display} as {alias}")),
+ None => Cow::Borrowed(human_display),
+ })
+}
+
impl ExecutionPlan for AggregateExec {
fn name(&self) -> &'static str {
"AggregateExec"
@@ -3021,6 +3045,147 @@ mod tests {
.map(Arc::new)
}
+ fn first_value_agg_expr(
+ schema: &SchemaRef,
+ column: &str,
+ alias: &str,
+ human_display: Option<&str>,
+ human_display_alias: Option<&str>,
+ ) -> Result<AggregateFunctionExpr> {
+ let mut builder =
+ AggregateExprBuilder::new(first_value_udaf(), vec![col(column,
schema)?])
+ .order_by(vec![PhysicalSortExpr {
+ expr: col(column, schema)?,
+ options: SortOptions::new(false, false),
+ }])
+ .schema(Arc::clone(schema))
+ .alias(alias);
+
+ if let Some(human_display) = human_display {
+ builder = builder.human_display(human_display);
+ }
+ if let Some(human_display_alias) = human_display_alias {
+ builder = builder.human_display_alias(human_display_alias);
+ }
+
+ builder.build()
+ }
+
+ #[test]
+ fn test_reverse_expr_preserves_aliased_human_display() -> Result<()> {
+ let schema = create_test_schema()?;
+ let agg = first_value_agg_expr(
+ &schema,
+ "b",
+ "agg",
+ Some("first_value(b) ORDER BY [b ASC NULLS LAST]"),
+ Some("agg"),
+ )?;
+
+ let reversed = agg.reverse_expr().expect("expected reverse expr");
+
+ assert_eq!(reversed.name(), "agg");
+ assert_eq!(reversed.human_display_alias(), Some("agg"));
+ assert_eq!(
+ format_tree_aggregate_expr(&reversed),
+ "last_value(b) ORDER BY [b DESC NULLS FIRST] as agg"
+ );
+ assert_eq!(
+ reversed.human_display(),
+ Some("last_value(b) ORDER BY [b DESC NULLS FIRST]")
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_reverse_expr_does_not_rewrite_column_names_in_human_display() ->
Result<()> {
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "first_value_col",
+ DataType::Int32,
+ true,
+ )]));
+ let agg = first_value_agg_expr(
+ &schema,
+ "first_value_col",
+ "agg",
+ Some(
+ "first_value(first_value_col) ORDER BY [first_value_col ASC
NULLS LAST]",
+ ),
+ Some("agg"),
+ )?;
+
+ let reversed = agg.reverse_expr().expect("expected reverse expr");
+
+ assert_eq!(reversed.name(), "agg");
+ assert_eq!(
+ reversed.human_display(),
+ Some(
+ "last_value(first_value_col) ORDER BY [first_value_col DESC
NULLS FIRST]"
+ )
+ );
+ assert_eq!(
+ format_tree_aggregate_expr(&reversed),
+ "last_value(first_value_col) ORDER BY [first_value_col DESC NULLS
FIRST] as agg"
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_empty_human_display_is_treated_as_absent() -> Result<()> {
+ let schema = create_test_schema()?;
+ let agg = first_value_agg_expr(&schema, "b", "agg", Some(""), None)?;
+
+ assert_eq!(agg.human_display(), None);
+ assert_eq!(format_tree_aggregate_expr(&agg), "agg");
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_human_display_alias_must_match_name() -> Result<()> {
+ let schema = create_test_schema()?;
+ let error = first_value_agg_expr(
+ &schema,
+ "b",
+ "agg",
+ Some("first_value(b) ORDER BY [b ASC NULLS LAST]"),
+ Some("other_alias"),
+ )
+ .unwrap_err();
+
+ assert!(
+ error
+ .to_string()
+ .contains("aggregate human_display_alias must match")
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_reverse_expr_preserves_non_aliased_display_path() -> Result<()> {
+ let schema = create_test_schema()?;
+ let agg = first_value_agg_expr(
+ &schema,
+ "b",
+ "first_value(b) ORDER BY [b ASC NULLS LAST]",
+ None,
+ None,
+ )?;
+
+ let reversed = agg.reverse_expr().expect("expected reverse expr");
+
+ assert_eq!(
+ reversed.name(),
+ "last_value(b) ORDER BY [b DESC NULLS FIRST]"
+ );
+ assert_eq!(reversed.human_display(), None);
+
+ Ok(())
+ }
+
// This function constructs the physical plan below,
//
// "AggregateExec: mode=Final, gby=[a@0 as a], aggr=[FIRST_VALUE(b)]",
diff --git a/datafusion/physical-plan/src/aggregates/row_hash.rs
b/datafusion/physical-plan/src/aggregates/row_hash.rs
index a55cf09c79..b4ac7d0605 100644
--- a/datafusion/physical-plan/src/aggregates/row_hash.rs
+++ b/datafusion/physical-plan/src/aggregates/row_hash.rs
@@ -21,8 +21,8 @@ use std::sync::Arc;
use std::task::{Context, Poll};
use std::vec;
-use super::AggregateExec;
use super::order::GroupOrdering;
+use super::{AggregateExec, format_human_display};
use crate::aggregates::group_values::{GroupByMetrics, GroupValues,
new_group_values};
use crate::aggregates::order::GroupOrderingFull;
use crate::aggregates::{
@@ -564,7 +564,11 @@ impl GroupedHashAggregateStream {
let agg_fn_names = aggregate_exprs
.iter()
- .map(|expr| expr.human_display())
+ .map(|expr| {
+ format_human_display(expr.human_display(),
expr.human_display_alias())
+ .map(|display| display.into_owned())
+ .unwrap_or_else(|| expr.name().to_string())
+ })
.collect::<Vec<_>>()
.join(", ");
let name = format!("GroupedHashAggregateStream[{partition}]
({agg_fn_names})");
diff --git a/datafusion/proto/src/physical_plan/mod.rs
b/datafusion/proto/src/physical_plan/mod.rs
index 5172a552fa..68a0b2a456 100644
--- a/datafusion/proto/src/physical_plan/mod.rs
+++ b/datafusion/proto/src/physical_plan/mod.rs
@@ -118,6 +118,58 @@ use crate::{convert_required, into_required};
pub mod from_proto;
pub mod to_proto;
+const HUMAN_DISPLAY_ALIAS_PREFIX: &str =
"\u{1f}datafusion_human_display_alias_v1:";
+
+fn encode_human_display_alias(human_display: &str, alias: &str) -> String {
+ format!(
+ "{HUMAN_DISPLAY_ALIAS_PREFIX}{}:{alias}{human_display}",
+ alias.len()
+ )
+}
+
+fn split_human_display_alias<'a>(
+ human_display: &'a str,
+ name: &'a str,
+) -> (&'a str, Option<&'a str>) {
+ if let Some(encoded) =
human_display.strip_prefix(HUMAN_DISPLAY_ALIAS_PREFIX)
+ && let Some((alias_len, encoded)) = encoded.split_once(':')
+ && let Ok(alias_len) = alias_len.parse::<usize>()
+ && let Some(alias) = encoded.get(..alias_len)
+ && let Some(human_display) = encoded.get(alias_len..)
+ && alias == name
+ && !human_display.is_empty()
+ {
+ return (human_display, Some(alias));
+ }
+
+ (human_display, None)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn split_human_display_alias_ignores_mismatched_alias() {
+ let encoded = encode_human_display_alias("sum(value)", "revenue");
+
+ assert_eq!(
+ split_human_display_alias(&encoded, "other"),
+ (encoded.as_str(), None)
+ );
+ }
+
+ #[test]
+ fn split_human_display_alias_keeps_malformed_prefix_literal() {
+ let display = format!("{HUMAN_DISPLAY_ALIAS_PREFIX}not-an-encoding");
+
+ assert_eq!(
+ split_human_display_alias(&display, "agg"),
+ (display.as_str(), None)
+ );
+ }
+}
+
/// Context threaded through physical-plan deserialization.
///
/// This bundles the stable per-call inputs for deserialization and the
@@ -1259,15 +1311,28 @@ impl protobuf::PhysicalPlanNode {
)?,
};
- AggregateExprBuilder::new(agg_udf,
input_phy_expr)
- .schema(Arc::clone(&physical_schema))
- .alias(name)
-
.human_display(agg_node.human_display.clone())
-
.with_ignore_nulls(agg_node.ignore_nulls)
- .with_distinct(agg_node.distinct)
- .order_by(order_bys)
- .build()
- .map(Arc::new)
+ let (human_display, human_display_alias) =
+ split_human_display_alias(
+ &agg_node.human_display,
+ name,
+ );
+ let builder = AggregateExprBuilder::new(
+ agg_udf,
+ input_phy_expr,
+ )
+ .schema(Arc::clone(&physical_schema))
+ .alias(name)
+ .with_ignore_nulls(agg_node.ignore_nulls)
+ .with_distinct(agg_node.distinct)
+ .order_by(order_bys)
+ .human_display(human_display);
+ let builder = if let Some(alias) =
human_display_alias
+ {
+ builder.human_display_alias(alias)
+ } else {
+ builder
+ };
+ builder.build().map(Arc::new)
}
})
.transpose()?
diff --git a/datafusion/proto/src/physical_plan/to_proto.rs
b/datafusion/proto/src/physical_plan/to_proto.rs
index 83c11cfc6b..ec8e168178 100644
--- a/datafusion/proto/src/physical_plan/to_proto.rs
+++ b/datafusion/proto/src/physical_plan/to_proto.rs
@@ -47,7 +47,7 @@ use datafusion_physical_plan::{Partitioning, PhysicalExpr,
WindowExpr};
use super::{
DefaultPhysicalProtoConverter, PhysicalExtensionCodec,
- PhysicalProtoConverterExtension,
+ PhysicalProtoConverterExtension, encode_human_display_alias,
};
use crate::protobuf::{
self, PhysicalSortExprNode, PhysicalSortExprNodeCollection,
@@ -71,6 +71,12 @@ pub fn serialize_physical_aggr_expr(
let name = aggr_expr.fun().name().to_string();
let mut buf = Vec::new();
codec.try_encode_udaf(aggr_expr.fun(), &mut buf)?;
+ let human_display = match (aggr_expr.human_display(),
aggr_expr.human_display_alias())
+ {
+ (Some(display), Some(alias)) => encode_human_display_alias(display,
alias),
+ (Some(display), None) => display.to_string(),
+ (None, _) => String::new(),
+ };
Ok(protobuf::PhysicalExprNode {
expr_id: None,
expr_type: Some(protobuf::physical_expr_node::ExprType::AggregateExpr(
@@ -81,7 +87,7 @@ pub fn serialize_physical_aggr_expr(
distinct: aggr_expr.is_distinct(),
ignore_nulls: aggr_expr.ignore_nulls(),
fun_definition: (!buf.is_empty()).then_some(buf),
- human_display: aggr_expr.human_display().to_string(),
+ human_display,
},
)),
})
diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
index fa342ae907..e21587cc62 100644
--- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
@@ -43,11 +43,12 @@ use datafusion::datasource::sink::DataSinkExec;
use datafusion::datasource::source::DataSourceExec;
use datafusion::execution::TaskContext;
use datafusion::functions_aggregate::count::count_udaf;
+use datafusion::functions_aggregate::first_last::first_value_udaf;
use datafusion::functions_aggregate::sum::sum_udaf;
use datafusion::functions_window::nth_value::nth_value_udwf;
use datafusion::functions_window::row_number::row_number_udwf;
use datafusion::logical_expr::{JoinType, Operator, Volatility, create_udf};
-use datafusion::physical_expr::aggregate::AggregateExprBuilder;
+use datafusion::physical_expr::aggregate::{AggregateExprBuilder,
AggregateFunctionExpr};
use datafusion::physical_expr::expressions::Literal;
use datafusion::physical_expr::window::{SlidingAggregateWindowExpr,
StandardWindowExpr};
use datafusion::physical_expr::{
@@ -2165,9 +2166,107 @@ async fn test_round_trip_human_display() -> Result<()> {
let sql = "select r_name, count(r_name) from region group by r_name";
roundtrip_test_sql_with_context(sql, &ctx).await?;
+ let sql = "select count(*) as count_star from region";
+ roundtrip_test_sql_with_context(sql, &ctx).await?;
+
+ Ok(())
+}
+
+#[test]
+fn test_round_trip_aliased_reverse_human_display() -> Result<()> {
+ let aggregate_expr = roundtrip_first_value_aggregate(
+ "agg",
+ "first_value(b) ORDER BY [b ASC NULLS LAST]",
+ Some("agg"),
+ )?;
+ let reversed = aggregate_expr
+ .reverse_expr()
+ .expect("expected reverse expr");
+
+ assert_eq!(reversed.name(), "agg");
+ assert_eq!(reversed.human_display_alias(), Some("agg"));
+ assert_eq!(
+ reversed.human_display(),
+ Some("last_value(b) ORDER BY [b DESC NULLS FIRST]")
+ );
+
Ok(())
}
+#[test]
+fn test_round_trip_human_display_alias_with_colon() -> Result<()> {
+ let aggregate_expr = roundtrip_first_value_aggregate(
+ "agg:one",
+ "first_value(b) ORDER BY [b ASC NULLS LAST]",
+ Some("agg:one"),
+ )?;
+
+ assert_eq!(aggregate_expr.name(), "agg:one");
+ assert_eq!(aggregate_expr.human_display_alias(), Some("agg:one"));
+ assert_eq!(
+ aggregate_expr.human_display(),
+ Some("first_value(b) ORDER BY [b ASC NULLS LAST]")
+ );
+
+ Ok(())
+}
+
+#[test]
+fn test_round_trip_non_aliased_human_display_ending_like_alias() -> Result<()>
{
+ let aggregate_expr =
+ roundtrip_first_value_aggregate("agg", "first_value(b) as agg", None)?;
+
+ assert_eq!(aggregate_expr.name(), "agg");
+ assert_eq!(
+ aggregate_expr.human_display(),
+ Some("first_value(b) as agg")
+ );
+ assert_eq!(aggregate_expr.human_display_alias(), None);
+
+ Ok(())
+}
+
+fn roundtrip_first_value_aggregate(
+ alias: &str,
+ human_display: &str,
+ human_display_alias: Option<&str>,
+) -> Result<Arc<AggregateFunctionExpr>> {
+ let schema = Arc::new(Schema::new(vec![Field::new("b", DataType::Int64,
true)]));
+ let mut builder =
+ AggregateExprBuilder::new(first_value_udaf(), vec![col("b", &schema)?])
+ .order_by(vec![PhysicalSortExpr {
+ expr: col("b", &schema)?,
+ options: SortOptions::new(false, false),
+ }])
+ .schema(Arc::clone(&schema))
+ .alias(alias)
+ .human_display(human_display);
+ if let Some(human_display_alias) = human_display_alias {
+ builder = builder.human_display_alias(human_display_alias);
+ }
+ let agg_expr = builder.build().map(Arc::new)?;
+
+ let plan = Arc::new(AggregateExec::try_new(
+ AggregateMode::Single,
+ PhysicalGroupBy::new(vec![], vec![], vec![], false),
+ vec![agg_expr],
+ vec![None],
+ Arc::new(EmptyExec::new(Arc::clone(&schema))),
+ schema,
+ )?);
+
+ let ctx = SessionContext::new();
+ let codec = DefaultPhysicalExtensionCodec {};
+ let proto_converter = DefaultPhysicalProtoConverter {};
+ let roundtrip_plan = roundtrip_test_and_return(plan, &ctx, &codec,
&proto_converter)?;
+ let aggregate = roundtrip_plan
+ .as_ref()
+ .downcast_ref::<AggregateExec>()
+ .expect("expected AggregateExec after roundtrip");
+
+ Ok(Arc::clone(&aggregate.aggr_expr()[0]))
+}
+
// Bug 2 of https://github.com/apache/datafusion/issues/16772
/// Test that PhysicalGroupBy groups field is correctly serialized/deserialized
/// for simple aggregates (no GROUP BY clause).
diff --git a/datafusion/sqllogictest/test_files/agg_func_substitute.slt
b/datafusion/sqllogictest/test_files/agg_func_substitute.slt
index e0199c8250..dd89e83492 100644
--- a/datafusion/sqllogictest/test_files/agg_func_substitute.slt
+++ b/datafusion/sqllogictest/test_files/agg_func_substitute.slt
@@ -79,9 +79,9 @@ logical_plan
03)----TableScan: multiple_ordered_table projection=[a, c]
physical_plan
01)ProjectionExec: expr=[a@0 as a, nth_value(multiple_ordered_table.c,Int64(1)
+ Int64(100)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]@1 as result]
-02)--AggregateExec: mode=FinalPartitioned, gby=[a@0 as a],
aggr=[nth_value(multiple_ordered_table.c,Int64(1) + Int64(100)) ORDER BY
[multiple_ordered_table.c ASC NULLS LAST]], ordering_mode=Sorted
+02)--AggregateExec: mode=FinalPartitioned, gby=[a@0 as a],
aggr=[nth_value(multiple_ordered_table.c, 101) ORDER BY
[multiple_ordered_table.c ASC NULLS LAST] as
nth_value(multiple_ordered_table.c,Int64(1) + Int64(100)) ORDER BY
[multiple_ordered_table.c ASC NULLS LAST]], ordering_mode=Sorted
03)----RepartitionExec: partitioning=Hash([a@0], 4), input_partitions=4,
preserve_order=true, sort_exprs=a@0 ASC NULLS LAST
-04)------AggregateExec: mode=Partial, gby=[a@0 as a],
aggr=[nth_value(multiple_ordered_table.c,Int64(1) + Int64(100)) ORDER BY
[multiple_ordered_table.c ASC NULLS LAST]], ordering_mode=Sorted
+04)------AggregateExec: mode=Partial, gby=[a@0 as a],
aggr=[nth_value(multiple_ordered_table.c, 101) ORDER BY
[multiple_ordered_table.c ASC NULLS LAST] as
nth_value(multiple_ordered_table.c,Int64(1) + Int64(100)) ORDER BY
[multiple_ordered_table.c ASC NULLS LAST]], ordering_mode=Sorted
05)--------RepartitionExec: partitioning=RoundRobinBatch(4),
input_partitions=1, maintains_sort_order=true
06)----------DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, c],
output_orderings=[[a@0 ASC NULLS LAST], [c@1 ASC NULLS LAST]], file_type=csv,
has_header=true
diff --git a/datafusion/sqllogictest/test_files/aggregate.slt
b/datafusion/sqllogictest/test_files/aggregate.slt
index b8009dfd57..e56eb9ef38 100644
--- a/datafusion/sqllogictest/test_files/aggregate.slt
+++ b/datafusion/sqllogictest/test_files/aggregate.slt
@@ -3915,9 +3915,9 @@ logical_plan
01)Aggregate: groupBy=[[]], aggr=[[min(CAST(aggregate_test_100.c2 AS Float64))
AS percentile_cont(Float64(0)) WITHIN GROUP [aggregate_test_100.c2 ASC NULLS
LAST]]]
02)--TableScan: aggregate_test_100 projection=[c2]
physical_plan
-01)AggregateExec: mode=Final, gby=[], aggr=[percentile_cont(Float64(0)) WITHIN
GROUP [aggregate_test_100.c2 ASC NULLS LAST]]
+01)AggregateExec: mode=Final, gby=[], aggr=[min(aggregate_test_100.c2) as
percentile_cont(Float64(0)) WITHIN GROUP [aggregate_test_100.c2 ASC NULLS LAST]]
02)--CoalescePartitionsExec
-03)----AggregateExec: mode=Partial, gby=[], aggr=[percentile_cont(Float64(0))
WITHIN GROUP [aggregate_test_100.c2 ASC NULLS LAST]]
+03)----AggregateExec: mode=Partial, gby=[], aggr=[min(aggregate_test_100.c2)
as percentile_cont(Float64(0)) WITHIN GROUP [aggregate_test_100.c2 ASC NULLS
LAST]]
04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
05)--------DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100_with_dates.csv]]},
projection=[c2], file_type=csv, has_header=true
@@ -3928,9 +3928,9 @@ logical_plan
01)Aggregate: groupBy=[[]], aggr=[[max(CAST(aggregate_test_100.c2 AS Float64))
AS percentile_cont(Float64(0)) WITHIN GROUP [aggregate_test_100.c2 DESC NULLS
FIRST]]]
02)--TableScan: aggregate_test_100 projection=[c2]
physical_plan
-01)AggregateExec: mode=Final, gby=[], aggr=[percentile_cont(Float64(0)) WITHIN
GROUP [aggregate_test_100.c2 DESC NULLS FIRST]]
+01)AggregateExec: mode=Final, gby=[], aggr=[max(aggregate_test_100.c2) as
percentile_cont(Float64(0)) WITHIN GROUP [aggregate_test_100.c2 DESC NULLS
FIRST]]
02)--CoalescePartitionsExec
-03)----AggregateExec: mode=Partial, gby=[], aggr=[percentile_cont(Float64(0))
WITHIN GROUP [aggregate_test_100.c2 DESC NULLS FIRST]]
+03)----AggregateExec: mode=Partial, gby=[], aggr=[max(aggregate_test_100.c2)
as percentile_cont(Float64(0)) WITHIN GROUP [aggregate_test_100.c2 DESC NULLS
FIRST]]
04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
05)--------DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100_with_dates.csv]]},
projection=[c2], file_type=csv, has_header=true
@@ -3941,9 +3941,9 @@ logical_plan
01)Aggregate: groupBy=[[]], aggr=[[min(CAST(aggregate_test_100.c2 AS Float64))
AS percentile_cont(aggregate_test_100.c2,Float64(0))]]
02)--TableScan: aggregate_test_100 projection=[c2]
physical_plan
-01)AggregateExec: mode=Final, gby=[],
aggr=[percentile_cont(aggregate_test_100.c2,Float64(0))]
+01)AggregateExec: mode=Final, gby=[], aggr=[min(aggregate_test_100.c2) as
percentile_cont(aggregate_test_100.c2,Float64(0))]
02)--CoalescePartitionsExec
-03)----AggregateExec: mode=Partial, gby=[],
aggr=[percentile_cont(aggregate_test_100.c2,Float64(0))]
+03)----AggregateExec: mode=Partial, gby=[], aggr=[min(aggregate_test_100.c2)
as percentile_cont(aggregate_test_100.c2,Float64(0))]
04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
05)--------DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100_with_dates.csv]]},
projection=[c2], file_type=csv, has_header=true
@@ -3954,9 +3954,9 @@ logical_plan
01)Aggregate: groupBy=[[]], aggr=[[max(CAST(aggregate_test_100.c2 AS Float64))
AS percentile_cont(aggregate_test_100.c2,Float64(1))]]
02)--TableScan: aggregate_test_100 projection=[c2]
physical_plan
-01)AggregateExec: mode=Final, gby=[],
aggr=[percentile_cont(aggregate_test_100.c2,Float64(1))]
+01)AggregateExec: mode=Final, gby=[], aggr=[max(aggregate_test_100.c2) as
percentile_cont(aggregate_test_100.c2,Float64(1))]
02)--CoalescePartitionsExec
-03)----AggregateExec: mode=Partial, gby=[],
aggr=[percentile_cont(aggregate_test_100.c2,Float64(1))]
+03)----AggregateExec: mode=Partial, gby=[], aggr=[max(aggregate_test_100.c2)
as percentile_cont(aggregate_test_100.c2,Float64(1))]
04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
05)--------DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100_with_dates.csv]]},
projection=[c2], file_type=csv, has_header=true
diff --git a/datafusion/sqllogictest/test_files/aggregates_simplify.slt
b/datafusion/sqllogictest/test_files/aggregates_simplify.slt
index 9aa3ecf7a2..c4055d1739 100644
--- a/datafusion/sqllogictest/test_files/aggregates_simplify.slt
+++ b/datafusion/sqllogictest/test_files/aggregates_simplify.slt
@@ -91,7 +91,7 @@ logical_plan
03)----TableScan: sum_simplify_t projection=[]
physical_plan
01)ProjectionExec: expr=[__common_expr_1@0 as sum(Int64(2) + Int64(1)),
__common_expr_1@0 as sum(Int64(3))]
-02)--AggregateExec: mode=Single, gby=[], aggr=[__common_expr_1]
+02)--AggregateExec: mode=Single, gby=[], aggr=[sum(3) as __common_expr_1]
03)----DataSourceExec: partitions=1, partition_sizes=[1]
@@ -170,9 +170,9 @@ physical_plan
02)--AggregateExec: mode=Final, gby=[], aggr=[sum(alias1), sum(alias2)]
03)----CoalescePartitionsExec
04)------AggregateExec: mode=Partial, gby=[], aggr=[sum(alias1), sum(alias2)]
-05)--------AggregateExec: mode=FinalPartitioned, gby=[alias1@0 as alias1],
aggr=[alias2]
+05)--------AggregateExec: mode=FinalPartitioned, gby=[alias1@0 as alias1],
aggr=[sum(__common_expr_1) as alias2]
06)----------RepartitionExec: partitioning=Hash([alias1@0], 4),
input_partitions=1
-07)------------AggregateExec: mode=Partial, gby=[__common_expr_1@0 as alias1],
aggr=[alias2]
+07)------------AggregateExec: mode=Partial, gby=[__common_expr_1@0 as alias1],
aggr=[sum(__common_expr_1) as alias2]
08)--------------ProjectionExec: expr=[column1@0 + 1 as __common_expr_1]
09)----------------DataSourceExec: partitions=1, partition_sizes=[1]
@@ -249,7 +249,7 @@ logical_plan
03)----TableScan: sum_simplify_t projection=[]
physical_plan
01)ProjectionExec: expr=[sum(random() + Int64(2))@1 > sum(random() +
Int64(1))@0 as sum(random() + Int64(1)) < sum(random() + Int64(2))]
-02)--AggregateExec: mode=Single, gby=[], aggr=[sum(random() + Int64(1)),
sum(random() + Int64(2))]
+02)--AggregateExec: mode=Single, gby=[], aggr=[sum(random() + 1) as
sum(random() + Int64(1)), sum(random() + 2) as sum(random() + Int64(2))]
03)----DataSourceExec: partitions=1, partition_sizes=[1]
# Checks grouped aggregates with explicit ORDER BY return deterministic row
order.
diff --git a/datafusion/sqllogictest/test_files/expr.slt
b/datafusion/sqllogictest/test_files/expr.slt
index 163730baae..51b7591b41 100644
--- a/datafusion/sqllogictest/test_files/expr.slt
+++ b/datafusion/sqllogictest/test_files/expr.slt
@@ -2436,7 +2436,7 @@ logical_plan
03)----TableScan: t projection=[a]
physical_plan
01)ProjectionExec: expr=[min(t.a) FILTER (WHERE t.a > Int64(1))@0 as x]
-02)--AggregateExec: mode=Single, gby=[], aggr=[min(t.a) FILTER (WHERE t.a >
Int64(1))]
+02)--AggregateExec: mode=Single, gby=[], aggr=[min(t.a) FILTER (WHERE t.a >
Float32(1)) as min(t.a) FILTER (WHERE t.a > Int64(1))]
03)----DataSourceExec: partitions=1, partition_sizes=[1]
diff --git a/datafusion/sqllogictest/test_files/group_by.slt
b/datafusion/sqllogictest/test_files/group_by.slt
index b313424951..8c055c25ca 100644
--- a/datafusion/sqllogictest/test_files/group_by.slt
+++ b/datafusion/sqllogictest/test_files/group_by.slt
@@ -4443,9 +4443,9 @@ physical_plan
04)------AggregateExec: mode=FinalPartitioned, gby=[c1@0 as c1],
aggr=[count(alias1), min(alias1), sum(alias2), max(alias3)]
05)--------RepartitionExec: partitioning=Hash([c1@0], 8), input_partitions=8
06)----------AggregateExec: mode=Partial, gby=[c1@0 as c1],
aggr=[count(alias1), min(alias1), sum(alias2), max(alias3)]
-07)------------AggregateExec: mode=FinalPartitioned, gby=[c1@0 as c1, alias1@1
as alias1], aggr=[alias2, alias3]
+07)------------AggregateExec: mode=FinalPartitioned, gby=[c1@0 as c1, alias1@1
as alias1], aggr=[sum(aggregate_test_100.c3) as alias2,
max(aggregate_test_100.c4) as alias3]
08)--------------RepartitionExec: partitioning=Hash([c1@0, alias1@1], 8),
input_partitions=8
-09)----------------AggregateExec: mode=Partial, gby=[c1@0 as c1, c2@1 as
alias1], aggr=[alias2, alias3]
+09)----------------AggregateExec: mode=Partial, gby=[c1@0 as c1, c2@1 as
alias1], aggr=[sum(aggregate_test_100.c3) as alias2, max(aggregate_test_100.c4)
as alias3]
10)------------------RepartitionExec: partitioning=RoundRobinBatch(8),
input_partitions=1
11)--------------------DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c1,
c2, c3, c4], file_type=csv, has_header=true
diff --git a/datafusion/sqllogictest/test_files/subquery.slt
b/datafusion/sqllogictest/test_files/subquery.slt
index 1ee92fc75a..f56a8a10d2 100644
--- a/datafusion/sqllogictest/test_files/subquery.slt
+++ b/datafusion/sqllogictest/test_files/subquery.slt
@@ -237,9 +237,9 @@ physical_plan
02)--HashJoinExec: mode=CollectLeft, join_type=Right, on=[(t2_id@1, t1_id@0)],
projection=[t1_id@2, sum(t2.t2_int * Float64(1)) + Int64(1)@0]
03)----CoalescePartitionsExec
04)------ProjectionExec: expr=[sum(t2.t2_int * Float64(1))@1 + 1 as
sum(t2.t2_int * Float64(1)) + Int64(1), t2_id@0 as t2_id]
-05)--------AggregateExec: mode=FinalPartitioned, gby=[t2_id@0 as t2_id],
aggr=[sum(t2.t2_int * Float64(1))]
+05)--------AggregateExec: mode=FinalPartitioned, gby=[t2_id@0 as t2_id],
aggr=[sum(t2.t2_int) as sum(t2.t2_int * Float64(1))]
06)----------RepartitionExec: partitioning=Hash([t2_id@0], 4),
input_partitions=4
-07)------------AggregateExec: mode=Partial, gby=[t2_id@0 as t2_id],
aggr=[sum(t2.t2_int * Float64(1))]
+07)------------AggregateExec: mode=Partial, gby=[t2_id@0 as t2_id],
aggr=[sum(t2.t2_int) as sum(t2.t2_int * Float64(1))]
08)--------------RepartitionExec: partitioning=RoundRobinBatch(4),
input_partitions=1
09)----------------DataSourceExec: partitions=1, partition_sizes=[2]
10)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
diff --git a/datafusion/sqllogictest/test_files/tpch/plans/q1.slt.part
b/datafusion/sqllogictest/test_files/tpch/plans/q1.slt.part
index 7e3617b1d5..10c229546b 100644
--- a/datafusion/sqllogictest/test_files/tpch/plans/q1.slt.part
+++ b/datafusion/sqllogictest/test_files/tpch/plans/q1.slt.part
@@ -50,9 +50,9 @@ physical_plan
01)SortPreservingMergeExec: [l_returnflag@0 ASC NULLS LAST, l_linestatus@1 ASC
NULLS LAST]
02)--SortExec: expr=[l_returnflag@0 ASC NULLS LAST, l_linestatus@1 ASC NULLS
LAST], preserve_partitioning=[true]
03)----ProjectionExec: expr=[l_returnflag@0 as l_returnflag, l_linestatus@1 as
l_linestatus, sum(lineitem.l_quantity)@2 as sum_qty,
sum(lineitem.l_extendedprice)@3 as sum_base_price, sum(lineitem.l_extendedprice
* Int64(1) - lineitem.l_discount)@4 as sum_disc_price,
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount * Int64(1) +
lineitem.l_tax)@5 as sum_charge, avg(lineitem.l_quantity)@6 as avg_qty,
avg(lineitem.l_extendedprice)@7 as avg_price, avg(lineitem.l_discount)@8 as av
[...]
-04)------AggregateExec: mode=FinalPartitioned, gby=[l_returnflag@0 as
l_returnflag, l_linestatus@1 as l_linestatus], aggr=[sum(lineitem.l_quantity),
sum(lineitem.l_extendedprice), sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount), sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount * Int64(1) + lineitem.l_tax), avg(lineitem.l_quantity),
avg(lineitem.l_extendedprice), avg(lineitem.l_discount), count(Int64(1))]
+04)------AggregateExec: mode=FinalPartitioned, gby=[l_returnflag@0 as
l_returnflag, l_linestatus@1 as l_linestatus], aggr=[sum(lineitem.l_quantity),
sum(lineitem.l_extendedprice), sum(__common_expr_1) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount),
sum(__common_expr_1 * Some(1),20,0 + lineitem.l_tax) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount * Int64(1) +
lineitem.l_tax), avg(lineitem.l_quantity), avg(lineitem.l_extendedprice),
avg(lineitem.l_disc [...]
05)--------RepartitionExec: partitioning=Hash([l_returnflag@0,
l_linestatus@1], 4), input_partitions=4
-06)----------AggregateExec: mode=Partial, gby=[l_returnflag@5 as l_returnflag,
l_linestatus@6 as l_linestatus], aggr=[sum(lineitem.l_quantity),
sum(lineitem.l_extendedprice), sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount), sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount * Int64(1) + lineitem.l_tax), avg(lineitem.l_quantity),
avg(lineitem.l_extendedprice), avg(lineitem.l_discount), count(Int64(1))]
+06)----------AggregateExec: mode=Partial, gby=[l_returnflag@5 as l_returnflag,
l_linestatus@6 as l_linestatus], aggr=[sum(lineitem.l_quantity),
sum(lineitem.l_extendedprice), sum(__common_expr_1) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount),
sum(__common_expr_1 * Some(1),20,0 + lineitem.l_tax) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount * Int64(1) +
lineitem.l_tax), avg(lineitem.l_quantity), avg(lineitem.l_extendedprice),
avg(lineitem.l_discount) [...]
07)------------ProjectionExec: expr=[l_extendedprice@0 * (Some(1),20,0 -
l_discount@1) as __common_expr_1, l_quantity@2 as l_quantity, l_extendedprice@0
as l_extendedprice, l_discount@1 as l_discount, l_tax@3 as l_tax,
l_returnflag@4 as l_returnflag, l_linestatus@5 as l_linestatus]
08)--------------FilterExec: l_shipdate@6 <= 1998-09-02,
projection=[l_extendedprice@1, l_discount@2, l_quantity@0, l_tax@3,
l_returnflag@4, l_linestatus@5]
09)----------------DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:0..18561749],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:18561749..37123498],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:37123498..55685247],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:55685247..74246996]]},
projection=[l_quantity, l_extendedprice, l_discount, l_tax [...]
diff --git a/datafusion/sqllogictest/test_files/tpch/plans/q10.slt.part
b/datafusion/sqllogictest/test_files/tpch/plans/q10.slt.part
index 62649148bf..33d5e273a0 100644
--- a/datafusion/sqllogictest/test_files/tpch/plans/q10.slt.part
+++ b/datafusion/sqllogictest/test_files/tpch/plans/q10.slt.part
@@ -72,9 +72,9 @@ physical_plan
01)SortPreservingMergeExec: [revenue@2 DESC], fetch=10
02)--SortExec: TopK(fetch=10), expr=[revenue@2 DESC],
preserve_partitioning=[true]
03)----ProjectionExec: expr=[c_custkey@0 as c_custkey, c_name@1 as c_name,
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)@7 as revenue,
c_acctbal@2 as c_acctbal, n_name@4 as n_name, c_address@5 as c_address,
c_phone@3 as c_phone, c_comment@6 as c_comment]
-04)------AggregateExec: mode=FinalPartitioned, gby=[c_custkey@0 as c_custkey,
c_name@1 as c_name, c_acctbal@2 as c_acctbal, c_phone@3 as c_phone, n_name@4 as
n_name, c_address@5 as c_address, c_comment@6 as c_comment],
aggr=[sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
+04)------AggregateExec: mode=FinalPartitioned, gby=[c_custkey@0 as c_custkey,
c_name@1 as c_name, c_acctbal@2 as c_acctbal, c_phone@3 as c_phone, n_name@4 as
n_name, c_address@5 as c_address, c_comment@6 as c_comment],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
05)--------RepartitionExec: partitioning=Hash([c_custkey@0, c_name@1,
c_acctbal@2, c_phone@3, n_name@4, c_address@5, c_comment@6], 4),
input_partitions=4
-06)----------AggregateExec: mode=Partial, gby=[c_custkey@0 as c_custkey,
c_name@1 as c_name, c_acctbal@4 as c_acctbal, c_phone@3 as c_phone, n_name@8 as
n_name, c_address@2 as c_address, c_comment@5 as c_comment],
aggr=[sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
+06)----------AggregateExec: mode=Partial, gby=[c_custkey@0 as c_custkey,
c_name@1 as c_name, c_acctbal@4 as c_acctbal, c_phone@3 as c_phone, n_name@8 as
n_name, c_address@2 as c_address, c_comment@5 as c_comment],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
07)------------HashJoinExec: mode=Partitioned, join_type=Inner,
on=[(c_nationkey@3, n_nationkey@0)], projection=[c_custkey@0, c_name@1,
c_address@2, c_phone@4, c_acctbal@5, c_comment@6, l_extendedprice@7,
l_discount@8, n_name@10]
08)--------------RepartitionExec: partitioning=Hash([c_nationkey@3], 4),
input_partitions=4
09)----------------HashJoinExec: mode=Partitioned, join_type=Inner,
on=[(o_orderkey@7, l_orderkey@0)], projection=[c_custkey@0, c_name@1,
c_address@2, c_nationkey@3, c_phone@4, c_acctbal@5, c_comment@6,
l_extendedprice@9, l_discount@10]
diff --git a/datafusion/sqllogictest/test_files/tpch/plans/q12.slt.part
b/datafusion/sqllogictest/test_files/tpch/plans/q12.slt.part
index b152fde02f..84a6598cb9 100644
--- a/datafusion/sqllogictest/test_files/tpch/plans/q12.slt.part
+++ b/datafusion/sqllogictest/test_files/tpch/plans/q12.slt.part
@@ -62,9 +62,9 @@ physical_plan
01)SortPreservingMergeExec: [l_shipmode@0 ASC NULLS LAST]
02)--SortExec: expr=[l_shipmode@0 ASC NULLS LAST], preserve_partitioning=[true]
03)----ProjectionExec: expr=[l_shipmode@0 as l_shipmode, sum(CASE WHEN
orders.o_orderpriority = Utf8("1-URGENT") OR orders.o_orderpriority =
Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END)@1 as high_line_count, sum(CASE
WHEN orders.o_orderpriority != Utf8("1-URGENT") AND orders.o_orderpriority !=
Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END)@2 as low_line_count]
-04)------AggregateExec: mode=FinalPartitioned, gby=[l_shipmode@0 as
l_shipmode], aggr=[sum(CASE WHEN orders.o_orderpriority = Utf8("1-URGENT") OR
orders.o_orderpriority = Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END),
sum(CASE WHEN orders.o_orderpriority != Utf8("1-URGENT") AND
orders.o_orderpriority != Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END)]
+04)------AggregateExec: mode=FinalPartitioned, gby=[l_shipmode@0 as
l_shipmode], aggr=[sum(CASE WHEN orders.o_orderpriority = 1-URGENT OR
orders.o_orderpriority = 2-HIGH THEN 1 ELSE 0 END) as sum(CASE WHEN
orders.o_orderpriority = Utf8("1-URGENT") OR orders.o_orderpriority =
Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END), sum(CASE WHEN
orders.o_orderpriority != 1-URGENT AND orders.o_orderpriority != 2-HIGH THEN 1
ELSE 0 END) as sum(CASE WHEN orders.o_orderpriority != Utf8("1-URGENT") AN [...]
05)--------RepartitionExec: partitioning=Hash([l_shipmode@0], 4),
input_partitions=4
-06)----------AggregateExec: mode=Partial, gby=[l_shipmode@0 as l_shipmode],
aggr=[sum(CASE WHEN orders.o_orderpriority = Utf8("1-URGENT") OR
orders.o_orderpriority = Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END),
sum(CASE WHEN orders.o_orderpriority != Utf8("1-URGENT") AND
orders.o_orderpriority != Utf8("2-HIGH") THEN Int64(1) ELSE Int64(0) END)]
+06)----------AggregateExec: mode=Partial, gby=[l_shipmode@0 as l_shipmode],
aggr=[sum(CASE WHEN orders.o_orderpriority = 1-URGENT OR orders.o_orderpriority
= 2-HIGH THEN 1 ELSE 0 END) as sum(CASE WHEN orders.o_orderpriority =
Utf8("1-URGENT") OR orders.o_orderpriority = Utf8("2-HIGH") THEN Int64(1) ELSE
Int64(0) END), sum(CASE WHEN orders.o_orderpriority != 1-URGENT AND
orders.o_orderpriority != 2-HIGH THEN 1 ELSE 0 END) as sum(CASE WHEN
orders.o_orderpriority != Utf8("1-URGENT") AND ord [...]
07)------------HashJoinExec: mode=Partitioned, join_type=Inner,
on=[(l_orderkey@0, o_orderkey@0)], projection=[l_shipmode@1, o_orderpriority@3]
08)--------------RepartitionExec: partitioning=Hash([l_orderkey@0], 4),
input_partitions=4
09)----------------FilterExec: (l_shipmode@4 = MAIL OR l_shipmode@4 = SHIP)
AND l_receiptdate@3 > l_commitdate@2 AND l_shipdate@1 < l_commitdate@2 AND
l_receiptdate@3 >= 1994-01-01 AND l_receiptdate@3 < 1995-01-01,
projection=[l_orderkey@0, l_shipmode@4]
diff --git a/datafusion/sqllogictest/test_files/tpch/plans/q14.slt.part
b/datafusion/sqllogictest/test_files/tpch/plans/q14.slt.part
index a9ac517f28..198e6676f8 100644
--- a/datafusion/sqllogictest/test_files/tpch/plans/q14.slt.part
+++ b/datafusion/sqllogictest/test_files/tpch/plans/q14.slt.part
@@ -42,9 +42,9 @@ logical_plan
08)--------TableScan: part projection=[p_partkey, p_type]
physical_plan
01)ProjectionExec: expr=[100 * CAST(sum(CASE WHEN part.p_type LIKE
Utf8("PROMO%") THEN lineitem.l_extendedprice * Int64(1) - lineitem.l_discount
ELSE Int64(0) END)@0 AS Float64) / CAST(sum(lineitem.l_extendedprice * Int64(1)
- lineitem.l_discount)@1 AS Float64) as promo_revenue]
-02)--AggregateExec: mode=Final, gby=[], aggr=[sum(CASE WHEN part.p_type LIKE
Utf8("PROMO%") THEN lineitem.l_extendedprice * Int64(1) - lineitem.l_discount
ELSE Int64(0) END), sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)]
+02)--AggregateExec: mode=Final, gby=[], aggr=[sum(CASE WHEN part.p_type LIKE
PROMO% THEN __common_expr_1 ELSE Some(0),38,4 END) as sum(CASE WHEN part.p_type
LIKE Utf8("PROMO%") THEN lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount ELSE Int64(0) END), sum(__common_expr_1) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
03)----CoalescePartitionsExec
-04)------AggregateExec: mode=Partial, gby=[], aggr=[sum(CASE WHEN part.p_type
LIKE Utf8("PROMO%") THEN lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount ELSE Int64(0) END), sum(lineitem.l_extendedprice * Int64(1)
- lineitem.l_discount)]
+04)------AggregateExec: mode=Partial, gby=[], aggr=[sum(CASE WHEN part.p_type
LIKE PROMO% THEN __common_expr_1 ELSE Some(0),38,4 END) as sum(CASE WHEN
part.p_type LIKE Utf8("PROMO%") THEN lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount ELSE Int64(0) END), sum(__common_expr_1) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
05)--------ProjectionExec: expr=[l_extendedprice@0 * (Some(1),20,0 -
l_discount@1) as __common_expr_1, p_type@2 as p_type]
06)----------HashJoinExec: mode=Partitioned, join_type=Inner,
on=[(l_partkey@0, p_partkey@0)], projection=[l_extendedprice@1, l_discount@2,
p_type@4]
07)------------RepartitionExec: partitioning=Hash([l_partkey@0], 4),
input_partitions=4
diff --git a/datafusion/sqllogictest/test_files/tpch/plans/q15.slt.part
b/datafusion/sqllogictest/test_files/tpch/plans/q15.slt.part
index 3de29211bc..388e473c00 100644
--- a/datafusion/sqllogictest/test_files/tpch/plans/q15.slt.part
+++ b/datafusion/sqllogictest/test_files/tpch/plans/q15.slt.part
@@ -78,17 +78,17 @@ physical_plan
06)----------DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/supplier.tbl]]},
projection=[s_suppkey, s_name, s_address, s_phone], file_type=csv,
has_header=false
07)--------ProjectionExec: expr=[l_suppkey@0 as supplier_no,
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)@1 as
total_revenue]
08)----------FilterExec: sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)@1 = scalar_subquery(<pending>)
-09)------------AggregateExec: mode=FinalPartitioned, gby=[l_suppkey@0 as
l_suppkey], aggr=[sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)]
+09)------------AggregateExec: mode=FinalPartitioned, gby=[l_suppkey@0 as
l_suppkey], aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 -
lineitem.l_discount) as sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)]
10)--------------RepartitionExec: partitioning=Hash([l_suppkey@0], 4),
input_partitions=4
-11)----------------AggregateExec: mode=Partial, gby=[l_suppkey@0 as
l_suppkey], aggr=[sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)]
+11)----------------AggregateExec: mode=Partial, gby=[l_suppkey@0 as
l_suppkey], aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 -
lineitem.l_discount) as sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)]
12)------------------FilterExec: l_shipdate@3 >= 1996-01-01 AND l_shipdate@3 <
1996-04-01, projection=[l_suppkey@0, l_extendedprice@1, l_discount@2]
13)--------------------DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:0..18561749],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:18561749..37123498],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:37123498..55685247],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:55685247..74246996]]},
projection=[l_suppkey, l_extendedprice, l_discount, l_ [...]
14)--AggregateExec: mode=Final, gby=[], aggr=[max(revenue0.total_revenue)]
15)----CoalescePartitionsExec
16)------AggregateExec: mode=Partial, gby=[],
aggr=[max(revenue0.total_revenue)]
17)--------ProjectionExec: expr=[sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)@1 as total_revenue]
-18)----------AggregateExec: mode=FinalPartitioned, gby=[l_suppkey@0 as
l_suppkey], aggr=[sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)]
+18)----------AggregateExec: mode=FinalPartitioned, gby=[l_suppkey@0 as
l_suppkey], aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 -
lineitem.l_discount) as sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)]
19)------------RepartitionExec: partitioning=Hash([l_suppkey@0], 4),
input_partitions=4
-20)--------------AggregateExec: mode=Partial, gby=[l_suppkey@0 as l_suppkey],
aggr=[sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
+20)--------------AggregateExec: mode=Partial, gby=[l_suppkey@0 as l_suppkey],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
21)----------------FilterExec: l_shipdate@3 >= 1996-01-01 AND l_shipdate@3 <
1996-04-01, projection=[l_suppkey@0, l_extendedprice@1, l_discount@2]
22)------------------DataSourceExec: file_groups={4 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:0..18561749],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:18561749..37123498],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:37123498..55685247],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:55685247..74246996]]},
projection=[l_suppkey, l_extendedprice, l_discount, l_sh [...]
diff --git a/datafusion/sqllogictest/test_files/tpch/plans/q19.slt.part
b/datafusion/sqllogictest/test_files/tpch/plans/q19.slt.part
index 72c21e060f..8e38ca5a56 100644
--- a/datafusion/sqllogictest/test_files/tpch/plans/q19.slt.part
+++ b/datafusion/sqllogictest/test_files/tpch/plans/q19.slt.part
@@ -65,9 +65,9 @@ logical_plan
09)----------TableScan: part projection=[p_partkey, p_brand, p_size,
p_container], partial_filters=[part.p_size >= Int32(1), part.p_brand =
Utf8View("Brand#12") AND part.p_container IN ([Utf8View("SM CASE"),
Utf8View("SM BOX"), Utf8View("SM PACK"), Utf8View("SM PKG")]) AND part.p_size
<= Int32(5) OR part.p_brand = Utf8View("Brand#23") AND part.p_container IN
([Utf8View("MED BAG"), Utf8View("MED BOX"), Utf8View("MED PKG"), Utf8View("MED
PACK")]) AND part.p_size <= Int32(10) OR part.p_bran [...]
physical_plan
01)ProjectionExec: expr=[sum(lineitem.l_extendedprice * Int64(1) -
lineitem.l_discount)@0 as revenue]
-02)--AggregateExec: mode=Final, gby=[], aggr=[sum(lineitem.l_extendedprice *
Int64(1) - lineitem.l_discount)]
+02)--AggregateExec: mode=Final, gby=[], aggr=[sum(lineitem.l_extendedprice *
Some(1),20,0 - lineitem.l_discount) as sum(lineitem.l_extendedprice * Int64(1)
- lineitem.l_discount)]
03)----CoalescePartitionsExec
-04)------AggregateExec: mode=Partial, gby=[],
aggr=[sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
+04)------AggregateExec: mode=Partial, gby=[],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
05)--------HashJoinExec: mode=Partitioned, join_type=Inner, on=[(l_partkey@0,
p_partkey@0)], filter=p_brand@1 = Brand#12 AND p_container@3 IN (SET) ([SM
CASE, SM BOX, SM PACK, SM PKG]) AND l_quantity@0 >= Some(100),15,2 AND
l_quantity@0 <= Some(1100),15,2 AND p_size@2 <= 5 OR p_brand@1 = Brand#23 AND
p_container@3 IN (SET) ([MED BAG, MED BOX, MED PKG, MED PACK]) AND l_quantity@0
>= Some(1000),15,2 AND l_quantity@0 <= Some(2000),15,2 AND p_size@2 <= 10 OR
p_brand@1 = Brand#34 AND p_contai [...]
06)----------RepartitionExec: partitioning=Hash([l_partkey@0], 4),
input_partitions=4
07)------------FilterExec: (l_quantity@1 >= Some(100),15,2 AND l_quantity@1 <=
Some(1100),15,2 OR l_quantity@1 >= Some(1000),15,2 AND l_quantity@1 <=
Some(2000),15,2 OR l_quantity@1 >= Some(2000),15,2 AND l_quantity@1 <=
Some(3000),15,2) AND (l_shipmode@5 = AIR OR l_shipmode@5 = AIR REG) AND
l_shipinstruct@4 = DELIVER IN PERSON, projection=[l_partkey@0, l_quantity@1,
l_extendedprice@2, l_discount@3]
diff --git a/datafusion/sqllogictest/test_files/tpch/plans/q3.slt.part
b/datafusion/sqllogictest/test_files/tpch/plans/q3.slt.part
index 7fec4e5f5d..ba56f10fab 100644
--- a/datafusion/sqllogictest/test_files/tpch/plans/q3.slt.part
+++ b/datafusion/sqllogictest/test_files/tpch/plans/q3.slt.part
@@ -61,7 +61,7 @@ physical_plan
01)SortPreservingMergeExec: [revenue@1 DESC, o_orderdate@2 ASC NULLS LAST],
fetch=10
02)--SortExec: TopK(fetch=10), expr=[revenue@1 DESC, o_orderdate@2 ASC NULLS
LAST], preserve_partitioning=[true]
03)----ProjectionExec: expr=[l_orderkey@0 as l_orderkey,
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)@3 as revenue,
o_orderdate@1 as o_orderdate, o_shippriority@2 as o_shippriority]
-04)------AggregateExec: mode=SinglePartitioned, gby=[l_orderkey@2 as
l_orderkey, o_orderdate@0 as o_orderdate, o_shippriority@1 as o_shippriority],
aggr=[sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
+04)------AggregateExec: mode=SinglePartitioned, gby=[l_orderkey@2 as
l_orderkey, o_orderdate@0 as o_orderdate, o_shippriority@1 as o_shippriority],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
05)--------HashJoinExec: mode=Partitioned, join_type=Inner, on=[(o_orderkey@0,
l_orderkey@0)], projection=[o_orderdate@1, o_shippriority@2, l_orderkey@3,
l_extendedprice@4, l_discount@5]
06)----------RepartitionExec: partitioning=Hash([o_orderkey@0], 4),
input_partitions=4
07)------------HashJoinExec: mode=Partitioned, join_type=Inner,
on=[(c_custkey@0, o_custkey@1)], projection=[o_orderkey@1, o_orderdate@3,
o_shippriority@4]
diff --git a/datafusion/sqllogictest/test_files/tpch/plans/q5.slt.part
b/datafusion/sqllogictest/test_files/tpch/plans/q5.slt.part
index d854001f3c..bda0586963 100644
--- a/datafusion/sqllogictest/test_files/tpch/plans/q5.slt.part
+++ b/datafusion/sqllogictest/test_files/tpch/plans/q5.slt.part
@@ -70,9 +70,9 @@ physical_plan
01)SortPreservingMergeExec: [revenue@1 DESC]
02)--SortExec: expr=[revenue@1 DESC], preserve_partitioning=[true]
03)----ProjectionExec: expr=[n_name@0 as n_name, sum(lineitem.l_extendedprice
* Int64(1) - lineitem.l_discount)@1 as revenue]
-04)------AggregateExec: mode=FinalPartitioned, gby=[n_name@0 as n_name],
aggr=[sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
+04)------AggregateExec: mode=FinalPartitioned, gby=[n_name@0 as n_name],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
05)--------RepartitionExec: partitioning=Hash([n_name@0], 4),
input_partitions=4
-06)----------AggregateExec: mode=Partial, gby=[n_name@2 as n_name],
aggr=[sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
+06)----------AggregateExec: mode=Partial, gby=[n_name@2 as n_name],
aggr=[sum(lineitem.l_extendedprice * Some(1),20,0 - lineitem.l_discount) as
sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
07)------------HashJoinExec: mode=Partitioned, join_type=Inner,
on=[(n_regionkey@3, r_regionkey@0)], projection=[l_extendedprice@0,
l_discount@1, n_name@2]
08)--------------RepartitionExec: partitioning=Hash([n_regionkey@3], 4),
input_partitions=4
09)----------------HashJoinExec: mode=Partitioned, join_type=Inner,
on=[(s_nationkey@2, n_nationkey@0)], projection=[l_extendedprice@0,
l_discount@1, n_name@4, n_regionkey@5]
diff --git a/datafusion/sqllogictest/test_files/tpch/plans/q8.slt.part
b/datafusion/sqllogictest/test_files/tpch/plans/q8.slt.part
index 7f160ce072..82de61c60b 100644
--- a/datafusion/sqllogictest/test_files/tpch/plans/q8.slt.part
+++ b/datafusion/sqllogictest/test_files/tpch/plans/q8.slt.part
@@ -93,9 +93,9 @@ physical_plan
01)SortPreservingMergeExec: [o_year@0 ASC NULLS LAST]
02)--SortExec: expr=[o_year@0 ASC NULLS LAST], preserve_partitioning=[true]
03)----ProjectionExec: expr=[o_year@0 as o_year, CAST(CAST(sum(CASE WHEN
all_nations.nation = Utf8("BRAZIL") THEN all_nations.volume ELSE Int64(0)
END)@1 AS Decimal128(12, 2)) / CAST(sum(all_nations.volume)@2 AS Decimal128(12,
2)) AS Decimal128(15, 2)) as mkt_share]
-04)------AggregateExec: mode=FinalPartitioned, gby=[o_year@0 as o_year],
aggr=[sum(CASE WHEN all_nations.nation = Utf8("BRAZIL") THEN all_nations.volume
ELSE Int64(0) END), sum(all_nations.volume)]
+04)------AggregateExec: mode=FinalPartitioned, gby=[o_year@0 as o_year],
aggr=[sum(CASE WHEN all_nations.nation = BRAZIL THEN all_nations.volume ELSE
Some(0),38,4 END) as sum(CASE WHEN all_nations.nation = Utf8("BRAZIL") THEN
all_nations.volume ELSE Int64(0) END), sum(all_nations.volume)]
05)--------RepartitionExec: partitioning=Hash([o_year@0], 4),
input_partitions=4
-06)----------AggregateExec: mode=Partial, gby=[o_year@0 as o_year],
aggr=[sum(CASE WHEN all_nations.nation = Utf8("BRAZIL") THEN all_nations.volume
ELSE Int64(0) END), sum(all_nations.volume)]
+06)----------AggregateExec: mode=Partial, gby=[o_year@0 as o_year],
aggr=[sum(CASE WHEN all_nations.nation = BRAZIL THEN all_nations.volume ELSE
Some(0),38,4 END) as sum(CASE WHEN all_nations.nation = Utf8("BRAZIL") THEN
all_nations.volume ELSE Int64(0) END), sum(all_nations.volume)]
07)------------ProjectionExec: expr=[date_part(YEAR, o_orderdate@0) as o_year,
l_extendedprice@1 * (Some(1),20,0 - l_discount@2) as volume, n_name@3 as nation]
08)--------------HashJoinExec: mode=Partitioned, join_type=Inner,
on=[(n_regionkey@3, r_regionkey@0)], projection=[o_orderdate@2,
l_extendedprice@0, l_discount@1, n_name@4]
09)----------------RepartitionExec: partitioning=Hash([n_regionkey@3], 4),
input_partitions=4
diff --git a/docs/source/library-user-guide/upgrading/54.0.0.md
b/docs/source/library-user-guide/upgrading/54.0.0.md
index 46b768e834..0117a776b2 100644
--- a/docs/source/library-user-guide/upgrading/54.0.0.md
+++ b/docs/source/library-user-guide/upgrading/54.0.0.md
@@ -25,6 +25,73 @@
in this section pertains to features and changes that have already been merged
to the main branch and are awaiting release in this version.
+### `AggregateFunctionExpr::human_display()` now returns `Option<&str>`
+
+`datafusion_physical_expr::aggregate::AggregateFunctionExpr::human_display()`
+now returns `Option<&str>` instead of `&str`.
+
+If your code read the display text directly, handle the `None` case and fall
+back to `name()` when needed:
+
+```rust
+let display = agg_expr.human_display().unwrap_or(agg_expr.name());
+```
+
+### Aggregate logical-to-physical lowering helpers are deprecated
+
+`create_aggregate_expr_with_name_and_maybe_filter` and
+`create_aggregate_expr_and_maybe_filter` are deprecated. Use
+`datafusion_physical_expr::aggregate::LoweredAggregateBuilder` for new code
that
+lowers a logical aggregate `Expr` into an `AggregateFunctionExpr`, filter, and
+order-by expressions.
+
+For example:
+
+```rust
+let lowered = LoweredAggregateBuilder::new(
+ expr,
+ logical_input_schema,
+ physical_input_schema,
+ execution_props,
+)
+.build()?;
+```
+
+`LoweredAggregateBuilder` returns a `LoweredAggregate` containing the aggregate
+physical expression, optional filter, and order-by expressions.
+
+### `Expr::unalias_nested()` preserves aliases with metadata
+
+`Expr::unalias_nested()` no longer removes aliases that carry non-empty
+`FieldMetadata`. This preserves user-provided output field metadata. Code that
+needs to remove all aliases, including aliases with metadata, should unwrap
+`Expr::Alias` explicitly.
+
+### Physical aggregate proto display may contain encoded alias data
+
+`PhysicalAggregateExprNode.human_display` may now contain an internal encoded
+prefix when an aggregate display has a separate output alias. DataFusion
decodes
+this when reading physical plans. Older readers that do not know this encoding
+may show the prefix text directly in diagnostics.
+
+### Physical `EXPLAIN` now shows lowered aggregate execution forms
+
+Physical `EXPLAIN` output is intended for diagnostics and may change between
+DataFusion versions. This release changes aggregate expression formatting in
+physical plans to show the lowered expression executed by the engine while
+keeping the visible output alias.
+
+Examples:
+
+- `count(*)` may now appear as `count(1) as count(*)`
+- simplified aggregates may show the lowered implementation, such as
+ `min(...) as percentile_cont(...)`
+- internal aggregate aliases may now show the underlying expression instead of
+ only the alias name
+
+Tests or diagnostics that compare physical `EXPLAIN` output exactly may need
+to update their expected strings.
+
### String/numeric comparison coercion now prefers numeric types
Previously, comparing a numeric column with a string value (e.g.,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
