[Enhancement] rank window function optimization, add partition topn (2)

[liuyehcf]

What type of PR is this：

• bug
• feature
• enhancement
• others

Which issues of this PR fixes ：

Fixes #5885

Enhancement

select * from (
    select *, rank() over (partition by v2 order by v3) as rk from t0
) sub_t0
where rk < 5;

For rank window function, including rank、dense_rank、row_number, if it has a related predicate (rk < 5), then it can be optimized by inserting a PartitionTopN operator before the Sort operator of window function.

The main purpose of PartitionTopN is to filter data, and it's output still remain unordered. It consists of three components:

• partitioner：Divide the input chunk based on the partition exprs
• sorter(topn)：Each partition has an instance of sorter and is sorted independently
• gather：fetch chunks from all sorters into one data stream, so the data is still unordered after gahtering. Moreover, gather is a only logical concetp, not an actual component

                                   ┌────► topn─────┐
                                   │               │
 (unordered)                       │               │                  (unordered)
 inputChunks ───────► partitioner ─┼────► topn ────┼─► gather ─────► outputChunks
                                   │               │
                                   │               │
                                   └────► topn ────┘

The implementation on the optimizer and the executor is a little different:

• In optimizer, for simplicity, we do not define a new pair of {Logical/Physical}PartitionTopNOperator but reuse the existing {Logical/Physical}TopNOperator by adding a new field partitionByExprs to record the partition by information. Besides, we need to pay attentation to the following things:
  • Make sure that we cannot derive sort property from PartitionTopN
  • Make sure that ExchangeNode not set limit if PartitionTopN
• In executor, we define a new pair of LocalPartitionTopN{Sink/Source}Operator, and we may use different implementation based on the field partitionByExprs
  • if partitionByExprs is unset or empty, then the original pair of PartitionSortSinkOperator/LocalMergeSortSourceOperator is used
  • if partitionByExprs is not empty, then pair of LocalPartitionTopN{Sink/Source}Operator is used

Tasks

• Support component partitioner
  • only support one partition expr right now
• Support PartitionTopN(this pr)
• Support row_number
• Support rank
  • by supporting TopN limit by rank
• Support dense_rank
  • by supporting TopN limit by dense_rank
• Support multi partition exprs

[wanpengfei-git]

[FE PR Coverage check]

😍 pass : 0 / 0 (0%)

[trueeyu]

There is a cost to move and compare items in chunk. If it is NullFirst, when there are many Null values, placing it in front may avoid the movement of some items of the Column int the TopN Stage.

[trueeyu]

Can be optimized later