Skip to content

Commit fe693bb

Browse files
zhyasszhyass
authored
refactor(storage): compact/recluster flow and clustering stats derivation (databendlabs#19754)
* fix: compact and recluster fix test * fix * fix * fix review comments fix review comments fix test fix test fix test * fix test * fix test * fix test * fix * add test
1 parent 1edc6ef commit fe693bb

47 files changed

Lines changed: 1098 additions & 1163 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎src/query/catalog/src/plan/datasource/datasource_plan.rs‎

Lines changed: 28 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,33 @@ use crate::plan::PushDownInfo;
2525
use crate::plan::datasource::datasource_info::DataSourceInfo;
2626
use crate::table_args::TableArgs;
2727

28+
#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Default)]
29+
pub struct BlockMetaOptions {
30+
// for merge_into target build.
31+
pub reserve_block_index: bool,
32+
// Whether to update stream columns.
33+
pub update_stream_columns: bool,
34+
// used to query internal columns.
35+
pub query_internal_columns: bool,
36+
}
37+
38+
impl BlockMetaOptions {
39+
pub fn set_reserve_block_index(mut self, reserve_block_index: bool) -> Self {
40+
self.reserve_block_index = reserve_block_index;
41+
self
42+
}
43+
44+
pub fn set_update_stream_columns(mut self, update_stream_columns: bool) -> Self {
45+
self.update_stream_columns = update_stream_columns;
46+
self
47+
}
48+
49+
pub fn set_query_internal_columns(mut self, query_internal_columns: bool) -> Self {
50+
self.query_internal_columns = query_internal_columns;
51+
self
52+
}
53+
}
54+
2855
#[derive(serde::Serialize, serde::Deserialize, Clone, Debug)]
2956
pub struct DataSourcePlan {
3057
pub source_info: DataSourceInfo,
@@ -39,8 +66,7 @@ pub struct DataSourcePlan {
3966
pub push_downs: Option<PushDownInfo>,
4067
pub internal_columns: Option<BTreeMap<FieldIndex, InternalColumn>>,
4168
pub base_block_ids: Option<Scalar>,
42-
// used for recluster to update stream columns
43-
pub update_stream_columns: bool,
69+
pub block_meta_options: BlockMetaOptions,
4470

4571
pub table_index: usize,
4672
pub scan_id: usize,

‎src/query/catalog/src/plan/datasource/mod.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,4 +16,5 @@ mod datasource_info;
1616
mod datasource_plan;
1717

1818
pub use datasource_info::*;
19+
pub use datasource_plan::BlockMetaOptions;
1920
pub use datasource_plan::DataSourcePlan;

‎src/query/catalog/src/plan/partition.rs‎

Lines changed: 9 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -371,53 +371,21 @@ pub struct ReclusterTask {
371371

372372
pub type BlockMetaWithHLL = (Arc<BlockMeta>, Option<RawBlockHLL>);
373373

374-
#[derive(Clone)]
375-
pub enum ReclusterParts {
376-
Recluster {
377-
tasks: Vec<ReclusterTask>,
378-
remained_blocks: Vec<BlockMetaWithHLL>,
379-
removed_segment_indexes: Vec<usize>,
380-
removed_segment_summary: Statistics,
381-
},
382-
Compact(Partitions),
374+
#[derive(Clone, Default)]
375+
pub struct ReclusterParts {
376+
pub tasks: Vec<ReclusterTask>,
377+
pub remained_blocks: Vec<BlockMetaWithHLL>,
378+
pub removed_segment_indexes: Vec<usize>,
379+
pub removed_segment_summary: Statistics,
383380
}
384381

385382
impl ReclusterParts {
386383
pub fn is_empty(&self) -> bool {
387-
match self {
388-
ReclusterParts::Recluster {
389-
tasks,
390-
remained_blocks,
391-
..
392-
} => tasks.is_empty() && remained_blocks.is_empty(),
393-
ReclusterParts::Compact(parts) => parts.is_empty(),
394-
}
395-
}
396-
397-
pub fn new_recluster_parts() -> Self {
398-
Self::Recluster {
399-
tasks: vec![],
400-
remained_blocks: vec![],
401-
removed_segment_indexes: vec![],
402-
removed_segment_summary: Statistics::default(),
403-
}
404-
}
405-
406-
pub fn new_compact_parts() -> Self {
407-
Self::Compact(Partitions::default())
384+
self.tasks.is_empty() && self.remained_blocks.is_empty()
408385
}
409386

410-
pub fn is_distributed(&self, ctx: Arc<dyn TableContext>) -> bool {
411-
match self {
412-
ReclusterParts::Recluster { tasks, .. } => tasks.len() > 1,
413-
ReclusterParts::Compact(_) => {
414-
(!ctx.get_cluster().is_empty())
415-
&& ctx
416-
.get_settings()
417-
.get_enable_distributed_compact()
418-
.unwrap_or(false)
419-
}
420-
}
387+
pub fn is_distributed(&self, _ctx: Arc<dyn TableContext>) -> bool {
388+
self.tasks.len() > 1
421389
}
422390
}
423391

‎src/query/ee/src/storages/fuse/operations/virtual_columns.rs‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -129,8 +129,7 @@ pub async fn prepare_refresh_virtual_column(
129129
let virtual_column_builder = VirtualColumnBuilder::try_create(ctx.clone(), source_schema)?;
130130

131131
let projection = Projection::Columns(field_indices);
132-
let block_reader =
133-
fuse_table.create_block_reader(ctx.clone(), projection, false, false, false)?;
132+
let block_reader = fuse_table.create_block_reader(ctx.clone(), projection, false)?;
134133

135134
let segment_reader = MetaReaders::segment_info_reader(fuse_table.get_operator(), table_schema);
136135

‎src/query/expression/src/block.rs‎

Lines changed: 0 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -689,24 +689,6 @@ impl DataBlock {
689689
res
690690
}
691691

692-
pub fn split_by_rows_if_needed_no_tail(&self, rows_per_block: usize) -> Vec<Self> {
693-
// Since rows_per_block represents the expected number of rows per block,
694-
// and the minimum number of rows per block is 0.8 * rows_per_block,
695-
// the maximum is taken as 1.8 * rows_per_block.
696-
let max_rows_per_block = (rows_per_block * 9).div_ceil(5);
697-
let mut res = vec![];
698-
let mut offset = 0;
699-
let mut remain_rows = self.num_rows;
700-
while remain_rows >= max_rows_per_block {
701-
let cut = self.slice(offset..(offset + rows_per_block));
702-
res.push(cut);
703-
offset += rows_per_block;
704-
remain_rows -= rows_per_block;
705-
}
706-
res.push(self.slice(offset..(offset + remain_rows)));
707-
res
708-
}
709-
710692
#[inline]
711693
pub fn merge_block(&mut self, block: DataBlock) {
712694
self.entries.reserve(block.num_columns());

‎src/query/expression/src/utils/block_thresholds.rs‎

Lines changed: 70 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@ use databend_common_io::constants::DEFAULT_BLOCK_COMPRESSED_SIZE;
1717
use databend_common_io::constants::DEFAULT_BLOCK_PER_SEGMENT;
1818
use databend_common_io::constants::DEFAULT_BLOCK_ROW_COUNT;
1919

20+
const MAX_BYTES_PER_BLOCK_FACTOR: usize = 2;
21+
2022
#[derive(Clone, Copy, Debug, serde::Serialize, serde::Deserialize)]
2123
pub struct BlockThresholds {
2224
pub max_rows_per_block: usize,
@@ -35,11 +37,11 @@ impl Default for BlockThresholds {
3537
fn default() -> BlockThresholds {
3638
BlockThresholds {
3739
max_rows_per_block: DEFAULT_BLOCK_ROW_COUNT,
38-
min_rows_per_block: (DEFAULT_BLOCK_ROW_COUNT * 4).div_ceil(5),
39-
max_bytes_per_block: DEFAULT_BLOCK_BUFFER_SIZE * 2,
40-
min_bytes_per_block: (DEFAULT_BLOCK_BUFFER_SIZE * 4).div_ceil(5),
40+
min_rows_per_block: Self::min_block_threshold(DEFAULT_BLOCK_ROW_COUNT),
41+
max_bytes_per_block: DEFAULT_BLOCK_BUFFER_SIZE * MAX_BYTES_PER_BLOCK_FACTOR,
42+
min_bytes_per_block: Self::min_block_threshold(DEFAULT_BLOCK_BUFFER_SIZE),
4143
max_compressed_per_block: DEFAULT_BLOCK_COMPRESSED_SIZE,
42-
min_compressed_per_block: (DEFAULT_BLOCK_COMPRESSED_SIZE * 4).div_ceil(5),
44+
min_compressed_per_block: Self::min_block_threshold(DEFAULT_BLOCK_COMPRESSED_SIZE),
4345
block_per_segment: DEFAULT_BLOCK_PER_SEGMENT,
4446
}
4547
}
@@ -54,15 +56,34 @@ impl BlockThresholds {
5456
) -> Self {
5557
BlockThresholds {
5658
max_rows_per_block,
57-
min_rows_per_block: (max_rows_per_block * 4).div_ceil(5),
58-
max_bytes_per_block: bytes_per_block * 2,
59-
min_bytes_per_block: (bytes_per_block * 4).div_ceil(5),
59+
min_rows_per_block: Self::min_block_threshold(max_rows_per_block),
60+
max_bytes_per_block: bytes_per_block * MAX_BYTES_PER_BLOCK_FACTOR,
61+
min_bytes_per_block: Self::min_block_threshold(bytes_per_block),
6062
max_compressed_per_block,
61-
min_compressed_per_block: (max_compressed_per_block * 4).div_ceil(5),
63+
min_compressed_per_block: Self::min_block_threshold(max_compressed_per_block),
6264
block_per_segment,
6365
}
6466
}
6567

68+
#[inline]
69+
pub fn set_rows_per_block(mut self, rows_per_block: usize) -> Self {
70+
self.max_rows_per_block = rows_per_block;
71+
self.min_rows_per_block = Self::min_block_threshold(rows_per_block);
72+
self
73+
}
74+
75+
#[inline]
76+
pub fn set_bytes_per_block(mut self, bytes_per_block: usize) -> Self {
77+
self.max_bytes_per_block = bytes_per_block * MAX_BYTES_PER_BLOCK_FACTOR;
78+
self.min_bytes_per_block = Self::min_block_threshold(bytes_per_block);
79+
self
80+
}
81+
82+
#[inline]
83+
pub fn min_block_threshold(value: usize) -> usize {
84+
(value * 4).div_ceil(5)
85+
}
86+
6687
#[inline]
6788
pub fn check_perfect_block(
6889
&self,
@@ -112,63 +133,80 @@ impl BlockThresholds {
112133
}
113134

114135
#[inline]
115-
pub fn calc_rows_for_compact(&self, total_bytes: usize, total_rows: usize) -> usize {
116-
if self.check_for_compact(total_rows, total_bytes) {
117-
return total_rows;
118-
}
136+
pub fn calc_compact_block_num(&self, total_rows: usize, total_bytes: usize) -> usize {
137+
let block_num_by_rows = if total_rows >= 2 * self.min_rows_per_block {
138+
(total_rows / self.max_rows_per_block).max(2)
139+
} else {
140+
1
141+
};
119142

120-
let block_num_by_rows = std::cmp::max(total_rows / self.min_rows_per_block, 1);
121-
let block_num_by_size = total_bytes / self.min_bytes_per_block;
122-
if block_num_by_rows >= block_num_by_size {
123-
return self.max_rows_per_block;
124-
}
125-
total_rows.div_ceil(block_num_by_size)
143+
let bytes_per_block = self.max_bytes_per_block / MAX_BYTES_PER_BLOCK_FACTOR;
144+
let block_num_by_bytes = if total_bytes >= 2 * self.min_bytes_per_block {
145+
(total_bytes / bytes_per_block).max(2)
146+
} else {
147+
1
148+
};
149+
150+
block_num_by_rows.max(block_num_by_bytes).min(total_rows)
126151
}
127152

128-
/// Calculates the optimal number of rows per block based on total data size and row count.
153+
/// Calculates the optimal rows and bytes per block based on total data size and row count.
129154
///
130155
/// # Parameters
131156
/// - `total_bytes`: The total size of the data in bytes.
132157
/// - `total_rows`: The total number of rows in the data.
133158
/// - `total_compressed`: The total compressed size of the data in bytes.
134159
///
135160
/// # Returns
136-
/// - The calculated number of rows per block that satisfies the thresholds.
161+
/// - `(rows_per_block, bytes_per_block)`: rows are used as the sort block size,
162+
/// and bytes are used by ordered compact to keep post-sort blocks near the
163+
/// recluster target.
137164
#[inline]
138165
pub fn calc_rows_for_recluster(
139166
&self,
140167
total_rows: usize,
141168
total_bytes: usize,
142169
total_compressed: usize,
143-
) -> usize {
170+
) -> (usize, usize) {
171+
debug_assert!(total_rows > 0);
172+
173+
let default_bytes_per_block = self
174+
.max_bytes_per_block
175+
.div_ceil(MAX_BYTES_PER_BLOCK_FACTOR);
144176
// Check if the data is compact enough to skip further calculations.
145177
if self.check_for_compact(total_rows, total_bytes)
146178
&& total_compressed < 2 * self.min_compressed_per_block
147179
{
148-
return total_rows;
180+
return (total_rows, default_bytes_per_block);
149181
}
150182

151183
let block_num_by_rows = std::cmp::max(total_rows / self.min_rows_per_block, 1);
152184
let block_num_by_compressed = total_compressed.div_ceil(self.max_compressed_per_block);
153185
// If row-based block count exceeds compressed-based block count, use max rows per block.
154186
if block_num_by_rows >= block_num_by_compressed {
155-
return self.max_rows_per_block;
187+
return (self.max_rows_per_block, default_bytes_per_block);
156188
}
157189

158190
let bytes_per_block = total_bytes.div_ceil(block_num_by_compressed);
159191
// Adjust the number of blocks based on block size thresholds.
160-
let max_bytes_per_block = self.max_bytes_per_block.min(400 * 1024 * 1024);
161-
let min_bytes_per_block = (self.min_bytes_per_block / 2).min(50 * 1024 * 1024);
162-
let block_nums = if bytes_per_block > max_bytes_per_block {
192+
let max_bytes_per_block =
193+
default_bytes_per_block + default_bytes_per_block.min(DEFAULT_BLOCK_BUFFER_SIZE);
194+
if bytes_per_block > max_bytes_per_block {
163195
// Case 1: If the block size is too bigger.
164-
total_bytes.div_ceil(max_bytes_per_block)
165-
} else if bytes_per_block < min_bytes_per_block {
196+
let bytes_per_block = max_bytes_per_block;
197+
let block_nums = total_bytes.div_ceil(bytes_per_block);
198+
(total_rows.div_ceil(block_nums).max(1), bytes_per_block)
199+
} else if bytes_per_block < self.min_bytes_per_block {
166200
// Case 2: If the block size is too smaller.
167-
total_bytes / min_bytes_per_block
201+
let bytes_per_block = self.min_bytes_per_block;
202+
let block_nums = std::cmp::max(total_bytes / bytes_per_block, 1);
203+
(total_rows.div_ceil(block_nums).max(1), bytes_per_block)
168204
} else {
169205
// Case 3: Otherwise, use the compressed-based block count.
170-
block_num_by_compressed
171-
};
172-
total_rows.div_ceil(block_nums.max(1)).max(1)
206+
(
207+
total_rows.div_ceil(block_num_by_compressed).max(1),
208+
bytes_per_block,
209+
)
210+
}
173211
}
174212
}

‎src/query/expression/tests/it/block.rs‎

Lines changed: 0 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -31,21 +31,6 @@ use databend_common_expression::types::StringType;
3131
use databend_common_expression::types::number::NumberScalar;
3232
use databend_common_expression::types::string::StringColumnBuilder;
3333

34-
#[test]
35-
fn test_split_block() {
36-
let value = "abc";
37-
let n = 10;
38-
let block = DataBlock::new_from_columns(vec![Column::String(
39-
StringColumnBuilder::repeat(value, n).build(),
40-
)]);
41-
let sizes = block
42-
.split_by_rows_if_needed_no_tail(3)
43-
.iter()
44-
.map(|b| b.num_rows())
45-
.collect::<Vec<_>>();
46-
assert_eq!(sizes, vec![3, 3, 4]);
47-
}
48-
4934
#[test]
5035
fn test_box_render_block() {
5136
let value = "abc";

‎src/query/expression/tests/it/block_thresholds.rs‎

Lines changed: 15 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -72,43 +72,41 @@ fn test_check_too_small() {
7272
}
7373

7474
#[test]
75-
fn test_calc_rows_for_compact() {
75+
fn test_calc_compact_block_num() {
7676
let t = default_thresholds();
7777

78-
assert_eq!(t.calc_rows_for_compact(500_000, 1000), 1000);
79-
80-
// Block number by rows wins → max_rows_per_block
81-
assert_eq!(
82-
t.calc_rows_for_compact(2_000_000, 10_000),
83-
t.max_rows_per_block
84-
);
85-
86-
// Size-based block number wins
87-
assert_eq!(t.calc_rows_for_compact(4_000_000, 2000), 400);
78+
assert_eq!(t.calc_compact_block_num(1000, 500_000), 1);
79+
assert_eq!(t.calc_compact_block_num(1800, 500_000), 2);
80+
assert_eq!(t.calc_compact_block_num(10_000, 500_000), 10);
81+
assert_eq!(t.calc_compact_block_num(1000, 1_800_000), 2);
82+
assert_eq!(t.calc_compact_block_num(1000, 4_000_000), 4);
8883
}
8984

9085
#[test]
9186
fn test_calc_rows_for_recluster() {
9287
let t = default_thresholds();
9388

9489
// compact enough to skip further calculations
95-
assert_eq!(t.calc_rows_for_recluster(1000, 500_000, 100_000), 1000);
90+
assert_eq!(
91+
t.calc_rows_for_recluster(1000, 500_000, 100_000),
92+
(1000, 1_000_000)
93+
);
9694

9795
// row-based block count exceeds compressed-based block count, use max rows per block.
9896
assert_eq!(
9997
t.calc_rows_for_recluster(10_000, 2_000_000, 100_000),
100-
t.max_rows_per_block
98+
(t.max_rows_per_block, 1_000_000)
10199
);
102100

103101
// Case 1: If the block size is too bigger.
104102
let result = t.calc_rows_for_recluster(4_000, 30_000_000, 600_000);
105-
assert_eq!(result, 267);
103+
assert_eq!(result, (267, 2_000_000));
106104

107105
// Case 2: If the block size is too smaller.
108-
let result = t.calc_rows_for_recluster(4_000, 2_000_000, 600_000);
109-
assert_eq!(result, 800);
106+
let result = t.calc_rows_for_recluster(4_000, 1_600_000, 600_000);
107+
assert_eq!(result, (2000, 800_000));
110108

111109
// Case 3: use the compressed-based block count.
112110
let result = t.calc_rows_for_recluster(4_000, 10_000_000, 600_000);
113-
assert_eq!(result, 667);
111+
assert_eq!(result, (667, 1_666_667));
114112
}

0 commit comments

Comments
 (0)