risingwave_meta/hummock/manager/
table_write_throughput_statistic.rs

1// Copyright 2024 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
15use std::collections::HashMap;
16use std::time::Duration;
17
18use risingwave_common::catalog::TableId;
19use tokio::time::Instant;
20
21// Coalesce bursts of commit completions instead of treating each completion as a sample.
22const MIN_SAMPLE_INTERVAL: Duration = Duration::from_secs(1);
23const BUCKET_COUNT: usize = 6;
24
25#[derive(Debug, Clone, Copy, Default)]
26struct ThroughputBucket {
27    bytes: f64,
28    seconds: f64,
29    peak: u64,
30    last_sample: Option<Instant>,
31}
32
33#[derive(Debug, Clone)]
34struct TableThroughput {
35    buckets: [ThroughputBucket; BUCKET_COUNT],
36    observed_since: Instant,
37    sample_start: Instant,
38    last_interval: Duration,
39    latest: Option<u64>,
40    pending_bytes: u64,
41}
42
43impl TableThroughput {
44    fn record(&mut self, bytes: u64, now: Instant, bucket_width: Duration) {
45        self.pending_bytes = self.pending_bytes.saturating_add(bytes);
46        let elapsed = now.duration_since(self.sample_start);
47        let index = (now.duration_since(self.observed_since).as_nanos() / bucket_width.as_nanos()
48            % BUCKET_COUNT as u128) as usize;
49        let bucket = &mut self.buckets[index];
50        // Only reset the slot being reused. Queries exclude stale slots by timestamp,
51        // so even a long gap needs neither a rotation loop nor synthetic zero samples.
52        if bucket
53            .last_sample
54            .is_some_and(|last| now.duration_since(last) >= bucket_width)
55        {
56            *bucket = ThroughputBucket::default();
57        }
58        let rate = self.pending_bytes as f64 / elapsed.max(MIN_SAMPLE_INTERVAL).as_secs_f64();
59        // Retain pending bounds too: they can already trigger split, even if a later
60        // completion averages the same bytes over a longer, colder interval.
61        // Round up so a fractional rate above a threshold is never classified as cold.
62        bucket.peak = bucket.peak.max(rate.ceil() as u64);
63        bucket.last_sample = Some(now);
64        if elapsed < MIN_SAMPLE_INTERVAL {
65            return;
66        }
67        bucket.bytes += self.pending_bytes as f64;
68        bucket.seconds += elapsed.as_secs_f64();
69        self.latest = Some(rate as u64);
70        self.pending_bytes = 0;
71        self.sample_start = now;
72        self.last_interval = elapsed;
73    }
74
75    fn window(&self, now: Instant, window: Duration) -> impl Iterator<Item = &ThroughputBucket> {
76        self.buckets.iter().filter(move |bucket| {
77            bucket
78                .last_sample
79                .is_some_and(|last| now.duration_since(last) <= window)
80        })
81    }
82}
83
84/// Successful Hummock ingress, independent of configured checkpoint periods.
85/// One six-bucket history serves all consumers. A bucket is one fifth of the retention
86/// window (at least one second). Whole boundary buckets extend history by at most one bucket.
87/// Completed rates represent whole observed commit intervals; pending ingress uses a
88/// conservative one-second bound. Long intervals cannot reveal bursts within them.
89/// Timer reads neither add zero observations nor advance the history.
90#[derive(Debug, Clone)]
91pub struct TableWriteThroughputStatisticManager {
92    tables: HashMap<TableId, TableThroughput>,
93    retention: Duration,
94    bucket_width: Duration,
95}
96
97impl TableWriteThroughputStatisticManager {
98    pub fn new(retention_secs: usize) -> Self {
99        let retention = Duration::from_secs(retention_secs.max(1) as u64);
100        Self {
101            tables: HashMap::new(),
102            retention,
103            bucket_width: (retention / (BUCKET_COUNT - 1) as u32).max(MIN_SAMPLE_INTERVAL),
104        }
105    }
106
107    pub fn record_commit(&mut self, table_id: TableId, bytes: u64, now: Instant) {
108        match self.tables.entry(table_id) {
109            std::collections::hash_map::Entry::Vacant(entry) => {
110                // The first commit has no known starting time. Establish a baseline rather
111                // than attributing its bytes to an invented checkpoint interval.
112                entry.insert(TableThroughput {
113                    buckets: [ThroughputBucket::default(); BUCKET_COUNT],
114                    observed_since: now,
115                    sample_start: now,
116                    last_interval: MIN_SAMPLE_INTERVAL,
117                    latest: None,
118                    pending_bytes: 0,
119                });
120            }
121            std::collections::hash_map::Entry::Occupied(mut entry) => {
122                entry.get_mut().record(bytes, now, self.bucket_width);
123            }
124        }
125    }
126
127    /// Historical peak after a full observed window, or None if history is missing/stale.
128    /// Anchor history at the last successful sample so slow, successful empty commits remain
129    /// useful between arrivals. Silence beyond the window or observed cadence is unknown.
130    pub fn max_write_throughput(&self, table_id: TableId, now: Instant) -> Option<u64> {
131        let table = self.tables.get(&table_id)?;
132        if table.sample_start.duration_since(table.observed_since) < self.retention
133            || now.duration_since(table.sample_start) > self.retention.max(table.last_interval)
134        {
135            return None;
136        }
137        let peak = table
138            .window(table.sample_start, self.retention)
139            .map(|bucket| bucket.peak)
140            .max()?;
141        // The one-second minimum sample interval makes pending bytes a rate bound.
142        // Include the same bound as latest_table_throughput so pending hot ingress
143        // cannot qualify as cold just because no later commit has closed its sample.
144        Some(peak.max(table.pending_bytes))
145    }
146
147    /// Latest completed rate, with a provisional one-second bound for pending ingress.
148    /// A burst of successful commits remains visible even without another completion to
149    /// close its sample. Reads never add zero observations or dilute a rate with silence.
150    pub fn latest_table_throughput(&self, table_id: TableId) -> Option<u64> {
151        let table = self.tables.get(&table_id)?;
152        if Instant::now().duration_since(table.sample_start)
153            > self.retention.max(table.last_interval)
154            || (table.latest.is_none() && table.pending_bytes == 0)
155        {
156            return None;
157        }
158        Some(table.latest.unwrap_or(0).max(table.pending_bytes))
159    }
160
161    /// Window-averaged ingress for metrics, using the same configured retention.
162    /// Scheduling uses `latest_table_throughput` and `max_write_throughput` instead.
163    /// Include whole boundary buckets and complete commit intervals; do not assume bytes
164    /// arrived uniformly within a slow commit interval or count unobserved silence as zero.
165    pub fn avg_write_throughput(&self, table_id: TableId) -> f64 {
166        let Some(table) = self.tables.get(&table_id) else {
167            return 0.0;
168        };
169        let now = Instant::now();
170        let (mut bytes, mut seconds) = table
171            .window(now, self.retention)
172            .fold((0.0, 0.0), |(bytes, seconds), bucket| {
173                (bytes + bucket.bytes, seconds + bucket.seconds)
174            });
175        let elapsed = now.duration_since(table.sample_start);
176        if table.pending_bytes > 0 && elapsed <= self.retention {
177            // Include bytes from completed commits even if their coalesced sample is still
178            // open. Reads do not create observations; the minimum interval bounds the rate.
179            bytes += table.pending_bytes as f64;
180            seconds += elapsed.max(MIN_SAMPLE_INTERVAL).as_secs_f64();
181        }
182        if seconds == 0.0 { 0.0 } else { bytes / seconds }
183    }
184
185    pub fn remove_table(&mut self, table_id: TableId) {
186        self.tables.remove(&table_id);
187    }
188}
189
190#[cfg(test)]
191mod tests {
192    use super::*;
193    use crate::hummock::test_utils::advance_time;
194
195    #[test]
196    fn test_peak_retention_at_bucket_boundaries() {
197        let default_retention = risingwave_common::config::meta::default::meta::table_write_throughput_retention_seconds();
198        assert_eq!(
199            TableWriteThroughputStatisticManager::new(default_retention).bucket_width,
200            Duration::from_secs(60)
201        );
202        let table = TableId::new(100);
203        let start = Instant::now();
204        for retention in [1_u64, 3, 10, 240, 300] {
205            let bucket_width = retention.div_ceil((BUCKET_COUNT - 1) as u64).max(1);
206            for offset in 1..=bucket_width {
207                let mut stats = TableWriteThroughputStatisticManager::new(retention as usize);
208                for second in 0..=retention + 2 * bucket_width + 1 {
209                    let now = start + Duration::from_secs(second);
210                    stats.record_commit(table, if second == offset { 32 } else { 0 }, now);
211                    let peak = stats.max_write_throughput(table, now);
212                    if second < retention {
213                        assert_eq!(peak, None);
214                    } else if second - offset < retention {
215                        assert_eq!(
216                            peak,
217                            Some(32),
218                            "retention={retention}, offset={offset}, second={second}"
219                        );
220                    } else if second - offset >= retention + bucket_width {
221                        assert_eq!(
222                            peak,
223                            Some(0),
224                            "retention={retention}, offset={offset}, second={second}"
225                        );
226                    }
227                }
228            }
229        }
230    }
231
232    #[tokio::test(start_paused = true)]
233    async fn test_window_rate_weights_elapsed_time() {
234        let table = TableId::new(100);
235        let mut stats = TableWriteThroughputStatisticManager::new(240);
236        stats.record_commit(table, 0, Instant::now());
237        advance_time(Duration::from_secs(1)).await;
238        stats.record_commit(table, 100, Instant::now());
239        advance_time(Duration::from_secs(99)).await;
240        stats.record_commit(table, 0, Instant::now());
241        // Two completions are not two equal votes: 100 bytes arrived over 100 seconds.
242        assert_eq!(stats.avg_write_throughput(table), 1.0);
243        // A burst completed within the minimum sample interval still contributes to
244        // the metric, even if no subsequent commit arrives to close that sample.
245        stats.record_commit(table, 900, Instant::now());
246        assert_eq!(stats.avg_write_throughput(table), 1000.0 / 101.0);
247        advance_time(Duration::from_secs(1)).await;
248        stats.record_commit(table, 0, Instant::now());
249        assert_eq!(stats.avg_write_throughput(table), 1000.0 / 101.0);
250    }
251
252    #[tokio::test(start_paused = true)]
253    async fn test_actual_intervals_and_bunched_commits() {
254        let table = TableId::new(100);
255        let mut stats = TableWriteThroughputStatisticManager::new(240);
256        stats.record_commit(table, 999, Instant::now());
257        assert_eq!(stats.latest_table_throughput(table), None);
258        // Pending ingress is visible even before the first full sample has completed.
259        stats.record_commit(table, 100, Instant::now());
260        assert_eq!(stats.latest_table_throughput(table), Some(100));
261        advance_time(Duration::from_secs(1)).await;
262        stats.record_commit(table, 0, Instant::now());
263        for seconds in [2, 10, 300, 1] {
264            advance_time(Duration::from_secs(seconds)).await;
265            stats.record_commit(table, 100 * seconds, Instant::now());
266            assert_eq!(stats.latest_table_throughput(table), Some(100));
267        }
268        // Repeated completions at the same instant retain their bytes without gaining votes.
269        for _ in 0..1000 {
270            stats.record_commit(table, 1, Instant::now());
271        }
272        assert_eq!(stats.latest_table_throughput(table), Some(1000));
273        assert_eq!(
274            stats.max_write_throughput(table, Instant::now()),
275            Some(1000)
276        );
277        advance_time(Duration::from_secs(1)).await;
278        stats.record_commit(table, 0, Instant::now());
279        assert_eq!(stats.latest_table_throughput(table), Some(1000));
280        for _ in 0..500 {
281            advance_time(Duration::from_secs(1)).await;
282            stats.record_commit(table, 0, Instant::now());
283        }
284
285        assert_eq!(stats.max_write_throughput(table, Instant::now()), Some(0));
286        stats.record_commit(table, 100, Instant::now());
287        advance_time(Duration::from_secs(10)).await;
288        stats.record_commit(table, 0, Instant::now());
289        assert_eq!(stats.latest_table_throughput(table), Some(10));
290        assert_eq!(
291            stats.max_write_throughput(table, Instant::now()),
292            Some(100),
293            "closing a sample must retain the pending rate that could already trigger split"
294        );
295        stats.record_commit(table, 100, Instant::now());
296        advance_time(Duration::from_secs(241)).await;
297        assert_eq!(stats.latest_table_throughput(table), None);
298        assert_eq!(stats.max_write_throughput(table, Instant::now()), None);
299    }
300
301    #[tokio::test(start_paused = true)]
302    async fn test_unknown_pause_resume_and_slow_idle() {
303        let table = TableId::new(100);
304        let mut stats = TableWriteThroughputStatisticManager::new(240);
305        assert_eq!(stats.max_write_throughput(table, Instant::now()), None);
306        stats.record_commit(table, 0, Instant::now());
307        advance_time(Duration::from_secs(300)).await;
308        assert_eq!(stats.max_write_throughput(table, Instant::now()), None);
309        stats.record_commit(table, 0, Instant::now());
310        assert_eq!(stats.max_write_throughput(table, Instant::now()), Some(0));
311        advance_time(Duration::from_secs(300)).await;
312        stats.record_commit(table, 0, Instant::now());
313        // Slow successful empty commits provide evidence. A timer between them must not
314        // require a shorter commit interval than the configured history window.
315        advance_time(Duration::from_secs(299)).await;
316        assert_eq!(stats.max_write_throughput(table, Instant::now()), Some(0));
317        assert_eq!(stats.latest_table_throughput(table), Some(0));
318        advance_time(Duration::from_secs(2)).await;
319        assert_eq!(stats.max_write_throughput(table, Instant::now()), None);
320        assert_eq!(stats.latest_table_throughput(table), None);
321        // A successful empty commit accounts for the elapsed interval. Subsequent backlog
322        // drain is real Hummock ingress and may legitimately turn the table hot again.
323        stats.record_commit(table, 0, Instant::now());
324        assert_eq!(stats.max_write_throughput(table, Instant::now()), Some(0));
325        advance_time(Duration::from_secs(1)).await;
326        stats.record_commit(table, 100, Instant::now());
327        assert_eq!(stats.latest_table_throughput(table), Some(100));
328        assert_eq!(stats.max_write_throughput(table, Instant::now()), Some(100));
329        stats.remove_table(table);
330        assert_eq!(stats.latest_table_throughput(table), None);
331    }
332}
risingwave_meta/hummock/manager/table_write_throughput_statistic.rs

risingwave_meta/hummock/manager/
table_write_throughput_statistic.rs
