chore(cubestore): Introduce cachestore/metastore.rocksdb_properties tables (#7055)

Author: ovr

rust/cubestore/cubestore-sql-tests/src/tests.rs

@@ -65,6 +65,8 @@ pub fn sql_tests() -> Vec<(&'static str, TestFn)> {
         t("column_escaping", column_escaping),
         t("information_schema", information_schema),
         t("system_query_cache", system_query_cache),
+        t("metastore_rocksdb_tables", metastore_rocksdb_tables),
+        t("cachestore_rocksdb_tables", cachestore_rocksdb_tables),
         t("case_column_escaping", case_column_escaping),
         t("inner_column_escaping", inner_column_escaping),
         t("convert_tz", convert_tz),
@@ -1395,9 +1397,18 @@ async fn system_query_cache(service: Box<dyn SqlClient>) {
         .exec_query("SELECT * FROM system.query_cache")
         .await
         .unwrap();
+}
+
+async fn metastore_rocksdb_tables(service: Box<dyn SqlClient>) {
+    service
+        .exec_query("SELECT * FROM metastore.rocksdb_properties")
+        .await
+        .unwrap();
+}
 
+async fn cachestore_rocksdb_tables(service: Box<dyn SqlClient>) {
     service
-        .exec_query("SELECT sql FROM system.query_cache;")
+        .exec_query("SELECT * FROM cachestore.rocksdb_properties")
         .await
         .unwrap();
 }

rust/cubestore/cubestore/src/cachestore/cache_rocksstore.rs

@@ -13,8 +13,8 @@ use std::collections::HashMap;
 use std::env;
 
 use crate::metastore::{
-    BaseRocksStoreFs, BatchPipe, DbTableRef, IdRow, MetaStoreEvent, MetaStoreFs, RocksStore,
-    RocksStoreDetails, RocksTable,
+    BaseRocksStoreFs, BatchPipe, DbTableRef, IdRow, MetaStoreEvent, MetaStoreFs, RocksPropertyRow,
+    RocksStore, RocksStoreDetails, RocksTable,
 };
 use crate::remotefs::LocalDirRemoteFs;
 use crate::util::WorkerLoop;
@@ -536,6 +536,7 @@ pub trait CacheStore: DIService + Send + Sync {
     // Force compaction for the whole RocksDB
     async fn compaction(&self) -> Result<(), CubeError>;
     async fn healthcheck(&self) -> Result<(), CubeError>;
+    async fn rocksdb_properties(&self) -> Result<Vec<RocksPropertyRow>, CubeError>;
 }
 
 #[async_trait]
@@ -1038,6 +1039,10 @@ impl CacheStore for RocksCacheStore {
 
         Ok(())
     }
+
+    async fn rocksdb_properties(&self) -> Result<Vec<RocksPropertyRow>, CubeError> {
+        self.store.rocksdb_properties()
+    }
 }
 
 crate::di_service!(RocksCacheStore, [CacheStore]);
@@ -1170,6 +1175,10 @@ impl CacheStore for ClusterCacheStoreClient {
     async fn healthcheck(&self) -> Result<(), CubeError> {
         panic!("CacheStore cannot be used on the worker node! healthcheck was used.")
     }
+
+    async fn rocksdb_properties(&self) -> Result<Vec<RocksPropertyRow>, CubeError> {
+        panic!("CacheStore cannot be used on the worker node! rocksdb_properties was used.")
+    }
 }
 
 crate::di_service!(ClusterCacheStoreClient, [CacheStore]);

rust/cubestore/cubestore/src/cachestore/lazy.rs

@@ -5,7 +5,7 @@ use crate::cachestore::{
     RocksCacheStore,
 };
 use crate::config::ConfigObj;
-use crate::metastore::{IdRow, MetaStoreEvent, MetaStoreFs};
+use crate::metastore::{IdRow, MetaStoreEvent, MetaStoreFs, RocksPropertyRow};
 use crate::CubeError;
 use async_trait::async_trait;
 use log::trace;
@@ -302,6 +302,10 @@ impl CacheStore for LazyRocksCacheStore {
     async fn healthcheck(&self) -> Result<(), CubeError> {
         self.init().await?.healthcheck().await
     }
+
+    async fn rocksdb_properties(&self) -> Result<Vec<RocksPropertyRow>, CubeError> {
+        self.init().await?.rocksdb_properties().await
+    }
 }
 
 crate::di_service!(LazyRocksCacheStore, [CacheStore]);

rust/cubestore/cubestore/src/metastore/mod.rs

@@ -1126,6 +1126,7 @@ pub trait MetaStore: DIService + Send + Sync {
     // Force compaction for the whole RocksDB
     async fn compaction(&self) -> Result<(), CubeError>;
     async fn healthcheck(&self) -> Result<(), CubeError>;
+    async fn rocksdb_properties(&self) -> Result<Vec<RocksPropertyRow>, CubeError>;
 
     async fn get_snapshots_list(&self) -> Result<Vec<SnapshotInfo>, CubeError>;
     async fn set_current_snapshot(&self, snapshot_id: u128) -> Result<(), CubeError>;
@@ -4203,6 +4204,10 @@ impl MetaStore for RocksMetaStore {
         Ok(())
     }
 
+    async fn rocksdb_properties(&self) -> Result<Vec<RocksPropertyRow>, CubeError> {
+        self.store.rocksdb_properties()
+    }
+
     async fn healthcheck(&self) -> Result<(), CubeError> {
         self.store.healthcheck().await?;
 

rust/cubestore/cubestore/src/metastore/rocks_store.rs

@@ -139,6 +139,12 @@ pub struct RocksSecondaryIndexValueTTLExtended {
     pub raw_size: u32,
 }
 
+#[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
+pub struct RocksPropertyRow {
+    pub key: String,
+    pub value: Option<String>,
+}
+
 #[derive(Debug)]
 pub enum RocksSecondaryIndexValue<'a> {
     Hash(&'a [u8]),
@@ -991,6 +997,55 @@ impl RocksStore {
         Ok(())
     }
 
+    pub fn rocksdb_properties(&self) -> Result<Vec<RocksPropertyRow>, CubeError> {
+        let to_collect = [
+            rocksdb::properties::BLOCK_CACHE_CAPACITY,
+            rocksdb::properties::BLOCK_CACHE_USAGE,
+            rocksdb::properties::BLOCK_CACHE_PINNED_USAGE,
+            rocksdb::properties::LEVELSTATS,
+            &rocksdb::properties::compression_ratio_at_level(0),
+            &rocksdb::properties::compression_ratio_at_level(1),
+            &rocksdb::properties::compression_ratio_at_level(2),
+            &rocksdb::properties::compression_ratio_at_level(3),
+            &rocksdb::properties::compression_ratio_at_level(4),
+            &rocksdb::properties::compression_ratio_at_level(6),
+            rocksdb::properties::DBSTATS,
+            // rocksdb::properties::SSTABLES,
+            rocksdb::properties::NUM_RUNNING_FLUSHES,
+            rocksdb::properties::COMPACTION_PENDING,
+            rocksdb::properties::NUM_RUNNING_COMPACTIONS,
+            rocksdb::properties::BACKGROUND_ERRORS,
+            rocksdb::properties::CUR_SIZE_ACTIVE_MEM_TABLE,
+            rocksdb::properties::CUR_SIZE_ALL_MEM_TABLES,
+            rocksdb::properties::SIZE_ALL_MEM_TABLES,
+            rocksdb::properties::NUM_ENTRIES_ACTIVE_MEM_TABLE,
+            rocksdb::properties::NUM_ENTRIES_IMM_MEM_TABLES,
+            rocksdb::properties::NUM_DELETES_ACTIVE_MEM_TABLE,
+            rocksdb::properties::NUM_DELETES_IMM_MEM_TABLES,
+            rocksdb::properties::ESTIMATE_NUM_KEYS,
+            rocksdb::properties::NUM_SNAPSHOTS,
+            rocksdb::properties::OLDEST_SNAPSHOT_TIME,
+            rocksdb::properties::NUM_LIVE_VERSIONS,
+            rocksdb::properties::ESTIMATE_LIVE_DATA_SIZE,
+            rocksdb::properties::LIVE_SST_FILES_SIZE,
+            rocksdb::properties::ESTIMATE_PENDING_COMPACTION_BYTES,
+            rocksdb::properties::ESTIMATE_TABLE_READERS_MEM,
+            rocksdb::properties::BASE_LEVEL,
+            rocksdb::properties::AGGREGATED_TABLE_PROPERTIES,
+        ];
+
+        let mut result = Vec::with_capacity(to_collect.len());
+
+        for property_name in to_collect {
+            result.push(RocksPropertyRow {
+                key: property_name.to_string_lossy().to_string(),
+                value: self.db.property_value(property_name)?,
+            })
+        }
+
+        Ok(result)
+    }
+
     pub async fn healthcheck(&self) -> Result<(), CubeError> {
         self.read_operation(move |_| {
             // read_operation will call getSnapshot, which is enough to test that RocksDB works

rust/cubestore/cubestore/src/queryplanner/info_schema/mod.rs

@@ -1,6 +1,7 @@
 mod info_schema_columns;
 mod info_schema_schemata;
 mod info_schema_tables;
+mod rocksdb_properties;
 mod system_cache;
 mod system_chunks;
 mod system_indexes;
@@ -15,6 +16,7 @@ mod system_tables;
 pub use info_schema_columns::*;
 pub use info_schema_schemata::*;
 pub use info_schema_tables::*;
+pub use rocksdb_properties::*;
 pub use system_cache::*;
 pub use system_chunks::*;
 pub use system_indexes::*;

rust/cubestore/cubestore/src/queryplanner/info_schema/rocksdb_properties.rs

@@ -0,0 +1,66 @@
+use crate::metastore::RocksPropertyRow;
+use crate::queryplanner::{InfoSchemaTableDef, InfoSchemaTableDefContext};
+use crate::CubeError;
+use arrow::array::{ArrayRef, StringArray};
+use arrow::datatypes::{DataType, Field};
+use async_trait::async_trait;
+use std::sync::Arc;
+
+pub struct RocksDBPropertiesTableDef {
+    for_cachestore: bool,
+}
+
+impl RocksDBPropertiesTableDef {
+    pub fn new_cachestore() -> Self {
+        Self {
+            for_cachestore: true,
+        }
+    }
+
+    pub fn new_metastore() -> Self {
+        Self {
+            for_cachestore: false,
+        }
+    }
+}
+
+#[async_trait]
+impl InfoSchemaTableDef for RocksDBPropertiesTableDef {
+    type T = RocksPropertyRow;
+
+    async fn rows(
+        &self,
+        ctx: InfoSchemaTableDefContext,
+        _limit: Option<usize>,
+    ) -> Result<Arc<Vec<Self::T>>, CubeError> {
+        Ok(Arc::new(if self.for_cachestore {
+            ctx.cache_store.rocksdb_properties().await?
+        } else {
+            ctx.meta_store.rocksdb_properties().await?
+        }))
+    }
+
+    fn schema(&self) -> Vec<Field> {
+        vec![
+            Field::new("id", DataType::Utf8, false),
+            Field::new("value", DataType::Utf8, true),
+        ]
+    }
+
+    fn columns(&self) -> Vec<Box<dyn Fn(Arc<Vec<Self::T>>) -> ArrayRef>> {
+        vec![
+            Box::new(|items| {
+                Arc::new(StringArray::from_iter_values(
+                    items.iter().map(|row| row.key.clone()),
+                ))
+            }),
+            Box::new(|items| {
+                Arc::new(StringArray::from_iter(
+                    items.iter().map(|row| row.value.as_ref()),
+                ))
+            }),
+        ]
+    }
+}
+
+crate::base_info_schema_table_def!(RocksDBPropertiesTableDef);

rust/cubestore/cubestore/src/queryplanner/mod.rs

@@ -30,10 +30,11 @@ use crate::metastore::table::{Table, TablePath};
 use crate::metastore::{IdRow, MetaStore};
 use crate::queryplanner::flatten_union::FlattenUnion;
 use crate::queryplanner::info_schema::{
-    ColumnsInfoSchemaTableDef, SchemataInfoSchemaTableDef, SystemCacheTableDef,
-    SystemChunksTableDef, SystemIndexesTableDef, SystemJobsTableDef, SystemPartitionsTableDef,
-    SystemQueueResultsTableDef, SystemQueueTableDef, SystemReplayHandlesTableDef,
-    SystemSnapshotsTableDef, SystemTablesTableDef, TablesInfoSchemaTableDef,
+    ColumnsInfoSchemaTableDef, RocksDBPropertiesTableDef, SchemataInfoSchemaTableDef,
+    SystemCacheTableDef, SystemChunksTableDef, SystemIndexesTableDef, SystemJobsTableDef,
+    SystemPartitionsTableDef, SystemQueueResultsTableDef, SystemQueueTableDef,
+    SystemReplayHandlesTableDef, SystemSnapshotsTableDef, SystemTablesTableDef,
+    TablesInfoSchemaTableDef,
 };
 use crate::queryplanner::now::MaterializeNow;
 use crate::queryplanner::planning::{choose_index_ext, ClusterSendNode};
@@ -385,6 +386,16 @@ impl ContextProvider for MetaStoreSchemaProvider {
                 self.cache_store.clone(),
                 InfoSchemaTable::SystemSnapshots,
             ))),
+            ("metastore", "rocksdb_properties") => Some(Arc::new(InfoSchemaTableProvider::new(
+                self.meta_store.clone(),
+                self.cache_store.clone(),
+                InfoSchemaTable::MetastoreRocksDBProperties,
+            ))),
+            ("cachestore", "rocksdb_properties") => Some(Arc::new(InfoSchemaTableProvider::new(
+                self.meta_store.clone(),
+                self.cache_store.clone(),
+                InfoSchemaTable::CachestoreRocksDBProperties,
+            ))),
             _ => None,
         })
     }
@@ -428,6 +439,8 @@ pub enum InfoSchemaTable {
     SystemReplayHandles,
     SystemCache,
     SystemSnapshots,
+    CachestoreRocksDBProperties,
+    MetastoreRocksDBProperties,
 }
 
 pub struct InfoSchemaTableDefContext {
@@ -504,6 +517,12 @@ impl InfoSchemaTable {
             InfoSchemaTable::SystemJobs => Box::new(SystemJobsTableDef),
             InfoSchemaTable::SystemCache => Box::new(SystemCacheTableDef),
             InfoSchemaTable::SystemSnapshots => Box::new(SystemSnapshotsTableDef),
+            InfoSchemaTable::CachestoreRocksDBProperties => {
+                Box::new(RocksDBPropertiesTableDef::new_cachestore())
+            }
+            InfoSchemaTable::MetastoreRocksDBProperties => {
+                Box::new(RocksDBPropertiesTableDef::new_metastore())
+            }
         }
     }
 

rust/cubestore/cubestore/src/queryplanner/test_utils.rs

@@ -10,7 +10,7 @@ use crate::metastore::source::{Source, SourceCredentials};
 use crate::metastore::table::{StreamOffset, Table, TablePath};
 use crate::metastore::{
     Chunk, ChunkMetaStoreTable, Column, IdRow, ImportFormat, Index, IndexDef, IndexMetaStoreTable,
-    MetaStore, Partition, PartitionData, PartitionMetaStoreTable, RowKey, Schema,
+    MetaStore, Partition, PartitionData, PartitionMetaStoreTable, RocksPropertyRow, RowKey, Schema,
     SchemaMetaStoreTable, TableMetaStoreTable, WAL,
 };
 use crate::table::Row;
@@ -732,6 +732,11 @@ impl MetaStore for MetaStoreMock {
     async fn healthcheck(&self) -> Result<(), CubeError> {
         panic!("MetaStore mock!")
     }
+
+    async fn rocksdb_properties(&self) -> Result<Vec<RocksPropertyRow>, CubeError> {
+        panic!("MetaStore mock!")
+    }
+
     async fn get_snapshots_list(&self) -> Result<Vec<SnapshotInfo>, CubeError> {
         panic!("MetaStore mock!")
     }
@@ -873,6 +878,10 @@ impl CacheStore for CacheStoreMock {
     async fn healthcheck(&self) -> Result<(), CubeError> {
         panic!("CacheStore mock!")
     }
+
+    async fn rocksdb_properties(&self) -> Result<Vec<RocksPropertyRow>, CubeError> {
+        panic!("CacheStore mock!")
+    }
 }
 
 crate::di_service!(CacheStoreMock, [CacheStore]);