Skip to content

Commit

Permalink
refactor(storage): distinguish SST id and object id (close #8434)
Browse files Browse the repository at this point in the history
  • Loading branch information
Liang Zhao committed Mar 9, 2023
1 parent f4817e9 commit af7511d
Show file tree
Hide file tree
Showing 35 changed files with 333 additions and 273 deletions.
18 changes: 9 additions & 9 deletions proto/hummock.proto
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,15 +10,14 @@ option java_package = "com.risingwave.proto";
option optimize_for = SPEED;

message SstableInfo {
uint64 id = 1;
KeyRange key_range = 2;
uint64 file_size = 3;
repeated uint32 table_ids = 4;
uint64 meta_offset = 5;
uint64 stale_key_count = 6;
uint64 total_key_count = 7;
// When a SST is divided, its divide_version will increase one.
uint64 divide_version = 8;
uint64 object_id = 1;
uint64 sst_id = 2;
KeyRange key_range = 3;
uint64 file_size = 4;
repeated uint32 table_ids = 5;
uint64 meta_offset = 6;
uint64 stale_key_count = 7;
uint64 total_key_count = 8;
uint64 min_epoch = 9;
uint64 max_epoch = 10;
uint64 uncompressed_file_size = 11;
Expand Down Expand Up @@ -64,6 +63,7 @@ message GroupConstruct {
uint64 parent_group_id = 2;
repeated uint32 table_ids = 3;
uint64 group_id = 4;
uint64 new_sst_start_id = 5;
}

message GroupMetaChange {
Expand Down
13 changes: 7 additions & 6 deletions src/ctl/src/cmd_impl/hummock/sst_dump.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ use risingwave_common::types::to_text::ToText;
use risingwave_common::util::epoch::Epoch;
use risingwave_common::util::iter_util::ZipEqFast;
use risingwave_frontend::TableCatalog;
use risingwave_hummock_sdk::HummockSstableId;
use risingwave_hummock_sdk::compaction_group::hummock_version_ext::HummockVersionExt;
use risingwave_hummock_sdk::key::FullKey;
use risingwave_object_store::object::BlockLocation;
Expand Down Expand Up @@ -73,14 +74,14 @@ pub async fn sst_dump(context: &CtlContext, args: SstDumpArgs) -> anyhow::Result
for level in version.get_combined_levels() {
for sstable_info in &level.table_infos {
if let Some(sst_id) = &args.sst_id {
if *sst_id == sstable_info.id {
if *sst_id == sstable_info.get_object_id() {
if args.print_level {
print_level(level);
}

sst_dump_via_sstable_store(
sstable_store,
sstable_info.id,
sstable_info.get_object_id(),
sstable_info.meta_offset,
sstable_info.file_size,
&table_data,
Expand All @@ -96,7 +97,7 @@ pub async fn sst_dump(context: &CtlContext, args: SstDumpArgs) -> anyhow::Result

sst_dump_via_sstable_store(
sstable_store,
sstable_info.id,
sstable_info.get_object_id(),
sstable_info.meta_offset,
sstable_info.file_size,
&table_data,
Expand All @@ -111,14 +112,14 @@ pub async fn sst_dump(context: &CtlContext, args: SstDumpArgs) -> anyhow::Result

pub async fn sst_dump_via_sstable_store(
sstable_store: &SstableStore,
sst_id: u64,
object_id: HummockSstableId,
meta_offset: u64,
file_size: u64,
table_data: &TableData,
args: &SstDumpArgs,
) -> anyhow::Result<()> {
let sstable_info = SstableInfo {
id: sst_id,
object_id: object_id,
meta_offset,
file_size,
..Default::default()
Expand All @@ -129,7 +130,7 @@ pub async fn sst_dump_via_sstable_store(
let sstable = sstable_cache.value().as_ref();
let sstable_meta = &sstable.meta;

println!("SST id: {}", sst_id);
println!("SST object id: {}", object_id);
println!("-------------------------------------");
println!("File Size: {}", sstable.estimate_size());

Expand Down
2 changes: 1 addition & 1 deletion src/meta/src/barrier/mod.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1022,7 +1022,7 @@ fn collect_synced_ssts(
.iter_mut()
.map(|grouped| {
let sst_info = std::mem::take(&mut grouped.sst).expect("field not None");
sst_to_worker.insert(sst_info.id, resp.worker_id);
sst_to_worker.insert(sst_info.get_object_id(), resp.worker_id);
ExtendedSstableInfo::new(
grouped.compaction_group_id,
sst_info,
Expand Down
14 changes: 7 additions & 7 deletions src/meta/src/hummock/compaction/level_selector.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -540,7 +540,7 @@ pub mod tests {
level_type: LevelType::Overlapping as i32,
total_file_size: sst.file_size,
uncompressed_file_size: sst.uncompressed_file_size,
sub_level_id: sst.id,
sub_level_id: sst.get_sst_id(),
table_infos: vec![sst],
});
}
Expand All @@ -563,7 +563,7 @@ pub mod tests {
.iter()
.map(|table| table.uncompressed_file_size)
.sum();
let sub_level_id = table_infos[0].id;
let sub_level_id = table_infos[0].get_sst_id();
levels.l0.as_mut().unwrap().total_file_size += total_file_size;
levels.l0.as_mut().unwrap().sub_levels.push(Level {
level_idx: 0,
Expand All @@ -583,7 +583,8 @@ pub mod tests {
epoch: u64,
) -> SstableInfo {
SstableInfo {
id,
object_id: id,
sst_id: id,
key_range: Some(KeyRange {
left: iterator_test_key_of_epoch(table_prefix, left, epoch),
right: iterator_test_key_of_epoch(table_prefix, right, epoch),
Expand All @@ -594,7 +595,6 @@ pub mod tests {
meta_offset: 0,
stale_key_count: 0,
total_key_count: 0,
divide_version: 0,
uncompressed_file_size: (right - left + 1) as u64,
min_epoch: 0,
max_epoch: 0,
Expand All @@ -613,7 +613,8 @@ pub mod tests {
max_epoch: u64,
) -> SstableInfo {
SstableInfo {
id,
object_id: id,
sst_id: id,
key_range: Some(KeyRange {
left: iterator_test_key_of_epoch(table_prefix, left, epoch),
right: iterator_test_key_of_epoch(table_prefix, right, epoch),
Expand All @@ -624,7 +625,6 @@ pub mod tests {
meta_offset: 0,
stale_key_count: 0,
total_key_count: 0,
divide_version: 0,
uncompressed_file_size: (right - left + 1) as u64,
min_epoch,
max_epoch,
Expand Down Expand Up @@ -730,7 +730,7 @@ pub mod tests {
) {
for i in &compact_task.input.input_levels {
for t in &i.table_infos {
assert!(level_handlers[i.level_idx as usize].is_pending_compact(&t.id));
assert!(level_handlers[i.level_idx as usize].is_pending_compact(&t.sst_id));
}
}
}
Expand Down
7 changes: 4 additions & 3 deletions src/meta/src/hummock/compaction/mod.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ pub mod compaction_config;
mod level_selector;
mod overlap_strategy;
use risingwave_common::catalog::TableOption;
use risingwave_hummock_sdk::compaction_group::StateTableId;
use risingwave_hummock_sdk::prost_key_range::KeyRangeExt;
use risingwave_pb::hummock::compact_task::{self, TaskStatus};

Expand All @@ -27,7 +28,7 @@ use std::sync::Arc;
use picker::{
LevelCompactionPicker, ManualCompactionPicker, MinOverlappingPicker, TierCompactionPicker,
};
use risingwave_hummock_sdk::{CompactionGroupId, HummockCompactionTaskId, HummockEpoch};
use risingwave_hummock_sdk::{CompactionGroupId, HummockCompactionTaskId, HummockEpoch, HummockSstableId};
use risingwave_pb::hummock::compaction_config::CompactionMode;
use risingwave_pb::hummock::hummock_version::Levels;
use risingwave_pb::hummock::{CompactTask, CompactionConfig, InputLevel, KeyRange, LevelType};
Expand Down Expand Up @@ -218,11 +219,11 @@ impl CompactStatus {
#[derive(Clone, Debug, PartialEq)]
pub struct ManualCompactionOption {
/// Filters out SSTs to pick. Has no effect if empty.
pub sst_ids: Vec<u64>,
pub sst_ids: Vec<HummockSstableId>,
/// Filters out SSTs to pick.
pub key_range: KeyRange,
/// Filters out SSTs to pick. Has no effect if empty.
pub internal_table_id: HashSet<u32>,
pub internal_table_id: HashSet<StateTableId>,
/// Input level.
pub level: usize,
}
Expand Down
34 changes: 17 additions & 17 deletions src/meta/src/hummock/compaction/picker/base_level_compaction_picker.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -259,8 +259,8 @@ pub mod tests {
let ret = picker
.pick_compaction(&levels, &levels_handler, &mut local_stats)
.unwrap();
assert_eq!(ret.input_levels[0].table_infos[0].id, 4);
assert_eq!(ret.input_levels[1].table_infos[0].id, 1);
assert_eq!(ret.input_levels[0].table_infos[0].get_sst_id(), 4);
assert_eq!(ret.input_levels[1].table_infos[0].get_sst_id(), 1);
ret.add_pending_task(0, &mut levels_handler);

// Cannot pick because sub-level[0] is pending.
Expand All @@ -272,7 +272,7 @@ pub mod tests {

levels.l0.as_mut().unwrap().sub_levels[0]
.table_infos
.retain(|table| table.id != 4);
.retain(|table| table.get_sst_id() != 4);
levels.l0.as_mut().unwrap().total_file_size -= ret.input_levels[0].table_infos[0].file_size;

levels_handler[0].remove_task(0);
Expand All @@ -282,9 +282,9 @@ pub mod tests {
.pick_compaction(&levels, &levels_handler, &mut local_stats)
.unwrap();
assert_eq!(ret.input_levels.len(), 4);
assert_eq!(ret.input_levels[0].table_infos[0].id, 7);
assert_eq!(ret.input_levels[1].table_infos[0].id, 6);
assert_eq!(ret.input_levels[2].table_infos[0].id, 5);
assert_eq!(ret.input_levels[0].table_infos[0].get_sst_id(), 7);
assert_eq!(ret.input_levels[1].table_infos[0].get_sst_id(), 6);
assert_eq!(ret.input_levels[2].table_infos[0].get_sst_id(), 5);
assert_eq!(ret.input_levels[3].table_infos.len(), 4);
ret.add_pending_task(1, &mut levels_handler);

Expand All @@ -302,9 +302,9 @@ pub mod tests {
.pick_compaction(&levels, &levels_handler, &mut local_stats)
.unwrap();
assert_eq!(ret.input_levels.len(), 4);
assert_eq!(ret.input_levels[0].table_infos[0].id, 7);
assert_eq!(ret.input_levels[1].table_infos[0].id, 6);
assert_eq!(ret.input_levels[2].table_infos[0].id, 5);
assert_eq!(ret.input_levels[0].table_infos[0].get_sst_id(), 7);
assert_eq!(ret.input_levels[1].table_infos[0].get_sst_id(), 6);
assert_eq!(ret.input_levels[2].table_infos[0].get_sst_id(), 5);
assert_eq!(ret.input_levels[3].table_infos.len(), 4);
}

Expand Down Expand Up @@ -362,7 +362,7 @@ pub mod tests {
ret.input_levels[1]
.table_infos
.iter()
.map(|t| t.id)
.map(|t| t.get_sst_id())
.collect_vec(),
vec![1]
);
Expand All @@ -371,7 +371,7 @@ pub mod tests {
ret.input_levels[0]
.table_infos
.iter()
.map(|t| t.id)
.map(|t| t.get_sst_id())
.collect_vec(),
vec![7, 8]
);
Expand Down Expand Up @@ -471,7 +471,7 @@ pub mod tests {
ret.input_levels[0]
.table_infos
.iter()
.map(|t| t.id)
.map(|t| t.get_sst_id())
.collect_vec(),
vec![3]
);
Expand Down Expand Up @@ -565,9 +565,9 @@ pub mod tests {
.pick_compaction(&levels, &levels_handler, &mut local_stats)
.unwrap();
assert_eq!(ret.input_levels.len(), 3);
assert_eq!(ret.input_levels[2].table_infos[0].id, 1);
assert_eq!(ret.input_levels[2].table_infos[1].id, 2);
assert_eq!(ret.input_levels[2].table_infos[2].id, 3);
assert_eq!(ret.input_levels[2].table_infos[0].get_sst_id(), 1);
assert_eq!(ret.input_levels[2].table_infos[1].get_sst_id(), 2);
assert_eq!(ret.input_levels[2].table_infos[2].get_sst_id(), 3);
levels.levels[0].table_infos[0].file_size += 1600 - levels.levels[0].total_file_size;
levels.levels[0].total_file_size = 1600;
let sub_level = &mut levels.l0.as_mut().unwrap().sub_levels[0];
Expand Down Expand Up @@ -649,7 +649,7 @@ pub mod tests {
let ret = picker
.pick_compaction(&levels, &levels_handler, &mut local_stats)
.unwrap();
assert_eq!(ret.input_levels[0].table_infos[0].id, 7);
assert_eq!(ret.input_levels[0].table_infos[0].get_sst_id(), 7);
assert_eq!(
3,
ret.input_levels.iter().filter(|l| l.level_idx == 0).count()
Expand All @@ -674,7 +674,7 @@ pub mod tests {
let ret = picker
.pick_compaction(&levels, &levels_handler, &mut local_stats)
.unwrap();
assert_eq!(ret.input_levels[0].table_infos[0].id, 6);
assert_eq!(ret.input_levels[0].table_infos[0].get_sst_id(), 6);
assert_eq!(
2,
ret.input_levels.iter().filter(|l| l.level_idx == 0).count()
Expand Down
Loading

0 comments on commit af7511d

Please sign in to comment.