From f9ad1c350c4d8c86cc873c13bcb8d2baa2678841 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dani=C3=ABl=20Heres?= Date: Sun, 30 Aug 2026 08:49:37 +0200 Subject: [PATCH 1/2] fix: Unnest must not keep its input's uniqueness `Unnest::try_new` copied the input's functional dependencies unchanged, with the comment "We can use the existing functional dependencies". That is not true for a list unnest, which turns one input row into several. A determinant that occurred once in the input can occur many times in the output. Optimizer rules that read those dependencies then produce wrong results. With a declared key: CREATE TABLE t_list (k INT, vals INT[], PRIMARY KEY (k)) AS VALUES (1, [10, 20, 30]), (2, [40]); CREATE TABLE t_join (k INT) AS VALUES (1), (2); SELECT u.k, u.v FROM (SELECT k, unnest(vals) AS v FROM t_list) u JOIN t_join j ON u.k = j.k; returns 2 rows instead of 4, because `eliminate_join` sees `u` as unique on `k` and rewrites the inner join into a semi join, which drops the repeated rows. Without the PRIMARY KEY the same query returns 4. The dependency still holds in the weaker sense: all rows produced from one input row share the determinant, so it determines the same columns. Downgrade it to `Dependency::Multi` rather than dropping it, which keeps it useful and stops it being read as a uniqueness guarantee. Unnesting a struct produces one row per input row, so those dependencies are left as they are. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01SgqwvctZdvR1ZCz2hbkEJC --- datafusion/expr/src/logical_plan/plan.rs | 11 ++++- .../test_files/functional_dependencies.slt | 46 +++++++++++++++++++ 2 files changed, 55 insertions(+), 2 deletions(-) diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs index 1a8cd81aa74bd..80a00dc8b6a28 100644 --- a/datafusion/expr/src/logical_plan/plan.rs +++ b/datafusion/expr/src/logical_plan/plan.rs @@ -4896,8 +4896,15 @@ impl Unnest { let metadata = input_schema.metadata().clone(); let df_schema = DFSchema::new_with_metadata(fields, metadata)?; - // We can use the existing functional dependencies: - let deps = input_schema.functional_dependencies().clone(); + // Unnesting a list turns one input row into several, so a determinant + // that occurred once in the input can now occur many times. It still + // determines the same columns, so downgrade the dependency instead of + // dropping it. Unnesting a struct keeps one row per input row, and so + // keeps the dependencies as they are. + let mut deps = input_schema.functional_dependencies().clone(); + if !list_columns.is_empty() { + deps = deps.with_dependency(Dependency::Multi); + } let schema = Arc::new(df_schema.with_functional_dependencies(deps)?); Ok(Unnest { diff --git a/datafusion/sqllogictest/test_files/functional_dependencies.slt b/datafusion/sqllogictest/test_files/functional_dependencies.slt index c49004190dc60..566dbe923f2f5 100644 --- a/datafusion/sqllogictest/test_files/functional_dependencies.slt +++ b/datafusion/sqllogictest/test_files/functional_dependencies.slt @@ -296,6 +296,52 @@ drop table t_null; statement ok drop table t_probe; +########## +## Unnest +########## + +# Unnesting a list turns one row into several, so the key of the input no +# longer identifies a row of the output. Trusting it here made the join below +# a semi join, which dropped the repeated rows. +statement ok +CREATE TABLE t_list (k INT, vals INT[], PRIMARY KEY (k)) AS VALUES (1, [10, 20, 30]), (2, [40]); + +statement ok +CREATE TABLE t_join (k INT) AS VALUES (1), (2); + +query II +SELECT u.k, u.v FROM (SELECT k, unnest(vals) AS v FROM t_list) u +JOIN t_join j ON u.k = j.k +ORDER BY u.k, u.v; +---- +1 10 +1 20 +1 30 +2 40 + +# Unnesting a struct keeps one row per input row, so the key still holds. +statement ok +CREATE TABLE t_struct (k INT, s STRUCT, PRIMARY KEY (k)) AS VALUES (1, {'a': 1, 'b': 2}), (2, {'a': 3, 'b': 4}); + +query TT +EXPLAIN SELECT DISTINCT k FROM (SELECT k, unnest(s) FROM t_struct) t; +---- +logical_plan +01)SubqueryAlias: t +02)--Projection: t_struct.k +03)----Unnest: lists[] structs[__unnest_placeholder(t_struct.s)] +04)------Projection: t_struct.k, t_struct.s AS __unnest_placeholder(t_struct.s) +05)--------TableScan: t_struct projection=[k, s] + +statement ok +drop table t_list; + +statement ok +drop table t_join; + +statement ok +drop table t_struct; + ########## ## Cleanup ########## From 9988cb902ee12b47c69cbdafcb8f738f12e9495f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dani=C3=ABl=20Heres?= Date: Sun, 30 Aug 2026 09:02:22 +0200 Subject: [PATCH 2/2] perf: Remove an aggregate whose GROUP BY covers a unique key When the GROUP BY expressions include a unique key of the input, every group holds exactly one row. The aggregate then produces one output row per input row, like a projection, and each aggregate function returns the value of that single row. DataFusion did not use this. With lineitem keyed by (l_orderkey, l_linenumber): SELECT l_orderkey, l_linenumber, sum(l_quantity) FROM lineitem GROUP BY l_orderkey, l_linenumber still built a hash table over all 60M rows, one row per group. This extends `EliminateGroupByConstant`, which already simplifies an aggregate based on what its GROUP BY contains and runs at the right point, so it costs no extra plan traversal. The aggregate becomes a projection: Projection: l_orderkey, l_linenumber, CAST(l_quantity AS Decimal128(25, 2)) TableScan: lineitem On TPC-H SF10 with the key declared, summing that result goes from 751 ms to 31 ms. A DISTINCT over a unique key, which the planner turns into an aggregate with no aggregate expressions, goes from 51 ms to 19 ms. The single-row value of each aggregate is known for min, max, sum, avg, first_value and last_value (the argument, cast to the aggregate's return type where that differs) and count (1, or 0 when the argument is NULL). Anything else keeps the aggregate. DISTINCT, ORDER BY and IGNORE NULLS make no difference to one row, but a FILTER can exclude it and leave the aggregate with no input at all, so those are left alone. An empty GROUP BY is not eligible: it returns one row for an empty input, which a projection would not. Grouping sets are not eligible either, since they do not group by every expression at once. Of the benchmark suites, only TPC-DS q54 changes: it drops a DISTINCT over customer's primary key. Its runtime is unchanged at SF1. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01SgqwvctZdvR1ZCz2hbkEJC --- .../src/eliminate_group_by_constant.rs | 163 +++++++++++++++- .../sqllogictest/test_files/aggregate.slt | 43 ++--- .../sqllogictest/test_files/distinct_on.slt | 2 +- .../test_files/functional_dependencies.slt | 91 ++++++++- .../sqllogictest/test_files/group_by.slt | 174 ++++++------------ 5 files changed, 314 insertions(+), 159 deletions(-) diff --git a/datafusion/optimizer/src/eliminate_group_by_constant.rs b/datafusion/optimizer/src/eliminate_group_by_constant.rs index f0efe96668dba..e3bcfc23eb3b0 100644 --- a/datafusion/optimizer/src/eliminate_group_by_constant.rs +++ b/datafusion/optimizer/src/eliminate_group_by_constant.rs @@ -15,20 +15,44 @@ // specific language governing permissions and limitations // under the License. -//! [`EliminateGroupByConstant`] removes constant and functionally redundant -//! expressions from `GROUP BY` clause +//! [`EliminateGroupByConstant`] simplifies a `GROUP BY` clause, and removes the +//! aggregate altogether when the grouping makes it redundant. use crate::optimizer::ApplyOrder; use crate::{OptimizerConfig, OptimizerRule}; use std::collections::HashSet; +use std::sync::Arc; -use datafusion_common::Result; +use arrow::datatypes::DataType; use datafusion_common::tree_node::Transformed; -use datafusion_expr::{Aggregate, Expr, LogicalPlan, LogicalPlanBuilder, Volatility}; - -/// Optimizer rule that removes constant expressions from `GROUP BY` clause -/// and places additional projection on top of aggregation, to preserve -/// original schema +use datafusion_common::{DFSchema, Dependency, Result, ScalarValue}; +use datafusion_expr::expr::AggregateFunction; +use datafusion_expr::{ + Aggregate, Expr, ExprSchemable, LogicalPlan, LogicalPlanBuilder, Volatility, cast, + lit, when, +}; + +/// Optimizer rule that simplifies aggregation based on its `GROUP BY` clause. +/// +/// Constant expressions are removed from the `GROUP BY`, with a projection on +/// top to preserve the original schema. +/// +/// When the `GROUP BY` covers a unique key of the input, every group holds +/// exactly one row, so the aggregate produces one output row per input row like +/// a projection, and each aggregate function returns that row's value: +/// +/// ```text +/// -- lineitem is keyed by (l_orderkey, l_linenumber) +/// SELECT l_orderkey, l_linenumber, sum(l_quantity) +/// FROM lineitem GROUP BY l_orderkey, l_linenumber +/// +/// -- becomes +/// SELECT l_orderkey, l_linenumber, CAST(l_quantity AS Decimal128(38, 2)) +/// FROM lineitem +/// ``` +/// +/// That also removes a `SELECT DISTINCT` over a unique key, which the planner +/// turns into an aggregate with no aggregate expressions. #[derive(Default, Debug)] pub struct EliminateGroupByConstant {} @@ -50,6 +74,10 @@ impl OptimizerRule for EliminateGroupByConstant { ) -> Result> { match plan { LogicalPlan::Aggregate(aggregate) => { + if let Some(projection) = eliminate_aggregate(&aggregate)? { + return Ok(Transformed::yes(projection)); + } + // Collect bare column references in GROUP BY let group_by_columns: HashSet<&datafusion_common::Column> = aggregate .group_expr @@ -103,6 +131,125 @@ impl OptimizerRule for EliminateGroupByConstant { } } +/// Replaces an aggregate whose `GROUP BY` covers a unique key of its input with +/// a projection, or returns `None` when that cannot be done. +fn eliminate_aggregate(aggregate: &Aggregate) -> Result> { + // An empty GROUP BY produces one row even for an empty input, which a + // projection would not. Grouping sets do not group by every expression at + // once, so a key among them proves nothing. + if aggregate.group_expr.is_empty() + || aggregate + .group_expr + .iter() + .any(|expr| matches!(expr, Expr::GroupingSet(_))) + || !group_by_covers_unique_key(&aggregate.input, &aggregate.group_expr) + { + return Ok(None); + } + + // The output schema has to survive untouched, so each aggregate is replaced + // by an expression of the same type aliased to the same name. + let output_fields = aggregate.schema.fields(); + let mut projection = aggregate.group_expr.clone(); + for (index, aggr_expr) in aggregate.aggr_expr.iter().enumerate() { + let field = &output_fields[aggregate.group_expr.len() + index]; + let Some(value) = + single_row_value(aggr_expr, field.data_type(), aggregate.input.schema())? + else { + return Ok(None); + }; + projection.push(value.alias(field.name())); + } + + let input = Arc::clone(&aggregate.input); + Ok(Some( + LogicalPlanBuilder::from(Arc::unwrap_or_clone(input)) + .project(projection)? + .build()?, + )) +} + +/// Returns true when the GROUP BY expressions include a unique key of the +/// input, so that every group holds exactly one row. +/// +/// Extra grouping expressions beyond the key are harmless: they can only split +/// groups further, and the groups are already singletons. +fn group_by_covers_unique_key(input: &LogicalPlan, group_expr: &[Expr]) -> bool { + let schema = input.schema(); + let grouped: HashSet = group_expr + .iter() + .filter_map(|expr| match expr { + Expr::Alias(alias) => alias.expr.as_ref().try_as_col(), + _ => expr.try_as_col(), + }) + .filter_map(|column| schema.maybe_index_of_column(column)) + .collect(); + if grouped.is_empty() { + return false; + } + + schema.functional_dependencies().iter().any(|dependency| { + // A nullable key does not identify rows: two rows can both be NULL and + // are then not distinguished by the grouping either. + let nullable = dependency.nullable + && dependency + .source_indices + .iter() + .any(|&index| schema.field(index).is_nullable()); + !nullable + && dependency.mode == Dependency::Single + // The dependency has to determine the whole row, not just part of it. + && dependency.target_indices.len() == schema.fields().len() + && dependency + .source_indices + .iter() + .all(|index| grouped.contains(index)) + }) +} + +/// The value an aggregate takes over a group of exactly one row, as an +/// expression over that row, or `None` when it cannot be expressed. +fn single_row_value( + aggr_expr: &Expr, + output_type: &DataType, + input_schema: &DFSchema, +) -> Result> { + let Expr::AggregateFunction(AggregateFunction { func, params }) = aggr_expr else { + return Ok(None); + }; + // A FILTER can exclude the single row, leaving the aggregate with no input + // at all, which is a different value for every function. + if params.filter.is_some() { + return Ok(None); + } + // DISTINCT, ORDER BY and IGNORE NULLS all make no difference to a group of + // one row: there is nothing to deduplicate or order, and a single NULL is + // skipped by the functions below anyway. + let [arg] = params.args.as_slice() else { + return Ok(None); + }; + + Ok(match func.name() { + // Over one row these all return that row's value, in the type the + // aggregate would have returned. + "min" | "max" | "sum" | "avg" | "first_value" | "last_value" => { + Some(if &arg.get_type(input_schema)? == output_type { + arg.clone() + } else { + cast(arg.clone(), output_type.clone()) + }) + } + // COUNT ignores NULLs, so it is 1 unless the single value is NULL. + "count" => Some(if arg.nullable(input_schema)? { + when(arg.clone().is_null(), lit(ScalarValue::Int64(Some(0)))) + .otherwise(lit(ScalarValue::Int64(Some(1))))? + } else { + lit(ScalarValue::Int64(Some(1))) + }), + _ => None, + }) +} + /// Checks if a GROUP BY expression is redundant (can be removed without /// changing grouping semantics). An expression is redundant if it is a /// deterministic function of constants and columns already present as bare diff --git a/datafusion/sqllogictest/test_files/aggregate.slt b/datafusion/sqllogictest/test_files/aggregate.slt index 6ee38c48f5f1b..6cd10d57c2f7c 100644 --- a/datafusion/sqllogictest/test_files/aggregate.slt +++ b/datafusion/sqllogictest/test_files/aggregate.slt @@ -7221,17 +7221,15 @@ EXPLAIN SELECT DISTINCT c3, min(c1) FROM aggregate_test_100 group by c3 limit 5; ---- logical_plan 01)Limit: skip=0, fetch=5 -02)--Aggregate: groupBy=[[aggregate_test_100.c3, min(aggregate_test_100.c1)]], aggr=[[]] -03)----Aggregate: groupBy=[[aggregate_test_100.c3]], aggr=[[min(aggregate_test_100.c1)]] -04)------TableScan: aggregate_test_100 projection=[c1, c3] +02)--Aggregate: groupBy=[[aggregate_test_100.c3]], aggr=[[min(aggregate_test_100.c1)]] +03)----TableScan: aggregate_test_100 projection=[c1, c3] physical_plan 01)CoalescePartitionsExec: fetch=5 -02)--AggregateExec: mode=SinglePartitioned, gby=[c3@0 as c3, min(aggregate_test_100.c1)@1 as min(aggregate_test_100.c1)], aggr=[], lim=[5] -03)----AggregateExec: mode=FinalPartitioned, gby=[c3@0 as c3], aggr=[min(aggregate_test_100.c1)] -04)------RepartitionExec: partitioning=Hash([c3@0], 4), input_partitions=4 -05)--------AggregateExec: mode=Partial, gby=[c3@1 as c3], aggr=[min(aggregate_test_100.c1)] -06)----------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 -07)------------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100_with_dates.csv]]}, projection=[c1, c3], file_type=csv, has_header=true +02)--AggregateExec: mode=FinalPartitioned, gby=[c3@0 as c3], aggr=[min(aggregate_test_100.c1)] +03)----RepartitionExec: partitioning=Hash([c3@0], 4), input_partitions=4 +04)------AggregateExec: mode=Partial, gby=[c3@1 as c3], aggr=[min(aggregate_test_100.c1)] +05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 +06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100_with_dates.csv]]}, projection=[c1, c3], file_type=csv, has_header=true # @@ -7348,23 +7346,18 @@ query TT EXPLAIN SELECT DISTINCT c3, c2 FROM aggregate_test_100 group by c2, c3 limit 3 offset 10; ---- logical_plan -01)Limit: skip=10, fetch=3 -02)--Aggregate: groupBy=[[aggregate_test_100.c3, aggregate_test_100.c2]], aggr=[[]] -03)----Projection: aggregate_test_100.c3, aggregate_test_100.c2 -04)------Aggregate: groupBy=[[aggregate_test_100.c2, aggregate_test_100.c3]], aggr=[[]] -05)--------TableScan: aggregate_test_100 projection=[c2, c3] +01)Projection: aggregate_test_100.c3, aggregate_test_100.c2 +02)--Limit: skip=10, fetch=3 +03)----Aggregate: groupBy=[[aggregate_test_100.c2, aggregate_test_100.c3]], aggr=[[]] +04)------TableScan: aggregate_test_100 projection=[c2, c3] physical_plan -01)GlobalLimitExec: skip=10, fetch=3 -02)--AggregateExec: mode=Final, gby=[c3@0 as c3, c2@1 as c2], aggr=[], lim=[13] -03)----CoalescePartitionsExec -04)------AggregateExec: mode=Partial, gby=[c3@0 as c3, c2@1 as c2], aggr=[], lim=[13] -05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 -06)----------ProjectionExec: expr=[c3@1 as c3, c2@0 as c2] -07)------------AggregateExec: mode=Final, gby=[c2@0 as c2, c3@1 as c3], aggr=[] -08)--------------CoalescePartitionsExec -09)----------------AggregateExec: mode=Partial, gby=[c2@0 as c2, c3@1 as c3], aggr=[] -10)------------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 -11)--------------------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100_with_dates.csv]]}, projection=[c2, c3], file_type=csv, has_header=true +01)ProjectionExec: expr=[c3@1 as c3, c2@0 as c2] +02)--GlobalLimitExec: skip=10, fetch=3 +03)----AggregateExec: mode=Final, gby=[c2@0 as c2, c3@1 as c3], aggr=[], lim=[13] +04)------CoalescePartitionsExec +05)--------AggregateExec: mode=Partial, gby=[c2@0 as c2, c3@1 as c3], aggr=[], lim=[13] +06)----------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 +07)------------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100_with_dates.csv]]}, projection=[c2, c3], file_type=csv, has_header=true query II SELECT DISTINCT c3, c2 FROM aggregate_test_100 group by c3, c2 order by c3, c2 limit 3 offset 10; diff --git a/datafusion/sqllogictest/test_files/distinct_on.slt b/datafusion/sqllogictest/test_files/distinct_on.slt index 0659b9c208f9c..3298e01d4ba9a 100644 --- a/datafusion/sqllogictest/test_files/distinct_on.slt +++ b/datafusion/sqllogictest/test_files/distinct_on.slt @@ -397,7 +397,7 @@ FROM aggregate_test_100 GROUP BY c1 ORDER BY c1, agg2; logical_plan 01)Projection: first_value(aggregate_test_100.c1) ORDER BY [aggregate_test_100.c1 ASC NULLS LAST, max(aggregate_test_100.c4) ASC NULLS LAST] AS c1, first_value(agg2) ORDER BY [aggregate_test_100.c1 ASC NULLS LAST, max(aggregate_test_100.c4) ASC NULLS LAST] AS agg2 02)--Sort: aggregate_test_100.c1 ASC NULLS LAST -03)----Aggregate: groupBy=[[aggregate_test_100.c1]], aggr=[[first_value(aggregate_test_100.c1) ORDER BY [aggregate_test_100.c1 ASC NULLS LAST, max(aggregate_test_100.c4) ASC NULLS LAST], first_value(max(aggregate_test_100.c4) AS agg2) ORDER BY [aggregate_test_100.c1 ASC NULLS LAST, max(aggregate_test_100.c4) ASC NULLS LAST]]] +03)----Projection: aggregate_test_100.c1, aggregate_test_100.c1 AS first_value(aggregate_test_100.c1) ORDER BY [aggregate_test_100.c1 ASC NULLS LAST, max(aggregate_test_100.c4) ASC NULLS LAST], max(aggregate_test_100.c4) AS agg2 AS first_value(agg2) ORDER BY [aggregate_test_100.c1 ASC NULLS LAST, max(aggregate_test_100.c4) ASC NULLS LAST] 04)------Aggregate: groupBy=[[aggregate_test_100.c1]], aggr=[[max(aggregate_test_100.c4)]] 05)--------TableScan: aggregate_test_100 projection=[c1, c4] diff --git a/datafusion/sqllogictest/test_files/functional_dependencies.slt b/datafusion/sqllogictest/test_files/functional_dependencies.slt index 566dbe923f2f5..a4bb92e38fe0a 100644 --- a/datafusion/sqllogictest/test_files/functional_dependencies.slt +++ b/datafusion/sqllogictest/test_files/functional_dependencies.slt @@ -101,9 +101,8 @@ query TT EXPLAIN SELECT DISTINCT p.x FROM t_pk p LEFT JOIN t_orders o ON p.x = o.x; ---- logical_plan -01)Aggregate: groupBy=[[p.x]], aggr=[[]] -02)--SubqueryAlias: p -03)----TableScan: t_pk projection=[x] +01)SubqueryAlias: p +02)--TableScan: t_pk projection=[x] statement ok drop table t_orders; @@ -169,9 +168,7 @@ logical_plan query TT EXPLAIN SELECT x FROM t_pk GROUP BY x, y; ---- -logical_plan -01)Aggregate: groupBy=[[t_pk.x]], aggr=[[]] -02)--TableScan: t_pk projection=[x] +logical_plan TableScan: t_pk projection=[x] # 3.2 Nullable UNIQUE: grouping by `x, y` is NOT the same as grouping by # `x` -- two NULL rows differ in `y` and belong in separate groups. @@ -213,9 +210,7 @@ SELECT x, y FROM t_pk GROUP BY x; query TT EXPLAIN SELECT x, y FROM t_pk GROUP BY x; ---- -logical_plan -01)Aggregate: groupBy=[[t_pk.x, t_pk.y]], aggr=[[]] -02)--TableScan: t_pk projection=[x, y] +logical_plan TableScan: t_pk projection=[x, y] # 4.2 Nullable UNIQUE: `x` does NOT determine `y`, so there is no # well-defined `y` for the `x = NULL` group. @@ -296,6 +291,84 @@ drop table t_null; statement ok drop table t_probe; +########## +## Aggregate elimination +########## + +# GROUP BY covers a unique key, so every group holds one row and each aggregate +# takes that row's value. The aggregate is replaced by a projection. +statement ok +CREATE TABLE t_agg (k INT, v INT, s TEXT, PRIMARY KEY (k)) AS VALUES +(1, 10, 'a'), +(2, NULL, NULL), +(3, 30, 'c'); + +query TT +EXPLAIN SELECT k, sum(v), min(v), max(s), avg(v), count(*), count(v) FROM t_agg GROUP BY k; +---- +logical_plan +01)Projection: t_agg.k, CAST(t_agg.v AS Int64) AS sum(t_agg.v), t_agg.v AS min(t_agg.v), t_agg.s AS max(t_agg.s), CAST(t_agg.v AS Float64) AS avg(t_agg.v), Int64(1) AS count(*), CASE WHEN t_agg.v IS NULL THEN Int64(0) ELSE Int64(1) END AS count(t_agg.v) +02)--TableScan: t_agg projection=[k, v, s] + +# NULLs keep the semantics of each aggregate: count skips them, the rest return +# NULL. +query IIITRII +SELECT k, sum(v), min(v), max(s), avg(v), count(*), count(v) FROM t_agg GROUP BY k ORDER BY k; +---- +1 10 10 a 10 1 1 +2 NULL NULL NULL NULL 1 0 +3 30 30 c 30 1 1 + +# Extra grouping expressions beyond the key only split groups further, and they +# are already single rows. +query TT +EXPLAIN SELECT k, v, count(*) FROM t_agg GROUP BY k, v; +---- +logical_plan +01)Projection: t_agg.k, t_agg.v, Int64(1) AS count(*) +02)--TableScan: t_agg projection=[k, v] + +# A FILTER can exclude the single row, leaving the aggregate with no input at +# all, so the aggregate stays. +query TT +EXPLAIN SELECT k, count(*) FILTER (WHERE v > 20) FROM t_agg GROUP BY k; +---- +logical_plan +01)Projection: t_agg.k, count(Int64(1)) FILTER (WHERE t_agg.v > Int64(20)) AS count(*) FILTER (WHERE t_agg.v > Int64(20)) +02)--Aggregate: groupBy=[[t_agg.k]], aggr=[[count(Int64(1)) FILTER (WHERE t_agg.v > Int32(20)) AS count(Int64(1)) FILTER (WHERE t_agg.v > Int64(20))]] +03)----TableScan: t_agg projection=[k, v] + +# Grouping sets do not group by every expression at once, so a key among them +# proves nothing. +query TT +EXPLAIN SELECT k, v, count(*) FROM t_agg GROUP BY ROLLUP(k, v); +---- +logical_plan +01)Projection: t_agg.k, t_agg.v, count(Int64(1)) AS count(*) +02)--Aggregate: groupBy=[[ROLLUP (t_agg.k, t_agg.v)]], aggr=[[count(Int64(1))]] +03)----TableScan: t_agg projection=[k, v] + +# An aggregate whose single-row value cannot be written as an expression keeps +# the aggregate. +query TT +EXPLAIN SELECT k, array_agg(v) FROM t_agg GROUP BY k; +---- +logical_plan +01)Aggregate: groupBy=[[t_agg.k]], aggr=[[array_agg(t_agg.v)]] +02)--TableScan: t_agg projection=[k, v] + +# Grouping by a non-key column keeps the aggregate. +query TT +EXPLAIN SELECT v, count(*) FROM t_agg GROUP BY v; +---- +logical_plan +01)Projection: t_agg.v, count(Int64(1)) AS count(*) +02)--Aggregate: groupBy=[[t_agg.v]], aggr=[[count(Int64(1))]] +03)----TableScan: t_agg projection=[v] + +statement ok +drop table t_agg; + ########## ## Unnest ########## diff --git a/datafusion/sqllogictest/test_files/group_by.slt b/datafusion/sqllogictest/test_files/group_by.slt index 38d1b7821451d..edae333d2223c 100644 --- a/datafusion/sqllogictest/test_files/group_by.slt +++ b/datafusion/sqllogictest/test_files/group_by.slt @@ -3411,18 +3411,14 @@ EXPLAIN SELECT s.sn, s.amount, 2*s.sn logical_plan 01)Sort: s.sn ASC NULLS LAST 02)--Projection: s.sn, s.amount, Int64(2) * CAST(s.sn AS Int64) -03)----Aggregate: groupBy=[[s.sn, s.amount]], aggr=[[]] -04)------SubqueryAlias: s -05)--------TableScan: sales_global_with_pk projection=[sn, amount] +03)----SubqueryAlias: s +04)------TableScan: sales_global_with_pk projection=[sn, amount] physical_plan 01)SortPreservingMergeExec: [sn@0 ASC NULLS LAST] 02)--SortExec: expr=[sn@0 ASC NULLS LAST], preserve_partitioning=[true] 03)----ProjectionExec: expr=[sn@0 as sn, amount@1 as amount, 2 * CAST(sn@0 AS Int64) as Int64(2) * s.sn] -04)------AggregateExec: mode=FinalPartitioned, gby=[sn@0 as sn, amount@1 as amount], aggr=[] -05)--------RepartitionExec: partitioning=Hash([sn@0, amount@1], 8), input_partitions=8 -06)----------AggregateExec: mode=Partial, gby=[sn@0 as sn, amount@1 as amount], aggr=[] -07)------------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1 -08)--------------DataSourceExec: partitions=1, partition_sizes=[2] +04)------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1 +05)--------DataSourceExec: partitions=1, partition_sizes=[2] query IRI SELECT s.sn, s.amount, 2*s.sn @@ -3621,24 +3617,16 @@ EXPLAIN SELECT * ---- logical_plan 01)Sort: l.sn ASC NULLS LAST -02)--Projection: l.zip_code, l.country, l.sn, l.ts, l.currency, l.amount, l.sum_amount -03)----Aggregate: groupBy=[[l.sn, l.zip_code, l.country, l.ts, l.currency, l.amount, l.sum_amount]], aggr=[[]] -04)------SubqueryAlias: l -05)--------Projection: l.zip_code, l.country, l.sn, l.ts, l.currency, l.amount, sum(l.amount) ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING AS sum_amount -06)----------WindowAggr: windowExpr=[[sum(CAST(l.amount AS Float64)) ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING]] -07)------------SubqueryAlias: l -08)--------------TableScan: sales_global_with_pk projection=[zip_code, country, sn, ts, currency, amount] +02)--SubqueryAlias: l +03)----Projection: l.zip_code, l.country, l.sn, l.ts, l.currency, l.amount, sum(l.amount) ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING AS sum_amount +04)------WindowAggr: windowExpr=[[sum(CAST(l.amount AS Float64)) ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING]] +05)--------SubqueryAlias: l +06)----------TableScan: sales_global_with_pk projection=[zip_code, country, sn, ts, currency, amount] physical_plan -01)SortPreservingMergeExec: [sn@2 ASC NULLS LAST] -02)--ProjectionExec: expr=[zip_code@1 as zip_code, country@2 as country, sn@0 as sn, ts@3 as ts, currency@4 as currency, amount@5 as amount, sum_amount@6 as sum_amount] -03)----SortExec: expr=[sn@0 ASC NULLS LAST], preserve_partitioning=[true] -04)------AggregateExec: mode=FinalPartitioned, gby=[sn@0 as sn, zip_code@1 as zip_code, country@2 as country, ts@3 as ts, currency@4 as currency, amount@5 as amount, sum_amount@6 as sum_amount], aggr=[] -05)--------RepartitionExec: partitioning=Hash([sn@0, zip_code@1, country@2, ts@3, currency@4, amount@5, sum_amount@6], 8), input_partitions=8 -06)----------AggregateExec: mode=Partial, gby=[sn@2 as sn, zip_code@0 as zip_code, country@1 as country, ts@3 as ts, currency@4 as currency, amount@5 as amount, sum_amount@6 as sum_amount], aggr=[] -07)------------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1 -08)--------------ProjectionExec: expr=[zip_code@0 as zip_code, country@1 as country, sn@2 as sn, ts@3 as ts, currency@4 as currency, amount@5 as amount, sum(l.amount) ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING@6 as sum_amount] -09)----------------BoundedWindowAggExec: wdw=[sum(l.amount) ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING: Field { "sum(l.amount) ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING": nullable Float64 }, frame: ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING], mode=[Sorted] -10)------------------DataSourceExec: partitions=1, partition_sizes=[2] +01)ProjectionExec: expr=[zip_code@0 as zip_code, country@1 as country, sn@2 as sn, ts@3 as ts, currency@4 as currency, amount@5 as amount, sum(l.amount) ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING@6 as sum_amount] +02)--SortExec: expr=[sn@2 ASC NULLS LAST], preserve_partitioning=[false] +03)----BoundedWindowAggExec: wdw=[sum(l.amount) ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING: Field { "sum(l.amount) ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING": nullable Float64 }, frame: ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING], mode=[Sorted] +04)------DataSourceExec: partitions=1, partition_sizes=[2] query ITIPTRR @@ -3965,14 +3953,12 @@ FROM multiple_ordered_table_with_pk GROUP BY c; ---- logical_plan -01)Aggregate: groupBy=[[multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.b]], aggr=[[sum(CAST(multiple_ordered_table_with_pk.d AS Int64))]] +01)Projection: multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.b, CAST(multiple_ordered_table_with_pk.d AS Int64) AS sum(multiple_ordered_table_with_pk.d) 02)--TableScan: multiple_ordered_table_with_pk projection=[b, c, d] physical_plan -01)AggregateExec: mode=FinalPartitioned, gby=[c@0 as c, b@1 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=PartiallySorted([0]) -02)--RepartitionExec: partitioning=Hash([c@0, b@1], 8), input_partitions=8, preserve_order=true, sort_exprs=c@0 ASC NULLS LAST -03)----AggregateExec: mode=Partial, gby=[c@1 as c, b@0 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=PartiallySorted([0]) -04)------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1, maintains_sort_order=true -05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[b, c, d], output_ordering=[c@1 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true +01)ProjectionExec: expr=[c@1 as c, b@0 as b, CAST(d@2 AS Int64) as sum(multiple_ordered_table_with_pk.d)] +02)--RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1, maintains_sort_order=true +03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[b, c, d], output_ordering=[c@1 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true # drop table multiple_ordered_table_with_pk statement ok @@ -4004,14 +3990,12 @@ FROM multiple_ordered_table_with_pk GROUP BY c; ---- logical_plan -01)Aggregate: groupBy=[[multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.b]], aggr=[[sum(CAST(multiple_ordered_table_with_pk.d AS Int64))]] +01)Projection: multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.b, CAST(multiple_ordered_table_with_pk.d AS Int64) AS sum(multiple_ordered_table_with_pk.d) 02)--TableScan: multiple_ordered_table_with_pk projection=[b, c, d] physical_plan -01)AggregateExec: mode=FinalPartitioned, gby=[c@0 as c, b@1 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=PartiallySorted([0]) -02)--RepartitionExec: partitioning=Hash([c@0, b@1], 8), input_partitions=8, preserve_order=true, sort_exprs=c@0 ASC NULLS LAST -03)----AggregateExec: mode=Partial, gby=[c@1 as c, b@0 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=PartiallySorted([0]) -04)------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1, maintains_sort_order=true -05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[b, c, d], output_ordering=[c@1 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true +01)ProjectionExec: expr=[c@1 as c, b@0 as b, CAST(d@2 AS Int64) as sum(multiple_ordered_table_with_pk.d)] +02)--RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1, maintains_sort_order=true +03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[b, c, d], output_ordering=[c@1 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true statement ok set datafusion.execution.target_partitions = 1; @@ -4025,15 +4009,9 @@ EXPLAIN SELECT c, sum1 GROUP BY c; ---- logical_plan -01)Aggregate: groupBy=[[multiple_ordered_table_with_pk.c, sum1]], aggr=[[]] -02)--Projection: multiple_ordered_table_with_pk.c, sum(multiple_ordered_table_with_pk.d) AS sum1 -03)----Aggregate: groupBy=[[multiple_ordered_table_with_pk.c]], aggr=[[sum(CAST(multiple_ordered_table_with_pk.d AS Int64))]] -04)------TableScan: multiple_ordered_table_with_pk projection=[c, d] -physical_plan -01)AggregateExec: mode=Single, gby=[c@0 as c, sum1@1 as sum1], aggr=[], ordering_mode=PartiallySorted([0]) -02)--ProjectionExec: expr=[c@0 as c, sum(multiple_ordered_table_with_pk.d)@1 as sum1] -03)----AggregateExec: mode=Single, gby=[c@0 as c], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=Sorted -04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[c, d], output_ordering=[c@0 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true +01)Projection: multiple_ordered_table_with_pk.c, CAST(multiple_ordered_table_with_pk.d AS Int64) AS sum1 +02)--TableScan: multiple_ordered_table_with_pk projection=[c, d] +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[c, CAST(d@4 AS Int64) as sum1], output_ordering=[c@0 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true query TT EXPLAIN SELECT c, sum1, SUM(b) OVER() as sumb @@ -4045,15 +4023,12 @@ EXPLAIN SELECT c, sum1, SUM(b) OVER() as sumb logical_plan 01)Projection: multiple_ordered_table_with_pk.c, sum1, sum(multiple_ordered_table_with_pk.b) ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS sumb 02)--WindowAggr: windowExpr=[[sum(CAST(multiple_ordered_table_with_pk.b AS Int64)) ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]] -03)----Projection: multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.b, sum(multiple_ordered_table_with_pk.d) AS sum1 -04)------Aggregate: groupBy=[[multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.b]], aggr=[[sum(CAST(multiple_ordered_table_with_pk.d AS Int64))]] -05)--------TableScan: multiple_ordered_table_with_pk projection=[b, c, d] +03)----Projection: multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.b, CAST(multiple_ordered_table_with_pk.d AS Int64) AS sum1 +04)------TableScan: multiple_ordered_table_with_pk projection=[b, c, d] physical_plan 01)ProjectionExec: expr=[c@0 as c, sum1@2 as sum1, sum(multiple_ordered_table_with_pk.b) ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@3 as sumb] 02)--WindowAggExec: wdw=[sum(multiple_ordered_table_with_pk.b) ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "sum(multiple_ordered_table_with_pk.b) ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: Int64, nullable: true }), frame: WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound: Following(UInt64(NULL)), is_causal: false }] -03)----ProjectionExec: expr=[c@0 as c, b@1 as b, sum(multiple_ordered_table_with_pk.d)@2 as sum1] -04)------AggregateExec: mode=Single, gby=[c@1 as c, b@0 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=PartiallySorted([0]) -05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[b, c, d], output_ordering=[c@1 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true +03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[c, b, CAST(d@4 AS Int64) as sum1], output_ordering=[c@0 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true query TT EXPLAIN SELECT lhs.c, rhs.c, lhs.sum1, rhs.sum1 @@ -4071,21 +4046,15 @@ logical_plan 01)Projection: lhs.c, rhs.c, lhs.sum1, rhs.sum1 02)--Inner Join: lhs.b = rhs.b 03)----SubqueryAlias: lhs -04)------Projection: multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.b, sum(multiple_ordered_table_with_pk.d) AS sum1 -05)--------Aggregate: groupBy=[[multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.b]], aggr=[[sum(CAST(multiple_ordered_table_with_pk.d AS Int64))]] -06)----------TableScan: multiple_ordered_table_with_pk projection=[b, c, d] -07)----SubqueryAlias: rhs -08)------Projection: multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.b, sum(multiple_ordered_table_with_pk.d) AS sum1 -09)--------Aggregate: groupBy=[[multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.b]], aggr=[[sum(CAST(multiple_ordered_table_with_pk.d AS Int64))]] -10)----------TableScan: multiple_ordered_table_with_pk projection=[b, c, d] +04)------Projection: multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.b, CAST(multiple_ordered_table_with_pk.d AS Int64) AS sum1 +05)--------TableScan: multiple_ordered_table_with_pk projection=[b, c, d] +06)----SubqueryAlias: rhs +07)------Projection: multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.b, CAST(multiple_ordered_table_with_pk.d AS Int64) AS sum1 +08)--------TableScan: multiple_ordered_table_with_pk projection=[b, c, d] physical_plan 01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(b@1, b@1)], projection=[c@0, c@3, sum1@2, sum1@5] -02)--ProjectionExec: expr=[c@0 as c, b@1 as b, sum(multiple_ordered_table_with_pk.d)@2 as sum1] -03)----AggregateExec: mode=Single, gby=[c@1 as c, b@0 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=PartiallySorted([0]) -04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[b, c, d], output_ordering=[c@1 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true -05)--ProjectionExec: expr=[c@0 as c, b@1 as b, sum(multiple_ordered_table_with_pk.d)@2 as sum1] -06)----AggregateExec: mode=Single, gby=[c@1 as c, b@0 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=PartiallySorted([0]) -07)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[b, c, d], output_ordering=[c@1 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[c, b, CAST(d@4 AS Int64) as sum1], output_ordering=[c@0 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true +03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[c, b, CAST(d@4 AS Int64) as sum1], output_ordering=[c@0 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true query TT EXPLAIN SELECT lhs.c, rhs.c, lhs.sum1, rhs.sum1 @@ -4102,22 +4071,16 @@ logical_plan 01)Projection: lhs.c, rhs.c, lhs.sum1, rhs.sum1 02)--Cross Join: 03)----SubqueryAlias: lhs -04)------Projection: multiple_ordered_table_with_pk.c, sum(multiple_ordered_table_with_pk.d) AS sum1 -05)--------Aggregate: groupBy=[[multiple_ordered_table_with_pk.c]], aggr=[[sum(CAST(multiple_ordered_table_with_pk.d AS Int64))]] -06)----------TableScan: multiple_ordered_table_with_pk projection=[c, d] -07)----SubqueryAlias: rhs -08)------Projection: multiple_ordered_table_with_pk.c, sum(multiple_ordered_table_with_pk.d) AS sum1 -09)--------Aggregate: groupBy=[[multiple_ordered_table_with_pk.c]], aggr=[[sum(CAST(multiple_ordered_table_with_pk.d AS Int64))]] -10)----------TableScan: multiple_ordered_table_with_pk projection=[c, d] +04)------Projection: multiple_ordered_table_with_pk.c, CAST(multiple_ordered_table_with_pk.d AS Int64) AS sum1 +05)--------TableScan: multiple_ordered_table_with_pk projection=[c, d] +06)----SubqueryAlias: rhs +07)------Projection: multiple_ordered_table_with_pk.c, CAST(multiple_ordered_table_with_pk.d AS Int64) AS sum1 +08)--------TableScan: multiple_ordered_table_with_pk projection=[c, d] physical_plan 01)ProjectionExec: expr=[c@0 as c, c@2 as c, sum1@1 as sum1, sum1@3 as sum1] 02)--CrossJoinExec -03)----ProjectionExec: expr=[c@0 as c, sum(multiple_ordered_table_with_pk.d)@1 as sum1] -04)------AggregateExec: mode=Single, gby=[c@0 as c], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=Sorted -05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[c, d], output_ordering=[c@0 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true -06)----ProjectionExec: expr=[c@0 as c, sum(multiple_ordered_table_with_pk.d)@1 as sum1] -07)------AggregateExec: mode=Single, gby=[c@0 as c], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=Sorted -08)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[c, d], output_ordering=[c@0 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true +03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[c, CAST(d@4 AS Int64) as sum1], output_ordering=[c@0 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true +04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[c, CAST(d@4 AS Int64) as sum1], output_ordering=[c@0 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true # we do not generate physical plan for Repartition yet (e.g Distribute By queries). query TT @@ -4129,9 +4092,8 @@ DISTRIBUTE BY a ---- logical_plan 01)Repartition: DistributeBy(multiple_ordered_table_with_pk.a) -02)--Projection: multiple_ordered_table_with_pk.a, multiple_ordered_table_with_pk.b, sum(multiple_ordered_table_with_pk.d) AS sum1 -03)----Aggregate: groupBy=[[multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.a, multiple_ordered_table_with_pk.b]], aggr=[[sum(CAST(multiple_ordered_table_with_pk.d AS Int64))]] -04)------TableScan: multiple_ordered_table_with_pk projection=[a, b, c, d] +02)--Projection: multiple_ordered_table_with_pk.a, multiple_ordered_table_with_pk.b, CAST(multiple_ordered_table_with_pk.d AS Int64) AS sum1 +03)----TableScan: multiple_ordered_table_with_pk projection=[a, b, d] physical_plan_error This feature is not implemented: Physical plan does not support DistributeBy partitioning # union with aggregate @@ -4146,20 +4108,14 @@ UNION ALL ---- logical_plan 01)Union -02)--Projection: multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.a, sum(multiple_ordered_table_with_pk.d) AS sum1 -03)----Aggregate: groupBy=[[multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.a]], aggr=[[sum(CAST(multiple_ordered_table_with_pk.d AS Int64))]] -04)------TableScan: multiple_ordered_table_with_pk projection=[a, c, d] -05)--Projection: multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.a, sum(multiple_ordered_table_with_pk.d) AS sum1 -06)----Aggregate: groupBy=[[multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.a]], aggr=[[sum(CAST(multiple_ordered_table_with_pk.d AS Int64))]] -07)------TableScan: multiple_ordered_table_with_pk projection=[a, c, d] +02)--Projection: multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.a, CAST(multiple_ordered_table_with_pk.d AS Int64) AS sum1 +03)----TableScan: multiple_ordered_table_with_pk projection=[a, c, d] +04)--Projection: multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.a, CAST(multiple_ordered_table_with_pk.d AS Int64) AS sum1 +05)----TableScan: multiple_ordered_table_with_pk projection=[a, c, d] physical_plan 01)UnionExec -02)--ProjectionExec: expr=[c@0 as c, a@1 as a, sum(multiple_ordered_table_with_pk.d)@2 as sum1] -03)----AggregateExec: mode=Single, gby=[c@1 as c, a@0 as a], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=Sorted -04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, c, d], output_orderings=[[a@0 ASC NULLS LAST], [c@1 ASC NULLS LAST]], constraints=[PrimaryKey([3])], file_type=csv, has_header=true -05)--ProjectionExec: expr=[c@0 as c, a@1 as a, sum(multiple_ordered_table_with_pk.d)@2 as sum1] -06)----AggregateExec: mode=Single, gby=[c@1 as c, a@0 as a], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=Sorted -07)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, c, d], output_orderings=[[a@0 ASC NULLS LAST], [c@1 ASC NULLS LAST]], constraints=[PrimaryKey([3])], file_type=csv, has_header=true +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[c, a, CAST(d@4 AS Int64) as sum1], output_orderings=[[a@1 ASC NULLS LAST], [c@0 ASC NULLS LAST]], constraints=[PrimaryKey([3])], file_type=csv, has_header=true +03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[c, a, CAST(d@4 AS Int64) as sum1], output_orderings=[[a@1 ASC NULLS LAST], [c@0 ASC NULLS LAST]], constraints=[PrimaryKey([3])], file_type=csv, has_header=true # table scan should be simplified. query TT @@ -4168,13 +4124,9 @@ EXPLAIN SELECT c, a, SUM(d) as sum1 GROUP BY c ---- logical_plan -01)Projection: multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.a, sum(multiple_ordered_table_with_pk.d) AS sum1 -02)--Aggregate: groupBy=[[multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.a]], aggr=[[sum(CAST(multiple_ordered_table_with_pk.d AS Int64))]] -03)----TableScan: multiple_ordered_table_with_pk projection=[a, c, d] -physical_plan -01)ProjectionExec: expr=[c@0 as c, a@1 as a, sum(multiple_ordered_table_with_pk.d)@2 as sum1] -02)--AggregateExec: mode=Single, gby=[c@1 as c, a@0 as a], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=Sorted -03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, c, d], output_orderings=[[a@0 ASC NULLS LAST], [c@1 ASC NULLS LAST]], constraints=[PrimaryKey([3])], file_type=csv, has_header=true +01)Projection: multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.a, CAST(multiple_ordered_table_with_pk.d AS Int64) AS sum1 +02)--TableScan: multiple_ordered_table_with_pk projection=[a, c, d] +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[c, a, CAST(d@4 AS Int64) as sum1], output_orderings=[[a@1 ASC NULLS LAST], [c@0 ASC NULLS LAST]], constraints=[PrimaryKey([3])], file_type=csv, has_header=true # limit should be simplified query TT @@ -4185,15 +4137,10 @@ EXPLAIN SELECT * LIMIT 5) ---- logical_plan -01)Projection: multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.a, sum(multiple_ordered_table_with_pk.d) AS sum1 +01)Projection: multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.a, CAST(multiple_ordered_table_with_pk.d AS Int64) AS sum1 02)--Limit: skip=0, fetch=5 -03)----Aggregate: groupBy=[[multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.a]], aggr=[[sum(CAST(multiple_ordered_table_with_pk.d AS Int64))]] -04)------TableScan: multiple_ordered_table_with_pk projection=[a, c, d] -physical_plan -01)ProjectionExec: expr=[c@0 as c, a@1 as a, sum(multiple_ordered_table_with_pk.d)@2 as sum1] -02)--GlobalLimitExec: skip=0, fetch=5 -03)----AggregateExec: mode=Single, gby=[c@1 as c, a@0 as a], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=Sorted -04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, c, d], output_orderings=[[a@0 ASC NULLS LAST], [c@1 ASC NULLS LAST]], constraints=[PrimaryKey([3])], file_type=csv, has_header=true +03)----TableScan: multiple_ordered_table_with_pk projection=[a, c, d], fetch=5 +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[c, a, CAST(d@4 AS Int64) as sum1], limit=5, output_orderings=[[a@1 ASC NULLS LAST], [c@0 ASC NULLS LAST]], constraints=[PrimaryKey([3])], file_type=csv, has_header=true statement ok set datafusion.execution.target_partitions = 8; @@ -5673,14 +5620,9 @@ EXPLAIN SELECT DISTINCT u.id LEFT JOIN user_orders o ON u.id = o.user_id; ---- logical_plan -01)Aggregate: groupBy=[[u.id]], aggr=[[]] -02)--SubqueryAlias: u -03)----TableScan: users_with_pk projection=[id] -physical_plan -01)AggregateExec: mode=FinalPartitioned, gby=[id@0 as id], aggr=[] -02)--RepartitionExec: partitioning=Hash([id@0], 4), input_partitions=1 -03)----AggregateExec: mode=Partial, gby=[id@0 as id], aggr=[] -04)------DataSourceExec: partitions=1, partition_sizes=[1] +01)SubqueryAlias: u +02)--TableScan: users_with_pk projection=[id] +physical_plan DataSourceExec: partitions=1, partition_sizes=[1] statement ok drop table users_with_pk;