feat(query): window partition by spill to disk

forsaken628 commented 2 months ago

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

Summary

part of #15328

Changes:

Local file direct io reading and writing
Implement TempDirManager to provide local temporary file management, including usage control and cleanup.
Add a new configuration item spill
Window supports spill to local disk

How to enable:

Check that the configuration item spill is configured correctly
set window_partition_spilling_to_disk_bytes_limit = 10 * 1024 * 1024 * 1024 (10G), this is the maximum disk space allowed for each request, when the conditions of spill are met, the local disk will be used in preference

Tests

[x] Unit Test
[ ] Logic Test
[ ] Benchmark Test
[ ] No Test - Explain why

Type of change

[ ] Bug Fix (non-breaking change which fixes an issue)
[x] 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

BohuTANG commented 2 months ago

Is there some benchmark numbers for this PR?

forsaken628 commented 2 months ago

Is there some benchmark numbers for this PR?

@BohuTANG Here just replace the storage, where can I go to find the right environment to do benchmark? If the test environment is not representative, benchmark has no meaning.

BohuTANG commented 2 months ago

Is there some benchmark numbers for this PR?

@BohuTANG Here just replace the storage, where can I go to find the right environment to do benchmark? If the test environment is not representative, benchmark has no meaning.

I think @Dousir9 has some cases to do benchmark, because #16448 also need it.

Dousir9 commented 1 month ago

@forsaken628 Could you provide a benchmark to compare the query execution time with and without local spill ? You can set window_partition_spilling_memory_ratio to 0, 30, 60 for testing.

forsaken628 commented 1 month ago

benchmark:

settings:

set max_memory_usage = 16*1024*1024*1024;
set window_partition_spilling_memory_ratio = 30;

sql

EXPLAIN ANALYZE SELECT
    l_orderkey,
    l_partkey,
    l_quantity,
    l_extendedprice,
    ROW_NUMBER() OVER (PARTITION BY l_orderkey ORDER BY l_extendedprice DESC) AS row_num,
    RANK() OVER (PARTITION BY l_orderkey ORDER BY l_extendedprice DESC) AS rank_num
FROM
    lineitem ignore_result;

result

set window_partition_spilling_to_disk_bytes_limit = 0;
        ├── hash keys: [l_orderkey]
        ├── estimated rows: 600037902.00
        ├── cpu time: 474.358152575s
        ├── wait time: 197.201014451s
        ├── output rows: 600.04 million
        ├── output bytes: 26.82 GiB
        ├── numbers remote spilled by write: 112
        ├── bytes remote spilled by write: 26.69 GiB
        ├── remote spilled time by write: 181.29s
        ├── numbers remote spilled by read: 1535
        ├── bytes remote spilled by read: 26.69 GiB
        ├── remote spilled time by read: 102.371s
46 rows explain in 49.291 sec. Processed 0 rows, 0 B (0 row/s, 0 B/s)

set window_partition_spilling_to_disk_bytes_limit = 10*1024*1024*1024;
        ├── hash keys: [l_orderkey]
        ├── estimated rows: 600037902.00
        ├── cpu time: 972.806314767s
        ├── wait time: 277.454963277s
        ├── output rows: 600.04 million
        ├── output bytes: 26.82 GiB
        ├── numbers remote spilled by write: 84
        ├── bytes remote spilled by write: 16.72 GiB
        ├── remote spilled time by write: 681.639s
        ├── numbers remote spilled by read: 1328
        ├── bytes remote spilled by read: 16.72 GiB
        ├── remote spilled time by read: 53.13s
        ├── numbers local spilled by write: 282
        ├── bytes local spilled by write: 10.00 GiB
        ├── local spilled time by write: 78.319s
        ├── numbers local spilled by read: 4160
        ├── bytes local spilled by read: 10.00 GiB
        ├── local spilled time by read: 32.624s
52 rows explain in 102.438 sec. Processed 0 rows, 0 B (0 row/s, 0 B/s)

        ├── hash keys: [l_orderkey]
        ├── estimated rows: 600037902.00
        ├── cpu time: 896.504639969s
        ├── wait time: 279.228803569s
        ├── output rows: 600.04 million
        ├── output bytes: 26.82 GiB
        ├── numbers remote spilled by write: 95
        ├── bytes remote spilled by write: 16.75 GiB
        ├── remote spilled time by write: 535.451s
        ├── numbers remote spilled by read: 1497
        ├── bytes remote spilled by read: 16.75 GiB
        ├── remote spilled time by read: 44.231s
        ├── numbers local spilled by write: 430
        ├── bytes local spilled by write: 10.00 GiB
        ├── local spilled time by write: 64.566s
        ├── numbers local spilled by read: 6315
        ├── bytes local spilled by read: 10.00 GiB
        ├── local spilled time by read: 21.033s
52 rows explain in 91.103 sec. Processed 0 rows, 0 B (0 row/s, 0 B/s)

set window_partition_spilling_to_disk_bytes_limit = 30*1024*1024*1024;
        ├── hash keys: [l_orderkey]
        ├── estimated rows: 600037902.00
        ├── cpu time: 421.353882878s
        ├── wait time: 200.590871919s
        ├── output rows: 600.04 million
        ├── output bytes: 26.82 GiB
        ├── numbers local spilled by write: 1137
        ├── bytes local spilled by write: 26.74 GiB
        ├── local spilled time by write: 142.206s
        ├── numbers local spilled by read: 17646
        ├── bytes local spilled by read: 26.74 GiB
        ├── local spilled time by read: 56.713s
46 rows explain in 48.832 sec. Processed 0 rows, 0 B (0 row/s, 0 B/s)

The results are rather strange, when using only remote spills or just local spills, the time consumption is relatively normal. But when using a mix of them it causes a big jump in cpu time instead of wait time. The local test environment has only one hard disk, so the longer time may be caused by io contention.

Dousir9 commented 1 month ago

@forsaken628 Thanks for the benchmark result, we can try to read and write different IO devices for local spill and remote spill to determine whether it is IO contention.

forsaken628 commented 1 month ago

The big jump in CPU time when using a mix of local and remote spill has disappeared for some reason. It can now only be explained as an environmental issue, so keep watching.

        ├── hash keys: [l_orderkey]
        ├── estimated rows: 600037902.00
        ├── cpu time: 509.153733888s
        ├── wait time: 173.043153894s
        ├── output rows: 600.04 million
        ├── output bytes: 26.82 GiB
        ├── numbers remote spilled by write: 216
        ├── bytes remote spilled by write: 16.75 GiB
        ├── remote spilled time by write: 153.346s
        ├── numbers remote spilled by read: 3440
        ├── bytes remote spilled by read: 16.75 GiB
        ├── remote spilled time by read: 49.122s
        ├── numbers local spilled by write: 622
        ├── bytes local spilled by write: 10.00 GiB
        ├── local spilled time by write: 55.315s
        ├── numbers local spilled by read: 9712
        ├── bytes local spilled by read: 10.00 GiB
        ├── local spilled time by read: 26.595s