feat(storage): refactor recluster

[zhyass]

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

Summary

This PR introduces several enhancements and optimizations related to the reclustering process:

1. Addition of RelOperator::Recluster:

   • Refactored the method for generating the physical plan for reclustering. Thus, the physical plan will be generated by PhysicalPlanBuilder.

2. Optimize Compact Execution:

   • Modified the optimize compact logic to directly perform reclustering when the target table is a clustered table. This change eliminates the need for a separate compact operation followed by reclustering, thereby reducing the overall execution time and complexity.

3. Segment-level Sorting:

   • Introduced a segment-level sorting mechanism. Even when block-level reclustering is not necessary, if there are overlapping segments, they will be sorted to create new segments. This ensures better data organization and improved query performance.

mysql> create table t(a int) cluster by(a) block_per_segment=2 row_per_block=3;
Query OK, 0 rows affected (0.06 sec)

mysql> insert into t values(0),(1),(2);
Query OK, 3 rows affected (0.08 sec)

mysql> insert into t values(6),(7),(8);
Query OK, 3 rows affected (0.07 sec)

mysql> optimize table t compact;
Query OK, 0 rows affected (0.24 sec)

mysql> insert into t values(3),(4),(5);
Query OK, 3 rows affected (0.07 sec)

mysql>  insert into t values(9),(10),(11);
Query OK, 3 rows affected (0.08 sec)

mysql> optimize table t compact segment;
Query OK, 0 rows affected (0.21 sec)

mysql> select * from clustering_information('default','t');
+-------------+----------------------------+-------------------+----------------------+------------------+---------------+-----------------------+
| cluster_key | timestamp                  | total_block_count | constant_block_count | average_overlaps | average_depth | block_depth_histogram |
+-------------+----------------------------+-------------------+----------------------+------------------+---------------+-----------------------+
| (a)         | 2024-07-18 10:35:48.426716 |                 4 |                    0 |              0.0 |           1.0 | {"00001":4}           |
+-------------+----------------------------+-------------------+----------------------+------------------+---------------+-----------------------+
1 row in set (0.05 sec)
Read 1 rows, 448.00 B in 0.025 sec., 40.34 rows/sec., 17.65 KiB/sec.

mysql> explain select * from t where a=6;
+--------------------------------------------------------------------------------------------------------------------------+
| explain                                                                                                                  |
+--------------------------------------------------------------------------------------------------------------------------+
| Filter                                                                                                                   |
| ├── output columns: [t.a (#0)]                                                                                           |
| ├── filters: [is_true(t.a (#0) = 6)]                                                                                     |
| ├── estimated rows: 1.00                                                                                                 |
| └── TableScan                                                                                                            |
|     ├── table: default.default.t                                                                                         |
|     ├── output columns: [a (#0)]                                                                                         |
|     ├── read rows: 3                                                                                                     |
|     ├── read size: < 1 KiB                                                                                               |
|     ├── partitions total: 4                                                                                              |
|     ├── partitions scanned: 1                                                                                            |
|     ├── pruning stats: [segments: <range pruning: 2 to 2>, blocks: <range pruning: 4 to 1, bloom pruning: 1 to 1>]       |
|     ├── push downs: [filters: [is_true(t.a (#0) = 6)], limit: NONE]                                                      |
|     └── estimated rows: 12.00                                                                                            |
+--------------------------------------------------------------------------------------------------------------------------+
14 rows in set (0.08 sec)
Read 0 rows, 0.00 B in 0.019 sec., 0 rows/sec., 0.00 B/sec.

mysql> optimize table t compact;
Query OK, 0 rows affected (0.21 sec)

mysql> select * from clustering_information('default','t');
+-------------+----------------------------+-------------------+----------------------+------------------+---------------+-----------------------+
| cluster_key | timestamp                  | total_block_count | constant_block_count | average_overlaps | average_depth | block_depth_histogram |
+-------------+----------------------------+-------------------+----------------------+------------------+---------------+-----------------------+
| (a)         | 2024-07-18 10:36:10.959885 |                 4 |                    0 |              0.0 |           1.0 | {"00001":4}           |
+-------------+----------------------------+-------------------+----------------------+------------------+---------------+-----------------------+
1 row in set (0.05 sec)
Read 1 rows, 448.00 B in 0.024 sec., 41.35 rows/sec., 18.09 KiB/sec.

mysql> explain select * from t where a=6;
+--------------------------------------------------------------------------------------------------------------------------+
| explain                                                                                                                  |
+--------------------------------------------------------------------------------------------------------------------------+
| Filter                                                                                                                   |
| ├── output columns: [t.a (#0)]                                                                                           |
| ├── filters: [is_true(t.a (#0) = 6)]                                                                                     |
| ├── estimated rows: 1.00                                                                                                 |
| └── TableScan                                                                                                            |
|     ├── table: default.default.t                                                                                         |
|     ├── output columns: [a (#0)]                                                                                         |
|     ├── read rows: 3                                                                                                     |
|     ├── read size: < 1 KiB                                                                                               |
|     ├── partitions total: 4                                                                                              |
|     ├── partitions scanned: 1                                                                                            |
|     ├── pruning stats: [segments: <range pruning: 2 to 1>, blocks: <range pruning: 2 to 1, bloom pruning: 1 to 1>]       |
|     ├── push downs: [filters: [is_true(t.a (#0) = 6)], limit: NONE]                                                      |
|     └── estimated rows: 12.00                                                                                            |
+--------------------------------------------------------------------------------------------------------------------------+
14 rows in set (0.05 sec)
Read 0 rows, 0.00 B in 0.008 sec., 0 rows/sec., 0.00 B/sec.

mysql> select segment_count, block_count, row_count from fuse_snapshot('default', 't');
+---------------+-------------+-----------+
| segment_count | block_count | row_count |
+---------------+-------------+-----------+
|             2 |           4 |        12 |
|             2 |           4 |        12 |
|             3 |           4 |        12 |
|             2 |           3 |         9 |
|             1 |           2 |         6 |
|             2 |           2 |         6 |
|             1 |           1 |         3 |
+---------------+-------------+-----------+
7 rows in set (0.05 sec)
Read 7 rows, 2.77 KiB in 0.026 sec., 538 rows/sec., 106.50 KiB/sec.

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 

[BohuTANG]

How much faster is the new recluster than the old?

[zhyass]

How much faster is the new recluster than the old?

The point of this pr is not to increase execution speed.

Instead, it modifies some of the logic of compact and recluster.

For example, recluster will implicitly execute compact. Compact on a cluster table will directly execute recluster instead of compact+recluster.