feat: read break down

rust/meta/src/hummock/integration_tests.rs

@@ -78,7 +78,7 @@ async fn get_hummock_storage() -> (HummockStorage, Arc<HummockManager<MemStore>>
     });
     let hummock_meta_client = Arc::new(get_hummock_meta_client().await);
     let obj_client = Arc::new(InMemObjectStore::new());
-    let sstable_store = Arc::new(SstableStore::new(obj_client.clone(), remote_dir));
+    let sstable_store = Arc::new(SstableStore::new(obj_client.clone(), remote_dir, None));
     let local_version_manager = Arc::new(LocalVersionManager::new(sstable_store.clone()));
     let storage = HummockStorage::with_default_stats(
         options.clone(),

rust/storage/src/hummock/compactor.rs

@@ -119,6 +119,7 @@ impl Compactor {
                             ))
                         },
                     )),
+                Some(context.stats.clone()),
             );
 
             let context_clone = context.clone();

rust/storage/src/hummock/iterator/concat_inner.rs

@@ -117,7 +117,7 @@ impl<TI: SSTableIteratorType> HummockIterator for ConcatIteratorInner<TI> {
 
         self.seek_idx(table_idx, Some(key)).await?;
         if !self.is_valid() {
-            // seek to next block
+            // seek to next table
             self.seek_idx(table_idx + 1, None).await?;
         }
         Ok(())

rust/storage/src/hummock/iterator/merge.rs

@@ -50,7 +50,7 @@ mod test {
             .map(|x| Box::new(x) as BoxedHummockIterator)
             .collect_vec();
 
-        let mut mi = MergeIterator::new(iters);
+        let mut mi = MergeIterator::new(iters, None);
         let mut i = 0;
         mi.rewind().await.unwrap();
         while mi.is_valid() {
@@ -84,7 +84,7 @@ mod test {
             .map(|x| Box::new(x) as BoxedHummockIterator)
             .collect_vec();
 
-        let mut mi = MergeIterator::new(iters);
+        let mut mi = MergeIterator::new(iters, None);
         let test_validator = &validators[2];
 
         // right edge case
@@ -131,7 +131,7 @@ mod test {
             Box::new(SSTableIterator::new(Arc::new(table1), sstable_store)),
         ];
 
-        let mut mi = MergeIterator::new(iters);
+        let mut mi = MergeIterator::new(iters, None);
 
         mi.rewind().await.unwrap();
         let mut count = 0;

rust/storage/src/hummock/iterator/merge_inner.rs

@@ -14,6 +14,7 @@
 //
 use std::collections::binary_heap::PeekMut;
 use std::collections::{BinaryHeap, LinkedList};
+use std::sync::Arc;
 
 use async_trait::async_trait;
 
@@ -22,6 +23,7 @@ use crate::hummock::iterator::{BoxedHummockIterator, HummockIterator};
 use crate::hummock::value::HummockValue;
 use crate::hummock::version_cmp::VersionedComparator;
 use crate::hummock::HummockResult;
+use crate::monitor::StateStoreMetrics;
 
 pub struct Node<'a, const DIRECTION: usize>(BoxedHummockIterator<'a>);
 
@@ -56,14 +58,21 @@ pub struct MergeIteratorInner<'a, const DIRECTION: usize> {
 
     /// The heap for merge sort.
     heap: BinaryHeap<Node<'a, DIRECTION>>,
+
+    /// Statistics.
+    stats: Option<Arc<StateStoreMetrics>>,
 }
 
 impl<'a, const DIRECTION: usize> MergeIteratorInner<'a, DIRECTION> {
     /// Caller should make sure that `iterators`'s direction is the same as `DIRECTION`.
-    pub fn new(iterators: impl IntoIterator<Item = BoxedHummockIterator<'a>>) -> Self {
+    pub fn new(
+        iterators: impl IntoIterator<Item = BoxedHummockIterator<'a>>,
+        stats: Option<Arc<StateStoreMetrics>>,
+    ) -> Self {
         Self {
             unused_iters: iterators.into_iter().collect(),
             heap: BinaryHeap::new(),
+            stats,
         }
     }
 
@@ -88,6 +97,18 @@ impl<'a, const DIRECTION: usize> MergeIteratorInner<'a, DIRECTION> {
 #[async_trait]
 impl<const DIRECTION: usize> HummockIterator for MergeIteratorInner<'_, DIRECTION> {
     async fn next(&mut self) -> HummockResult<()> {
+        let timer = if self.stats.is_some() {
+            Some(
+                self.stats
+                    .as_ref()
+                    .unwrap()
+                    .iter_merge_next_duration
+                    .start_timer(),
+            )
+        } else {
+            None
+        };
+
         let mut node = self.heap.peek_mut().expect("no inner iter");
 
         node.0.next().await?;
@@ -100,6 +121,10 @@ impl<const DIRECTION: usize> HummockIterator for MergeIteratorInner<'_, DIRECTIO
             drop(node);
         }
 
+        if let Some(timer) = timer {
+            timer.observe_duration();
+        }
+
         Ok(())
     }
 
@@ -123,9 +148,26 @@ impl<const DIRECTION: usize> HummockIterator for MergeIteratorInner<'_, DIRECTIO
     }
 
     async fn seek(&mut self, key: &[u8]) -> HummockResult<()> {
+        let timer = if self.stats.is_some() {
+            Some(
+                self.stats
+                    .as_ref()
+                    .unwrap()
+                    .iter_merge_seek_duration
+                    .start_timer(),
+            )
+        } else {
+            None
+        };
+
         self.reset_heap();
         futures::future::try_join_all(self.unused_iters.iter_mut().map(|x| x.seek(key))).await?;
         self.build_heap();
+
+        if let Some(timer) = timer {
+            timer.observe_duration();
+        }
+
         Ok(())
     }
 }

rust/storage/src/hummock/iterator/reverse_merge.rs

@@ -53,7 +53,7 @@ mod test {
             .map(|x| Box::new(x) as BoxedHummockIterator)
             .collect_vec();
 
-        let mut mi = ReverseMergeIterator::new(iters);
+        let mut mi = ReverseMergeIterator::new(iters, None);
         let mut i = 0;
         mi.rewind().await.unwrap();
         while mi.is_valid() {
@@ -90,7 +90,7 @@ mod test {
             .map(|x| Box::new(x) as BoxedHummockIterator)
             .collect_vec();
 
-        let mut mi = ReverseMergeIterator::new(iters);
+        let mut mi = ReverseMergeIterator::new(iters, None);
         let test_validator = &validators[2];
 
         // right edge case
@@ -137,7 +137,7 @@ mod test {
             Box::new(ReverseSSTableIterator::new(Arc::new(table0), sstable_store)),
         ];
 
-        let mut mi = ReverseMergeIterator::new(iters);
+        let mut mi = ReverseMergeIterator::new(iters, None);
 
         mi.rewind().await.unwrap();
         let mut count = 0;

rust/storage/src/hummock/iterator/reverse_user.rs

@@ -303,7 +303,7 @@ mod tests {
             .map(|x| Box::new(x) as BoxedHummockIterator)
             .collect_vec();
 
-        let mi = ReverseMergeIterator::new(iters);
+        let mi = ReverseMergeIterator::new(iters, None);
         let mut ui = ReverseUserIterator::new(mi, (Unbounded, Unbounded));
         ui.rewind().await.unwrap();
 
@@ -355,7 +355,7 @@ mod tests {
             .map(|x| Box::new(x) as BoxedHummockIterator)
             .collect_vec();
 
-        let mi = ReverseMergeIterator::new(iters);
+        let mi = ReverseMergeIterator::new(iters, None);
         let mut ui = ReverseUserIterator::new(mi, (Unbounded, Unbounded));
         let test_validator = &validators[2];
 
@@ -418,7 +418,7 @@ mod tests {
             )),
             Box::new(SSTableIterator::new(Arc::new(table1), sstable_store)),
         ];
-        let mi = ReverseMergeIterator::new(iters);
+        let mi = ReverseMergeIterator::new(iters, None);
         let mut ui = ReverseUserIterator::new(mi, (Unbounded, Unbounded));
 
         ui.rewind().await.unwrap();
@@ -463,7 +463,7 @@ mod tests {
             Arc::new(table),
             sstable_store,
         ))];
-        let mi = ReverseMergeIterator::new(iters);
+        let mi = ReverseMergeIterator::new(iters, None);
 
         let begin_key = Included(user_key(iterator_test_key_of_epoch(0, 2, 0).as_slice()).to_vec());
         let end_key = Included(user_key(iterator_test_key_of_epoch(0, 7, 0).as_slice()).to_vec());
@@ -543,7 +543,7 @@ mod tests {
             Arc::new(table),
             sstable_store,
         ))];
-        let mi = ReverseMergeIterator::new(iters);
+        let mi = ReverseMergeIterator::new(iters, None);
 
         let begin_key = Excluded(user_key(iterator_test_key_of_epoch(0, 2, 0).as_slice()).to_vec());
         let end_key = Included(user_key(iterator_test_key_of_epoch(0, 7, 0).as_slice()).to_vec());
@@ -624,7 +624,7 @@ mod tests {
             Arc::new(table),
             sstable_store,
         ))];
-        let mi = ReverseMergeIterator::new(iters);
+        let mi = ReverseMergeIterator::new(iters, None);
         let end_key = Included(user_key(iterator_test_key_of_epoch(0, 7, 0).as_slice()).to_vec());
 
         let mut ui = ReverseUserIterator::new(mi, (Unbounded, end_key));
@@ -703,7 +703,7 @@ mod tests {
             Arc::new(table),
             sstable_store,
         ))];
-        let mi = ReverseMergeIterator::new(iters);
+        let mi = ReverseMergeIterator::new(iters, None);
         let begin_key = Included(user_key(iterator_test_key_of_epoch(0, 2, 0).as_slice()).to_vec());
 
         let mut ui = ReverseUserIterator::new(mi, (begin_key, Unbounded));
@@ -795,7 +795,7 @@ mod tests {
             Arc::new(clone_sst(&table)),
             sstable_store,
         ))];
-        let rsi = ReverseMergeIterator::new(iters);
+        let rsi = ReverseMergeIterator::new(iters, None);
         let mut ruki = ReverseUserIterator::new(rsi, (start_bound, end_bound));
         let num_puts: usize = truth
             .iter()

rust/storage/src/hummock/iterator/test_utils.rs

@@ -343,7 +343,7 @@ pub fn mock_sstable_store() -> SstableStoreRef {
 
 pub fn mock_sstable_store_with_object_store(object_store: ObjectStoreRef) -> SstableStoreRef {
     let path = "test".to_string();
-    Arc::new(SstableStore::new(object_store, path))
+    Arc::new(SstableStore::new(object_store, path, None))
 }
 
 #[cfg(test)]

rust/storage/src/hummock/iterator/user.rs

@@ -277,7 +277,7 @@ mod tests {
             .map(|x| Box::new(x) as BoxedHummockIterator)
             .collect_vec();
 
-        let mi = MergeIterator::new(iters);
+        let mi = MergeIterator::new(iters, None);
         let mut ui = UserIterator::for_test(mi, (Unbounded, Unbounded));
         ui.rewind().await.unwrap();
 
@@ -318,7 +318,7 @@ mod tests {
             .map(|x| Box::new(x) as BoxedHummockIterator)
             .collect_vec();
 
-        let mi = MergeIterator::new(iters);
+        let mi = MergeIterator::new(iters, None);
         let mut ui = UserIterator::for_test(mi, (Unbounded, Unbounded));
         let test_validator = &validators[2];
 
@@ -385,7 +385,7 @@ mod tests {
                 sstable_store.clone(),
             )),
         ];
-        let mi = MergeIterator::new(iters);
+        let mi = MergeIterator::new(iters, None);
         let mut ui = UserIterator::for_test(mi, (Unbounded, Unbounded));
         ui.rewind().await.unwrap();
 
@@ -427,7 +427,7 @@ mod tests {
             Arc::new(table),
             sstable_store,
         ))];
-        let mi = MergeIterator::new(iters);
+        let mi = MergeIterator::new(iters, None);
 
         let begin_key = Included(user_key(iterator_test_key_of_epoch(0, 2, 0).as_slice()).to_vec());
         let end_key = Included(user_key(iterator_test_key_of_epoch(0, 7, 0).as_slice()).to_vec());
@@ -507,7 +507,7 @@ mod tests {
             Arc::new(table),
             sstable_store,
         ))];
-        let mi = MergeIterator::new(iters);
+        let mi = MergeIterator::new(iters, None);
 
         let begin_key = Included(user_key(iterator_test_key_of_epoch(0, 2, 0).as_slice()).to_vec());
         let end_key = Excluded(user_key(iterator_test_key_of_epoch(0, 7, 0).as_slice()).to_vec());
@@ -588,7 +588,7 @@ mod tests {
             Arc::new(table),
             sstable_store,
         ))];
-        let mi = MergeIterator::new(iters);
+        let mi = MergeIterator::new(iters, None);
         let end_key = Included(user_key(iterator_test_key_of_epoch(0, 7, 0).as_slice()).to_vec());
 
         let mut ui = UserIterator::for_test(mi, (Unbounded, end_key));
@@ -671,7 +671,7 @@ mod tests {
             Arc::new(table),
             sstable_store,
         ))];
-        let mi = MergeIterator::new(iters);
+        let mi = MergeIterator::new(iters, None);
         let begin_key = Included(user_key(iterator_test_key_of_epoch(0, 2, 0).as_slice()).to_vec());
 
         let mut ui = UserIterator::for_test(mi, (begin_key, Unbounded));