You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "mustafasrepo (via GitHub)" <gi...@apache.org> on 2024/01/26 14:45:38 UTC
[PR] Cache common referred expression at the window input [arrow-datafusion]
mustafasrepo opened a new pull request, #9009:
URL: https://github.com/apache/arrow-datafusion/pull/9009
## 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 #.
## Rationale for this change
The [PR8960](https://github.com/apache/arrow-datafusion/pull/8960) retracted 2 of the window tests. This PR fixes these retracted tests also adds a new feature for caching common expressions at the window input.
As an example consider the following query
```sql
SELECT c3,
SUM(c9) OVER(ORDER BY c3+c4 DESC) as sum1,
SUM(c9) OVER(ORDER BY c3+c4 ASC) as sum2
FROM aggregate_test_100
```
which will generate following logical plan.
```
+ Projection: aggregate_test_100.c3, SUM(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c3 + aggregate_test_100.c4 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS sum1, SUM(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c3 + aggregate_test_100.c4 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS sum2
+ --Limit: skip=0, fetch=5
+ ----WindowAggr: windowExpr=[[SUM(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c3 + aggregate_test_100.c4aggregate_test_100.c4aggregate_test_100.c3 AS aggregate_test_100.c3 + aggregate_test_100.c4 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS SUM(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c3 + aggregate_test_100.c4 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
+ ------WindowAggr: windowExpr=[[SUM(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c3 + aggregate_test_100.c4aggregate_test_100.c4aggregate_test_100.c3 AS aggregate_test_100.c3 + aggregate_test_100.c4 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS SUM(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c3 + aggregate_test_100.c4 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
+ --------Projection: aggregate_test_100.c3 + aggregate_test_100.c4 AS aggregate_test_100.c3 + aggregate_test_100.c4aggregate_test_100.c4aggregate_test_100.c3, aggregate_test_100.c3, aggregate_test_100.c9
+ ----------TableScan: aggregate_test_100 projection=[c3, c4, c9]
```
where expression `c3+c4` is computed in the first `Projection` and its result (which is a column) is used in subsequent `WindowAggr`. This is done with the `CommonSubexprEliminate` rule.
However, for the following query
```sql
SELECT c3,
SUM(c9) OVER(ORDER BY c3+c4 DESC, c9 DESC, c2 ASC) as sum1,
SUM(c9) OVER(ORDER BY c3+c4 ASC, c9 ASC ) as sum2
FROM aggregate_test_100
```
datafusion generates following plan:
```
+ Projection: aggregate_test_100.c3, SUM(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c3 + aggregate_test_100.c4 DESC NULLS FIRST, aggregate_test_100.c9 DESC NULLS FIRST, aggregate_test_100.c2 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS sum1, SUM(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c3 + aggregate_test_100.c4 ASC NULLS LAST, aggregate_test_100.c9 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS sum2
+ --WindowAggr: windowExpr=[[SUM(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c3 + aggregate_test_100.c4 ASC NULLS LAST, aggregate_test_100.c9 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
+ ----Projection: aggregate_test_100.c3, aggregate_test_100.c4, aggregate_test_100.c9, SUM(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c3 + aggregate_test_100.c4 DESC NULLS FIRST, aggregate_test_100.c9 DESC NULLS FIRST, aggregate_test_100.c2 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+ ------WindowAggr: windowExpr=[[SUM(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c3 + aggregate_test_100.c4 DESC NULLS FIRST, aggregate_test_100.c9 DESC NULLS FIRST, aggregate_test_100.c2 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
+ --------TableScan: aggregate_test_100 projection=[c2, c3, c4, c9]
```
where computation `c3+c4` couldn't cache with a `Projection` before first `WindowAggr`. The reason is that, each `WindowAggr` refers to `c3+c4` once. Hence `CommonSubExpr` rule doesn't think "removing it is helpful".
<!--
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?
This PR fixes above problem so that `CommonSubExpr` considers consecutive window operators during common sub expression substitute analysis. With this analysis we can generate following logical plan for the second query:
```
+ Projection: aggregate_test_100.c3, SUM(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c3 + aggregate_test_100.c4 DESC NULLS FIRST, aggregate_test_100.c9 DESC NULLS FIRST, aggregate_test_100.c2 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS sum1, SUM(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c3 + aggregate_test_100.c4 ASC NULLS LAST, aggregate_test_100.c9 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS sum2
+ --WindowAggr: windowExpr=[[SUM(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c3 + aggregate_test_100.c4aggregate_test_100.c4aggregate_test_100.c3 AS aggregate_test_100.c3 + aggregate_test_100.c4 ASC NULLS LAST, aggregate_test_100.c9 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS SUM(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c3 + aggregate_test_100.c4 ASC NULLS LAST, aggregate_test_100.c9 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
+ ----Projection: aggregate_test_100.c3 + aggregate_test_100.c4aggregate_test_100.c4aggregate_test_100.c3, aggregate_test_100.c3, aggregate_test_100.c9, SUM(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c3 + aggregate_test_100.c4 DESC NULLS FIRST, aggregate_test_100.c9 DESC NULLS FIRST, aggregate_test_100.c2 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+ ------WindowAggr: windowExpr=[[SUM(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c3 + aggregate_test_100.c4aggregate_test_100.c4aggregate_test_100.c3 AS aggregate_test_100.c3 + aggregate_test_100.c4 DESC NULLS FIRST, aggregate_test_100.c9 DESC NULLS FIRST, aggregate_test_100.c2 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS SUM(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c3 + aggregate_test_100.c4 DESC NULLS FIRST, aggregate_test_100.c9 DESC NULLS FIRST, aggregate_test_100.c2 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
+ --------Projection: aggregate_test_100.c3 + aggregate_test_100.c4 AS aggregate_test_100.c3 + aggregate_test_100.c4aggregate_test_100.c4aggregate_test_100.c3, aggregate_test_100.c2, aggregate_test_100.c3, aggregate_test_100.c9
+ ----------TableScan: aggregate_test_100 projection=[c2, c3, c4, c9]
```
where common computation is cached with `Projection` before first window.
<!--
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?
<!--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache common referred expression at the window input [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #9009:
URL: https://github.com/apache/arrow-datafusion/pull/9009
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache common referred expression at the window input [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #9009:
URL: https://github.com/apache/arrow-datafusion/pull/9009#issuecomment-1914589292
> However, didn't cached (Previous behaviour). However, I don't think we will generate an invalid plan. I added your example as a test case also in this PR.
Thank you
>
Thank you for this contribution @mustafasrepo -- this PR looks good to me. Also, I found the description on this PR very clear and well written. Thank you very much 🙏
One thought I had was will there be a problem if there is a subquery that would end up with a nested WindowAggExec that could be incorrectly optimized away 🤔
Something like
SELECT c3,
SUM(c9) OVER(ORDER BY c3+c4 ASC) as sum2,
sum1,
FROM (
SELECT c3, c4, c9,
SUM(c9) OVER(ORDER BY c3+c4 DESC) as sum1,
FROM aggregate_test_100
)
I think, in these cases, we will generate a sub-optimal plan, where a complex expression is calculated more than once by subsequent operators. However, didn't cached (Previous behaviour). However, I don't think we will generate an invalid plan. I added your example as a test case also in this PR.
> I think as a future PR, we can analyze plan from top down to count expression referral count, for better calculating referral counts across plan.
Yes, I agree there is no need to optimize this case as part of this PR, and since it gives correct results, lets 🚀
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache common referred expression at the window input [arrow-datafusion]
Posted by "mustafasrepo (via GitHub)" <gi...@apache.org>.
mustafasrepo commented on PR #9009:
URL: https://github.com/apache/arrow-datafusion/pull/9009#issuecomment-1914173891
> Thank you for this contribution @mustafasrepo -- this PR looks good to me. Also, I found the description on this PR very clear and well written. Thank you very much 🙏
>
> One thought I had was will there be a problem if there is a subquery that would end up with a nested WindowAggExec that could be incorrectly optimized away 🤔
>
> Something like
>
> ```sql
> SELECT c3,
> SUM(c9) OVER(ORDER BY c3+c4 ASC) as sum2,
> sum1,
> FROM (
> SELECT c3, c4, c9,
> SUM(c9) OVER(ORDER BY c3+c4 DESC) as sum1,
> FROM aggregate_test_100
> )
> ```
I think, in these cases, we will generate a sub-optimal plan, where a complex expression is calculated more than once by subsequent operators. However, didn't cached (Previous behaviour). However, I don't think we will generate an invalid plan. I added your example as a test case also in this PR.
I think as a future PR, we can analyze plan from top down to count expression referral count, for better calculating referral counts across plan.
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache common referred expression at the window input [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #9009:
URL: https://github.com/apache/arrow-datafusion/pull/9009#discussion_r1468170121
##########
datafusion/optimizer/src/common_subexpr_eliminate.rs:
##########
@@ -166,25 +166,67 @@ impl CommonSubexprEliminate {
window: &Window,
config: &dyn OptimizerConfig,
) -> Result<LogicalPlan> {
- let Window {
- input,
- window_expr,
- schema,
- } = window;
+ let mut window_exprs = vec![];
+ let mut arrays_per_window = vec![];
let mut expr_set = ExprSet::new();
- let input_schema = Arc::clone(input.schema());
- let arrays =
- to_arrays(window_expr, input_schema, &mut expr_set, ExprMask::Normal)?;
+ // Get all window expressions inside the consecutive window operators.
Review Comment:
Perhaps we can add a comment here about why this is recursively looking down into all window operations (e.g. because they all get the same input schema and append on some window functions, but the window functions can't refer to previous window functions).
I think perhaps you could reuse the (very nicely written) description from this PR which explains it very well
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache common referred expression at the window input [arrow-datafusion]
Posted by "haohuaijin (via GitHub)" <gi...@apache.org>.
haohuaijin commented on code in PR #9009:
URL: https://github.com/apache/arrow-datafusion/pull/9009#discussion_r1468353780
##########
datafusion/optimizer/src/common_subexpr_eliminate.rs:
##########
@@ -166,25 +166,67 @@ impl CommonSubexprEliminate {
window: &Window,
config: &dyn OptimizerConfig,
) -> Result<LogicalPlan> {
- let Window {
- input,
- window_expr,
- schema,
- } = window;
+ let mut window_exprs = vec![];
+ let mut arrays_per_window = vec![];
let mut expr_set = ExprSet::new();
- let input_schema = Arc::clone(input.schema());
- let arrays =
- to_arrays(window_expr, input_schema, &mut expr_set, ExprMask::Normal)?;
+ // Get all window expressions inside the consecutive window operators.
+ // At this stage all window operators should be consecutive.
+ let mut plan = LogicalPlan::Window(window.clone());
+ while let LogicalPlan::Window(window) = plan {
+ let Window {
+ input, window_expr, ..
+ } = window;
+ plan = input.as_ref().clone();
+
+ let input_schema = Arc::clone(input.schema());
+ let arrays =
+ to_arrays(&window_expr, input_schema, &mut expr_set, ExprMask::Normal)?;
+
+ window_exprs.push(window_expr);
+ arrays_per_window.push(arrays);
+ }
- let (mut new_expr, new_input) =
- self.rewrite_expr(&[window_expr], &[&arrays], input, &expr_set, config)?;
+ let mut window_exprs = window_exprs
+ .iter()
+ .map(|expr| expr.as_slice())
+ .collect::<Vec<_>>();
+ let arrays_per_window = arrays_per_window
+ .iter()
+ .map(|arrays| arrays.as_slice())
+ .collect::<Vec<_>>();
- Ok(LogicalPlan::Window(Window {
- input: Arc::new(new_input),
- window_expr: pop_expr(&mut new_expr)?,
- schema: schema.clone(),
- }))
+ assert_eq!(window_exprs.len(), arrays_per_window.len());
+ let (mut new_expr, new_input) = self.rewrite_expr(
+ &window_exprs,
+ &arrays_per_window,
+ &plan,
+ &expr_set,
+ config,
+ )?;
+ assert_eq!(window_exprs.len(), new_expr.len());
+
+ // Construct consecutive window operator, with their corresponding new window expressions.
+ plan = new_input;
+ while let Some(new_window_expr) = new_expr.pop() {
+ // Since `new_expr` and `window_exprs` length are same. We can safely `.unwrap` here.
+ let orig_window_expr = window_exprs.pop().unwrap();
+ assert_eq!(new_window_expr.len(), orig_window_expr.len());
+
+ // Rename new re-written window expressions with original name (bu giving alias)
Review Comment:
```suggestion
// Rename new re-written window expressions with original name (by giving alias)
```
--
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@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org