jayzhan211 commented on code in PR #27113:
URL: https://github.com/apache/datafusion/pull/27113#discussion_r4505235914
##########
datafusion/common/src/functional_dependencies.rs:
##########
@@ -486,42 +478,55 @@ pub fn aggregate_functional_dependencies(
&& source_indices.contains(input_idx)
// A key column can be in the GROUP BY list more than one
// time (`x, x AS k`). Count it one time.
- && !new_source_input_indices.contains(&Some(*input_idx))
+ && !grouped_source_indices.contains(&Some(*input_idx))
{
- new_source_indices.push(idx);
- new_source_input_indices.push(Some(*input_idx));
+ source_output_indices.push(idx);
+ grouped_source_indices.push(Some(*input_idx));
}
}
- let new_target_indices = get_target_indices_from_dependencies(
+ let determined_input_indices =
get_target_indices_from_dependencies(
func_dependencies.iter(),
- &new_source_input_indices,
- );
- let mode = if existing_target_indices == new_target_indices
- && new_target_indices.is_some()
- {
- // If dependency covers all GROUP BY expressions, mode will be
`Single`:
- Dependency::Single
+ &grouped_source_indices,
+ )
+ .unwrap_or_default();
+ // Dependencies are tracked for columns only, so a computed
+ // expression is never counted as determined, even when its inputs
+ // are.
+ let mut determined_output_indices = vec![];
+ for (output_idx, input_idx) in
group_by_input_indices.iter().enumerate() {
+ if let Some(input_idx) = input_idx
+ && determined_input_indices.contains(input_idx)
+ {
+ determined_output_indices.push(output_idx);
+ }
Review Comment:
As written, the PR turns queries that return correct results on 55.2.0 into
planning errors: any GROUP BY with a computed expression next to a key that
came through a join. For example, this returns `1 2` / `2 1` on 55.2.0 and main
and fails to plan with this PR:
```sql
SELECT i_item_sk, c
FROM (SELECT i_item_sk, upper(i_desc) AS u, count(*) AS c
FROM item JOIN sales ON i_item_sk = s_item_sk
GROUP BY i_item_sk, upper(i_desc)) q
GROUP BY i_item_sk;
```
`GROUP BY i_item_sk, i_item_sk + 1` and `GROUP BY i_item_sk, 'x'` fail the
same way. Proving these doesn't need dependencies on expressions. A
deterministic expression is determined whenever every column it reads is, so
it's enough to pass the columns each computed GROUP BY expression reads. With
the change below, the full sqllogictest suite passes, the three queries above
work again, and `s_date + 1` and `random()` are still treated as not
determined. It changes the signature of `aggregate_functional_dependencies`, so
it needs the `api change` label and a line in the upgrade guide. An
uncorrelated scalar subquery in GROUP BY still errors, because the subquery
check is conservative; correlated ones don't run on 55.2.0 either.
<details>
<summary>Change in functional_dependencies.rs and plan.rs</summary>
```diff
--- a/datafusion/common/src/functional_dependencies.rs
+++ b/datafusion/common/src/functional_dependencies.rs
@@ -445,9 +445,15 @@
/// even though its output name may equal the column's name.
+///
+/// `computed_input_indices[i]` lists the input fields that a deterministic
+/// computed GROUP BY expression `i` reads, or is `None` for a column
+/// reference or a volatile expression. A computed expression whose fields
+/// are all determined by a determinant is determined by it too.
pub fn aggregate_functional_dependencies(
aggr_input_schema: &DFSchema,
group_by_input_indices: &[Option<usize>],
+ computed_input_indices: &[Option<Vec<usize>>],
aggr_schema: &DFSchema,
) -> FunctionalDependencies {
@@ -489,14 +495,18 @@
- // Dependencies are tracked for columns only, so a computed
- // expression is never counted as determined, even when its
inputs
- // are.
+ // A deterministic computed expression is determined when every
+ // field it reads is.
let mut determined_output_indices = vec![];
for (output_idx, input_idx) in
group_by_input_indices.iter().enumerate() {
- if let Some(input_idx) = input_idx
- && determined_input_indices.contains(input_idx)
- {
+ let determined = match (input_idx,
&computed_input_indices[output_idx]) {
+ (Some(input_idx), _) =>
determined_input_indices.contains(input_idx),
+ (None, Some(read)) => read
+ .iter()
+ .all(|idx| determined_input_indices.contains(idx)),
+ (None, None) => false,
+ };
+ if determined {
determined_output_indices.push(output_idx);
}
}
--- a/datafusion/expr/src/logical_plan/plan.rs
+++ b/datafusion/expr/src/logical_plan/plan.rs
@@ -4529,13 +4529,23 @@ fn calc_func_dependencies_for_aggregate(
- let group_by_input_indices = grouping_set_to_exprlist(group_expr)?
- .into_iter()
+ let group_exprs = grouping_set_to_exprlist(group_expr)?;
+ let group_by_input_indices = group_exprs
+ .iter()
.map(|item| passthrough_field_index(item, input.schema()))
.collect::<Vec<_>>();
+ let computed_input_indices = group_exprs
+ .iter()
+ .zip(&group_by_input_indices)
+ .map(|(item, input_idx)| match input_idx {
+ Some(_) => None,
+ None => deterministic_input_indices(item, input.schema()),
+ })
+ .collect::<Vec<_>>();
let aggregate_func_dependencies = aggregate_functional_dependencies(
input.schema(),
&group_by_input_indices,
+ &computed_input_indices,
aggr_schema,
);
@@ -4544,6 +4554,30 @@
+/// Returns the input fields that `expr` reads, or `None` if `expr` is
volatile,
+/// contains a subquery or reads a column that is not an input field.
+fn deterministic_input_indices(expr: &Expr, schema: &DFSchema) ->
Option<Vec<usize>> {
+ let has_subquery = expr
+ .exists(|e| {
+ Ok(matches!(
+ e,
+ Expr::ScalarSubquery(_)
+ | Expr::Exists(_)
+ | Expr::InSubquery(_)
+ | Expr::SetComparison(_)
+ | Expr::OuterReferenceColumn(..)
+ ))
+ })
+ .unwrap_or(true);
+ if expr.is_volatile() || has_subquery {
+ return None;
+ }
+ expr.column_refs()
+ .into_iter()
+ .map(|column| schema.index_of_column(column).ok())
+ .collect()
+}
```
</details>
<details>
<summary>Tests to add after 8.4 (8.6 fails if every computed expression
counts as determined, 8.7 fails without the volatile check)</summary>
```
# 8.5 `upper(i_desc)` reads only `i_desc`, which `i_item_sk` determines, so
# `i_item_sk` is a key of the inner aggregate and `c` can be selected.
query II
SELECT i_item_sk, c
FROM (SELECT i_item_sk, upper(i_desc) AS u, count(*) AS c
FROM item JOIN sales ON i_item_sk = s_item_sk
GROUP BY i_item_sk, upper(i_desc)) q
GROUP BY i_item_sk ORDER BY i_item_sk;
----
1 2
2 1
# 8.6 `s_date + 1` reads `s_date`, which `i_item_sk` does not determine.
statement error
SELECT i_item_sk, c
FROM (SELECT i_item_sk, s_date + 1 AS d, count(*) AS c
FROM item JOIN sales ON i_item_sk = s_item_sk
GROUP BY i_item_sk, s_date + 1) q
GROUP BY i_item_sk;
----
DataFusion error: Error during planning: Column in SELECT must be in GROUP
BY or an aggregate function: While expanding wildcard, column "q.c" must appear
in the GROUP BY clause or must be part of an aggregate function, currently only
"q.i_item_sk" appears in the SELECT clause satisfies this requirement
# 8.7 A volatile expression is never determined: the join stays an inner
join.
query TT
EXPLAIN SELECT sum(o_amount) FROM orders
JOIN (SELECT i_item_sk FROM item JOIN sales ON i_item_sk = s_item_sk
GROUP BY i_item_sk, random() > 0.5) g
ON o_item_sk = g.i_item_sk;
----
logical_plan
01)Aggregate: groupBy=[[]], aggr=[[sum(CAST(orders.o_amount AS Int64))]]
02)--Projection: orders.o_amount
03)----Inner Join: orders.o_item_sk = g.i_item_sk
04)------TableScan: orders projection=[o_item_sk, o_amount]
05)------SubqueryAlias: g
06)--------Projection: item.i_item_sk
07)----------Aggregate: groupBy=[[item.i_item_sk, random() > Float64(0.5),
item.i_desc]], aggr=[[]]
08)------------Projection: item.i_item_sk, item.i_desc
09)--------------Inner Join: item.i_item_sk = sales.s_item_sk
10)----------------TableScan: item projection=[i_item_sk, i_desc]
11)----------------TableScan: sales projection=[s_item_sk]
```
</details>
##########
datafusion/sqllogictest/test_files/functional_dependencies.slt:
##########
@@ -706,6 +706,113 @@ logical_plan
04)------Aggregate: groupBy=[[t_uniq.x]], aggr=[[count(Int64(1))]]
05)--------TableScan: t_uniq projection=[x]
+##########
+## GROUP BY a key and a column the key does not determine
+##
+## `item` has PRIMARY KEY `i_item_sk`. Joined with `sales`, item 1 has two
+## dates, so `GROUP BY i_item_sk, s_date` forms the groups (1, 100), (1, 101)
+## and (2, 100): the pair is a key of the output, `i_item_sk` alone is not.
+##########
+
+statement ok
+CREATE TABLE item (i_item_sk INT, i_desc VARCHAR, PRIMARY KEY (i_item_sk))
+AS VALUES (1, 'a'), (2, 'b');
+
+statement ok
+CREATE TABLE sales (s_item_sk INT, s_date INT) AS VALUES (1, 100), (1, 101),
(2, 100);
+
+statement ok
+CREATE TABLE orders (o_item_sk INT, o_amount INT) AS VALUES (1, 10), (2, 20);
+
+statement ok
+CREATE VIEW item_dates AS
+SELECT i_item_sk, s_date FROM item JOIN sales ON i_item_sk = s_item_sk
+GROUP BY i_item_sk, s_date;
+
+# 8.1 Each order row joins once per matching group, so the order of item 1
+# counts twice. The join must stay an inner join, not become a semi join.
+query I
+SELECT sum(o_amount) FROM orders JOIN item_dates ON o_item_sk = i_item_sk;
+----
+40
+
+query TT
+EXPLAIN SELECT sum(o_amount) FROM orders JOIN item_dates ON o_item_sk =
i_item_sk;
+----
+logical_plan
+01)Aggregate: groupBy=[[]], aggr=[[sum(CAST(orders.o_amount AS Int64))]]
+02)--Projection: orders.o_amount
+03)----Inner Join: orders.o_item_sk = item_dates.i_item_sk
+04)------TableScan: orders projection=[o_item_sk, o_amount]
+05)------SubqueryAlias: item_dates
+06)--------Projection: item.i_item_sk
+07)----------Aggregate: groupBy=[[item.i_item_sk, sales.s_date]], aggr=[[]]
+08)------------Projection: item.i_item_sk, sales.s_date
+09)--------------Inner Join: item.i_item_sk = sales.s_item_sk
+10)----------------TableScan: item projection=[i_item_sk]
+11)----------------TableScan: sales projection=[s_item_sk, s_date]
+
+# 8.2 GROUP BY the key alone: `i_item_sk` is a key of the output, so the
+# join is still converted to a semi join.
+query I
+SELECT sum(o_amount) FROM orders
+JOIN (SELECT i_item_sk FROM item JOIN sales ON i_item_sk = s_item_sk GROUP BY
i_item_sk) f
+ ON o_item_sk = f.i_item_sk;
+----
+30
+
+query TT
+EXPLAIN SELECT sum(o_amount) FROM orders
+JOIN (SELECT i_item_sk FROM item JOIN sales ON i_item_sk = s_item_sk GROUP BY
i_item_sk) f
+ ON o_item_sk = f.i_item_sk;
+----
+logical_plan
+01)Aggregate: groupBy=[[]], aggr=[[sum(CAST(orders.o_amount AS Int64))]]
+02)--Projection: orders.o_amount
+03)----LeftSemi Join: orders.o_item_sk = f.i_item_sk
+04)------TableScan: orders projection=[o_item_sk, o_amount]
+05)------SubqueryAlias: f
+06)--------Aggregate: groupBy=[[item.i_item_sk]], aggr=[[]]
+07)----------LeftSemi Join: item.i_item_sk = sales.s_item_sk
+08)------------TableScan: item projection=[i_item_sk]
+09)------------TableScan: sales projection=[s_item_sk]
+
+# 8.3 `i_item_sk` does not determine `s_date`, so an outer GROUP BY must keep
Review Comment:
Optional: the issue also lists the DISTINCT and window-frame symptoms, and
no test covers them yet. Both of these fail on main (3 rows, and `1 1 / 1 2 / 2
3`) and pass with this PR. They could go right after 8.3:
```
# `i_item_sk` is not a key of `item_dates`: the DISTINCT must stay, and the
# default window frame must stay RANGE (peers of item 1 count together).
query I
SELECT DISTINCT i_item_sk FROM item_dates ORDER BY i_item_sk;
----
1
2
query II
SELECT i_item_sk, count(*) OVER (ORDER BY i_item_sk) FROM item_dates
ORDER BY i_item_sk;
----
1 2
1 2
2 3
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
