Implement LogDb count api (opendata-oss#449)

Implements the count API for LogDb. I've removed the approximate count option, which seems kind of useless. Although there are opportunities for higher-level caching, this first pass just relis on the  block cache. Each count query is divided into one portion which uses slatedb BlockStats on the written SSTs, and one portion which scans keys in memtables and does a direct count using a range scan.

Author: hachikuji

Cargo.lock

@@ -29,6 +29,7 @@ mixtrics.workspace = true
 tokio.workspace = true
 tokio-util.workspace = true
 tracing.workspace = true
+ulid.workspace = true
 uuid.workspace = true
 serde.workspace = true
 arc-swap = { version = "1", optional = true }

common/Cargo.toml

@@ -18,6 +18,7 @@ pub use storage::factory::{
     create_object_store, create_storage_read,
 };
 pub use storage::loader::{LoadMetadata, LoadResult, LoadSpec, Loadable, Loader};
+pub use storage::sst_blocks::{BlockOpCounts, CountResult, count_in_range};
 pub use storage::{
     CheckpointInfo, MergeRecordOp, PutRecordOp, Record, Storage, StorageError, StorageIterator,
     StorageRead, StorageResult, Ttl, WriteOptions, WriteResult,

common/src/lib.rs

@@ -4,6 +4,7 @@ pub mod in_memory;
 pub mod loader;
 pub mod metrics_recorder;
 pub mod slate;
+pub mod sst_blocks;
 pub mod util;
 
 use std::sync::Arc;

common/src/storage/mod.rs

@@ -495,7 +495,10 @@ mod tests {
         storage.flush().await.unwrap();
 
         // Create reader and verify data
-        let reader = DbReader::builder(path, object_store).build().await.unwrap();
+        let reader = DbReader::builder(path, object_store.clone())
+            .build()
+            .await
+            .unwrap();
         let storage_reader = SlateDbStorageReader::new(Arc::new(reader));
 
         let record = storage_reader.get(Bytes::from("key1")).await.unwrap();
@@ -535,7 +538,10 @@ mod tests {
         storage.flush().await.unwrap();
 
         // Create reader and scan data
-        let reader = DbReader::builder(path, object_store).build().await.unwrap();
+        let reader = DbReader::builder(path, object_store.clone())
+            .build()
+            .await
+            .unwrap();
         let storage_reader = SlateDbStorageReader::new(Arc::new(reader));
 
         let mut iter = storage_reader
@@ -577,7 +583,10 @@ mod tests {
         storage.flush().await.unwrap();
 
         // Create reader while writer is still open - this should NOT cause fencing error
-        let reader = DbReader::builder(path, object_store).build().await.unwrap();
+        let reader = DbReader::builder(path, object_store.clone())
+            .build()
+            .await
+            .unwrap();
         let storage_reader = SlateDbStorageReader::new(Arc::new(reader));
 
         // Reader can read the data
@@ -604,7 +613,7 @@ mod tests {
         let path = "/test/ttl_db";
         let clock = Arc::new(MockSystemClock::new());
 
-        let db = DbBuilder::new(path, object_store)
+        let db = DbBuilder::new(path, object_store.clone())
             .with_settings(Settings {
                 default_ttl: Some(30_000),
                 ..Default::default()
@@ -681,7 +690,7 @@ mod tests {
 
         let merge_op: Arc<dyn MergeOperator> = Arc::new(ConcatMergeOperator);
         let slate_merge_op = SlateDbStorage::merge_operator_adapter(merge_op);
-        let db = DbBuilder::new(path, object_store)
+        let db = DbBuilder::new(path, object_store.clone())
             .with_settings(Settings {
                 default_ttl: Some(30_000),
                 ..Default::default()