feat(storage): replace hummock protobuf strcut with rust struct

src/ctl/src/cmd_impl/hummock/sst_dump.rs

@@ -28,9 +28,9 @@ use risingwave_common::util::value_encoding::column_aware_row_encoding::ColumnAw
 use risingwave_common::util::value_encoding::{BasicSerde, EitherSerde, ValueRowDeserializer};
 use risingwave_frontend::TableCatalog;
 use risingwave_hummock_sdk::key::FullKey;
+use risingwave_hummock_sdk::version::{Level, SstableInfo};
 use risingwave_hummock_sdk::HummockSstableObjectId;
 use risingwave_object_store::object::{ObjectMetadata, ObjectStoreImpl};
-use risingwave_pb::hummock::{Level, SstableInfo};
 use risingwave_rpc_client::MetaClient;
 use risingwave_storage::hummock::value::HummockValue;
 use risingwave_storage::hummock::{
@@ -161,7 +161,7 @@ pub async fn sst_dump(context: &CtlContext, args: SstDumpArgs) -> anyhow::Result
 }
 
 fn print_level(level: &Level, sst_info: &SstableInfo) {
-    println!("Level Type: {}", level.level_type);
+    println!("Level Type: {}", level.level_type.as_str_name());
     println!("Level Idx: {}", level.level_idx);
     if level.level_idx == 0 {
         println!("L0 Sub-Level Idx: {}", level.sub_level_id);

src/ctl/src/cmd_impl/hummock/validate_version.rs

@@ -20,11 +20,11 @@ use itertools::Itertools;
 use risingwave_common::util::epoch::Epoch;
 use risingwave_hummock_sdk::compaction_group::hummock_version_ext;
 use risingwave_hummock_sdk::key::{FullKey, UserKey};
-use risingwave_hummock_sdk::version::{HummockVersion, HummockVersionDelta};
+use risingwave_hummock_sdk::version::{HummockVersion, HummockVersionDelta, SstableInfo};
 use risingwave_hummock_sdk::{version_archive_dir, HummockSstableObjectId, HummockVersionId};
 use risingwave_object_store::object::ObjectStoreRef;
 use risingwave_pb::hummock::group_delta::DeltaType;
-use risingwave_pb::hummock::{HummockVersionArchive, SstableInfo};
+use risingwave_pb::hummock::HummockVersionArchive;
 use risingwave_rpc_client::HummockMetaClient;
 use risingwave_storage::hummock::value::HummockValue;
 use risingwave_storage::hummock::{Block, BlockHolder, BlockIterator, SstableStoreRef};
@@ -109,8 +109,7 @@ async fn print_user_key_in_version(
             .chain(cg.levels.iter())
         {
             for sstable_info in &level.table_infos {
-                use risingwave_hummock_sdk::key_range::KeyRange;
-                let key_range: KeyRange = sstable_info.key_range.as_ref().unwrap().into();
+                let key_range = sstable_info.key_range.as_ref().unwrap();
                 let left_user_key = FullKey::decode(&key_range.left);
                 let right_user_key = FullKey::decode(&key_range.right);
                 if left_user_key.user_key > *target_key || *target_key > right_user_key.user_key {

src/frontend/src/catalog/system_catalog/rw_catalog/rw_hummock_version.rs

@@ -107,7 +107,7 @@ fn version_to_compaction_group_rows(version: &HummockVersion) -> Vec<RwHummockVe
             version_id: version.id as _,
             max_committed_epoch: version.max_committed_epoch as _,
             safe_epoch: version.visible_table_safe_epoch() as _,
-            compaction_group: json!(cg).into(),
+            compaction_group: json!(cg.to_protobuf()).into(),
         })
         .collect()
 }
@@ -125,8 +125,8 @@ fn version_to_sstable_rows(version: HummockVersion) -> Vec<RwHummockSstable> {
                     level_id: level.level_idx as _,
                     sub_level_id: (level.level_idx == 0).then_some(level.sub_level_id as _),
                     level_type: level.level_type as _,
-                    key_range_left: key_range.left,
-                    key_range_right: key_range.right,
+                    key_range_left: key_range.left.to_vec(),
+                    key_range_right: key_range.right.to_vec(),
                     right_exclusive: key_range.right_exclusive,
                     file_size: sst.file_size as _,
                     meta_offset: sst.meta_offset as _,

src/frontend/src/catalog/system_catalog/rw_catalog/rw_hummock_version_deltas.rs

@@ -12,8 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
+
 use risingwave_common::types::{Fields, JsonbVal};
 use risingwave_frontend_macro::system_catalog;
+use risingwave_pb::hummock::hummock_version_delta::PbGroupDeltas;
 use serde_json::json;
 
 use crate::catalog::system_catalog::SysCatalogReaderImpl;
@@ -41,7 +44,12 @@ async fn read(reader: &SysCatalogReaderImpl) -> Result<Vec<RwHummockVersionDelta
             max_committed_epoch: d.max_committed_epoch as _,
             safe_epoch: d.visible_table_safe_epoch() as _,
             trivial_move: d.trivial_move,
-            group_deltas: json!(d.group_deltas).into(),
+            group_deltas: json!(d
+                .group_deltas
+                .into_iter()
+                .map(|(group_id, deltas)| (group_id, PbGroupDeltas::from(deltas)))
+                .collect::<HashMap<u64, PbGroupDeltas>>())
+            .into(),
         })
         .collect();
     Ok(rows)

src/meta/service/Cargo.toml

@@ -16,6 +16,7 @@ normal = ["workspace-hack"]
 [dependencies]
 anyhow = "1"
 async-trait = "0.1"
+bytes = { version = "1" }
 either = "1"
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
 itertools = { workspace = true }

src/meta/service/src/hummock_service.rs

@@ -18,6 +18,7 @@ use std::time::Duration;
 use futures::StreamExt;
 use itertools::Itertools;
 use risingwave_common::catalog::{TableId, SYS_CATALOG_START_ID};
+use risingwave_hummock_sdk::key_range::KeyRange;
 use risingwave_hummock_sdk::version::HummockVersionDelta;
 use risingwave_meta::manager::MetadataManager;
 use risingwave_pb::hummock::get_compaction_score_response::PickerInfo;
@@ -85,7 +86,7 @@ impl HummockManagerService for HummockServiceImpl {
         let current_version = self.hummock_manager.get_current_version().await;
         Ok(Response::new(GetCurrentVersionResponse {
             status: None,
-            current_version: Some(current_version.to_protobuf()),
+            current_version: Some(current_version.into()),
         }))
     }
 
@@ -101,7 +102,7 @@ impl HummockManagerService for HummockServiceImpl {
             ))
             .await?;
         Ok(Response::new(ReplayVersionDeltaResponse {
-            version: Some(version.to_protobuf()),
+            version: Some(version.into()),
             modified_compaction_groups: compaction_groups,
         }))
     }
@@ -123,7 +124,7 @@ impl HummockManagerService for HummockServiceImpl {
     ) -> Result<Response<DisableCommitEpochResponse>, Status> {
         let version = self.hummock_manager.disable_commit_epoch().await;
         Ok(Response::new(DisableCommitEpochResponse {
-            current_version: Some(version.to_protobuf()),
+            current_version: Some(version.into()),
         }))
     }
 
@@ -139,8 +140,8 @@ impl HummockManagerService for HummockServiceImpl {
         let resp = ListVersionDeltasResponse {
             version_deltas: Some(PbHummockVersionDeltas {
                 version_deltas: version_deltas
-                    .iter()
-                    .map(HummockVersionDelta::to_protobuf)
+                    .into_iter()
+                    .map(HummockVersionDelta::into)
                     .collect(),
             }),
         };
@@ -224,6 +225,7 @@ impl HummockManagerService for HummockServiceImpl {
         &self,
         request: Request<TriggerManualCompactionRequest>,
     ) -> Result<Response<TriggerManualCompactionResponse>, Status> {
+        use bytes::Bytes;
         let request = request.into_inner();
         let compaction_group_id = request.compaction_group_id;
         let mut option = ManualCompactionOption {
@@ -234,8 +236,12 @@ impl HummockManagerService for HummockServiceImpl {
 
         // rewrite the key_range
         match request.key_range {
-            Some(key_range) => {
-                option.key_range = key_range;
+            Some(pb_key_range) => {
+                option.key_range = KeyRange {
+                    left: Bytes::from(pb_key_range.left),
+                    right: Bytes::from(pb_key_range.right),
+                    right_exclusive: pb_key_range.right_exclusive,
+                };
             }
 
             None => {
@@ -426,7 +432,7 @@ impl HummockManagerService for HummockServiceImpl {
         let req = request.into_inner();
         let version = self.hummock_manager.pin_version(req.context_id).await?;
         Ok(Response::new(PinVersionResponse {
-            pinned_version: Some(version.to_protobuf()),
+            pinned_version: Some(version.into()),
         }))
     }
 
@@ -464,7 +470,7 @@ impl HummockManagerService for HummockServiceImpl {
     ) -> Result<Response<RiseCtlGetCheckpointVersionResponse>, Status> {
         let checkpoint_version = self.hummock_manager.get_checkpoint_version().await;
         Ok(Response::new(RiseCtlGetCheckpointVersionResponse {
-            checkpoint_version: Some(checkpoint_version.to_protobuf()),
+            checkpoint_version: Some(checkpoint_version.into()),
         }))
     }
 

src/meta/service/src/notification_service.rs

@@ -300,7 +300,7 @@ impl NotificationServiceImpl {
 
         Ok(MetaSnapshot {
             tables,
-            hummock_version: Some(hummock_version.to_protobuf()),
+            hummock_version: Some(hummock_version.into()),
             version: Some(SnapshotVersion {
                 catalog_version,
                 ..Default::default()

src/meta/src/backup_restore/restore.rs

@@ -97,7 +97,7 @@ async fn restore_hummock_version(
     );
     let checkpoint_path = version_checkpoint_path(hummock_storage_directory);
     let checkpoint = PbHummockVersionCheckpoint {
-        version: Some(hummock_version.to_protobuf()),
+        version: Some(hummock_version.into()),
         // Ignore stale objects. Full GC will clear them.
         stale_objects: Default::default(),
     };

src/meta/src/barrier/mod.rs

@@ -1176,13 +1176,13 @@ fn collect_commit_epoch_info(
             let sst_info = grouped.sst.expect("field not None");
             sst_to_worker.insert(sst_info.get_object_id(), resp.worker_id);
             LocalSstableInfo::new(
-                sst_info,
+                sst_info.into(),
                 from_prost_table_stats_map(grouped.table_stats_map),
             )
         });
         synced_ssts.extend(ssts_iter);
         table_watermarks.push(resp.table_watermarks);
-        old_value_ssts.extend(resp.old_value_sstables);
+        old_value_ssts.extend(resp.old_value_sstables.into_iter().map(|s| s.into()));
     }
     let new_table_fragment_info = if let Command::CreateStreamingJob {
         table_fragments, ..
@@ -1230,10 +1230,7 @@ fn collect_commit_epoch_info(
                     watermarks
                         .into_iter()
                         .map(|(table_id, watermarks)| {
-                            (
-                                TableId::new(table_id),
-                                TableWatermarks::from_protobuf(&watermarks),
-                            )
+                            (TableId::new(table_id), TableWatermarks::from(&watermarks))
                         })
                         .collect()
                 })

src/meta/src/hummock/compaction/mod.rs

@@ -17,6 +17,7 @@
 pub mod compaction_config;
 mod overlap_strategy;
 use risingwave_common::catalog::{TableId, TableOption};
+use risingwave_hummock_sdk::version::{CompactTask, Levels};
 use risingwave_pb::hummock::compact_task::{self, TaskType};
 
 mod picker;
@@ -28,8 +29,7 @@ use std::sync::Arc;
 use picker::{LevelCompactionPicker, TierCompactionPicker};
 use risingwave_hummock_sdk::{can_concat, CompactionGroupId, HummockCompactionTaskId};
 use risingwave_pb::hummock::compaction_config::CompactionMode;
-use risingwave_pb::hummock::hummock_version::Levels;
-use risingwave_pb::hummock::{CompactTask, CompactionConfig, LevelType};
+use risingwave_pb::hummock::{CompactionConfig, LevelType, PbCompactTask};
 pub use selector::CompactionSelector;
 
 use self::selector::{EmergencySelector, LocalSelectorStatistic};
@@ -177,6 +177,12 @@ impl CompactStatus {
     }
 
     /// Declares a task as either succeeded, failed or canceled.
+    pub fn report_compact_task_pb(&mut self, compact_task: &PbCompactTask) {
+        for level in &compact_task.input_ssts {
+            self.level_handlers[level.level_idx as usize].remove_task(compact_task.task_id);
+        }
+    }
+
     pub fn report_compact_task(&mut self, compact_task: &CompactTask) {
         for level in &compact_task.input_ssts {
             self.level_handlers[level.level_idx as usize].remove_task(compact_task.task_id);

src/meta/src/hummock/compaction/overlap_strategy.rs

@@ -17,9 +17,9 @@ use std::fmt::Debug;
 use std::ops::Range;
 
 use itertools::Itertools;
-use risingwave_hummock_sdk::key_range::KeyRangeCommon;
+use risingwave_hummock_sdk::key_range::{KeyRange, KeyRangeCommon};
+use risingwave_hummock_sdk::version::SstableInfo;
 use risingwave_hummock_sdk::KeyComparator;
-use risingwave_pb::hummock::{KeyRange, SstableInfo};
 
 pub trait OverlapInfo: Debug {
     fn check_overlap(&self, a: &SstableInfo) -> bool;

src/meta/src/hummock/compaction/picker/base_level_compaction_picker.rs

@@ -17,8 +17,8 @@ use std::sync::Arc;
 
 use itertools::Itertools;
 use risingwave_hummock_sdk::compaction_group::hummock_version_ext::HummockLevelsExt;
-use risingwave_pb::hummock::hummock_version::Levels;
-use risingwave_pb::hummock::{CompactionConfig, InputLevel, Level, LevelType, OverlappingLevel};
+use risingwave_hummock_sdk::version::{InputLevel, Level, Levels, OverlappingLevel};
+use risingwave_pb::hummock::{CompactionConfig, LevelType};
 
 use super::min_overlap_compaction_picker::NonOverlapSubLevelPicker;
 use super::{
@@ -51,7 +51,7 @@ impl CompactionPicker for LevelCompactionPicker {
         if l0.sub_levels.is_empty() {
             return None;
         }
-        if l0.sub_levels[0].level_type != LevelType::Nonoverlapping as i32
+        if l0.sub_levels[0].level_type != LevelType::Nonoverlapping
             && l0.sub_levels[0].table_infos.len() > 1
         {
             stats.skip_by_overlapping += 1;
@@ -231,7 +231,7 @@ impl LevelCompactionPicker {
                 .into_iter()
                 .map(|table_infos| InputLevel {
                     level_idx: 0,
-                    level_type: LevelType::Nonoverlapping as i32,
+                    level_type: LevelType::Nonoverlapping,
                     table_infos,
                 })
                 .collect_vec();
@@ -403,7 +403,7 @@ pub mod tests {
 
         let levels = vec![Level {
             level_idx: 1,
-            level_type: LevelType::Nonoverlapping as i32,
+            level_type: LevelType::Nonoverlapping,
             table_infos: vec![
                 generate_table(3, 1, 0, 50, 1),
                 generate_table(4, 1, 150, 200, 1),
@@ -466,7 +466,7 @@ pub mod tests {
         let mut picker = create_compaction_picker_for_test();
         let levels = vec![Level {
             level_idx: 1,
-            level_type: LevelType::Nonoverlapping as i32,
+            level_type: LevelType::Nonoverlapping,
             table_infos: vec![],
             total_file_size: 0,
             sub_level_id: 0,
@@ -530,7 +530,7 @@ pub mod tests {
         let mut levels = Levels {
             levels: vec![Level {
                 level_idx: 1,
-                level_type: LevelType::Nonoverlapping as i32,
+                level_type: LevelType::Nonoverlapping,
                 table_infos: vec![
                     generate_table(1, 1, 100, 399, 2),
                     generate_table(2, 1, 400, 699, 2),
@@ -578,7 +578,7 @@ pub mod tests {
         ]);
         // We can set level_type only because the input above is valid.
         for s in &mut l0.sub_levels {
-            s.level_type = LevelType::Nonoverlapping as i32;
+            s.level_type = LevelType::Nonoverlapping;
         }
         let levels = Levels {
             l0: Some(l0),

src/meta/src/hummock/compaction/picker/emergency_compaction_picker.rs

@@ -14,7 +14,7 @@
 
 use std::sync::Arc;
 
-use risingwave_pb::hummock::hummock_version::Levels;
+use risingwave_hummock_sdk::version::Levels;
 use risingwave_pb::hummock::{CompactionConfig, LevelType};
 
 use super::{
@@ -55,22 +55,20 @@ impl EmergencyCompactionPicker {
         let overlapping_count = l0
             .sub_levels
             .iter()
-            .filter(|level| level.level_type == LevelType::Overlapping as i32)
+            .filter(|level| level.level_type == LevelType::Overlapping)
             .count();
         let no_overlap_count = l0
             .sub_levels
             .iter()
             .filter(|level| {
-                level.level_type == LevelType::Nonoverlapping as i32
-                    && level.vnode_partition_count == 0
+                level.level_type == LevelType::Nonoverlapping && level.vnode_partition_count == 0
             })
             .count();
         let partitioned_count = l0
             .sub_levels
             .iter()
             .filter(|level| {
-                level.level_type == LevelType::Nonoverlapping as i32
-                    && level.vnode_partition_count > 0
+                level.level_type == LevelType::Nonoverlapping && level.vnode_partition_count > 0
             })
             .count();
         // We trigger `EmergencyCompactionPicker` only when some unexpected condition cause the number of l0 levels increase and the origin strategy