feat(ctl): support list key-value pairs

Signed-off-by: Alex Chi <[email protected]>

rust/ctl/Cargo.toml

@@ -5,7 +5,9 @@ edition = "2021"
 
 [dependencies]
 anyhow = "1"
+bytes = "1"
 clap = { version = "3", features = ["derive"] }
+risingwave_common = { path = "../common" }
 risingwave_pb = { path = "../prost" }
 risingwave_rpc_client = { path = "../rpc_client" }
 risingwave_storage = { path = "../storage" }

rust/ctl/src/cmd_impl.rs

@@ -1,2 +1,4 @@
 mod list_version;
 pub use list_version::*;
+mod list_kv;
+pub use list_kv::*;

rust/ctl/src/cmd_impl/list_kv.rs

@@ -0,0 +1,16 @@
+use risingwave_storage::StateStore;
+
+use crate::common::HummockServiceOpts;
+
+pub async fn list_kv() -> anyhow::Result<()> {
+    let hummock_opts = HummockServiceOpts::from_env()?;
+    let hummock = hummock_opts.create_hummock_store().await?;
+    // TODO: support speficy epoch
+    tracing::info!("using u64::MAX as epoch");
+
+    for (k, v) in hummock.scan::<_, Vec<u8>>(.., None, u64::MAX).await? {
+        println!("{:?} => {:?}", k, v);
+    }
+
+    Ok(())
+}

rust/ctl/src/common.rs

@@ -1,2 +1,4 @@
 mod meta_service;
 pub use meta_service::*;
+mod hummock_service;
+pub use hummock_service::*;

rust/ctl/src/common/hummock_service.rs

@@ -0,0 +1,61 @@
+use std::env;
+use std::sync::Arc;
+
+use anyhow::{anyhow, Result};
+use risingwave_common::config::StorageConfig;
+use risingwave_storage::hummock::HummockStateStore;
+use risingwave_storage::monitor::{MonitoredStateStore, StateStoreStats};
+use risingwave_storage::StateStoreImpl;
+
+use super::MetaServiceOpts;
+
+pub struct HummockServiceOpts {
+    pub meta_opts: MetaServiceOpts,
+    pub hummock_url: String,
+}
+
+impl HummockServiceOpts {
+    /// Recover hummock service options from env variable
+    ///
+    /// Currently, we will read these variables for meta:
+    ///
+    /// * `RW_HUMMOCK_URL`: meta service address
+    pub fn from_env() -> Result<Self> {
+        let meta_opts = MetaServiceOpts::from_env()?;
+        let hummock_url = env::var("RW_HUMMOCK_URL").unwrap_or_else(|_| {
+            const DEFAULT_ADDR: &str = "hummock+minio://hummock:[email protected]:9301/hummock001";
+            tracing::warn!(
+                "`RW_HUMMOCK_URL` not found, using default hummock URL {}",
+                DEFAULT_ADDR
+            );
+            DEFAULT_ADDR.to_string()
+        });
+        Ok(Self {
+            meta_opts,
+            hummock_url,
+        })
+    }
+
+    pub async fn create_hummock_store(&self) -> Result<MonitoredStateStore<HummockStateStore>> {
+        let meta_client = self.meta_opts.create_meta_client().await?;
+
+        // FIXME: allow specify custom config
+        let config = StorageConfig::default();
+
+        tracing::info!("using Hummock config: {:#?}", config);
+
+        let state_store_impl = StateStoreImpl::new(
+            &self.hummock_url,
+            Arc::new(config),
+            meta_client,
+            Arc::new(StateStoreStats::unused()),
+        )
+        .await?;
+
+        if let StateStoreImpl::HummockStateStore(hummock_state_store) = state_store_impl {
+            Ok(hummock_state_store)
+        } else {
+            Err(anyhow!("only Hummock state store is supported in risectl"))
+        }
+    }
+}

rust/ctl/src/common/meta_service.rs

@@ -17,7 +17,7 @@ impl MetaServiceOpts {
     /// Currently, we will read these variables for meta:
     ///
     /// * `RW_META_ADDR`: meta service address
-    pub fn from_env() -> Result<MetaServiceOpts> {
+    pub fn from_env() -> Result<Self> {
         let meta_addr = env::var("RW_META_ADDR").unwrap_or_else(|_| {
             const DEFAULT_ADDR: &str = "http://127.0.0.1:5690";
             tracing::warn!(

rust/ctl/src/lib.rs

@@ -14,12 +14,15 @@ struct Cli {
 enum Commands {
     /// list latest Hummock version on meta node
     ListVersion,
+    /// list all Hummock key-value pairs
+    ListKv,
 }
 
 pub async fn start() {
     let cli = Cli::parse();
 
     match &cli.command {
         Commands::ListVersion => cmd_impl::list_version().await.unwrap(),
+        Commands::ListKv => cmd_impl::list_kv().await.unwrap(),
     }
 }