risingwave_hummock_sdk/
compact_task.rs

1// Copyright 2024 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::{BTreeMap, HashMap};
16
17use itertools::Itertools;
18use risingwave_common::catalog::TableId;
19use risingwave_pb::hummock::compact_task::{PbTaskStatus, PbTaskType, TaskStatus, TaskType};
20use risingwave_pb::hummock::subscribe_compaction_event_request::PbReportTask;
21use risingwave_pb::hummock::{
22    CompactTaskAssignment as PbCompactTaskAssignment, LevelType, PbCompactTask, PbKeyRange,
23    PbSstableFilterLayout, PbSstableFilterType, PbTableOption, PbTableSchema, PbTableStats,
24    PbValidationTask,
25};
26use risingwave_pb::id::WorkerId;
27
28use crate::compaction_group::StateTableId;
29use crate::key_range::KeyRange;
30use crate::level::InputLevel;
31use crate::sstable_info::SstableInfo;
32use crate::table_watermark::{TableWatermarks, WatermarkSerdeType};
33use crate::{CompactionGroupId, HummockContextId, HummockSstableObjectId};
34
35#[derive(Clone, PartialEq, Default, Debug)]
36pub struct CompactTask {
37    /// SSTs selected by meta, which will be removed from LSM after compaction succeeds.
38    ///
39    /// Each SST's `table_ids` comes from version metadata, where truncate/drop table deltas prune
40    /// table ids before later compaction tasks are picked.
41    pub input_ssts: Vec<InputLevel>,
42    /// In ideal case, the compaction will generate `splits.len()` tables which have key range
43    /// corresponding to that in `splits`, respectively
44    pub splits: Vec<KeyRange>,
45    /// compaction output, which will be added to `target_level` of LSM after compaction
46    pub sorted_output_ssts: Vec<SstableInfo>,
47    /// task id assigned by hummock storage service
48    pub task_id: u64,
49    /// compaction output will be added to `target_level` of LSM after compaction
50    pub target_level: u32,
51    pub gc_delete_keys: bool,
52    /// Lbase in LSM
53    pub base_level: u32,
54    pub task_status: PbTaskStatus,
55    /// compaction group the task belongs to.
56    pub compaction_group_id: CompactionGroupId,
57    /// compaction group id when the compaction task is created
58    pub compaction_group_version_id: u64,
59    /// Table ids that still exist in the version when the compaction task is picked.
60    ///
61    /// This may be broader than the table ids touched by this task's input SSTs. Task-local table
62    /// ids should be read from `input_ssts[*].table_ids`.
63    pub existing_table_ids: Vec<TableId>,
64    pub compression_algorithm: u32,
65    pub target_file_size: u64,
66    pub compaction_filter_mask: u32,
67    pub table_options: BTreeMap<TableId, PbTableOption>,
68    pub current_epoch_time: u64,
69    pub target_sub_level_id: u64,
70    /// Identifies whether the task is `space_reclaim`, if the `compact_task_type` increases, it will be refactored to enum
71    pub task_type: PbTaskType,
72    /// Deprecated. use `table_vnode_partition` instead;
73    pub split_by_state_table: bool,
74    /// Compaction needs to cut the state table every time 1/weight of vnodes in the table have been processed.
75    /// Deprecated. use `table_vnode_partition` instead;
76    pub split_weight_by_vnode: u32,
77    pub table_vnode_partition: BTreeMap<TableId, u32>,
78    /// The table watermark of any table id. In compaction we only use the table watermarks on safe epoch,
79    /// so we only need to include the table watermarks on safe epoch to reduce the size of metadata.
80    pub pk_prefix_table_watermarks: BTreeMap<TableId, TableWatermarks>,
81
82    pub non_pk_prefix_table_watermarks: BTreeMap<TableId, TableWatermarks>,
83    pub value_table_watermarks: BTreeMap<TableId, TableWatermarks>,
84
85    pub table_schemas: BTreeMap<TableId, PbTableSchema>,
86
87    pub max_sub_compaction: u32,
88
89    pub blocked_xor_filter_kv_count_threshold: Option<u64>,
90
91    pub max_vnode_key_range_bytes: Option<u64>,
92
93    pub sstable_filter_type: PbSstableFilterType,
94
95    pub sstable_filter_layout: PbSstableFilterLayout,
96}
97
98#[derive(Clone, PartialEq, Default, Debug)]
99pub struct CompactTaskAssignment {
100    pub compact_task: CompactTask,
101    /// Worker selected to execute the task. Legacy assignments may contain zero.
102    pub context_id: HummockContextId,
103}
104
105impl CompactTask {
106    pub fn is_trivial_move_task(&self) -> bool {
107        if self.task_type != TaskType::Dynamic && self.task_type != TaskType::Emergency {
108            return false;
109        }
110
111        if self.input_ssts.len() != 2 || self.input_ssts[0].level_type != LevelType::Nonoverlapping
112        {
113            return false;
114        }
115
116        // it may be a manual compaction task
117        if self.input_ssts[0].level_idx == self.input_ssts[1].level_idx
118            && self.input_ssts[0].level_idx > 0
119        {
120            return false;
121        }
122
123        if self.input_ssts[1].level_idx == self.target_level
124            && self.input_ssts[1].table_infos.is_empty()
125        {
126            return true;
127        }
128
129        false
130    }
131
132    pub fn task_label(&self) -> &'static str {
133        if self.is_trivial_reclaim() {
134            return "trivial-space-reclaim";
135        }
136
137        if self.is_trivial_move_task() {
138            return "trivial-move";
139        }
140
141        "normal"
142    }
143
144    pub fn is_trivial_reclaim(&self) -> bool {
145        // Currently all VnodeWatermark tasks are trivial reclaim.
146        if self.task_type == TaskType::VnodeWatermark {
147            return true;
148        }
149        self.input_ssts
150            .iter()
151            .flat_map(|level| level.table_infos.iter())
152            .all(|sst| sst.table_ids.is_empty())
153    }
154}
155
156impl CompactTask {
157    // The compact task may need to reclaim key with TTL
158    pub fn contains_ttl(&self) -> bool {
159        self.table_options
160            .iter()
161            .any(|(_, table_option)| table_option.retention_seconds.is_some_and(|ttl| ttl > 0))
162    }
163
164    // The compact task may need to reclaim key with range tombstone
165    pub fn contains_range_tombstone(&self) -> bool {
166        self.read_input_ssts()
167            .any(|sst| sst.range_tombstone_count > 0)
168    }
169
170    // The compact task may need to reclaim key with split sst
171    pub fn contains_split_sst(&self) -> bool {
172        self.read_input_ssts()
173            .any(|sst| sst.sst_id.as_raw_id() != sst.object_id.as_raw_id())
174    }
175
176    /// Returns sorted and deduplicated table ids from this task's normalized input SST metadata.
177    pub fn get_table_ids_from_input_ssts(&self) -> impl Iterator<Item = StateTableId> + use<> {
178        self.input_ssts
179            .iter()
180            .flat_map(|level| level.table_infos.iter())
181            .flat_map(|sst| sst.table_ids.iter().copied())
182            .sorted()
183            .dedup()
184    }
185
186    pub fn is_expired(&self, compaction_group_version_id_expected: u64) -> bool {
187        is_compaction_task_expired(
188            self.compaction_group_version_id,
189            compaction_group_version_id_expected,
190        )
191    }
192
193    /// Resolves the physical filter layout for one output SST.
194    ///
195    /// The caller should pass an output-SST-level key-count estimate. This differs from the legacy
196    /// task-level check below: a large compaction task can produce many small output SSTs, and those
197    /// outputs should be allowed to use plain filters when each output is below the threshold.
198    pub fn sstable_filter_layout_for_output(
199        &self,
200        estimated_output_key_count: u64,
201    ) -> PbSstableFilterLayout {
202        match self.sstable_filter_layout {
203            PbSstableFilterLayout::Plain => PbSstableFilterLayout::Plain,
204            PbSstableFilterLayout::Blocked => PbSstableFilterLayout::Blocked,
205            PbSstableFilterLayout::Auto | PbSstableFilterLayout::Unspecified => {
206                if crate::filter_utils::should_use_blocked_xor_filter_by_kv_count(
207                    estimated_output_key_count,
208                    self.blocked_xor_filter_kv_count_threshold,
209                ) {
210                    PbSstableFilterLayout::Blocked
211                } else {
212                    PbSstableFilterLayout::Plain
213                }
214            }
215        }
216    }
217
218    /// Returns the effective vnode key-range hint limit (in bytes) for this compaction task.
219    ///
220    /// The hint is only meaningful when all input SSTs belong to a single table.
221    pub fn effective_max_vnode_key_range_bytes(&self) -> Option<usize> {
222        let limit = self.max_vnode_key_range_bytes.filter(|&v| v > 0)? as usize;
223        (self.get_table_ids_from_input_ssts().count() == 1).then_some(limit)
224    }
225
226    /// Returns input SSTs that should be read by the compactor.
227    ///
228    /// SST entries with empty `table_ids` are ignored defensively and should not be read by the
229    /// compactor.
230    pub fn read_input_ssts(&self) -> impl Iterator<Item = &SstableInfo> {
231        self.input_ssts
232            .iter()
233            .flat_map(|level| level.read_sstable_infos())
234    }
235}
236
237fn split_watermark_serde_types(
238    pb_compact_task: &PbCompactTask,
239) -> (
240    BTreeMap<TableId, TableWatermarks>,
241    BTreeMap<TableId, TableWatermarks>,
242    BTreeMap<TableId, TableWatermarks>,
243) {
244    let mut pk_prefix_table_watermarks = BTreeMap::default();
245    let mut non_pk_prefix_table_watermarks = BTreeMap::default();
246    let mut value_table_watermarks = BTreeMap::default();
247    for (table_id, pbwatermark) in &pb_compact_task.table_watermarks {
248        let watermark = TableWatermarks::from(pbwatermark);
249        match watermark.watermark_type {
250            WatermarkSerdeType::PkPrefix => {
251                pk_prefix_table_watermarks.insert(*table_id, watermark);
252            }
253            WatermarkSerdeType::NonPkPrefix => {
254                non_pk_prefix_table_watermarks.insert(*table_id, watermark);
255            }
256            WatermarkSerdeType::Value => {
257                value_table_watermarks.insert(*table_id, watermark);
258            }
259        }
260    }
261    (
262        pk_prefix_table_watermarks,
263        non_pk_prefix_table_watermarks,
264        value_table_watermarks,
265    )
266}
267
268pub fn is_compaction_task_expired(
269    compaction_group_version_id_in_task: u64,
270    compaction_group_version_id_expected: u64,
271) -> bool {
272    compaction_group_version_id_in_task != compaction_group_version_id_expected
273}
274
275impl From<PbCompactTask> for CompactTask {
276    fn from(pb_compact_task: PbCompactTask) -> Self {
277        let (pk_prefix_table_watermarks, non_pk_prefix_table_watermarks, value_table_watermarks) =
278            split_watermark_serde_types(&pb_compact_task);
279        #[expect(deprecated)]
280        Self {
281            input_ssts: pb_compact_task
282                .input_ssts
283                .into_iter()
284                .map(InputLevel::from)
285                .collect_vec(),
286            splits: pb_compact_task
287                .splits
288                .into_iter()
289                .map(|pb_keyrange| KeyRange {
290                    left: pb_keyrange.left.into(),
291                    right: pb_keyrange.right.into(),
292                    right_exclusive: pb_keyrange.right_exclusive,
293                })
294                .collect_vec(),
295            sorted_output_ssts: pb_compact_task
296                .sorted_output_ssts
297                .into_iter()
298                .map(SstableInfo::from)
299                .collect_vec(),
300            task_id: pb_compact_task.task_id,
301            target_level: pb_compact_task.target_level,
302            gc_delete_keys: pb_compact_task.gc_delete_keys,
303            base_level: pb_compact_task.base_level,
304            task_status: TaskStatus::try_from(pb_compact_task.task_status).unwrap(),
305            compaction_group_id: pb_compact_task.compaction_group_id,
306            existing_table_ids: pb_compact_task.existing_table_ids,
307            compression_algorithm: pb_compact_task.compression_algorithm,
308            target_file_size: pb_compact_task.target_file_size,
309            compaction_filter_mask: pb_compact_task.compaction_filter_mask,
310            table_options: pb_compact_task.table_options,
311            current_epoch_time: pb_compact_task.current_epoch_time,
312            target_sub_level_id: pb_compact_task.target_sub_level_id,
313            task_type: PbTaskType::try_from(pb_compact_task.task_type).unwrap(),
314            split_by_state_table: pb_compact_task.split_by_state_table,
315            split_weight_by_vnode: pb_compact_task.split_weight_by_vnode,
316            table_vnode_partition: pb_compact_task.table_vnode_partition,
317            pk_prefix_table_watermarks,
318            non_pk_prefix_table_watermarks,
319            value_table_watermarks,
320            table_schemas: pb_compact_task.table_schemas,
321            max_sub_compaction: pb_compact_task.max_sub_compaction,
322            compaction_group_version_id: pb_compact_task.compaction_group_version_id,
323            blocked_xor_filter_kv_count_threshold: pb_compact_task.max_kv_count_for_xor16,
324            max_vnode_key_range_bytes: pb_compact_task.max_vnode_key_range_bytes,
325            sstable_filter_type: PbSstableFilterType::try_from(pb_compact_task.sstable_filter_type)
326                .unwrap_or(PbSstableFilterType::SstableFilterUnspecified),
327            sstable_filter_layout: PbSstableFilterLayout::try_from(
328                pb_compact_task.sstable_filter_layout,
329            )
330            .unwrap_or(PbSstableFilterLayout::Auto),
331        }
332    }
333}
334
335impl From<&PbCompactTask> for CompactTask {
336    fn from(pb_compact_task: &PbCompactTask) -> Self {
337        let (pk_prefix_table_watermarks, non_pk_prefix_table_watermarks, value_table_watermarks) =
338            split_watermark_serde_types(pb_compact_task);
339        #[expect(deprecated)]
340        Self {
341            input_ssts: pb_compact_task
342                .input_ssts
343                .iter()
344                .map(InputLevel::from)
345                .collect_vec(),
346            splits: pb_compact_task
347                .splits
348                .iter()
349                .map(|pb_keyrange| KeyRange {
350                    left: pb_keyrange.left.clone().into(),
351                    right: pb_keyrange.right.clone().into(),
352                    right_exclusive: pb_keyrange.right_exclusive,
353                })
354                .collect_vec(),
355            sorted_output_ssts: pb_compact_task
356                .sorted_output_ssts
357                .iter()
358                .map(SstableInfo::from)
359                .collect_vec(),
360            task_id: pb_compact_task.task_id,
361            target_level: pb_compact_task.target_level,
362            gc_delete_keys: pb_compact_task.gc_delete_keys,
363            base_level: pb_compact_task.base_level,
364            task_status: TaskStatus::try_from(pb_compact_task.task_status).unwrap(),
365            compaction_group_id: pb_compact_task.compaction_group_id,
366            existing_table_ids: pb_compact_task.existing_table_ids.clone(),
367            compression_algorithm: pb_compact_task.compression_algorithm,
368            target_file_size: pb_compact_task.target_file_size,
369            compaction_filter_mask: pb_compact_task.compaction_filter_mask,
370            table_options: pb_compact_task
371                .table_options
372                .iter()
373                .map(|(table_id, v)| (*table_id, *v))
374                .collect(),
375            current_epoch_time: pb_compact_task.current_epoch_time,
376            target_sub_level_id: pb_compact_task.target_sub_level_id,
377            task_type: PbTaskType::try_from(pb_compact_task.task_type).unwrap(),
378            split_by_state_table: pb_compact_task.split_by_state_table,
379            split_weight_by_vnode: pb_compact_task.split_weight_by_vnode,
380            table_vnode_partition: pb_compact_task
381                .table_vnode_partition
382                .iter()
383                .map(|(table_id, v)| (*table_id, *v))
384                .collect(),
385            pk_prefix_table_watermarks,
386            non_pk_prefix_table_watermarks,
387            value_table_watermarks,
388            table_schemas: pb_compact_task
389                .table_schemas
390                .iter()
391                .map(|(table_id, v)| (*table_id, v.clone()))
392                .collect(),
393            max_sub_compaction: pb_compact_task.max_sub_compaction,
394            compaction_group_version_id: pb_compact_task.compaction_group_version_id,
395            blocked_xor_filter_kv_count_threshold: pb_compact_task.max_kv_count_for_xor16,
396            max_vnode_key_range_bytes: pb_compact_task.max_vnode_key_range_bytes,
397            sstable_filter_type: PbSstableFilterType::try_from(pb_compact_task.sstable_filter_type)
398                .unwrap_or(PbSstableFilterType::SstableFilterUnspecified),
399            sstable_filter_layout: PbSstableFilterLayout::try_from(
400                pb_compact_task.sstable_filter_layout,
401            )
402            .unwrap_or(PbSstableFilterLayout::Auto),
403        }
404    }
405}
406
407impl From<CompactTask> for PbCompactTask {
408    fn from(compact_task: CompactTask) -> Self {
409        #[expect(deprecated)]
410        Self {
411            input_ssts: compact_task
412                .input_ssts
413                .into_iter()
414                .map(|input_level| input_level.into())
415                .collect_vec(),
416            splits: compact_task
417                .splits
418                .into_iter()
419                .map(|keyrange| PbKeyRange {
420                    left: keyrange.left.into(),
421                    right: keyrange.right.into(),
422                    right_exclusive: keyrange.right_exclusive,
423                })
424                .collect_vec(),
425            sorted_output_ssts: compact_task
426                .sorted_output_ssts
427                .into_iter()
428                .map(|sst| sst.into())
429                .collect_vec(),
430            task_id: compact_task.task_id,
431            target_level: compact_task.target_level,
432            gc_delete_keys: compact_task.gc_delete_keys,
433            base_level: compact_task.base_level,
434            task_status: compact_task.task_status.into(),
435            compaction_group_id: compact_task.compaction_group_id,
436            existing_table_ids: compact_task.existing_table_ids,
437            compression_algorithm: compact_task.compression_algorithm,
438            target_file_size: compact_task.target_file_size,
439            compaction_filter_mask: compact_task.compaction_filter_mask,
440            table_options: compact_task.table_options,
441            current_epoch_time: compact_task.current_epoch_time,
442            target_sub_level_id: compact_task.target_sub_level_id,
443            task_type: compact_task.task_type.into(),
444            split_weight_by_vnode: compact_task.split_weight_by_vnode,
445            table_vnode_partition: compact_task.table_vnode_partition,
446            table_watermarks: compact_task
447                .pk_prefix_table_watermarks
448                .into_iter()
449                .chain(compact_task.non_pk_prefix_table_watermarks)
450                .chain(compact_task.value_table_watermarks)
451                .map(|(table_id, table_watermark)| (table_id, table_watermark.into()))
452                .collect(),
453            split_by_state_table: compact_task.split_by_state_table,
454            table_schemas: compact_task.table_schemas,
455            max_sub_compaction: compact_task.max_sub_compaction,
456            compaction_group_version_id: compact_task.compaction_group_version_id,
457            max_kv_count_for_xor16: compact_task.blocked_xor_filter_kv_count_threshold,
458            max_vnode_key_range_bytes: compact_task.max_vnode_key_range_bytes,
459            sstable_filter_type: compact_task.sstable_filter_type.into(),
460            sstable_filter_layout: compact_task.sstable_filter_layout.into(),
461        }
462    }
463}
464
465impl From<&CompactTask> for PbCompactTask {
466    fn from(compact_task: &CompactTask) -> Self {
467        #[expect(deprecated)]
468        Self {
469            input_ssts: compact_task
470                .input_ssts
471                .iter()
472                .map(|input_level| input_level.into())
473                .collect_vec(),
474            splits: compact_task
475                .splits
476                .iter()
477                .map(|keyrange| PbKeyRange {
478                    left: keyrange.left.to_vec(),
479                    right: keyrange.right.to_vec(),
480                    right_exclusive: keyrange.right_exclusive,
481                })
482                .collect_vec(),
483            sorted_output_ssts: compact_task
484                .sorted_output_ssts
485                .iter()
486                .map(|sst| sst.into())
487                .collect_vec(),
488            task_id: compact_task.task_id,
489            target_level: compact_task.target_level,
490            gc_delete_keys: compact_task.gc_delete_keys,
491            base_level: compact_task.base_level,
492            task_status: compact_task.task_status.into(),
493            compaction_group_id: compact_task.compaction_group_id,
494            existing_table_ids: compact_task.existing_table_ids.clone(),
495            compression_algorithm: compact_task.compression_algorithm,
496            target_file_size: compact_task.target_file_size,
497            compaction_filter_mask: compact_task.compaction_filter_mask,
498            table_options: compact_task.table_options.clone(),
499            current_epoch_time: compact_task.current_epoch_time,
500            target_sub_level_id: compact_task.target_sub_level_id,
501            task_type: compact_task.task_type.into(),
502            split_weight_by_vnode: compact_task.split_weight_by_vnode,
503            table_vnode_partition: compact_task.table_vnode_partition.clone(),
504            table_watermarks: compact_task
505                .pk_prefix_table_watermarks
506                .iter()
507                .chain(compact_task.non_pk_prefix_table_watermarks.iter())
508                .chain(compact_task.value_table_watermarks.iter())
509                .map(|(table_id, table_watermark)| (*table_id, table_watermark.into()))
510                .collect(),
511            split_by_state_table: compact_task.split_by_state_table,
512            table_schemas: compact_task.table_schemas.clone(),
513            max_sub_compaction: compact_task.max_sub_compaction,
514            compaction_group_version_id: compact_task.compaction_group_version_id,
515            max_kv_count_for_xor16: compact_task.blocked_xor_filter_kv_count_threshold,
516            max_vnode_key_range_bytes: compact_task.max_vnode_key_range_bytes,
517            sstable_filter_type: compact_task.sstable_filter_type.into(),
518            sstable_filter_layout: compact_task.sstable_filter_layout.into(),
519        }
520    }
521}
522
523impl From<PbCompactTaskAssignment> for CompactTaskAssignment {
524    fn from(assignment: PbCompactTaskAssignment) -> Self {
525        Self {
526            compact_task: CompactTask::from(assignment.compact_task.unwrap()),
527            context_id: assignment.context_id,
528        }
529    }
530}
531
532impl From<&PbCompactTaskAssignment> for CompactTaskAssignment {
533    fn from(assignment: &PbCompactTaskAssignment) -> Self {
534        Self {
535            compact_task: CompactTask::from(assignment.compact_task.as_ref().unwrap()),
536            context_id: assignment.context_id,
537        }
538    }
539}
540
541impl From<CompactTaskAssignment> for PbCompactTaskAssignment {
542    fn from(assignment: CompactTaskAssignment) -> Self {
543        Self {
544            compact_task: Some(assignment.compact_task.into()),
545            context_id: assignment.context_id,
546        }
547    }
548}
549
550impl From<&CompactTaskAssignment> for PbCompactTaskAssignment {
551    fn from(assignment: &CompactTaskAssignment) -> Self {
552        Self {
553            compact_task: Some((&assignment.compact_task).into()),
554            context_id: assignment.context_id,
555        }
556    }
557}
558
559#[derive(Clone, PartialEq, Default)]
560pub struct ValidationTask {
561    pub sst_infos: Vec<SstableInfo>,
562    pub sst_id_to_worker_id: HashMap<HummockSstableObjectId, WorkerId>,
563}
564
565impl From<PbValidationTask> for ValidationTask {
566    fn from(pb_validation_task: PbValidationTask) -> Self {
567        Self {
568            sst_infos: pb_validation_task
569                .sst_infos
570                .into_iter()
571                .map(SstableInfo::from)
572                .collect_vec(),
573            sst_id_to_worker_id: pb_validation_task.sst_id_to_worker_id,
574        }
575    }
576}
577
578impl From<ValidationTask> for PbValidationTask {
579    fn from(validation_task: ValidationTask) -> Self {
580        Self {
581            sst_infos: validation_task
582                .sst_infos
583                .into_iter()
584                .map(|sst| sst.into())
585                .collect_vec(),
586            sst_id_to_worker_id: validation_task.sst_id_to_worker_id,
587        }
588    }
589}
590
591#[derive(Clone, PartialEq, Default, Debug)]
592pub struct ReportTask {
593    pub table_stats_change: HashMap<TableId, PbTableStats>,
594    pub task_id: u64,
595    pub task_status: TaskStatus,
596    pub sorted_output_ssts: Vec<SstableInfo>,
597    pub object_timestamps: HashMap<HummockSstableObjectId, u64>,
598}
599
600impl From<PbReportTask> for ReportTask {
601    fn from(value: PbReportTask) -> Self {
602        Self {
603            table_stats_change: value.table_stats_change,
604            task_id: value.task_id,
605            task_status: PbTaskStatus::try_from(value.task_status).unwrap(),
606            sorted_output_ssts: value
607                .sorted_output_ssts
608                .into_iter()
609                .map(SstableInfo::from)
610                .collect_vec(),
611            object_timestamps: value.object_timestamps,
612        }
613    }
614}
615
616impl From<ReportTask> for PbReportTask {
617    fn from(value: ReportTask) -> Self {
618        Self {
619            table_stats_change: value.table_stats_change,
620            task_id: value.task_id,
621            task_status: value.task_status.into(),
622            sorted_output_ssts: value
623                .sorted_output_ssts
624                .into_iter()
625                .map(|sst| sst.into())
626                .collect_vec(),
627            object_timestamps: value.object_timestamps,
628        }
629    }
630}
631
632#[cfg(test)]
633mod tests {
634    use risingwave_common::catalog::TableId;
635    use risingwave_pb::hummock::compact_task::TaskType;
636    use risingwave_pb::hummock::{PbCompactTask, PbLevelType, PbSstableFilterLayout};
637
638    use super::CompactTask;
639    use crate::level::InputLevel;
640    use crate::sstable_info::{SstableInfo, SstableInfoInner};
641
642    fn test_sstable(sst_id: u64, table_ids: Vec<TableId>) -> SstableInfo {
643        SstableInfo::from(SstableInfoInner {
644            object_id: sst_id.into(),
645            sst_id: sst_id.into(),
646            table_ids,
647            ..Default::default()
648        })
649    }
650
651    fn test_read_property_sstable(table_ids: Vec<TableId>) -> SstableInfo {
652        SstableInfo::from(SstableInfoInner {
653            object_id: 1.into(),
654            sst_id: 2.into(),
655            table_ids,
656            range_tombstone_count: 1,
657            total_key_count: 100,
658            ..Default::default()
659        })
660    }
661
662    #[test]
663    fn test_blocked_xor_filter_kv_count_threshold_roundtrip() {
664        let pb = PbCompactTask {
665            max_kv_count_for_xor16: Some(123),
666            ..Default::default()
667        };
668
669        let task = CompactTask::from(&pb);
670        assert_eq!(task.blocked_xor_filter_kv_count_threshold, Some(123));
671
672        let pb2 = PbCompactTask::from(&task);
673        assert_eq!(pb2.max_kv_count_for_xor16, Some(123));
674    }
675
676    #[test]
677    fn test_sstable_filter_layout_for_output() {
678        let task = CompactTask {
679            sstable_filter_layout: PbSstableFilterLayout::Auto,
680            blocked_xor_filter_kv_count_threshold: Some(100),
681            ..Default::default()
682        };
683
684        assert_eq!(
685            task.sstable_filter_layout_for_output(100),
686            PbSstableFilterLayout::Plain
687        );
688        assert_eq!(
689            task.sstable_filter_layout_for_output(101),
690            PbSstableFilterLayout::Blocked
691        );
692
693        let task = CompactTask {
694            sstable_filter_layout: PbSstableFilterLayout::Plain,
695            blocked_xor_filter_kv_count_threshold: Some(100),
696            ..Default::default()
697        };
698
699        assert_eq!(
700            task.sstable_filter_layout_for_output(101),
701            PbSstableFilterLayout::Plain
702        );
703
704        let task = CompactTask {
705            sstable_filter_layout: PbSstableFilterLayout::Blocked,
706            blocked_xor_filter_kv_count_threshold: Some(100),
707            ..Default::default()
708        };
709
710        assert_eq!(
711            task.sstable_filter_layout_for_output(0),
712            PbSstableFilterLayout::Blocked
713        );
714    }
715
716    #[test]
717    fn test_empty_table_ids_are_reclaim_and_have_no_input_table_ids() {
718        let task = CompactTask {
719            input_ssts: vec![InputLevel {
720                table_infos: vec![test_sstable(1, vec![])],
721                ..Default::default()
722            }],
723            existing_table_ids: vec![TableId::new(1)],
724            ..Default::default()
725        };
726
727        assert!(task.get_table_ids_from_input_ssts().next().is_none());
728        assert!(task.is_trivial_reclaim());
729    }
730
731    #[test]
732    fn test_read_properties_ignore_empty_table_ids() {
733        let task = CompactTask {
734            input_ssts: vec![InputLevel {
735                table_infos: vec![
736                    test_read_property_sstable(vec![]),
737                    SstableInfo::from(SstableInfoInner {
738                        object_id: 3.into(),
739                        sst_id: 3.into(),
740                        table_ids: vec![TableId::new(1)],
741                        total_key_count: 1,
742                        ..Default::default()
743                    }),
744                ],
745                ..Default::default()
746            }],
747            ..Default::default()
748        };
749
750        assert!(!task.contains_range_tombstone());
751        assert!(!task.contains_split_sst());
752
753        let task = CompactTask {
754            input_ssts: vec![InputLevel {
755                table_infos: vec![test_read_property_sstable(vec![TableId::new(1)])],
756                ..Default::default()
757            }],
758            ..Default::default()
759        };
760
761        assert!(task.contains_range_tombstone());
762        assert!(task.contains_split_sst());
763    }
764
765    #[test]
766    fn test_task_label_for_trivial_move_and_reclaim() {
767        let trivial_move_task = CompactTask {
768            input_ssts: vec![
769                InputLevel {
770                    level_idx: 1,
771                    level_type: PbLevelType::Nonoverlapping,
772                    table_infos: vec![
773                        test_sstable(10, vec![TableId::new(1)]),
774                        test_sstable(11, vec![]),
775                    ],
776                },
777                InputLevel {
778                    level_idx: 2,
779                    level_type: PbLevelType::Nonoverlapping,
780                    table_infos: vec![],
781                },
782            ],
783            target_level: 2,
784            task_type: TaskType::Dynamic,
785            ..Default::default()
786        };
787
788        assert!(!trivial_move_task.is_trivial_reclaim());
789        assert_eq!(trivial_move_task.task_label(), "trivial-move");
790
791        let trivial_reclaim_task = CompactTask {
792            input_ssts: vec![
793                InputLevel {
794                    level_idx: 1,
795                    level_type: PbLevelType::Nonoverlapping,
796                    table_infos: vec![test_sstable(10, vec![])],
797                },
798                InputLevel {
799                    level_idx: 2,
800                    level_type: PbLevelType::Nonoverlapping,
801                    table_infos: vec![],
802                },
803            ],
804            target_level: 2,
805            task_type: TaskType::Dynamic,
806            ..Default::default()
807        };
808
809        assert!(trivial_reclaim_task.is_trivial_reclaim());
810        assert!(trivial_reclaim_task.is_trivial_move_task());
811        assert_eq!(trivial_reclaim_task.task_label(), "trivial-space-reclaim");
812    }
813}
risingwave_hummock_sdk/compact_task.rs

risingwave_hummock_sdk/
compact_task.rs
