fix(metrics): RowGroupLastRowCachedReader metrics (#4418)

fix/reader-metrics:
 Refactor cache hit/miss logic and update metrics in mito2

 - Simplify cache retrieval logic in CacheManager by removing inline update_hit_miss function call.
 - Add separate functions for incrementing cache hit and miss metrics.
 - Update RowGroupLastRowCachedReader to use new cache hit/miss functions and refactor to new helper methods for creating Hit and Miss variants.

src/mito2/src/cache.rs

@@ -179,10 +179,7 @@ impl CacheManager {
     ) -> Option<Arc<SelectorResultValue>> {
         self.selector_result_cache
             .as_ref()
-            .and_then(|selector_result_cache| {
-                let value = selector_result_cache.get(selector_key);
-                update_hit_miss(value, SELECTOR_RESULT_TYPE)
-            })
+            .and_then(|selector_result_cache| selector_result_cache.get(selector_key))
     }
 
     /// Puts result of the selector into the cache.
@@ -209,6 +206,16 @@ impl CacheManager {
     }
 }
 
+/// Increases selector cache miss metrics.
+pub fn selector_result_cache_miss() {
+    CACHE_MISS.with_label_values(&[SELECTOR_RESULT_TYPE]).inc()
+}
+
+/// Increases selector cache hit metrics.
+pub fn selector_result_cache_hit() {
+    CACHE_HIT.with_label_values(&[SELECTOR_RESULT_TYPE]).inc()
+}
+
 /// Builder to construct a [CacheManager].
 #[derive(Default)]
 pub struct CacheManagerBuilder {

src/mito2/src/read/last_row.rs

@@ -19,7 +19,10 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use store_api::storage::TimeSeriesRowSelector;
 
-use crate::cache::{CacheManagerRef, SelectorResultKey, SelectorResultValue};
+use crate::cache::{
+    selector_result_cache_hit, selector_result_cache_miss, CacheManagerRef, SelectorResultKey,
+    SelectorResultValue,
+};
 use crate::error::Result;
 use crate::read::{Batch, BatchReader, BoxedBatchReader};
 use crate::sst::file::FileId;
@@ -92,7 +95,7 @@ impl RowGroupLastRowCachedReader {
         };
 
         let Some(cache_manager) = cache_manager else {
-            return Self::Miss(RowGroupLastRowReader::new(key, row_group_reader, None));
+            return Self::new_miss(key, row_group_reader, None);
         };
         if let Some(value) = cache_manager.get_selector_result(&key) {
             let schema_matches = value.projection
@@ -102,22 +105,34 @@ impl RowGroupLastRowCachedReader {
                     .projection_indices();
             if schema_matches {
                 // Schema matches, use cache batches.
-                Self::Hit(LastRowCacheReader { value, idx: 0 })
+                Self::new_hit(value)
             } else {
-                Self::Miss(RowGroupLastRowReader::new(
-                    key,
-                    row_group_reader,
-                    Some(cache_manager),
-                ))
+                Self::new_miss(key, row_group_reader, Some(cache_manager))
             }
         } else {
-            Self::Miss(RowGroupLastRowReader::new(
-                key,
-                row_group_reader,
-                Some(cache_manager),
-            ))
+            Self::new_miss(key, row_group_reader, Some(cache_manager))
         }
     }
+
+    /// Creates new Hit variant and updates metrics.
+    fn new_hit(value: Arc<SelectorResultValue>) -> Self {
+        selector_result_cache_hit();
+        Self::Hit(LastRowCacheReader { value, idx: 0 })
+    }
+
+    /// Creates new Miss variant and updates metrics.
+    fn new_miss(
+        key: SelectorResultKey,
+        row_group_reader: RowGroupReader,
+        cache_manager: Option<CacheManagerRef>,
+    ) -> Self {
+        selector_result_cache_miss();
+        Self::Miss(RowGroupLastRowReader::new(
+            key,
+            row_group_reader,
+            cache_manager,
+        ))
+    }
 }
 
 #[async_trait]