feat(query): TopN window operator

[forsaken628]

I hereby agree to the terms of the CLA available at: https://docs.databend.com/dev/policies/cla/

Summary

fixes: #16394

This PR replaces row-by-row hash with sorting and then hashing only once per partition. the increased overhead of sort is offset by the reduced overhead of copy and hash. It looks like this optimisation is highly generalisable.

Tests

• Unit Test
• Logic Test
• Benchmark Test
• No Test - Explain why

Type of change

• Bug Fix (non-breaking change which fixes an issue)
• New Feature (non-breaking change which adds functionality)
• Breaking Change (fix or feature that could cause existing functionality not to work as expected)
• Documentation Update
• Refactoring
• Performance Improvement
• Other (please describe):

---

This change is 

[github-actions]

Docker Image for PR

• tag: pr-16726-23a65eb-1730218502

note: this image tag is only available for internal use,
please check the internal doc for more details.

[github-actions]

ClickBench Report

• hits: https://benchmark.databend.com/clickbench/pr/16726/11577501445/hits.html
• internal: https://benchmark.databend.com/clickbench/pr/16726/11577501445/internal.html
• load: https://benchmark.databend.com/clickbench/pr/16726/11577501445/load.html
• tpch: https://benchmark.databend.com/clickbench/pr/16726/11577501445/tpch.html

[sundy-li]

@forsaken628 Is there any perf results for window topn to share?

[github-actions]

Docker Image for PR

• tag: pr-16726-96169f1-1730810767

note: this image tag is only available for internal use,
please check the internal doc for more details.

[forsaken628]

benchmark:

SELECT * FROM (
  SELECT number, rank() OVER ( PARTITION BY number % 3 ORDER BY number ) AS c 
  FROM numbers(1000000)
) WHERE c < 3

this pr
version: pr-16726-96169f1-1730810767
size: Small
Total Execution Time: 23ms

WindowPartition
estimated rows: 1000000.00
cpu time: 48.109029ms
wait time: 812.222µs
output rows: 96

Window
cpu time: 28.382µs

main
version: cloud dev latest
size: Small
Total Execution Time: 127ms

WindowPartition
estimated rows: 1000000.00
cpu time: 100.431594ms
wait time: 1.078021ms
output rows: 1 million

Window
cpu time: 46.962569ms

[forsaken628]

benchmark:

create table random_tab (a int16, b String) Engine = Random;
explain ANALYZE select * from (select a % 1000, rank() over (PARTITION by a % 1000 ORDER by b) rk from (select * from random_tab limit 10000000)) where rk < 3;

this pr
version: pr-16726-96169f1-1730810767
size: Small

Window
├── output columns: [random_tab.a (#0), random_tab.b (#1), rank_part_0 (#2), rank() OVER ( PARTITION BY a % 1000 ORDER BY b ) (#3)]
├── aggregate function: [rank]
├── partition by: [rank_part_0]
├── order by: [b]
├── frame: [Range: Preceding(None) ~ CurrentRow]
├── cpu time: 10.049676ms
├── output rows: 64 thousand
├── output bytes: 1.67 MiB
└── WindowPartition
    ├── output columns: [random_tab.a (#0), random_tab.b (#1), rank_part_0 (#2)]
    ├── hash keys: [rank_part_0]
    ├── estimated rows: 0.00
    ├── cpu time: 1.941062005s
    ├── wait time: 3.682208ms
    ├── output rows: 64 thousand
    ├── output bytes: 1.18 MiB
    └── EvalScalar
        ├── output columns: [random_tab.a (#0), random_tab.b (#1), rank_part_0 (#2)]
        ├── expressions: [random_tab.a (#0) % 1000]
        ├── estimated rows: 0.00
        ├── cpu time: 27.624338ms
        ├── output rows: 10 million
        ├── output bytes: 184.77 MiB

main
version: cloud dev latest
size: Small

Window
├── output columns: [random_tab.a (#0), random_tab.b (#1), rank_part_0 (#2), rank() OVER ( PARTITION BY a % 1000 ORDER BY b ) (#3)]
├── aggregate function: [rank]
├── partition by: [rank_part_0]
├── order by: [b]
├── frame: [Range: Preceding(None) ~ CurrentRow]
├── cpu time: 912.368447ms
├── output rows: 10 million
├── output bytes: 261.07 MiB
└── WindowPartition
    ├── output columns: [random_tab.a (#0), random_tab.b (#1), rank_part_0 (#2)]
    ├── hash keys: [rank_part_0]
    ├── estimated rows: 0.00
    ├── cpu time: 2.927932431s
    ├── wait time: 4.966152ms
    ├── output rows: 10 million
    ├── output bytes: 184.78 MiB
    └── EvalScalar
        ├── output columns: [random_tab.a (#0), random_tab.b (#1), rank_part_0 (#2)]
        ├── expressions: [random_tab.a (#0) % 1000]
        ├── estimated rows: 0.00
        ├── cpu time: 27.556557ms
        ├── output rows: 10 million
        ├── output bytes: 184.77 MiB

[forsaken628]

benchmark:

create table random_tab (a int16, b String) Engine = Random;
explain ANALYZE select * from (select a % 100000, rank() over (PARTITION by a % 100000 ORDER by b) rk from (select * from random_tab limit 10000000)) where rk < 1000;

this pr
version: pr-16726-96169f1-1730810767
size: Small

Window
├── output columns: [random_tab.a (#0), random_tab.b (#1), rank_part_0 (#2), rank() OVER ( PARTITION BY a % 100000 ORDER BY b ) (#3)]
├── aggregate function: [rank]
├── partition by: [rank_part_0]
├── order by: [b]
├── frame: [Range: Preceding(None) ~ CurrentRow]
├── cpu time: 576.44352ms
├── output rows: 5.02 million
├── output bytes: 150.06 MiB
└── WindowPartition
    ├── output columns: [random_tab.a (#0), random_tab.b (#1), rank_part_0 (#2)]
    ├── hash keys: [rank_part_0]
    ├── estimated rows: 0.00
    ├── cpu time: 2.87370872s
    ├── wait time: 5.195948ms
    ├── output rows: 5.02 million
    ├── output bytes: 111.80 MiB
    └── EvalScalar
        ├── output columns: [random_tab.a (#0), random_tab.b (#1), rank_part_0 (#2)]
        ├── expressions: [random_tab.a (#0) % 100000]
        ├── estimated rows: 0.00
        ├── cpu time: 27.332388ms
        ├── output rows: 10 million
        ├── output bytes: 222.92 MiB

main
version: cloud dev latest
size: Small

Window
├── output columns: [random_tab.a (#0), random_tab.b (#1), rank_part_0 (#2), rank() OVER ( PARTITION BY a % 100000 ORDER BY b ) (#3)]
├── aggregate function: [rank]
├── partition by: [rank_part_0]
├── order by: [b]
├── frame: [Range: Preceding(None) ~ CurrentRow]
├── cpu time: 935.685435ms
├── output rows: 10 million
├── output bytes: 299.22 MiB
└── WindowPartition
    ├── output columns: [random_tab.a (#0), random_tab.b (#1), rank_part_0 (#2)]
    ├── hash keys: [rank_part_0]
    ├── estimated rows: 0.00
    ├── cpu time: 3.020992844s
    ├── wait time: 6.188028ms
    ├── output rows: 10 million
    ├── output bytes: 222.92 MiB
    └── EvalScalar
        ├── output columns: [random_tab.a (#0), random_tab.b (#1), rank_part_0 (#2)]
        ├── expressions: [random_tab.a (#0) % 100000]
        ├── estimated rows: 0.00
        ├── cpu time: 23.836146ms
        ├── output rows: 10 million
        ├── output bytes: 222.92 MiB