feat(storage): improve optimize and recluster

[zhyass]

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

Summary

1. Add limit for recluster.

ALTER TABLE [IF EXISTS] <name> RECLUSTER [FINAL] [WHERE condition] [LIMIT <segment_count>]

The option LIMIT sets the maximum number of segments to be recluster. Databend will select the newest segments. The default segment_ount limit is max_threads * 4.

2. Add memory usage limit for recluster

3. Sort the blockmeta by cluster_statistics during compact

4. after optimize compact, do recluster

5. Optimize recluster, serialize block in parallel during recluster.

create table test_order (
    id bigint,
    id1 bigint,
    id2 bigint,
    id3 bigint,
    id4 bigint,
    id5 bigint,
    id6 bigint,
    id7 bigint,
    
    s1 varchar,
    s2 varchar,
    s3 varchar,
    s4 varchar,
    s5 varchar,
    s6 varchar,
    s7 varchar,
    s8 varchar,
    s9 varchar,
    s10 varchar,
    s11 varchar,
    s12 varchar,
    s13 varchar,
    
    d1 DECIMAL(20, 8),
    d2 DECIMAL(20, 8),
    d3 DECIMAL(20, 8),
    d4 DECIMAL(20, 8),
    d5 DECIMAL(20, 8),
    d6 DECIMAL(30, 8),
    d7 DECIMAL(30, 8),
    d8 DECIMAL(30, 8),
    d9 DECIMAL(30, 8),
    d10 DECIMAL(30, 8),
    
    insert_time datetime,
    insert_time1 datetime,
    insert_time2 datetime,
    insert_time3 datetime,
    
    i int
) CLUSTER BY(id,insert_time);


create table random_source (
    id bigint,
    id1 bigint,
    id2 bigint,
    id3 bigint,
    id4 bigint,
    id5 bigint,
    id6 bigint,
    id7 bigint,
    
    s1 varchar,
    s2 varchar,
    s3 varchar,
    s4 varchar,
    s5 varchar,
    s6 varchar,
    s7 varchar,
    s8 varchar,
    s9 varchar,
    s10 varchar,
    s11 varchar,
    s12 varchar,
    s13 varchar,
    
    d1 DECIMAL(20, 8),
    d2 DECIMAL(20, 8),
    d3 DECIMAL(20, 8),
    d4 DECIMAL(20, 8),
    d5 DECIMAL(20, 8),
    d6 DECIMAL(30, 8),
    d7 DECIMAL(30, 8),
    d8 DECIMAL(30, 8),
    d9 DECIMAL(30, 8),
    d10 DECIMAL(30, 8),
    
    insert_time datetime,
    insert_time1 datetime,
    insert_time2 datetime,
    insert_time3 datetime,
    
    i int
) Engine = Random;

insert into test_order select * from random_source limit 50000000;

Before optimize:

mysql> optimize table test_order compact;
Query OK, 32000000 rows affected (3 min 10.62 sec)

After:

mysql> optimize table test_order compact;
Query OK, 32000000 rows affected (55.94 sec)

The recluster Pipeline is as follows:
┌──────────┐     ┌───────────────┐     ┌─────────┐
│FuseSource├────►│CompoundBlockOp├────►│SortMerge├────┐
└──────────┘     └───────────────┘     └─────────┘    │
┌──────────┐     ┌───────────────┐     ┌─────────┐    │     ┌──────────────┐     ┌─────────┐
│FuseSource├────►│CompoundBlockOp├────►│SortMerge├────┤────►│MultiSortMerge├────►│Resize(N)├───┐
└──────────┘     └───────────────┘     └─────────┘    │     └──────────────┘     └─────────┘   │
┌──────────┐     ┌───────────────┐     ┌─────────┐    │                                        │
│FuseSource├────►│CompoundBlockOp├────►│SortMerge├────┘                                        │
└──────────┘     └───────────────┘     └─────────┘                                             │
┌──────────────────────────────────────────────────────────────────────────────────────────────┘
│         ┌──────────────┐
│    ┌───►│SerializeBlock├───┐
│    │    └──────────────┘   │
│    │    ┌──────────────┐   │    ┌─────────┐    ┌────────────────┐     ┌─────────────────┐     ┌──────────┐
└───►│───►│SerializeBlock├───┤───►│Resize(1)├───►│SerializeSegment├────►│TableMutationAggr├────►│CommitSink│
     │    └──────────────┘   │    └─────────┘    └────────────────┘     └─────────────────┘     └──────────┘
     │    ┌──────────────┐   │
     └───►│SerializeBlock├───┘
          └──────────────┘

Closes #11799

[ZhiHanZ]

wonder about the mechanism, does the purge still needs to store candidate in memory and delete all candidates in a old segment?

[zhyass]

wonder about the mechanism, does the purge still needs to store candidate in memory and delete all candidates in a old segment?

will select and purge the oldest snapshots.

[BohuTANG]

Summary(By llmchain.rs)

• Added support for "limit" parameter in OPTIMIZE TABLE and ALTER TABLE RECLUSTER statements
  The code changes added support for the "limit" parameter in the ReclusterTable action, OptimizeTableStmt struct, and Display implementation of OptimizeTableStmt. The changes also added support for parsing the "limit" parameter in the OptimizeTableStmt and added new test cases for OPTIMIZE TABLE statement with different options and limits.

• Changed "dry_run_limit" parameter to "dry_run" boolean parameter
  The code changes modified the dry_run_limit parameter to a dry_run boolean parameter in the vacuum_handler.rs and handler.rs files. The changes also added a constant DRY_RUN_LIMIT and modified the function signature of do_vacuum to take a boolean dry_run instead of an optional dry_run_limit.

• Modified function signatures to take a Vec instead of a slice &[DataBlock]
  The code changes modified the function signatures of compact_final in transform_block_compact.rs, transform_block_compact_for_copy.rs, and transform_compact.rs to take a Vec<DataBlock> instead of a slice &[DataBlock]. The changes also modified the consume_event function to check if the output port can push before pushing the next data block.

• Added support for TransformSerializeBlock and TransformSerializeSegment
  The code changes added support for TransformSerializeBlock and TransformSerializeSegment in various files, including transform_append.rs, transform_serialize_data.rs, and replace.rs. The changes also removed the unused AppendTransform and added a new module mutation_meta.

• Added limit parameter to various functions
  The code changes added a limit parameter to the compact method in table.rs, the do_purge function in gc.rs, and the do_recluster function in recluster.rs. The changes also modified the implementation of the do_recluster function to use the limit parameter to limit the number of segment locations processed at a time.

• Changed various function parameters and return types
  The code changes changed the dry_run_limit parameter to a boolean dry_run parameter in various files, including vacuum_handler.rs, handler.rs, and gc.rs. The changes also changed the return type of apply_delete function in merge_into_mutator.rs and the function signature of compact_table function in hive_table.rs.

[flaneur2020]

wonder about the mechanism, does the purge still needs to store candidate in memory and delete all candidates in a old segment?

IMO purge do not need scan the world, but only need just tailing the older snapshot & segment & block files.

If I understands correctly, the block files have some kind of time ordering, if a block file's creation time earlier than the earliest active snapshot & not included in this snapshot, then it can be safely purged.

[zhyass]

IMO purge do not need scan the world, but only need just tailing the older snapshot & segment & block files.

If I understands correctly, the block files have some kind of time ordering, if a block file's creation time earlier than the earliest active snapshot & not included in this snapshot, then it can be safely purged.

This can be used by purge orphan blocks in vacuum table.

However, we cannot guarantee the accuracy of the file's creation time and the creation time of block is generated earlier than the creation time of snapshot.

[github-actions]

Docker Image for PR

• tag: pr-11850-f689982

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