window functions: a prototype

[akuzm]

Some thoughts for internal discussion. The main task is here: #1469

What we need for a meaningful prototype:

• single-threaded, partitioning via sort
• the only supported frame is ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
• only plain aggregate functions
• standard grammar (at least I want to be able to write count() over (partition by x order by y)).
• works properly across blocks
• can be used as an argument to other SELECT expressions (this is our ClickHouse-specific non-standard part, normally SQL doesn't let you reference the SELECT list). Will require some moderate complications in ExpressionActions (a separate type of action, can't be calculated by ExpressionTransform so we have to split the pipeline and insert WindowTransform there). Can be postponed. An example: SUM(col) / SUM(SUM(col)) OVER (PARTITION BY col2) aggregate function over WINDOW is not supported. #19857

Rough edges to smooth out after that:

• proper grammar for declaring windows

• implement RANGE frame and make it default. It is default in other DBs, so people get confused when they try CH and don't get the expected results because we only support ROWS. The default range should suffice for now. This is going to be complicated, because a group can span an arbitrary number of past and future blocks, and we can't output the group until it ends. This requires adding a separate kind of processor + some convenient way to address an arbitrarily large sequence of rows spanning many blocks (tuplestore in Postgres terms). We're going to need it all anyway, to support a moving frame start, and a frame end that is farther than current row.

• consider the source order not to perform redundant sorting -- this one might require massive refactoring window functions: use storage sorting key #19289

• work efficiently w/multiple windows (choose an order of windows that allows to sort less) Try to find sort order that satisfies multiple windows #19779

Advanced features:

• multithreaded calculation. This can be naturally implemented over hash-based partitioning.
• frame start that moves forward
  • subtraction from aggregate function state (the inverse of IAggregateFunction::add)
• frame end that is further than current row
  • this one is really bad, needs a separate "processor" and some tricky buffering, because each input block does not necessarily produce an output block
• by combining the above, we get sliding window (ROWS BETWEEN <X> PRECEDING AND <X> FOLLOWING)
• RANGE OFFSET frame, for sliding window based on time
• window functions that are not aggregate functions, e.g. rank()
  • some of them might require full read-behind + read-ahead? -- scary

References:

https://www.postgresql.org/docs/current/sql-select.html#SQL-WINDOW
https://www.postgresql.org/docs/devel/sql-expressions.html#SYNTAX-WINDOW-FUNCTIONS
https://www.postgresql.org/docs/devel/functions-window.html
https://www.postgresql.org/docs/devel/tutorial-window.html

https://dev.mysql.com/doc/refman/8.0/en/window-function-descriptions.html
https://dev.mysql.com/doc/refman/8.0/en/window-functions-usage.html
https://dev.mysql.com/doc/refman/8.0/en/window-functions-frames.html

Some drafts:

#18222
#18455
#19022 WINDOW clause
#19299 GROUPS frame

[filimonov]

'Crash tests' use case. Imagine a fast trading dataset.

Every second you have up to 10 million transactions with a micro/nanosecond resolution.

The task is to calculate the sum and the number of transactions in the exactly X seconds window prepending the transaction. (not on the second boundary but back from the current event). Single select can extract up to several weeks of data.

[akuzm]

'Crash tests' use case. Imagine a fast trading dataset.

Every second you have up to 10 million transactions with a micro/nanosecond resolution.

The task is to calculate the sum and the number of transactions in the exactly X seconds window prepending the transaction. (not on the second boundary but back from the current event). Single select can extract up to several weeks of data.

This is sum over (order by transaction_time range between 'X seconds' preceding and current row), right? Didn't think about how we are going to implement RANGE yet, but it's probably relatively straightforward.

This also needs the "frame start that moves forward" part.

If there is no partitioning, we will have to also invent some method of parallelizing the computation -- this is the part I dislike the most, we'll have to split the stream based on column values, and also have a correct overlap so that the frame fits.

10M records per second * 30 days per month * 24 * 60 * 60 seconds a day = 25T records, probably about 100TB of data compressed? No point in sorting this amount of data, so the primary key must match the order by and we must understand it and not sort again -- this is the "consider the source sort order" part.

[filimonov]

10M records per second * 30 days per month * 24 * 60 * 60 seconds a day = 25T records, probably about 100TB of data compressed?

Real numbers are smaller (that 10M records per second is peak), but let's say the order of magnitude is correct, and yes the data is sorted by timestamp.

Also, I can cite one nice person who uses ClickHouse together with Splunk:

we need

• a cumulative sum per ordered group (== streamstats sum by ...). I have seen avg too
• a cumulative count per ordered group (== streamstats count by )
• equivalent to streamstats first
• equivalent to streamstats time_window=
  https://docs.splunk.com/Documentation/SplunkCloud/8.1.2009/SearchReference/Streamstats

it is possible to emulate the first 3 in CH but the usability / productivity is poor.

It seems Splunk just read the data in the 'proper' sequence and calculate some stats in a streaming manner.

[akuzm]

Do I get it right that the frame in Splunk ('window' in their terms) always ends with current row?

a cumulative sum per ordered group (== streamstats sum by ...). I have seen avg too
a cumulative count per ordered group (== streamstats count by )

This is just count/sum/whatever over (partition by ...).

equivalent to streamstats first

Somewhat counterintuitively, this is any(...) over (partition by ...). Maybe we need to add first as an alias for it.

equivalent to streamstats time_window=

This is over (partition by ... range between <time interval> preceding and current row), same as your example above.

[akuzm]

I think we'll have to make a test with some simplified popular examples of window functions, and also the CH way of emulating it.

It will help to see how much the usability improves, and how much the performance changes. Some CH features may be more convenient and fast than window functions, e.g. LIMIT BY.

[vkingnew]

row_number
rank
dense_rank
lead/lag
current not support as of 21.1.3.32

[alexey-milovidov]

@vkingnew That's true, it is not developed yet.