remove option

Author: yuhao-su

src/stream/src/cache/managed_lru.rs

@@ -39,24 +39,22 @@ pub struct ManagedLruCache<K, V, S = DefaultHasher, A: Clone + Allocator = Globa
     /// The heap size of keys/values
     kv_heap_size: usize,
     /// The metrics of memory usage
-    memory_usage_metrics: Option<IntGauge>,
+    memory_usage_metrics: IntGauge,
     // Metrics info
-    metrics_info: Option<MetricsInfo>,
+    metrics_info: MetricsInfo,
     /// The size reported last time
     last_reported_size_bytes: usize,
 }
 
 impl<K, V, S, A: Clone + Allocator> Drop for ManagedLruCache<K, V, S, A> {
     fn drop(&mut self) {
-        if let Some(metrics) = &self.memory_usage_metrics {
-            metrics.set(0.into());
-        }
-        if let Some(info) = &self.metrics_info {
-            info.metrics
-                .stream_memory_usage
-                .remove_label_values(&[&info.table_id, &info.actor_id, &info.desc])
-                .unwrap();
-        }
+        let info = &self.metrics_info;
+        self.memory_usage_metrics.set(0.into());
+
+        info.metrics
+            .stream_memory_usage
+            .remove_label_values(&[&info.table_id, &info.actor_id, &info.desc])
+            .unwrap();
     }
 }
 
@@ -66,15 +64,16 @@ impl<K: Hash + Eq + EstimateSize, V: EstimateSize, S: BuildHasher, A: Clone + Al
     pub fn new_inner(
         inner: LruCache<K, V, S, A>,
         watermark_epoch: Arc<AtomicU64>,
-        metrics_info: Option<MetricsInfo>,
+        metrics_info: MetricsInfo,
     ) -> Self {
-        let memory_usage_metrics = metrics_info.as_ref().map(|info| {
-            info.metrics.stream_memory_usage.with_label_values(&[
-                &info.table_id,
-                &info.actor_id,
-                &info.desc,
-            ])
-        });
+        let memory_usage_metrics = metrics_info
+            .metrics
+            .stream_memory_usage
+            .with_label_values(&[
+                &metrics_info.table_id,
+                &metrics_info.actor_id,
+                &metrics_info.desc,
+            ]);
 
         Self {
             inner,
@@ -219,9 +218,7 @@ impl<K: Hash + Eq + EstimateSize, V: EstimateSize, S: BuildHasher, A: Clone + Al
         if self.kv_heap_size.abs_diff(self.last_reported_size_bytes)
             > REPORT_SIZE_EVERY_N_KB_CHANGE << 10
         {
-            if let Some(metrics) = self.memory_usage_metrics.as_ref() {
-                metrics.set(self.kv_heap_size as _);
-            }
+            self.memory_usage_metrics.set(self.kv_heap_size as _);
             self.last_reported_size_bytes = self.kv_heap_size;
             true
         } else {
@@ -234,7 +231,7 @@ pub fn new_unbounded<K: Hash + Eq + EstimateSize, V: EstimateSize>(
     watermark_epoch: Arc<AtomicU64>,
     metrics_info: MetricsInfo,
 ) -> ManagedLruCache<K, V> {
-    ManagedLruCache::new_inner(LruCache::unbounded(), watermark_epoch, Some(metrics_info))
+    ManagedLruCache::new_inner(LruCache::unbounded(), watermark_epoch, metrics_info)
 }
 
 pub fn new_with_hasher_in<
@@ -251,7 +248,7 @@ pub fn new_with_hasher_in<
     ManagedLruCache::new_inner(
         LruCache::unbounded_with_hasher_in(hasher, alloc),
         watermark_epoch,
-        Some(metrics_info),
+        metrics_info,
     )
 }
 
@@ -263,7 +260,7 @@ pub fn new_with_hasher<K: Hash + Eq + EstimateSize, V: EstimateSize, S: BuildHas
     ManagedLruCache::new_inner(
         LruCache::unbounded_with_hasher(hasher),
         watermark_epoch,
-        Some(metrics_info),
+        metrics_info,
     )
 }
 
@@ -274,15 +271,15 @@ pub struct MutGuard<'a, V: EstimateSize> {
     // The total size of a collection
     total_size: &'a mut usize,
     last_reported_size_bytes: &'a mut usize,
-    memory_usage_metrics: &'a mut Option<IntGauge>,
+    memory_usage_metrics: &'a mut IntGauge,
 }
 
 impl<'a, V: EstimateSize> MutGuard<'a, V> {
     pub fn new(
         inner: &'a mut V,
         total_size: &'a mut usize,
         last_reported_size_bytes: &'a mut usize,
-        memory_usage_metrics: &'a mut Option<IntGauge>,
+        memory_usage_metrics: &'a mut IntGauge,
     ) -> Self {
         let original_val_size = inner.estimated_size();
         Self {
@@ -298,9 +295,7 @@ impl<'a, V: EstimateSize> MutGuard<'a, V> {
         if self.total_size.abs_diff(*self.last_reported_size_bytes)
             > REPORT_SIZE_EVERY_N_KB_CHANGE << 10
         {
-            if let Some(metrics) = self.memory_usage_metrics.as_ref() {
-                metrics.set(*self.total_size as _);
-            }
+            self.memory_usage_metrics.set(*self.total_size as _);
             *self.last_reported_size_bytes = *self.total_size;
             true
         } else {
@@ -340,15 +335,15 @@ pub struct UnsafeMutGuard<V: EstimateSize> {
     // The total size of a collection
     total_size: NonNull<usize>,
     last_reported_size_bytes: NonNull<usize>,
-    memory_usage_metrics: NonNull<Option<IntGauge>>,
+    memory_usage_metrics: NonNull<IntGauge>,
 }
 
 impl<V: EstimateSize> UnsafeMutGuard<V> {
     pub fn new(
         inner: &mut V,
         total_size: &mut usize,
         last_reported_size_bytes: &mut usize,
-        memory_usage_metrics: &mut Option<IntGauge>,
+        memory_usage_metrics: &mut IntGauge,
     ) -> Self {
         let original_val_size = inner.estimated_size();
         Self {