Add window expression part 1 - logical and physical planning, structure, to/from proto, and explain, for empty over clause only #334

jimexist · 2021-05-13T10:52:22Z

Which issue does this PR close?

We'd like to support window function in three or more steps:

Support window functions with basic logical planning and physical planning #359 basic structure
Support window functions with empty OVER clause #298 empty over clause
Support window functions with PARTITION BY clause #299 with partition clause
Support window functions with order by #360 with order by
Support window functions with window frame #361 with window frame

Closes #359.

This is partly re #298.

Work to be done next: finish #298 and #299

Rationale for this change

this is a very stretch version of adding window function constructs to the planner, proto, etc.

What changes are included in this PR?

logical expression for window expression
physical expression for window expression
basic exec plan structure
basic from/to proto parsing
support explain parsing now

> explain select c1, sum(c3) over () from test;
+--------------+------------------------------------------------------------------+
| plan_type    | plan                                                             |
+--------------+------------------------------------------------------------------+
| logical_plan | Projection: #c1, #SUM(c3)                                        |
|              |   WindowAggr: windowExpr=[[SUM(#c3)]] partitionBy=[], orderBy=[] |
|              |     TableScan: test projection=None                              |
+--------------+------------------------------------------------------------------+
1 row in set. Query took 0 seconds.

> explain select c1, c3, sum(c3 + 2) over () from test;
+--------------+--------------------------------------------------------------------------------+
| plan_type    | plan                                                                           |
+--------------+--------------------------------------------------------------------------------+
| logical_plan | Projection: #c1, #c3, #SUM(c3 Plus Int64(2))                                   |
|              |   WindowAggr: windowExpr=[[SUM(#c3 Plus Int64(2))]] partitionBy=[], orderBy=[] |
|              |     TableScan: test projection=None                                            |
+--------------+--------------------------------------------------------------------------------+
1 row in set. Query took 0 seconds.

> explain select c1, c3, sum(c3) over (), max(c3) over (), avg(c3) over () from test;
+--------------+--------------------------------------------------------------------------------------+
| plan_type    | plan                                                                                 |
+--------------+--------------------------------------------------------------------------------------+
| logical_plan | Projection: #c1, #c3, #SUM(c3), #MAX(c3), #AVG(c3)                                   |
|              |   WindowAggr: windowExpr=[[SUM(#c3), MAX(#c3), AVG(#c3)]] partitionBy=[], orderBy=[] |
|              |     TableScan: test projection=None                                                  |
+--------------+--------------------------------------------------------------------------------------+
1 row in set. Query took 0 seconds.

Are there any user-facing changes?

datafusion/src/optimizer/hash_build_probe_order.rs

datafusion/src/physical_plan/planner.rs

alamb

This looks like a cool start @jimexist 👍

datafusion/src/logical_plan/builder.rs

jimexist · 2021-05-16T17:00:41Z

I wonder which approach makes more sense:

To implement proto and serde as an API first and leave a lot of unimplemented stubs
To try to cover a minimal working end to end feature and leave advanced use cases like window frames for later

Dandandan · 2021-05-16T19:30:32Z

@jimexist I think it would be perfectly fine if more "advanced" features like window frames etc. are not supported, just as not all window functions have to be supported.
If we could just support e.g. ROW_NUMBER() OVER () or MAX(x) OVER () with default spec that would be great already.

jimexist · 2021-05-19T14:58:58Z

@Dandandan @alamb PTAL

alamb · 2021-05-19T19:00:42Z

I will try to review this later today but I may not get to it until tomorrow

Dandandan · 2021-05-19T19:16:45Z

ballista/rust/core/proto/ballista.proto

+
+message WindowExprNode {
+  oneof window_function {
+    AggregateFunction aggr_function = 1;


I checked whether this makes sense to reuse aggregate functions for window expressions - I think it does! E.g. PostgreSQL also says:
https://www.postgresql.org/docs/9.3/functions-window.html

In addition to these functions, any built-in or user-defined aggregate function can be used as a window function (see Section 9.20 for a list of the built-in aggregates). Aggregate functions act as window functions only when an OVER clause follows the call; otherwise they act as regular aggregates.

yes in general three types of things can be used:

aggregation

UDAF

built in window function

for both 1. and 2. they are not order sensitive, but for 3 we'll have to take sort into account

[postgres] # explain select c1, count(c3) over (partition by c1 order by c3) from test; QUERY PLAN ------------------------------------------------------------------ WindowAgg (cost=6.32..8.32 rows=100 width=12) -> Sort (cost=6.32..6.57 rows=100 width=4) Sort Key: c1, c3 -> Seq Scan on test (cost=0.00..3.00 rows=100 width=4) (4 rows)

[postgres] # explain select c1, first_value(c3) over (partition by c1 order by c3) from test; QUERY PLAN ------------------------------------------------------------------ WindowAgg (cost=6.32..8.32 rows=100 width=6) -> Sort (cost=6.32..6.57 rows=100 width=4) Sort Key: c1, c3 -> Seq Scan on test (cost=0.00..3.00 rows=100 width=4) (4 rows)

IMO only the second time we'll really need to sort by c3

also fun thing to notice:

[postgres] # explain analyze select c1, sum(c3) over (partition by c1 order by c3), avg(c3) over (partition by c1 order by c3 desc) from test; QUERY PLAN -------------------------------------------------------------------------------------------------------------------------- WindowAgg (cost=11.64..13.64 rows=100 width=44) (actual time=1.287..1.373 rows=100 loops=1) -> Sort (cost=11.64..11.89 rows=100 width=36) (actual time=1.281..1.292 rows=100 loops=1) Sort Key: c1, c3 Sort Method: quicksort Memory: 31kB -> WindowAgg (cost=6.32..8.32 rows=100 width=36) (actual time=1.051..1.174 rows=100 loops=1) -> Sort (cost=6.32..6.57 rows=100 width=4) (actual time=0.221..0.231 rows=100 loops=1) Sort Key: c1, c3 DESC Sort Method: quicksort Memory: 29kB -> Seq Scan on test (cost=0.00..3.00 rows=100 width=4) (actual time=0.010..0.028 rows=100 loops=1) Planning Time: 0.087 ms Execution Time: 1.437 ms (11 rows)

I checked whether this makes sense to reuse aggregate functions for window expressions - I think it does! E.g. PostgreSQL also says:
https://www.postgresql.org/docs/9.3/functions-window.html

In addition to these functions, any built-in or user-defined aggregate function can be used as a window function (see Section 9.20 for a list of the built-in aggregates). Aggregate functions act as window functions only when an OVER clause follows the call; otherwise they act as regular aggregates.

it is very useful for analytics, e.g. if you want to know (in an employee table with name, department, and salary) the list of employees in each department with salary above average.

I believe count(..) over order by .. also needs to be sorted, it will do a count over the window, which means a running count (over sorted rows) by default.
But yeah very useful for analytics indeed 👍

My comment above was more about re-using the same functions over here - as I thought we might not want to support every aggregation function here too. But for me it sounds like a good idea to reuse them. Maybe @alamb has some ideas about it as well

I think SQL is confusing in this area -- as @jimexist says, all "normal" aggregate functions (e.g. sum, count, etc) are also valid window functions, but the reverse is not true. You can't use window functions (e.g. LAG, LEAD, etc) outside of a window clause.

Thus I think representing window functions as a new type of function, as this PR does, makes the most sense. They are different enough (e.g. require information on the incoming windows) that trying to wrangle them into the same structures as normal aggregates seems like it will get messy. Long term I would expect we have a UDWF (user defined window function) api as well.

Ideally the physical implementation for sum / count / etc can be mostly reused but in the plans I think they are different enough to warrant different plan structures.

I believe count(..) over order by .. also needs to be sorted, it will do a count over the window, which means a running count (over sorted rows) by default.

Good point! Indeed.

alamb · 2021-05-20T13:11:51Z

Checking it out...

alamb

I went over this PR quite carefully. Thank you very much @jimexist for the contribution! ❤️
❤️ -- this PR looks to be in great shape.

All I think this PR needs is a few more tests and it could be merged in.

I am not familiar with the ballista code, but it looked ok to me. @andygrove do you have any suggestions of who might be interested in those changes?

alamb · 2021-05-20T13:19:14Z

ballista/rust/core/proto/ballista.proto

+
+message WindowExprNode {
+  oneof window_function {
+    AggregateFunction aggr_function = 1;


I think SQL is confusing in this area -- as @jimexist says, all "normal" aggregate functions (e.g. sum, count, etc) are also valid window functions, but the reverse is not true. You can't use window functions (e.g. LAG, LEAD, etc) outside of a window clause.

Thus I think representing window functions as a new type of function, as this PR does, makes the most sense. They are different enough (e.g. require information on the incoming windows) that trying to wrangle them into the same structures as normal aggregates seems like it will get messy. Long term I would expect we have a UDWF (user defined window function) api as well.

Ideally the physical implementation for sum / count / etc can be mostly reused but in the plans I think they are different enough to warrant different plan structures.

datafusion/src/logical_plan/builder.rs

datafusion/src/physical_plan/window_functions.rs

alamb · 2021-05-20T13:44:23Z

datafusion/src/physical_plan/windows.rs

+            )));
+        }
+
+        // window needs to operate on a single partition currently


👍

Eventually it would be cool to push the partitioning expressions into a RepartitionExec so that we can execute the window functions in parallel on different windows but that is definitely an optimization for the future (not this initial PR) :)

alamb · 2021-05-20T13:45:55Z

datafusion/src/sql/planner.rs

@@ -2641,13 +2701,23 @@ mod tests {
    }

    #[test]
-    fn over_not_supported() {
+    fn empty_over() {


datafusion/src/sql/planner.rs

andygrove · 2021-05-20T13:55:57Z

ballista/rust/core/src/serde/logical_plan/from_proto.rs

+                //     .iter()
+                //     .map(|expr| expr.try_into())
+                //     .collect::<Result<Vec<_>, _>>()?;
+                // // FIXME parse the window_frame data


It looks this is still WIP? Is the plan to complete this as part of this PR?

I think @jimexist plans to implement the feature in several PRs as outlined in the PR description

jorgecarleitao · 2021-05-20T14:02:11Z

I haven't had the time to go through yet;

Since this is a large decomposable change, a suggestion here is to create a branch and merge this on that branch, so that we do not have incomplete features in master and allow PRs to that branch without an issue. Then merge the branch into master once its first iteration is ready (e.g. logical and 1 physical plan).

(This is something that I wished to have when implementing large features; it gives the time to review in parts without the risk of having incomplete code in master)

With that said, I am also fine risk it and merg to master; @jimexist do you have any preference how would you prefer to work on this here? :)

alamb · 2021-05-20T14:42:22Z

I am supportive of putting this directly on master -- we have various other "not yet implemented" functionality in DataFusion and I think we can handle any potential confusion with additional documentation

jimexist · 2021-05-20T14:43:42Z

I haven't had the time to go through yet;

Since this is a large decomposable change, a suggestion here is to create a branch and merge this on that branch, so that we do not have incomplete features in master and allow PRs to that branch without an issue. Then merge the branch into master once its first iteration is ready (e.g. logical and 1 physical plan).

(This is something that I wished to have when implementing large features; it gives the time to review in parts without the risk of having incomplete code in master)

With that said, I am also fine risk it and merg to master; @jimexist do you have any preference how would you prefer to work on this here? :)

I wish to:

add unit tests
fix some small comments, per @alamb, like
1. adding filter clause to leave room for future room without API disruption,
2. adding nth_value and tile and leaving them unimplemented and leave room for good first PR
merge it into master as it

3 because although I agree with having a feature branch, the maintenance of continuous rebasing would be troublesome, and I can estimate the whole series of window function implementations take > 1 month. if I can keep the merged code isolated and structurally complete then there's little worry about breaking changes in future: I believe this PR:

is isolated because the explain works (thus mostly the API is determined), while actually executing the window SQL returns zero rows, due to unimplemented exec plan, which I intend to do next
is structurally complete if reviewers can focus on the phase and timing of the window clause in the query planner. future changes of e.g. adding one or more sort phases requires only modifications within the .window fn and thus can be non-intrusive

Co-authored-by: Andrew Lamb <[email protected]>

jorgecarleitao

Amazing design and implementation. I left two small comments, but it is regardless ready to merge.

Thank you very much, @jimexist ; great work.

jorgecarleitao · 2021-05-21T02:41:40Z

datafusion/src/logical_plan/builder.rs

+    /// - https://github.com/apache/arrow-datafusion/issues/361 with window frame
+    pub fn window(
+        &self,
+        window_expr: impl IntoIterator<Item = Expr>,


I am not sure we should use IntoIterator<Item = Expr> for every field with 6 fields. This will produce a version of the compiled function for every combination, which IMO adds an unnecessary compile time and binary size.

IntoIterator is more relevant when we want to avoid an extra allocation; these are very small vectors.

tracked in #372

jorgecarleitao · 2021-05-21T02:43:36Z

datafusion/src/physical_plan/aggregates.rs

@@ -183,7 +182,7 @@ static TIMESTAMPS: &[DataType] = &[
 ];

 /// the signatures supported by the function `fun`.
-fn signature(fun: &AggregateFunction) -> Signature {
+pub fn signature(fun: &AggregateFunction) -> Signature {


pub(super) or pub(crate) instead?

tracked in #373

codecov-commenter · 2021-05-21T07:24:35Z

Codecov Report

Merging #334 (6941151) into master (913bf86) will decrease coverage by 0.95%.
The diff coverage is 36.84%.

@@            Coverage Diff             @@
##           master     #334      +/-   ##
==========================================
- Coverage   75.89%   74.94%   -0.96%     
==========================================
  Files         144      146       +2     
  Lines       23771    24314     +543     
==========================================
+ Hits        18040    18221     +181     
- Misses       5731     6093     +362

Impacted Files	Coverage Δ
...lista/rust/core/src/serde/logical_plan/to_proto.rs	`62.37% <0.00%> (-5.46%)`	⬇️
datafusion/src/optimizer/constant_folding.rs	`91.63% <0.00%> (-0.30%)`	⬇️
datafusion/src/optimizer/hash_build_probe_order.rs	`60.50% <0.00%> (-1.57%)`	⬇️
datafusion/src/optimizer/projection_push_down.rs	`91.11% <0.00%> (-7.63%)`	⬇️
datafusion/src/optimizer/utils.rs	`47.76% <0.00%> (-2.05%)`	⬇️
datafusion/src/physical_plan/mod.rs	`82.75% <0.00%> (-1.95%)`	⬇️
datafusion/src/physical_plan/planner.rs	`76.51% <0.00%> (-4.11%)`	⬇️
datafusion/src/physical_plan/sort.rs	`92.07% <ø> (ø)`
datafusion/src/physical_plan/windows.rs	`0.00% <0.00%> (ø)`
datafusion/tests/sql.rs	`99.89% <ø> (ø)`
... and 12 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 913bf86...6941151. Read the comment docs.

jimexist mentioned this pull request May 13, 2021

Support window functions with empty OVER clause #298

Closed

jimexist force-pushed the add-window-expr branch from cfbc925 to 919a687 Compare May 13, 2021 14:50

Dandandan reviewed May 13, 2021

View reviewed changes

datafusion/src/optimizer/hash_build_probe_order.rs Outdated Show resolved Hide resolved

Dandandan reviewed May 13, 2021

View reviewed changes

datafusion/src/physical_plan/planner.rs Outdated Show resolved Hide resolved

alamb reviewed May 14, 2021

View reviewed changes

datafusion/src/logical_plan/builder.rs Outdated Show resolved Hide resolved

jimexist force-pushed the add-window-expr branch 2 times, most recently from 7ae445a to 11e9541 Compare May 15, 2021 15:46

jimexist force-pushed the add-window-expr branch from 01f53a8 to 5fcdb8f Compare May 17, 2021 00:43

jimexist changed the title ~~Add window expr~~ Add window expr (part I - to only support empty over () clauses) May 17, 2021

jimexist force-pushed the add-window-expr branch 7 times, most recently from ad02da7 to a300aae Compare May 19, 2021 14:41

jimexist changed the title ~~Add window expr (part I - to only support empty over () clauses)~~ Add window expression part 1 - logical and physical planning, structure, to/from proto, and explain, for empty over clause only May 19, 2021

jimexist marked this pull request as ready for review May 19, 2021 14:58

Dandandan reviewed May 19, 2021

View reviewed changes

jimexist force-pushed the add-window-expr branch from 761e4f7 to 8cb0343 Compare May 20, 2021 01:07

alamb reviewed May 20, 2021

View reviewed changes

andygrove reviewed May 20, 2021

View reviewed changes

Jiayu Liu and others added 15 commits May 21, 2021 10:46

fix unused imports

a0b7526

fix clippy

5c4d92d

fix unit test

3ee87aa

Update datafusion/src/logical_plan/builder.rs

f70c739

Co-authored-by: Andrew Lamb <[email protected]>

Update datafusion/src/logical_plan/builder.rs

831c069

Co-authored-by: Andrew Lamb <[email protected]>

Update datafusion/src/physical_plan/window_functions.rs

0cbca53

Co-authored-by: Andrew Lamb <[email protected]>

Update datafusion/src/physical_plan/window_functions.rs

abf08cd

Co-authored-by: Andrew Lamb <[email protected]>

adding more built-in functions

8b486d5

adding filter by todo

0d2a214

enrich unit test

a1eae86

update

f5e64de

add more tests

c36c04a

fix test

4e792e1

fix unit test

880b94f

fix error

bc2271d

jimexist force-pushed the add-window-expr branch from 1b9442b to bc2271d Compare May 21, 2021 02:47

jorgecarleitao approved these changes May 21, 2021

View reviewed changes

This was referenced May 21, 2021

reduce usage of IntoIterator<Item = Expr> in logical plan builder window fn #372

Closed

change from pub(super) or pub(crate) when reusing fns in aggregate.rs in window function implementations #373

Closed

Jiayu Liu added 5 commits May 21, 2021 12:36

fix unit test

1ecae8f

fix unit test

5d96e52

use upper case

bb57c76

fix unit test

2af2a27

comment out test

6941151

alamb merged commit db4f098 into apache:master May 21, 2021

jimexist deleted the add-window-expr branch May 21, 2021 15:52

houqp added ballista datafusion Changes in the datafusion crate enhancement New feature or request labels Jul 29, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add window expression part 1 - logical and physical planning, structure, to/from proto, and explain, for empty over clause only #334

Add window expression part 1 - logical and physical planning, structure, to/from proto, and explain, for empty over clause only #334

jimexist commented May 13, 2021 •

edited by alamb

Loading

alamb left a comment

jimexist commented May 16, 2021

Dandandan commented May 16, 2021

jimexist commented May 19, 2021

alamb commented May 19, 2021

Dandandan May 19, 2021

jimexist May 20, 2021

jimexist May 20, 2021

jimexist May 20, 2021

jimexist May 20, 2021

Dandandan May 20, 2021 •

edited

Loading

Dandandan May 20, 2021

alamb May 20, 2021

jimexist May 20, 2021

alamb commented May 20, 2021

alamb left a comment

alamb May 20, 2021

alamb May 20, 2021

alamb May 20, 2021

andygrove May 20, 2021

alamb May 20, 2021

jorgecarleitao commented May 20, 2021

alamb commented May 20, 2021

jimexist commented May 20, 2021 •

edited

Loading

jorgecarleitao left a comment

jorgecarleitao May 21, 2021

jimexist May 21, 2021

jorgecarleitao May 21, 2021

jimexist May 21, 2021

codecov-commenter commented May 21, 2021

Add window expression part 1 - logical and physical planning, structure, to/from proto, and explain, for empty over clause only #334

Add window expression part 1 - logical and physical planning, structure, to/from proto, and explain, for empty over clause only #334

Conversation

jimexist commented May 13, 2021 • edited by alamb Loading

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are there any user-facing changes?

alamb left a comment

Choose a reason for hiding this comment

jimexist commented May 16, 2021

Dandandan commented May 16, 2021

jimexist commented May 19, 2021

alamb commented May 19, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Dandandan May 20, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb commented May 20, 2021

alamb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jorgecarleitao commented May 20, 2021

alamb commented May 20, 2021

jimexist commented May 20, 2021 • edited Loading

jorgecarleitao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov-commenter commented May 21, 2021

Codecov Report

jimexist commented May 13, 2021 •

edited by alamb

Loading

Dandandan May 20, 2021 •

edited

Loading

jimexist commented May 20, 2021 •

edited

Loading