refactor(storage): replace PbHummockVersion with new HummockVersion struct

src/ctl/src/cmd_impl/hummock/list_version_deltas.rs

@@ -25,6 +25,6 @@ pub async fn list_version_deltas(
     let resp = meta_client
         .list_version_deltas(start_id, num_epochs, HummockEpoch::MAX)
         .await?;
-    println!("{:#?}", resp.version_deltas);
+    println!("{:#?}", resp);
     Ok(())
 }

src/ctl/src/cmd_impl/hummock/pause_resume.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use risingwave_hummock_sdk::compaction_group::hummock_version_ext::HummockVersionUpdateExt;
+use risingwave_hummock_sdk::version::HummockVersion;
 use risingwave_hummock_sdk::HummockEpoch;
 
 use crate::CtlContext;
@@ -51,11 +51,13 @@ pub async fn resume_version_checkpoint(context: &CtlContext) -> anyhow::Result<(
 /// added/removed for what reason (row deletion/compaction/etc.).
 pub async fn replay_version(context: &CtlContext) -> anyhow::Result<()> {
     let meta_client = context.meta_client().await?;
-    let mut base_version = meta_client
-        .risectl_get_checkpoint_hummock_version()
-        .await?
-        .checkpoint_version
-        .unwrap();
+    let mut base_version = HummockVersion::from_rpc_protobuf(
+        &meta_client
+            .risectl_get_checkpoint_hummock_version()
+            .await?
+            .checkpoint_version
+            .unwrap(),
+    );
     println!("replay starts");
     println!("base version {}", base_version.id);
     let delta_fetch_size = 100;
@@ -65,10 +67,10 @@ pub async fn replay_version(context: &CtlContext) -> anyhow::Result<()> {
             .list_version_deltas(current_delta_id, delta_fetch_size, HummockEpoch::MAX)
             .await
             .unwrap();
-        if deltas.version_deltas.is_empty() {
+        if deltas.is_empty() {
             break;
         }
-        for delta in deltas.version_deltas {
+        for delta in deltas {
             if delta.prev_id != base_version.id {
                 eprintln!("missing delta log for version {}", base_version.id);
                 break;

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

@@ -28,7 +28,6 @@ use risingwave_common::util::iter_util::ZipEqFast;
 use risingwave_common::util::value_encoding::column_aware_row_encoding::ColumnAwareSerde;
 use risingwave_common::util::value_encoding::{BasicSerde, EitherSerde, ValueRowDeserializer};
 use risingwave_frontend::TableCatalog;
-use risingwave_hummock_sdk::compaction_group::hummock_version_ext::HummockVersionExt;
 use risingwave_hummock_sdk::key::FullKey;
 use risingwave_hummock_sdk::HummockSstableObjectId;
 use risingwave_object_store::object::{ObjectMetadata, ObjectStoreImpl};
@@ -75,7 +74,7 @@ pub async fn sst_dump(context: &CtlContext, args: SstDumpArgs) -> anyhow::Result
         let hummock = context
             .hummock_store(HummockServiceOpts::from_env(args.data_dir.clone())?)
             .await?;
-        let version = hummock.inner().get_pinned_version().version();
+        let version = hummock.inner().get_pinned_version().version().clone();
         let sstable_store = hummock.sstable_store();
         for level in version.get_combined_levels() {
             for sstable_info in &level.table_infos {

src/frontend/Cargo.toml

@@ -56,6 +56,7 @@ risingwave_common = { workspace = true }
 risingwave_common_service = { workspace = true }
 risingwave_connector = { workspace = true }
 risingwave_expr = { workspace = true }
+risingwave_hummock_sdk = { workspace = true }
 risingwave_pb = { workspace = true }
 risingwave_rpc_client = { workspace = true }
 risingwave_source = { workspace = true }

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

@@ -16,7 +16,7 @@ use risingwave_common::catalog::RW_CATALOG_SCHEMA_NAME;
 use risingwave_common::error::Result;
 use risingwave_common::row::OwnedRow;
 use risingwave_common::types::{DataType, ScalarImpl};
-use risingwave_pb::hummock::HummockVersion;
+use risingwave_hummock_sdk::version::HummockVersion;
 use serde_json::json;
 
 use crate::catalog::system_catalog::{BuiltinTable, SysCatalogReaderImpl};

src/frontend/src/meta_client.rs

@@ -15,14 +15,15 @@
 use std::collections::HashMap;
 
 use risingwave_common::system_param::reader::SystemParamsReader;
+use risingwave_hummock_sdk::version::{HummockVersion, HummockVersionDelta};
 use risingwave_pb::backup_service::MetaSnapshotMetadata;
 use risingwave_pb::catalog::Table;
 use risingwave_pb::common::WorkerNode;
 use risingwave_pb::ddl_service::DdlProgress;
 use risingwave_pb::hummock::write_limits::WriteLimit;
 use risingwave_pb::hummock::{
     BranchedObject, CompactTaskAssignment, CompactTaskProgress, CompactionGroupInfo,
-    HummockSnapshot, HummockVersion, HummockVersionDelta,
+    HummockSnapshot,
 };
 use risingwave_pb::meta::cancel_creating_jobs_request::PbJobs;
 use risingwave_pb::meta::list_actor_states_response::ActorState;
@@ -223,15 +224,12 @@ impl FrontendMetaClient for FrontendMetaClientImpl {
         self.0
             .risectl_get_checkpoint_hummock_version()
             .await
-            .map(|v| v.checkpoint_version.unwrap())
+            .map(|v| HummockVersion::from_rpc_protobuf(&v.checkpoint_version.unwrap()))
     }
 
     async fn list_version_deltas(&self) -> Result<Vec<HummockVersionDelta>> {
         // FIXME #8612: there can be lots of version deltas, so better to fetch them by pages and refactor `SysRowSeqScanExecutor` to yield multiple chunks.
-        self.0
-            .list_version_deltas(0, u32::MAX, u64::MAX)
-            .await
-            .map(|v| v.version_deltas)
+        self.0.list_version_deltas(0, u32::MAX, u64::MAX).await
     }
 
     async fn list_branched_objects(&self) -> Result<Vec<BranchedObject>> {

src/frontend/src/test_utils.rs

@@ -31,6 +31,7 @@ use risingwave_common::catalog::{
 use risingwave_common::error::{ErrorCode, Result};
 use risingwave_common::system_param::reader::SystemParamsReader;
 use risingwave_common::util::column_index_mapping::ColIndexMapping;
+use risingwave_hummock_sdk::version::{HummockVersion, HummockVersionDelta};
 use risingwave_pb::backup_service::MetaSnapshotMetadata;
 use risingwave_pb::catalog::table::OptionalAssociatedSourceId;
 use risingwave_pb::catalog::{
@@ -45,7 +46,7 @@ use risingwave_pb::ddl_service::{
 use risingwave_pb::hummock::write_limits::WriteLimit;
 use risingwave_pb::hummock::{
     BranchedObject, CompactTaskAssignment, CompactTaskProgress, CompactionGroupInfo,
-    HummockSnapshot, HummockVersion, HummockVersionDelta,
+    HummockSnapshot,
 };
 use risingwave_pb::meta::cancel_creating_jobs_request::PbJobs;
 use risingwave_pb::meta::list_actor_states_response::ActorState;

src/jni_core/src/hummock_iterator.rs

@@ -24,6 +24,7 @@ use risingwave_common::util::select_all;
 use risingwave_common::util::value_encoding::column_aware_row_encoding::ColumnAwareSerde;
 use risingwave_common::util::value_encoding::{BasicSerde, EitherSerde, ValueRowDeserializer};
 use risingwave_hummock_sdk::key::{prefixed_range_with_vnode, TableKeyRange};
+use risingwave_hummock_sdk::version::HummockVersion;
 use risingwave_object_store::object::build_remote_object_store;
 use risingwave_object_store::object::object_metrics::ObjectStoreMetrics;
 use risingwave_pb::java_binding::key_range::Bound;
@@ -84,7 +85,10 @@ impl HummockJavaBindingIterator {
 
         let mut streams = Vec::with_capacity(read_plan.vnode_ids.len());
         let key_range = read_plan.key_range.unwrap();
-        let pin_version = PinnedVersion::new(read_plan.version.unwrap(), unbounded_channel().0);
+        let pin_version = PinnedVersion::new(
+            HummockVersion::from_rpc_protobuf(&read_plan.version.unwrap()),
+            unbounded_channel().0,
+        );
         let table_id = read_plan.table_id.into();
 
         for vnode in read_plan.vnode_ids {

src/meta/model_v2/Cargo.toml

@@ -14,6 +14,7 @@ ignored = ["workspace-hack"]
 normal = ["workspace-hack"]
 
 [dependencies]
+risingwave_hummock_sdk = { workspace = true }
 risingwave_pb = { workspace = true }
 sea-orm = { version = "0.12.0", features = [
     "sqlx-mysql",

src/meta/model_v2/src/hummock_version_delta.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use risingwave_pb::hummock::HummockVersionDelta;
+use risingwave_hummock_sdk::version::HummockVersionDelta;
 use sea_orm::entity::prelude::*;
 use sea_orm::FromJsonQueryResult;
 use serde::{Deserialize, Serialize};

src/meta/service/Cargo.toml

@@ -23,6 +23,7 @@ rand = "0.8"
 regex = "1"
 risingwave_common = { workspace = true }
 risingwave_connector = { workspace = true }
+risingwave_hummock_sdk = { workspace = true }
 risingwave_meta = { workspace = true }
 risingwave_meta_model_v2 = { workspace = true }
 risingwave_pb = { 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, NON_RESERVED_SYS_CATALOG_ID};
+use risingwave_hummock_sdk::version::HummockVersionDelta;
 use risingwave_pb::hummock::get_compaction_score_response::PickerInfo;
 use risingwave_pb::hummock::hummock_manager_service_server::HummockManagerService;
 use risingwave_pb::hummock::subscribe_compaction_event_request::Event as RequestEvent;
@@ -29,6 +30,7 @@ use crate::hummock::compaction::selector::ManualCompactionOption;
 use crate::hummock::{HummockManagerRef, VacuumManagerRef};
 use crate::manager::FragmentManagerRef;
 use crate::RwReceiverStream;
+
 pub struct HummockServiceImpl {
     hummock_manager: HummockManagerRef,
     vacuum_manager: VacuumManagerRef,
@@ -83,7 +85,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),
+            current_version: Some(current_version.to_protobuf()),
         }))
     }
 
@@ -94,10 +96,12 @@ impl HummockManagerService for HummockServiceImpl {
         let req = request.into_inner();
         let (version, compaction_groups) = self
             .hummock_manager
-            .replay_version_delta(req.version_delta.unwrap())
+            .replay_version_delta(HummockVersionDelta::from_rpc_protobuf(
+                &req.version_delta.unwrap(),
+            ))
             .await?;
         Ok(Response::new(ReplayVersionDeltaResponse {
-            version: Some(version),
+            version: Some(version.to_protobuf()),
             modified_compaction_groups: compaction_groups,
         }))
     }
@@ -119,7 +123,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),
+            current_version: Some(version.to_protobuf()),
         }))
     }
 
@@ -133,7 +137,12 @@ impl HummockManagerService for HummockServiceImpl {
             .list_version_deltas(req.start_id, req.num_limit, req.committed_epoch_limit)
             .await?;
         let resp = ListVersionDeltasResponse {
-            version_deltas: Some(version_deltas),
+            version_deltas: Some(PbHummockVersionDeltas {
+                version_deltas: version_deltas
+                    .iter()
+                    .map(HummockVersionDelta::to_protobuf)
+                    .collect(),
+            }),
         };
         Ok(Response::new(resp))
     }
@@ -460,7 +469,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),
+            checkpoint_version: Some(checkpoint_version.to_protobuf()),
         }))
     }
 

src/meta/service/src/notification_service.rs

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

src/meta/src/backup_restore/meta_snapshot_builder.rs

@@ -19,11 +19,11 @@ use anyhow::anyhow;
 use risingwave_backup::error::{BackupError, BackupResult};
 use risingwave_backup::meta_snapshot_v1::{ClusterMetadata, MetaSnapshotV1};
 use risingwave_backup::MetaSnapshotId;
-use risingwave_hummock_sdk::compaction_group::hummock_version_ext::HummockVersionUpdateExt;
+use risingwave_hummock_sdk::version::{HummockVersion, HummockVersionDelta};
 use risingwave_pb::catalog::{
     Connection, Database, Function, Index, Schema, Sink, Source, Table, View,
 };
-use risingwave_pb::hummock::{HummockVersion, HummockVersionDelta, HummockVersionStats};
+use risingwave_pb::hummock::HummockVersionStats;
 use risingwave_pb::meta::SystemParams;
 use risingwave_pb::user::UserInfo;
 
@@ -46,10 +46,10 @@ impl<S: MetaStore> MetaSnapshotV1Builder<S> {
         }
     }
 
-    pub async fn build<D: Future<Output = HummockVersion>>(
+    pub async fn build(
         &mut self,
         id: MetaSnapshotId,
-        hummock_version_builder: D,
+        hummock_version_builder: impl Future<Output = HummockVersion>,
     ) -> BackupResult<()> {
         self.snapshot.format_version = VERSION;
         self.snapshot.id = id;
@@ -169,7 +169,8 @@ mod tests {
     use risingwave_backup::error::BackupError;
     use risingwave_backup::meta_snapshot_v1::MetaSnapshotV1;
     use risingwave_common::system_param::system_params_for_test;
-    use risingwave_pb::hummock::{HummockVersion, HummockVersionStats};
+    use risingwave_hummock_sdk::version::HummockVersion;
+    use risingwave_pb::hummock::HummockVersionStats;
 
     use crate::backup_restore::meta_snapshot_builder;
     use crate::manager::model::SystemParamsModel;

src/meta/src/backup_restore/meta_snapshot_builder_v2.rs

@@ -20,9 +20,8 @@ use itertools::Itertools;
 use risingwave_backup::error::{BackupError, BackupResult};
 use risingwave_backup::meta_snapshot_v2::{MetaSnapshotV2, MetadataV2};
 use risingwave_backup::MetaSnapshotId;
-use risingwave_hummock_sdk::compaction_group::hummock_version_ext::HummockVersionUpdateExt;
+use risingwave_hummock_sdk::version::{HummockVersion, HummockVersionDelta};
 use risingwave_meta_model_v2 as model_v2;
-use risingwave_pb::hummock::{HummockVersion, HummockVersionDelta};
 use sea_orm::{EntityTrait, QueryOrder, TransactionTrait};
 
 use crate::controller::SqlMetaStore;
@@ -42,10 +41,10 @@ impl MetaSnapshotV2Builder {
         }
     }
 
-    pub async fn build<D: Future<Output = HummockVersion>>(
+    pub async fn build(
         &mut self,
         id: MetaSnapshotId,
-        hummock_version_builder: D,
+        hummock_version_builder: impl Future<Output = HummockVersion>,
     ) -> BackupResult<()> {
         self.snapshot.format_version = VERSION;
         self.snapshot.id = id;

src/meta/src/backup_restore/restore.rs

@@ -19,10 +19,11 @@ use risingwave_backup::meta_snapshot::Metadata;
 use risingwave_backup::storage::{MetaSnapshotStorage, MetaSnapshotStorageRef};
 use risingwave_backup::MetaSnapshotId;
 use risingwave_common::config::{MetaBackend, ObjectStoreConfig};
+use risingwave_hummock_sdk::version::HummockVersion;
 use risingwave_hummock_sdk::version_checkpoint_path;
 use risingwave_object_store::object::build_remote_object_store;
 use risingwave_object_store::object::object_metrics::ObjectStoreMetrics;
-use risingwave_pb::hummock::{HummockVersion, HummockVersionCheckpoint};
+use risingwave_pb::hummock::PbHummockVersionCheckpoint;
 
 use crate::backup_restore::restore_impl::v1::{LoaderV1, WriterModelV1ToMetaStoreV1};
 use crate::backup_restore::restore_impl::v2::{LoaderV2, WriterModelV2ToMetaStoreV2};
@@ -83,8 +84,8 @@ async fn restore_hummock_version(
         .await,
     );
     let checkpoint_path = version_checkpoint_path(hummock_storage_directory);
-    let checkpoint = HummockVersionCheckpoint {
-        version: Some(hummock_version.clone()),
+    let checkpoint = PbHummockVersionCheckpoint {
+        version: Some(hummock_version.to_protobuf()),
         // Ignore stale objects. Full GC will clear them.
         stale_objects: Default::default(),
     };
@@ -206,7 +207,8 @@ mod tests {
     use risingwave_backup::meta_snapshot_v1::{ClusterMetadata, MetaSnapshotV1};
     use risingwave_backup::storage::MetaSnapshotStorage;
     use risingwave_common::config::{MetaBackend, SystemConfig};
-    use risingwave_pb::hummock::{HummockVersion, HummockVersionStats};
+    use risingwave_hummock_sdk::version::HummockVersion;
+    use risingwave_pb::hummock::HummockVersionStats;
     use risingwave_pb::meta::SystemParams;
 
     use crate::backup_restore::restore::restore_impl;