You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "UBarney (via GitHub)" <gi...@apache.org> on 2025/04/28 09:30:39 UTC
[PR] fix: Allow ORDER BY aggregates not present in SELECT list [datafusion]
UBarney opened a new pull request, #15876:
URL: https://github.com/apache/datafusion/pull/15876
## 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 #15875.
## Rationale for this change
<!--
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?
If `sorts.expr` contains agg_func not selected. We first add them to LogicalPlan:Aggreate , then let `sort.children` output this columns
<!--
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?
No
<!--
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.
-->
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Allow ORDER BY aggregates not present in SELECT list [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on code in PR #15876:
URL: https://github.com/apache/datafusion/pull/15876#discussion_r2077729484
##########
datafusion/expr/src/logical_plan/builder.rs:
##########
@@ -797,26 +807,146 @@ impl LogicalPlanBuilder {
}
// remove pushed down sort columns
- let new_expr = schema.columns().into_iter().map(Expr::Column).collect();
+ let sort_output = schema.columns().into_iter().map(Expr::Column).collect();
+
+ let plan = Arc::unwrap_or_clone(self.plan);
+
+ let (plan, agg_func_to_col, sorts) = if missing_agg_funcs.is_empty() {
+ (plan, HashMap::new(), sorts)
+ } else {
+ {
+ let (plan, agg_func_to_col) =
+ Self::add_missing_agg_funcs_to_logical_agg(plan, &missing_agg_funcs)?;
+
+ let sorts = sorts
+ .iter()
+ .map(|x| {
+ Self::replace_subexpr_to_col(&x.expr, &agg_func_to_col)
+ .map(|expr| x.with_expr(expr))
+ })
+ .collect::<Result<Vec<_>, _>>()?;
+
+ (plan, agg_func_to_col, sorts)
+ }
+ };
+
+ // we need downstream filter/project return missing col(agg_funcs)
+ missing_cols.extend(agg_func_to_col.into_values());
let is_distinct = false;
- let plan = Self::add_missing_columns(
- Arc::unwrap_or_clone(self.plan),
- &missing_cols,
- is_distinct,
- )?;
+ let plan = Self::add_missing_columns(plan, &missing_cols, is_distinct)?;
let sort_plan = LogicalPlan::Sort(Sort {
expr: normalize_sorts(sorts, &plan)?,
input: Arc::new(plan),
fetch,
});
- Projection::try_new(new_expr, Arc::new(sort_plan))
+ Projection::try_new(sort_output, Arc::new(sort_plan))
.map(LogicalPlan::Projection)
.map(Self::new)
}
+ fn replace_subexpr_to_col(
+ expr: &Expr,
+ func_to_col: &HashMap<Expr, Column>,
+ ) -> Result<Expr> {
+ Ok(expr
+ .clone()
+ .transform_down(|nested_expr| {
+ if let Some(col) = func_to_col.get(&nested_expr) {
+ Ok(Transformed::yes(Expr::Column(col.clone())))
+ } else {
+ Ok(Transformed::no(nested_expr))
+ }
+ })?
+ .data)
+ }
+
+ fn add_missing_agg_funcs_to_logical_agg(
+ plan: LogicalPlan,
+ missing_agg_funcs: &IndexSet<Expr>,
+ ) -> Result<(LogicalPlan, HashMap<Expr, Column>)> {
+ let mut agg_func_to_output_col: HashMap<Expr, Column> =
+ HashMap::with_capacity(missing_agg_funcs.len());
+
+ if missing_agg_funcs.is_empty() {
+ return Ok((plan, agg_func_to_output_col));
+ }
+
+ let plan = plan
+ .transform_down(|plan| {
Review Comment:
This query would pass in branches that don't contain the code changes introduced in this PR.
```
√ devhomeinsp ~/c/datafusion > ./target/release/datafusion-cli
DataFusion CLI v46.0.1
> create table t(a int);
0 row(s) fetched.
Elapsed 0.010 seconds.
> select * from t, (select max(a) from t) order by max(a);
+---+----------+
| a | max(t.a) |
+---+----------+
+---+----------+
0 row(s) fetched.
Elapsed 0.013 seconds.
> CREATE TABLE order_by_agg_func_not_selected (
g INT,
o INT
) AS VALUES
(1, -1),
(2, -2),
(3, -3);
0 row(s) fetched.
Elapsed 0.001 seconds.
> SELECT g, count(o) FROM order_by_agg_func_not_selected group by g order by min(o);
type_coercion
caused by
Schema error: No field named order_by_agg_func_not_selected.o. Valid fields are order_by_agg_func_not_selected.g, "count(order_by_agg_func_not_selected.o)".
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Allow ORDER BY aggregates not present in SELECT list [datafusion]
Posted by "jonahgao (via GitHub)" <gi...@apache.org>.
jonahgao commented on code in PR #15876:
URL: https://github.com/apache/datafusion/pull/15876#discussion_r2079814968
##########
datafusion/expr/src/logical_plan/builder.rs:
##########
@@ -797,26 +807,146 @@ impl LogicalPlanBuilder {
}
// remove pushed down sort columns
- let new_expr = schema.columns().into_iter().map(Expr::Column).collect();
+ let sort_output = schema.columns().into_iter().map(Expr::Column).collect();
+
+ let plan = Arc::unwrap_or_clone(self.plan);
+
+ let (plan, agg_func_to_col, sorts) = if missing_agg_funcs.is_empty() {
+ (plan, HashMap::new(), sorts)
+ } else {
+ {
+ let (plan, agg_func_to_col) =
+ Self::add_missing_agg_funcs_to_logical_agg(plan, &missing_agg_funcs)?;
+
+ let sorts = sorts
+ .iter()
+ .map(|x| {
+ Self::replace_subexpr_to_col(&x.expr, &agg_func_to_col)
+ .map(|expr| x.with_expr(expr))
+ })
+ .collect::<Result<Vec<_>, _>>()?;
+
+ (plan, agg_func_to_col, sorts)
+ }
+ };
+
+ // we need downstream filter/project return missing col(agg_funcs)
+ missing_cols.extend(agg_func_to_col.into_values());
let is_distinct = false;
- let plan = Self::add_missing_columns(
- Arc::unwrap_or_clone(self.plan),
- &missing_cols,
- is_distinct,
- )?;
+ let plan = Self::add_missing_columns(plan, &missing_cols, is_distinct)?;
let sort_plan = LogicalPlan::Sort(Sort {
expr: normalize_sorts(sorts, &plan)?,
input: Arc::new(plan),
fetch,
});
- Projection::try_new(new_expr, Arc::new(sort_plan))
+ Projection::try_new(sort_output, Arc::new(sort_plan))
.map(LogicalPlan::Projection)
.map(Self::new)
}
+ fn replace_subexpr_to_col(
+ expr: &Expr,
+ func_to_col: &HashMap<Expr, Column>,
+ ) -> Result<Expr> {
+ Ok(expr
+ .clone()
+ .transform_down(|nested_expr| {
+ if let Some(col) = func_to_col.get(&nested_expr) {
+ Ok(Transformed::yes(Expr::Column(col.clone())))
+ } else {
+ Ok(Transformed::no(nested_expr))
+ }
+ })?
+ .data)
+ }
+
+ fn add_missing_agg_funcs_to_logical_agg(
+ plan: LogicalPlan,
+ missing_agg_funcs: &IndexSet<Expr>,
+ ) -> Result<(LogicalPlan, HashMap<Expr, Column>)> {
+ let mut agg_func_to_output_col: HashMap<Expr, Column> =
+ HashMap::with_capacity(missing_agg_funcs.len());
+
+ if missing_agg_funcs.is_empty() {
+ return Ok((plan, agg_func_to_output_col));
+ }
+
+ let plan = plan
+ .transform_down(|plan| {
Review Comment:
We need to first check whether the current query has a GROUP BY or if there are other aggregate functions in the select list. If there are, we can add the missing aggregate functions. Otherwise, we should throw an error. The current approach does not correctly perform this check because it cannot confirm whether a `LogicalPlan::Aggregate` node in the plan comes from the SELECT list or somewhere else.
I created [another example](https://github.com/jonahgao/datafusion/commit/aec8ea022d87915fb23c14b0b7cb39fce59e779a) using DataFrame to illustrate this problem. In this example, the missing `min(a)` is added to the right side of the join. It's a confusing behavior for me.
```sh
Projection: b, test.a [b:Utf8, a:Utf8]
Sort: min(test.a) ASC NULLS LAST [b:Utf8, a:Utf8, min(test.a):Utf8;N]
Inner Join: b = test.a [b:Utf8, a:Utf8, min(test.a):Utf8;N]
Projection: test.a AS b [b:Utf8]
TableScan: test projection=[a] [a:Utf8]
Aggregate: groupBy=[[test.a]], aggr=[[min(test.a)]] [a:Utf8, min(test.a):Utf8;N]
TableScan: test projection=[a] [a:Utf8]
```
At an earlier stage, such as during select_to_plan, we are able to know the exact select list.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Allow ORDER BY aggregates not present in SELECT list [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on code in PR #15876:
URL: https://github.com/apache/datafusion/pull/15876#discussion_r2083447146
##########
datafusion/expr/src/logical_plan/builder.rs:
##########
@@ -797,26 +807,146 @@ impl LogicalPlanBuilder {
}
// remove pushed down sort columns
- let new_expr = schema.columns().into_iter().map(Expr::Column).collect();
+ let sort_output = schema.columns().into_iter().map(Expr::Column).collect();
+
+ let plan = Arc::unwrap_or_clone(self.plan);
+
+ let (plan, agg_func_to_col, sorts) = if missing_agg_funcs.is_empty() {
+ (plan, HashMap::new(), sorts)
+ } else {
+ {
+ let (plan, agg_func_to_col) =
+ Self::add_missing_agg_funcs_to_logical_agg(plan, &missing_agg_funcs)?;
+
+ let sorts = sorts
+ .iter()
+ .map(|x| {
+ Self::replace_subexpr_to_col(&x.expr, &agg_func_to_col)
+ .map(|expr| x.with_expr(expr))
+ })
+ .collect::<Result<Vec<_>, _>>()?;
+
+ (plan, agg_func_to_col, sorts)
+ }
+ };
+
+ // we need downstream filter/project return missing col(agg_funcs)
+ missing_cols.extend(agg_func_to_col.into_values());
let is_distinct = false;
- let plan = Self::add_missing_columns(
- Arc::unwrap_or_clone(self.plan),
- &missing_cols,
- is_distinct,
- )?;
+ let plan = Self::add_missing_columns(plan, &missing_cols, is_distinct)?;
let sort_plan = LogicalPlan::Sort(Sort {
expr: normalize_sorts(sorts, &plan)?,
input: Arc::new(plan),
fetch,
});
- Projection::try_new(new_expr, Arc::new(sort_plan))
+ Projection::try_new(sort_output, Arc::new(sort_plan))
.map(LogicalPlan::Projection)
.map(Self::new)
}
+ fn replace_subexpr_to_col(
+ expr: &Expr,
+ func_to_col: &HashMap<Expr, Column>,
+ ) -> Result<Expr> {
+ Ok(expr
+ .clone()
+ .transform_down(|nested_expr| {
+ if let Some(col) = func_to_col.get(&nested_expr) {
+ Ok(Transformed::yes(Expr::Column(col.clone())))
+ } else {
+ Ok(Transformed::no(nested_expr))
+ }
+ })?
+ .data)
+ }
+
+ fn add_missing_agg_funcs_to_logical_agg(
+ plan: LogicalPlan,
+ missing_agg_funcs: &IndexSet<Expr>,
+ ) -> Result<(LogicalPlan, HashMap<Expr, Column>)> {
+ let mut agg_func_to_output_col: HashMap<Expr, Column> =
+ HashMap::with_capacity(missing_agg_funcs.len());
+
+ if missing_agg_funcs.is_empty() {
+ return Ok((plan, agg_func_to_output_col));
+ }
+
+ let plan = plan
+ .transform_down(|plan| {
Review Comment:
I fix this problem by adding missing_agg_funcs to agg only when there are filter and project operators between sort and agg operators, otherwise report an error.
```
// OK
Sort: min(employee_csv.salary)
Filter
Project
Aggregate: groupBy=[[employee_csv.state]], aggr=[[max(employee_csv.salary)]]
// Error
Sort: min(employee_csv.salary)
Join
Aggregate
TableScan
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Allow ORDER BY aggregates not present in SELECT list [datafusion]
Posted by "jonahgao (via GitHub)" <gi...@apache.org>.
jonahgao commented on code in PR #15876:
URL: https://github.com/apache/datafusion/pull/15876#discussion_r2075625947
##########
datafusion/expr/src/logical_plan/builder.rs:
##########
@@ -797,26 +807,146 @@ impl LogicalPlanBuilder {
}
// remove pushed down sort columns
- let new_expr = schema.columns().into_iter().map(Expr::Column).collect();
+ let sort_output = schema.columns().into_iter().map(Expr::Column).collect();
+
+ let plan = Arc::unwrap_or_clone(self.plan);
+
+ let (plan, agg_func_to_col, sorts) = if missing_agg_funcs.is_empty() {
+ (plan, HashMap::new(), sorts)
+ } else {
+ {
+ let (plan, agg_func_to_col) =
+ Self::add_missing_agg_funcs_to_logical_agg(plan, &missing_agg_funcs)?;
+
+ let sorts = sorts
+ .iter()
+ .map(|x| {
+ Self::replace_subexpr_to_col(&x.expr, &agg_func_to_col)
+ .map(|expr| x.with_expr(expr))
+ })
+ .collect::<Result<Vec<_>, _>>()?;
+
+ (plan, agg_func_to_col, sorts)
+ }
+ };
+
+ // we need downstream filter/project return missing col(agg_funcs)
+ missing_cols.extend(agg_func_to_col.into_values());
let is_distinct = false;
- let plan = Self::add_missing_columns(
- Arc::unwrap_or_clone(self.plan),
- &missing_cols,
- is_distinct,
- )?;
+ let plan = Self::add_missing_columns(plan, &missing_cols, is_distinct)?;
let sort_plan = LogicalPlan::Sort(Sort {
expr: normalize_sorts(sorts, &plan)?,
input: Arc::new(plan),
fetch,
});
- Projection::try_new(new_expr, Arc::new(sort_plan))
+ Projection::try_new(sort_output, Arc::new(sort_plan))
.map(LogicalPlan::Projection)
.map(Self::new)
}
+ fn replace_subexpr_to_col(
+ expr: &Expr,
+ func_to_col: &HashMap<Expr, Column>,
+ ) -> Result<Expr> {
+ Ok(expr
+ .clone()
+ .transform_down(|nested_expr| {
+ if let Some(col) = func_to_col.get(&nested_expr) {
+ Ok(Transformed::yes(Expr::Column(col.clone())))
+ } else {
+ Ok(Transformed::no(nested_expr))
+ }
+ })?
+ .data)
+ }
+
+ fn add_missing_agg_funcs_to_logical_agg(
+ plan: LogicalPlan,
+ missing_agg_funcs: &IndexSet<Expr>,
+ ) -> Result<(LogicalPlan, HashMap<Expr, Column>)> {
+ let mut agg_func_to_output_col: HashMap<Expr, Column> =
+ HashMap::with_capacity(missing_agg_funcs.len());
+
+ if missing_agg_funcs.is_empty() {
+ return Ok((plan, agg_func_to_output_col));
+ }
+
+ let plan = plan
+ .transform_down(|plan| {
Review Comment:
We should only consider aggregates from the select list of the current query. Using traversal does not guarantee this, meaning that we might encounter aggregate functions not originating from the select list, leading to a bad logical plan. For example, the following query is illegal, but it can pass in this PR.
```sh
> create table t(a int);
0 row(s) fetched.
Elapsed 0.007 seconds.
> select * from t, (select max(a) from t) order by max(a);
+---+----------+
| a | max(t.a) |
+---+----------+
+---+----------+
0 row(s) fetched.
Elapsed 0.017 seconds.
```
There may be more complex cases with similar problems.
I think a safer approach is to directly add the missing aggregate functions to the select list at an earlier stage. That is, transform `select min(a) from t order by max(a)` into `select min(a), max(a) from t order by max(a)` , similar to what was done in #14180.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Allow ORDER BY aggregates not present in SELECT list [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on code in PR #15876:
URL: https://github.com/apache/datafusion/pull/15876#discussion_r2079325033
##########
datafusion/expr/src/logical_plan/builder.rs:
##########
@@ -797,26 +807,146 @@ impl LogicalPlanBuilder {
}
// remove pushed down sort columns
- let new_expr = schema.columns().into_iter().map(Expr::Column).collect();
+ let sort_output = schema.columns().into_iter().map(Expr::Column).collect();
+
+ let plan = Arc::unwrap_or_clone(self.plan);
+
+ let (plan, agg_func_to_col, sorts) = if missing_agg_funcs.is_empty() {
+ (plan, HashMap::new(), sorts)
+ } else {
+ {
+ let (plan, agg_func_to_col) =
+ Self::add_missing_agg_funcs_to_logical_agg(plan, &missing_agg_funcs)?;
+
+ let sorts = sorts
+ .iter()
+ .map(|x| {
+ Self::replace_subexpr_to_col(&x.expr, &agg_func_to_col)
+ .map(|expr| x.with_expr(expr))
+ })
+ .collect::<Result<Vec<_>, _>>()?;
+
+ (plan, agg_func_to_col, sorts)
+ }
+ };
+
+ // we need downstream filter/project return missing col(agg_funcs)
+ missing_cols.extend(agg_func_to_col.into_values());
let is_distinct = false;
- let plan = Self::add_missing_columns(
- Arc::unwrap_or_clone(self.plan),
- &missing_cols,
- is_distinct,
- )?;
+ let plan = Self::add_missing_columns(plan, &missing_cols, is_distinct)?;
let sort_plan = LogicalPlan::Sort(Sort {
expr: normalize_sorts(sorts, &plan)?,
input: Arc::new(plan),
fetch,
});
- Projection::try_new(new_expr, Arc::new(sort_plan))
+ Projection::try_new(sort_output, Arc::new(sort_plan))
.map(LogicalPlan::Projection)
.map(Self::new)
}
+ fn replace_subexpr_to_col(
+ expr: &Expr,
+ func_to_col: &HashMap<Expr, Column>,
+ ) -> Result<Expr> {
+ Ok(expr
+ .clone()
+ .transform_down(|nested_expr| {
+ if let Some(col) = func_to_col.get(&nested_expr) {
+ Ok(Transformed::yes(Expr::Column(col.clone())))
+ } else {
+ Ok(Transformed::no(nested_expr))
+ }
+ })?
+ .data)
+ }
+
+ fn add_missing_agg_funcs_to_logical_agg(
+ plan: LogicalPlan,
+ missing_agg_funcs: &IndexSet<Expr>,
+ ) -> Result<(LogicalPlan, HashMap<Expr, Column>)> {
+ let mut agg_func_to_output_col: HashMap<Expr, Column> =
+ HashMap::with_capacity(missing_agg_funcs.len());
+
+ if missing_agg_funcs.is_empty() {
+ return Ok((plan, agg_func_to_output_col));
+ }
+
+ let plan = plan
+ .transform_down(|plan| {
Review Comment:
> I think a safer approach is to directly add the missing aggregate functions to the select list at an earlier stage. That is, transform select min(a) from t order by max(a) into select min(a), max(a) from t order by max(a) , similar to what was done in #14180.
Why we should "at an earlier stage" to do this transformation ?
In this PR. For query plan like
```
// select max(employee_csv.salary) from employee_csv group by employee_csv.state order by min(employee_csv.salary)
Sort: min(employee_csv.salary)
Aggregate: groupBy=[[employee_csv.state]], aggr=[[max(employee_csv.salary)]]
```
The plan will be transformed into:
```
Projection: col(max(employee_csv.salary)) // in final result set, it only output `max(employee_csv.salary)`.
Sort: col(min(employee_csv.salary)) // it's a column produced by agg
Aggregate: groupBy=[[employee_csv.state]], aggr=[[max(employee_csv.salary), min(employee_csv.salary)]]
```
The proposed logic change to `sort_with_limit` makes sense since it handles similar functionality:
- find missing aggreate function should output
- add to operator
- add project
> transform select min(a) from t order by max(a) into select min(a), max(a) from t order by max(a)
I think it should be transformed into "select "min(a)" from (select min(a) , max(a) from t order by max(a))" ?
> We should only consider aggregates from the select list of the current query. Using traversal does not guarantee this, meaning that we might encounter aggregate functions not originating from the select list, leading to a bad logical plan.
Could you explain this in more detail? According to my understanding, if there are aggregate functions in the ORDER BY clause, they should be placed in the aggregation operator below (which is what this PR does).
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Allow ORDER BY aggregates not present in SELECT list [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #15876: fix: Allow ORDER BY aggregates not present in SELECT list
URL: https://github.com/apache/datafusion/pull/15876
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Allow ORDER BY aggregates not present in SELECT list [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on code in PR #15876:
URL: https://github.com/apache/datafusion/pull/15876#discussion_r2079325033
##########
datafusion/expr/src/logical_plan/builder.rs:
##########
@@ -797,26 +807,146 @@ impl LogicalPlanBuilder {
}
// remove pushed down sort columns
- let new_expr = schema.columns().into_iter().map(Expr::Column).collect();
+ let sort_output = schema.columns().into_iter().map(Expr::Column).collect();
+
+ let plan = Arc::unwrap_or_clone(self.plan);
+
+ let (plan, agg_func_to_col, sorts) = if missing_agg_funcs.is_empty() {
+ (plan, HashMap::new(), sorts)
+ } else {
+ {
+ let (plan, agg_func_to_col) =
+ Self::add_missing_agg_funcs_to_logical_agg(plan, &missing_agg_funcs)?;
+
+ let sorts = sorts
+ .iter()
+ .map(|x| {
+ Self::replace_subexpr_to_col(&x.expr, &agg_func_to_col)
+ .map(|expr| x.with_expr(expr))
+ })
+ .collect::<Result<Vec<_>, _>>()?;
+
+ (plan, agg_func_to_col, sorts)
+ }
+ };
+
+ // we need downstream filter/project return missing col(agg_funcs)
+ missing_cols.extend(agg_func_to_col.into_values());
let is_distinct = false;
- let plan = Self::add_missing_columns(
- Arc::unwrap_or_clone(self.plan),
- &missing_cols,
- is_distinct,
- )?;
+ let plan = Self::add_missing_columns(plan, &missing_cols, is_distinct)?;
let sort_plan = LogicalPlan::Sort(Sort {
expr: normalize_sorts(sorts, &plan)?,
input: Arc::new(plan),
fetch,
});
- Projection::try_new(new_expr, Arc::new(sort_plan))
+ Projection::try_new(sort_output, Arc::new(sort_plan))
.map(LogicalPlan::Projection)
.map(Self::new)
}
+ fn replace_subexpr_to_col(
+ expr: &Expr,
+ func_to_col: &HashMap<Expr, Column>,
+ ) -> Result<Expr> {
+ Ok(expr
+ .clone()
+ .transform_down(|nested_expr| {
+ if let Some(col) = func_to_col.get(&nested_expr) {
+ Ok(Transformed::yes(Expr::Column(col.clone())))
+ } else {
+ Ok(Transformed::no(nested_expr))
+ }
+ })?
+ .data)
+ }
+
+ fn add_missing_agg_funcs_to_logical_agg(
+ plan: LogicalPlan,
+ missing_agg_funcs: &IndexSet<Expr>,
+ ) -> Result<(LogicalPlan, HashMap<Expr, Column>)> {
+ let mut agg_func_to_output_col: HashMap<Expr, Column> =
+ HashMap::with_capacity(missing_agg_funcs.len());
+
+ if missing_agg_funcs.is_empty() {
+ return Ok((plan, agg_func_to_output_col));
+ }
+
+ let plan = plan
+ .transform_down(|plan| {
Review Comment:
> I think a safer approach is to directly add the missing aggregate functions to the select list at an earlier stage. That is, transform select min(a) from t order by max(a) into select min(a), max(a) from t order by max(a) , similar to what was done in #14180.
Why we should "at an earlier stage" to do this transformation ?
In this PR. For query plan like
```
// select max(employee_csv.salary) from employee_csv group by employee_csv.state order by min(employee_csv.salary)
Sort: min(employee_csv.salary)
Aggregate: groupBy=[[employee_csv.state]], aggr=[[max(employee_csv.salary)]]
```
The plan will be transformed into:
```
Projection: col(max(employee_csv.salary)) // in final result set, it only output `max(employee_csv.salary)`.
Sort: col(min(employee_csv.salary)) // it's a column produced by agg
Aggregate: groupBy=[[employee_csv.state]], aggr=[[max(employee_csv.salary), min(employee_csv.salary)]]
```
The proposed logic change to `sort_with_limit` makes sense since it handles similar functionality:
- find missing aggreate function should output
- add to operator
- add project
> transform select min(a) from t order by max(a) into select min(a), max(a) from t order by max(a)
I think it should be transformed into "select "min(a)" from (select min(a) , max(a) from t order by max(a))" ?
> We should only consider aggregates from the select list of the current query. Using traversal does not guarantee this, meaning that we might encounter aggregate functions not originating from the select list, leading to a bad logical plan.
Could you explain this in more detail? According to my understanding, if we "encounter aggregate functions not originating from the select list" (such as order by), we should be placed in the aggregation operator below (which is what this PR does).
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Allow ORDER BY aggregates not present in SELECT list [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #15876:
URL: https://github.com/apache/datafusion/pull/15876#issuecomment-3076487145
Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Allow ORDER BY aggregates not present in SELECT list [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney closed pull request #15876: fix: Allow ORDER BY aggregates not present in SELECT list
URL: https://github.com/apache/datafusion/pull/15876
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Allow ORDER BY aggregates not present in SELECT list [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15876:
URL: https://github.com/apache/datafusion/pull/15876#issuecomment-2840609788
FYI @jonahgao -- would you possible have time to review this PR?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: Allow ORDER BY aggregates not present in SELECT list [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on code in PR #15876:
URL: https://github.com/apache/datafusion/pull/15876#discussion_r2065894121
##########
datafusion/sql/tests/cases/plan_to_sql.rs:
##########
@@ -345,10 +345,10 @@ fn roundtrip_statement_with_dialect_2() -> Result<(), DataFusionError> {
#[test]
fn roundtrip_statement_with_dialect_3() -> Result<(), DataFusionError> {
roundtrip_statement_with_dialect_helper!(
- sql: "select min(ta.j1_id) as j1_min, max(tb.j1_max) from j1 ta, (select distinct max(ta.j1_id) as j1_max from j1 ta order by max(ta.j1_id)) tb order by min(ta.j1_id) limit 10;",
+ sql: "select min(ta.j1_id) as j1_min, max(tb.j1_max) from j1 ta, (select distinct max(ta.j1_id) as j1_max from j1 ta ) tb order by min(ta.j1_id) limit 10;",
Review Comment:
Removed the `ORDER BY` clause.
In the `main` branch, a query structured like the following already failed to generate a **valid** plan since it can not figure out sort by `distinct` output..
```
> CREATE TABLE t (g INT, o INT );
0 row(s) fetched.
Elapsed 0.020 seconds.
> explain select DISTINCT max(g) as c from t GROUP by o order by max(g);
Internal error: Unsupported logical plan: Distinct should be replaced to Aggregate.
This was likely caused by a bug in DataFusion's code and we would welcome that you file an bug report in our issue tracker
>
```
In this PR it will generate error since it try push agg func to `distinct`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org