risingwave_storage/hummock/store/
table_change_log_manager.rs

1// Copyright 2026 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::future::Future;
16use std::iter;
17use std::pin::Pin;
18use std::sync::Arc;
19
20use futures::FutureExt;
21use futures::future::Shared;
22use moka::sync::Cache;
23use risingwave_common::id::TableId;
24use risingwave_hummock_sdk::change_log::TableChangeLogs;
25use risingwave_rpc_client::HummockMetaClient;
26
27use crate::hummock::{HummockError, HummockResult};
28use crate::monitor::HummockStateStoreMetrics;
29
30type InflightResult = Shared<Pin<Box<dyn Future<Output = HummockResult<TableChangeLogs>> + Send>>>;
31
32#[derive(Eq, Hash, PartialEq)]
33struct CacheKey {
34    table_id: TableId,
35    epoch_range: (u64, u64),
36    include_epoch_only: bool,
37    limit: Option<u32>,
38}
39
40/// A naive cache to reduce number of RPC sent to meta node.
41pub struct TableChangeLogManager {
42    cache: Cache<CacheKey, InflightResult>,
43    hummock_meta_client: Arc<dyn HummockMetaClient>,
44    metrics: Arc<HummockStateStoreMetrics>,
45}
46
47impl TableChangeLogManager {
48    pub fn new(
49        capacity: u64,
50        hummock_meta_client: Arc<dyn HummockMetaClient>,
51        metrics: Arc<HummockStateStoreMetrics>,
52    ) -> Self {
53        let cache = Cache::builder().max_capacity(capacity).build();
54        Self {
55            cache,
56            hummock_meta_client,
57            metrics,
58        }
59    }
60
61    async fn get_or_insert(
62        &self,
63        table_id: TableId,
64        epoch_range: (u64, u64),
65        include_epoch_only: bool,
66        limit: Option<u32>,
67        fetch: impl Future<Output = HummockResult<TableChangeLogs>> + Send + 'static,
68    ) -> HummockResult<TableChangeLogs> {
69        let entry = self
70            .cache
71            .entry(CacheKey {
72                table_id,
73                epoch_range,
74                include_epoch_only,
75                limit,
76            })
77            .or_insert_with_if(
78                || fetch.boxed().shared(),
79                |inflight| {
80                    if let Some(result) = inflight.peek() {
81                        return result.is_err();
82                    }
83                    false
84                },
85            );
86        if entry.is_fresh() {
87            self.metrics.table_change_log_cache_miss.inc();
88        } else {
89            self.metrics.table_change_log_cache_hit.inc();
90        }
91        entry.value().clone().await
92    }
93
94    /// Fetches table change logs for the given `table_id` and `epoch_range`.
95    ///
96    /// - If the end value of `epoch_range` is not `u64::MAX`, attempts to retrieve logs from the cache; if not cached, fetches via an RPC to the meta node and stores the result in the cache.
97    /// - If the end value of `epoch_range` is `u64::MAX`, always fetches table change logs directly from the meta node (bypassing the cache).
98    ///
99    /// Both the start and end values of `epoch_range` are inclusive.
100    ///
101    /// IMPORTANT: The caller must guarantee that the current max committed epoch is at least as large as the end of the provided `epoch_range`, if it's not `u64::MAX`.
102    /// Otherwise, the cache may serve outdated results: as new epochs are committed beyond the current maximum, subsequent RPC calls for the same `epoch_range`
103    /// could retrieve different or additional change logs that were not present in the previously cached result. For example, if you request logs for
104    /// `epoch_range = (1, N)` when the current max committed epoch is `M < N`, committing epochs `M+1` through `N` would make the cache inconsistent with reality;
105    /// future fetches for `(1, N)` could return new or updated information absent from the previous cache entry.
106    pub async fn fetch_table_change_logs(
107        &self,
108        table_id: TableId,
109        epoch_range: (u64, u64),
110        include_epoch_only: bool,
111        limit: Option<u32>,
112    ) -> HummockResult<TableChangeLogs> {
113        let _timer = self.metrics.table_change_log_fetch_latency.start_timer();
114        let hummock_meta_client = self.hummock_meta_client.clone();
115        let fetch = async move {
116            hummock_meta_client
117                .get_table_change_logs(
118                    include_epoch_only,
119                    Some(epoch_range.0),
120                    Some(epoch_range.1),
121                    Some(iter::once(table_id).collect()),
122                    false,
123                    limit,
124                )
125                .await
126                .map_err(HummockError::meta_error)
127        };
128        if epoch_range.1 == u64::MAX {
129            return fetch.await;
130        }
131        self.get_or_insert(table_id, epoch_range, include_epoch_only, limit, fetch)
132            .await
133    }
134}
risingwave_storage/hummock/store/table_change_log_manager.rs

risingwave_storage/hummock/store/
table_change_log_manager.rs
