risingwave_storage/hummock/shared_buffer/
mod.rs

1// Copyright 2022 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15pub(crate) const TEST_TABLE_ID: TableId = TableId::new(233);
16
17use std::fmt::Debug;
18use std::sync::Arc;
19
20use risingwave_common::metrics::{
21    LabelGuardedHistogram, LabelGuardedIntCounter, LabelGuardedIntGauge,
22    LazyLabelGuardedIntCounter, LazyLabelGuardedIntGauge, UintGauge,
23};
24use risingwave_pb::id::{FragmentId, TableId};
25
26use crate::monitor::HummockStateStoreMetrics;
27
28pub mod shared_buffer_batch;
29
30pub(crate) struct TableMemoryMetrics {
31    imm_total_size: LabelGuardedIntGauge,
32    imm_count: LabelGuardedIntGauge,
33    replicated_imm_size: Option<UintGauge>,
34    pub write_batch_tuple_counts: LabelGuardedIntCounter,
35    pub write_batch_duration: LabelGuardedHistogram,
36    pub write_batch_size: LabelGuardedHistogram,
37    pub mem_table_spill_counts: LazyLabelGuardedIntCounter,
38    pub old_value_size: LazyLabelGuardedIntGauge,
39}
40
41impl TableMemoryMetrics {
42    pub(super) fn new(
43        metrics: &HummockStateStoreMetrics,
44        table_id: TableId,
45        fragment_id: FragmentId,
46        is_replicated: bool,
47    ) -> Self {
48        let table_id_string = if is_replicated {
49            format!("{} replicated", table_id)
50        } else {
51            table_id.to_string()
52        };
53        let fragment_labels_vec = vec![table_id_string.clone(), fragment_id.to_string()];
54        let fragment_labels = fragment_labels_vec.as_slice();
55        let table_labels_vec = vec![table_id_string];
56        let table_labels = table_labels_vec.as_slice();
57        Self {
58            imm_total_size: metrics
59                .per_table_imm_size
60                .with_guarded_label_values(fragment_labels),
61            imm_count: metrics
62                .per_table_imm_count
63                .with_guarded_label_values(table_labels),
64            replicated_imm_size: is_replicated.then(|| metrics.replicated_imm_size.clone()),
65            write_batch_tuple_counts: metrics
66                .write_batch_tuple_counts
67                .with_guarded_label_values(table_labels),
68            write_batch_duration: metrics
69                .write_batch_duration
70                .with_guarded_label_values(table_labels),
71            write_batch_size: metrics
72                .write_batch_size
73                .with_guarded_label_values(table_labels),
74            mem_table_spill_counts: metrics
75                .mem_table_spill_counts
76                .lazy_guarded_metrics(table_labels_vec.clone()),
77            old_value_size: metrics
78                .old_value_size
79                .lazy_guarded_metrics(table_labels_vec),
80        }
81    }
82
83    pub(super) fn for_test() -> Arc<Self> {
84        Self::new(
85            &HummockStateStoreMetrics::unused(),
86            TEST_TABLE_ID,
87            FragmentId::default(),
88            false,
89        )
90        .into()
91    }
92
93    pub(super) fn inc_imm(&self, imm_size: usize) {
94        self.imm_total_size.add(imm_size as _);
95        self.imm_count.inc();
96        if let Some(replicated_imm_size) = &self.replicated_imm_size {
97            replicated_imm_size.add(imm_size as _);
98        }
99    }
100
101    pub(super) fn dec_imm(&self, imm_size: usize) {
102        self.imm_total_size.sub(imm_size as _);
103        self.imm_count.dec();
104        if let Some(replicated_imm_size) = &self.replicated_imm_size {
105            replicated_imm_size.sub(imm_size as _);
106        }
107    }
108}
109
110impl Debug for TableMemoryMetrics {
111    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
112        f.debug_struct("TableMemoryMetrics").finish()
113    }
114}
risingwave_storage/hummock/shared_buffer/mod.rs

risingwave_storage/hummock/shared_buffer/
mod.rs
