feat(meta): add hummock config relevant tables to rw_catalog

proto/hummock.proto

@@ -633,6 +633,19 @@ message ListBranchedObjectResponse {
  repeated BranchedObject branched_objects = 1;
 }
 
+message ListActiveWriteLimitRequest {}
+
+message ListActiveWriteLimitResponse {
+ // < compaction group id, write limit info >
+ map<uint64, WriteLimits.WriteLimit> write_limits = 1;
+}
+
+message ListHummockMetaConfigRequest {}
+
+message ListHummockMetaConfigResponse {
+ map<string, string> configs = 1;
+}
+
 service HummockManagerService {
  rpc UnpinVersionBefore(UnpinVersionBeforeRequest) returns (UnpinVersionBeforeResponse);
  rpc GetCurrentVersion(GetCurrentVersionRequest) returns (GetCurrentVersionResponse);
@@ -664,6 +677,8 @@ service HummockManagerService {
  rpc RiseCtlListCompactionStatus(RiseCtlListCompactionStatusRequest) returns (RiseCtlListCompactionStatusResponse);
  rpc SubscribeCompactionEvent(stream SubscribeCompactionEventRequest) returns (stream SubscribeCompactionEventResponse);
  rpc ListBranchedObject(ListBranchedObjectRequest) returns (ListBranchedObjectResponse);
+ rpc ListActiveWriteLimit(ListActiveWriteLimitRequest) returns (ListActiveWriteLimitResponse);
+ rpc ListHummockMetaConfig(ListHummockMetaConfigRequest) returns (ListHummockMetaConfigResponse);
 }
 
 message CompactionConfig {

src/frontend/src/catalog/system_catalog/mod.rs

@@ -409,6 +409,8 @@ prepare_sys_catalog! {
  { BuiltinCatalog::Table(&RW_HUMMOCK_CHECKPOINT_VERSION), read_hummock_checkpoint_version await },
  { BuiltinCatalog::Table(&RW_HUMMOCK_VERSION_DELTAS), read_hummock_version_deltas await },
  { BuiltinCatalog::Table(&RW_HUMMOCK_BRANCHED_OBJECTS), read_hummock_branched_objects await },
+ { BuiltinCatalog::Table(&RW_HUMMOCK_COMPACTION_GROUP_CONFIGS), read_hummock_compaction_group_configs await },
+ { BuiltinCatalog::Table(&RW_HUMMOCK_META_CONFIGS), read_hummock_meta_configs await},
 }
 
 #[cfg(test)]

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

@@ -20,6 +20,8 @@ mod rw_ddl_progress;
 mod rw_fragments;
 mod rw_functions;
 mod rw_hummock_branched_objects;
+mod rw_hummock_compaction_group_configs;
+mod rw_hummock_meta_configs;
 mod rw_hummock_pinned_snapshots;
 mod rw_hummock_pinned_versions;
 mod rw_hummock_version;
@@ -51,6 +53,8 @@ pub use rw_ddl_progress::*;
 pub use rw_fragments::*;
 pub use rw_functions::*;
 pub use rw_hummock_branched_objects::*;
+pub use rw_hummock_compaction_group_configs::*;
+pub use rw_hummock_meta_configs::*;
 pub use rw_hummock_pinned_snapshots::*;
 pub use rw_hummock_pinned_versions::*;
 pub use rw_hummock_version::*;

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

@@ -0,0 +1,72 @@
+// Copyright 2023 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use itertools::Itertools;
+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 serde_json::json;
+
+use crate::catalog::system_catalog::{BuiltinTable, SysCatalogReaderImpl};
+
+pub const RW_HUMMOCK_COMPACTION_GROUP_CONFIGS: BuiltinTable = BuiltinTable {
+ name: "rw_hummock_compaction_group_configs",
+ schema: RW_CATALOG_SCHEMA_NAME,
+ columns: &[
+ (DataType::Int64, "id"),
+ (DataType::Int64, "parent_id"),
+ (DataType::Jsonb, "member_tables"),
+ (DataType::Jsonb, "compaction_config"),
+ (DataType::Jsonb, "active_write_limit"),
+ ],
+ pk: &[0],
+};
+
+impl SysCatalogReaderImpl {
+ pub async fn read_hummock_compaction_group_configs(&self) -> Result<Vec<OwnedRow>> {
+ let info = self
+ .meta_client
+ .list_hummock_compaction_group_configs()
+ .await?;
+ let mut write_limits = self.meta_client.list_hummock_active_write_limits().await?;
+ let mut rows = info
+ .into_iter()
+ .map(|i| {
+ let active_write_limit = write_limits
+ .remove(&i.id)
+ .map(|w| ScalarImpl::Jsonb(json!(w).into()));
+ OwnedRow::new(vec![
+ Some(ScalarImpl::Int64(i.id as _)),
+ Some(ScalarImpl::Int64(i.parent_id as _)),
+ Some(ScalarImpl::Jsonb(json!(i.member_table_ids).into())),
+ Some(ScalarImpl::Jsonb(json!(i.compaction_config).into())),
+ active_write_limit,
+ ])
+ })
+ .collect_vec();
+ // As compaction group configs and active write limits are fetched via two RPCs, it's possible there's inconsistency.
+ // Just leave unknown field blank.
+ rows.extend(write_limits.into_iter().map(|(cg, w)| {
+ OwnedRow::new(vec![
+ Some(ScalarImpl::Int64(cg as _)),
+ None,
+ None,
+ None,
+ Some(ScalarImpl::Jsonb(json!(w).into())),
+ ])
+ }));
+ Ok(rows)
+ }
+}

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

@@ -0,0 +1,50 @@
+// Copyright 2023 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use itertools::Itertools;
+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 crate::catalog::system_catalog::{BuiltinTable, SysCatalogReaderImpl};
+
+pub const RW_HUMMOCK_META_CONFIGS: BuiltinTable = BuiltinTable {
+ name: "rw_hummock_meta_configs",
+ schema: RW_CATALOG_SCHEMA_NAME,
+ columns: &[
+ (DataType::Varchar, "config_name"),
+ (DataType::Varchar, "config_value"),
+ ],
+ pk: &[0],
+};
+
+impl SysCatalogReaderImpl {
+ pub async fn read_hummock_meta_configs(&self) -> Result<Vec<OwnedRow>> {
+ let configs = self
+ .meta_client
+ .list_hummock_meta_configs()
+ .await?
+ .into_iter()
+ .sorted()
+ .map(|(k, v)| {
+ OwnedRow::new(vec![
+ Some(ScalarImpl::Utf8(k.into())),
+ Some(ScalarImpl::Utf8(v.into())),
+ ])
+ })
+ .collect_vec();
+ Ok(configs)
+ }
+}

src/frontend/src/meta_client.rs

@@ -18,8 +18,9 @@ use risingwave_common::system_param::reader::SystemParamsReader;
 use risingwave_pb::backup_service::MetaSnapshotMetadata;
 use risingwave_pb::catalog::Table;
 use risingwave_pb::ddl_service::DdlProgress;
+use risingwave_pb::hummock::write_limits::WriteLimit;
 use risingwave_pb::hummock::{
- BranchedObject, HummockSnapshot, HummockVersion, HummockVersionDelta,
+ BranchedObject, CompactionGroupInfo, HummockSnapshot, HummockVersion, HummockVersionDelta,
 };
 use risingwave_pb::meta::cancel_creating_jobs_request::PbJobs;
 use risingwave_pb::meta::list_actor_states_response::ActorState;
@@ -86,6 +87,12 @@ pub trait FrontendMetaClient: Send + Sync {
  async fn list_version_deltas(&self) -> Result<Vec<HummockVersionDelta>>;
 
  async fn list_branched_objects(&self) -> Result<Vec<BranchedObject>>;
+
+ async fn list_hummock_compaction_group_configs(&self) -> Result<Vec<CompactionGroupInfo>>;
+
+ async fn list_hummock_active_write_limits(&self) -> Result<HashMap<u64, WriteLimit>>;
+
+ async fn list_hummock_meta_configs(&self) -> Result<HashMap<String, String>>;
 }
 
 pub struct FrontendMetaClientImpl(pub MetaClient);
@@ -214,4 +221,16 @@ impl FrontendMetaClient for FrontendMetaClientImpl {
  async fn list_branched_objects(&self) -> Result<Vec<BranchedObject>> {
  self.0.list_branched_object().await
  }
+
+ async fn list_hummock_compaction_group_configs(&self) -> Result<Vec<CompactionGroupInfo>> {
+ self.0.risectl_list_compaction_group().await
+ }
+
+ async fn list_hummock_active_write_limits(&self) -> Result<HashMap<u64, WriteLimit>> {
+ self.0.list_active_write_limit().await
+ }
+
+ async fn list_hummock_meta_configs(&self) -> Result<HashMap<String, String>> {
+ self.0.list_hummock_meta_config().await
+ }
 }

src/frontend/src/test_utils.rs

@@ -35,8 +35,9 @@ use risingwave_pb::catalog::{
  PbDatabase, PbFunction, PbIndex, PbSchema, PbSink, PbSource, PbTable, PbView, Table,
 };
 use risingwave_pb::ddl_service::{create_connection_request, DdlProgress};
+use risingwave_pb::hummock::write_limits::WriteLimit;
 use risingwave_pb::hummock::{
- BranchedObject, HummockSnapshot, HummockVersion, HummockVersionDelta,
+ BranchedObject, CompactionGroupInfo, HummockSnapshot, HummockVersion, HummockVersionDelta,
 };
 use risingwave_pb::meta::cancel_creating_jobs_request::PbJobs;
 use risingwave_pb::meta::list_actor_states_response::ActorState;
@@ -850,6 +851,18 @@ impl FrontendMetaClient for MockFrontendMetaClient {
  async fn list_branched_objects(&self) -> RpcResult<Vec<BranchedObject>> {
  unimplemented!()
  }
+
+ async fn list_hummock_compaction_group_configs(&self) -> RpcResult<Vec<CompactionGroupInfo>> {
+ unimplemented!()
+ }
+
+ async fn list_hummock_active_write_limits(&self) -> RpcResult<HashMap<u64, WriteLimit>> {
+ unimplemented!()
+ }
+
+ async fn list_hummock_meta_configs(&self) -> RpcResult<HashMap<String, String>> {
+ unimplemented!()
+ }
 }
 
 #[cfg(test)]

src/meta/src/backup_restore/backup_manager.rs

@@ -12,11 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashSet;
 use std::sync::Arc;
 use std::time::Instant;
 
 use arc_swap::ArcSwap;
-use itertools::Itertools;
 use risingwave_backup::error::BackupError;
 use risingwave_backup::storage::{BoxedMetaSnapshotStorage, ObjectStoreMetaSnapshotStorage};
 use risingwave_backup::{MetaBackupJobId, MetaSnapshotId, MetaSnapshotManifest};
@@ -307,16 +307,15 @@ impl BackupManager {
  }
 
  /// List all `SSTables` required by backups.
- pub fn list_pinned_ssts(&self) -> Vec<HummockSstableObjectId> {
+ pub fn list_pinned_ssts(&self) -> HashSet<HummockSstableObjectId> {
  self.backup_store
  .load()
  .0
  .manifest()
  .snapshot_metadata
  .iter()
  .flat_map(|s| s.ssts.clone())
- .dedup()
- .collect_vec()
+ .collect()
  }
 
  pub fn manifest(&self) -> Arc<MetaSnapshotManifest> {

src/meta/src/hummock/vacuum.rs

@@ -164,8 +164,7 @@ impl VacuumManager {
  &self,
  objects_to_delete: &mut Vec<HummockSstableObjectId>,
  ) -> MetaResult<()> {
- let reject: HashSet<HummockSstableObjectId> =
- self.backup_manager.list_pinned_ssts().into_iter().collect();
+ let reject = self.backup_manager.list_pinned_ssts();
  // Ack these SSTs immediately, because they tend to be pinned for long time.
  // They will be GCed during full GC when they are no longer pinned.
  let to_ack = objects_to_delete

src/meta/src/rpc/service/hummock_service.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::time::Duration;
 
 use futures::StreamExt;
@@ -48,6 +48,18 @@ impl HummockServiceImpl {
  }
 }
 
+macro_rules! fields_to_kvs {
+ ($struct:ident, $($field:ident),*) => {
+ {
+ let mut kvs = HashMap::default();
+ $(
+ kvs.insert(stringify!($field).to_string(), $struct.$field.to_string());
+ )*
+ kvs
+ }
+ }
+}
+
 #[async_trait::async_trait]
 impl HummockManagerService for HummockServiceImpl {
  type SubscribeCompactionEventStream = RwReceiverStream<SubscribeCompactionEventResponse>;
@@ -536,4 +548,63 @@ impl HummockManagerService for HummockServiceImpl {
  branched_objects,
  }))
  }
+
+ async fn list_active_write_limit(
+ &self,
+ _request: Request<ListActiveWriteLimitRequest>,
+ ) -> Result<Response<ListActiveWriteLimitResponse>, Status> {
+ Ok(Response::new(ListActiveWriteLimitResponse {
+ write_limits: self.hummock_manager.write_limits().await,
+ }))
+ }
+
+ async fn list_hummock_meta_config(
+ &self,
+ _request: Request<ListHummockMetaConfigRequest>,
+ ) -> Result<Response<ListHummockMetaConfigResponse>, Status> {
+ let opt = &self.hummock_manager.env.opts;
+ let configs = fields_to_kvs!(
+ opt,
+ vacuum_interval_sec,
+ vacuum_spin_interval_ms,
+ hummock_version_checkpoint_interval_sec,
+ min_delta_log_num_for_hummock_version_checkpoint,
+ min_sst_retention_time_sec,
+ full_gc_interval_sec,
+ collect_gc_watermark_spin_interval_sec,
+ periodic_compaction_interval_sec,
+ periodic_space_reclaim_compaction_interval_sec,
+ periodic_ttl_reclaim_compaction_interval_sec,
+ periodic_tombstone_reclaim_compaction_interval_sec,
+ periodic_split_compact_group_interval_sec,
+ split_group_size_limit,
+ min_table_split_size,
+ do_not_config_object_storage_lifecycle,
+ partition_vnode_count,
+ table_write_throughput_threshold,
+ min_table_split_write_throughput,
+ compaction_task_max_heartbeat_interval_secs
+ );
+ Ok(Response::new(ListHummockMetaConfigResponse { configs }))
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::collections::HashMap;
+ #[test]
+ fn test_fields_to_kvs() {
+ struct S {
+ foo: u64,
+ bar: String,
+ }
+ let s = S {
+ foo: 15,
+ bar: "foobar".to_string(),
+ };
+ let kvs: HashMap<String, String> = fields_to_kvs!(s, foo, bar);
+ assert_eq!(kvs.len(), 2);
+ assert_eq!(kvs.get("foo").unwrap(), "15");
+ assert_eq!(kvs.get("bar").unwrap(), "foobar");
+ }
 }