

首页  思维导图  详情



存算一体 StarRocks Compaction源码梳理

2026-03-16 22:02:33   0  举报





仅支持查看

AI智能生成

无

其他技术

模版推荐

作者其他创作

大纲/内容

BE-Server启动StorageEngine::start_bg_threads()

enable_event_based_compaction_framework

周期性扫描Tablets，创建compaction_candidates

（选择最高有candidate执行compaction） _compaction_manager->schedule()

双通道更新CandidateRowset机制

定时兜底：(分析Tablet，创建compaction_candidates) compaction_check()

_compaction_check_one_round()

TabletManager::get_next_batch_tablets

compacManager.update_tablet

Tablet::need_compaction()

_enable_compaction 可手动关闭compaction

enable_size_tiered_compaction_strategy

默认是使用size-tiered-compaction

（将被选中的tablet添加到_compaction_candidates中） udpate_candidates

更新对应Compaction排队指标

(通知下游消费者进行compaction) CompactionManager::_notify()

（event_based_compaction_framework）CompactionManager::_dispatch_worker()

dispatch_map是基于事件compaction框架的核心

对rowset进行新增修改等操作，会触发udpate_tablet_async动作

周期性读取dispatch_map，触发update_tablet

tablet->need_compaction()

update_candidates

notify

(调度Compaction) Compaction_manager->_schedule()

pick_candidate

从compaction_candidates中筛选出score最高的tablet

check_precondition for candate

check can run more

定时时间段可配置禁止base compaction

分compactionType检测条件

创建compaction任务

Tablet::create_compaction_task

compaction_policy -> create_compaction(tablet)

compaction_task -> start()

Compaction Policy

DefaultCompactionPolicy

need_compaction

cal_cumulative_point（earliest version）

计算出cumulative compaction的起始节点

（1）segment有重叠（2）或者rowset版本区间有间隔

_cumulative_point

pick_rs_2_cumulative_compact

pick_candidate_rowset

将候选Rowset用endVersion排序asc

迭代候选Rowset，选出最终用于Compaction的Rowset

（1）遇到version不与transientRowset连续的Rowset

（2）遇到了被delete的版本

（3）遇到了一个已经被合并到大Version Rowset

（4）上述规则都未匹配

当前CompactionScore大于maxCumulative阈值

将当前迭代到的Rowset加入transientRowset

完全迭代完所有Rowset

transientRowsets所占CompactionScore满足CompacCondition

不满足条件且长时间未执行过cumulativeCompaction

pick_rs_2_base_compact

pick_candidate_rowsets

确定baseSize和cumulative_total_size

确定compaction_score

查看cumulativeCompaction和baseCompaction哪一个点score更高就选择对应的

SizeTieredCompactionPolicy

need_compaction

pick_rs_to_size_tiered_compact

pick_all_candicate_rowset

BaseCompaction 前置判断

如果超过阈值未BaseCompaction，就设置强制BaseC为true

如果tablet中delete谓语的数量超过了10%的version数量，就设置强制BaseC为true

定义SizeTieredCompaction中的Level

迭代CandidateRowsets

通过rowset实际磁盘大小，计算levelSize

遇到导致版本不连续的Rowset

遇到正在被compacted的rowset

通过segments数量和levelSize 加权计算LevelCompactionScore

创建SizeTieredLevel

创建完成后清空transientRowsets

遇到缺失的version

遇到delete谓词

如果transientRowset startVersion从0开始，则不做任何动作

transientRowset startVersion不为0

从后向前，把version连续的多个小Level合并到transientRowsets中

将合并后的transientRowsets创建出一个新level，加到order_levels末尾

清理transientRowsets后将当前delete Rowset加入transienRowsets

rowsetSize  <（levelSize / 阈值倍数）

固化当前transientRowsets到Level中，将rowset添加到新的Level中去

迭代完CandidateRowsets后，如果transientRowset不为空，就兜底把这部分创建新Level

选择compactionScore最高的Level，取其rowsets作为最终的inputRowset输出

检查inputRowsets的version连续性

如果inputRowsets开头第一个rowset的version是0，则compactionType为BaseCompaction

_cal_compaction_score

create_compaction

取出待compact的input_rowsets中最小和最大version作为output_version的左右范围

将input_rowsets设置为正在comacting更新状态

CompactionTaskFactory::create_compaction_task()

拿到maxSchemVersion对应的schema

schema_version

根据不同条件选择不同的CompactionAlgorithm和CompactionTask

init the compaction task

CompactionUtils

CompactionAlgorithm

CompactionUtils::choose_compaction_algorithm

CompactionTask

VerticalCompactionTask

HorizontalCompactionTask

 Collect

Get Started

源网荷储一体化平台架构

 Collect

Get Started

ETC门架数据采集图示（机电一体化）

 Collect

Get Started

AI人工智能智算平台技术架构

 Collect

Get Started

集团信息化、业财一体化系统平台架构





0 条评论

下一页