feat(watermark): topNExec state cleaning

[yuhao-su]

Plain top-n:

Ascending

We need to range delete records above the watermark, while currently, we can only delete those below the watermark.

Descending

This is more tricky. When watermark w arrives, we need to locate the smallest record with the first column value larger than w as the n-th records. We can clean all records with the watermark column smaller than the watermark value of (n+offset+limit)-th records.

In this case, two functions from the storage layer might be needed,

• locate the first record smaller than x, return the position.
• get the value at position pos

Group top-n:

For group topn with group key a, order by b, to clean the state, range delete for b is needed. (currently we can only range delete with a give watermark in the first column)

[wcy-fdu]

Operators generally access storage through StateTable, could you please give the definition of such two interface in StateTable or relational semantics?

[BugenZhao]

May I ask about the use cases for ordering by the watermark (usually time) column in ascending manner? 🤔

[yuhao-su]

May I ask about the use cases for ordering by the watermark (usually time) column in ascending manner? 🤔

I'm not sure about the exact use case. But it is possible for users to write such queries. cc @TennyZhuang @fuyufjh @st1page

[st1page]

Emmm... I can not think up any use case for now. Let's remain this issue and see if it is required.

[yuhao-su]

Could you please give the definition of such two interface in StateTable or relational semantics?

Iet's wait to see the use case before decide a general interface.

[st1page]

Let's wait some real-world requeriment about the topN's cleaning