risingwave_storage/hummock/event_handler/
refiller.rs

1// Copyright 2023 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, HashSet};
16use std::sync::Arc;
17use std::time::Duration;
18
19use risingwave_common::bitmap::Bitmap;
20use risingwave_common::config::streaming::CacheRefillPolicy;
21use risingwave_common::license::Feature;
22use risingwave_hummock_sdk::compaction_group::hummock_version_ext::SstDeltaInfo;
23use risingwave_pb::id::TableId;
24use thiserror_ext::AsReport;
25use tokio::sync::Semaphore;
26use tokio::task::JoinHandle;
27
28use crate::hummock::SstableStoreRef;
29use crate::hummock::table_cache_metadata::TableCacheMetadata;
30pub use crate::monitor::{CacheRefillMetrics, GLOBAL_CACHE_REFILL_METRICS};
31use crate::opts::StorageOpts;
32
33// Foyer tasks consume immutable snapshots of the handler-owned policy and vnode maps.
34mod foyer;
35
36#[derive(Debug)]
37pub struct CacheRefillConfig {
38    /// Cache refill timeout.
39    pub timeout: Duration,
40
41    /// Data file cache refill levels.
42    pub data_refill_levels: HashSet<u32>,
43
44    /// Meta file cache refill concurrency.
45    pub meta_refill_concurrency: usize,
46
47    /// Data file cache refill concurrency.
48    pub concurrency: usize,
49
50    /// Data file cache refill unit (blocks).
51    pub unit: usize,
52
53    /// Data file cache reill unit threshold.
54    ///
55    /// Only units whose admit rate > threshold will be refilled.
56    pub threshold: f64,
57
58    /// Skip recent filter.
59    pub skip_recent_filter: bool,
60
61    /// Skip inheritance filter.
62    pub skip_inheritance_filter: bool,
63}
64
65impl CacheRefillConfig {
66    pub fn from_storage_opts(options: &StorageOpts) -> Self {
67        let data_refill_levels = match Feature::ElasticDiskCache.check_available() {
68            Ok(_) => options
69                .cache_refill_data_refill_levels
70                .iter()
71                .copied()
72                .collect(),
73            Err(e) => {
74                tracing::warn!(error = %e.as_report(), "ElasticDiskCache is not available.");
75                HashSet::new()
76            }
77        };
78
79        Self {
80            timeout: Duration::from_millis(options.cache_refill_timeout_ms),
81            data_refill_levels,
82            concurrency: options.cache_refill_concurrency,
83            meta_refill_concurrency: options.cache_refill_meta_refill_concurrency,
84            unit: options.cache_refill_unit,
85            threshold: options.cache_refill_threshold,
86            skip_recent_filter: options.cache_refill_skip_recent_filter,
87            skip_inheritance_filter: options.cache_refill_skip_inheritance_filter,
88        }
89    }
90}
91
92#[cfg(test)]
93type SpawnRefillTask =
94    Arc<dyn Fn(Vec<SstDeltaInfo>, CacheRefillContext) -> JoinHandle<()> + Send + Sync + 'static>;
95
96pub type TableCacheRefillContextMap = HashMap<TableId, TableCacheRefillContext>;
97
98/// Per-table metadata captured for a refill task to decide whether an sstable block should be
99/// refilled. Mutable runtime state used to build this snapshot is owned by the event handler.
100#[derive(Clone)]
101pub struct TableCacheRefillContext {
102    /// Vnodes covered by local streaming read versions on this compute node.
103    pub streaming_vnode_bitmap: Option<Bitmap>,
104    /// Vnodes served by this compute node according to the serving vnode mapping.
105    pub serving_vnode_bitmap: Option<Bitmap>,
106    /// Effective refill policy after applying the default policy and per-table overrides.
107    pub policy: CacheRefillPolicy,
108}
109
110/// Current CN metadata and its Foyer projection for monitor/debugging APIs.
111#[derive(Clone)]
112pub struct TableCacheRefillMonitorSnapshot {
113    pub contexts: TableCacheRefillContextMap,
114    pub policies: HashMap<TableId, CacheRefillPolicy>,
115    pub default_policy: CacheRefillPolicy,
116    pub streaming_table_vnode_mapping: HashMap<TableId, Bitmap>,
117    pub serving_table_vnode_mapping: HashMap<TableId, Bitmap>,
118}
119
120/// Admits SSTs using borrowed CN metadata, then runs Foyer tasks with immutable snapshots.
121pub(crate) struct CacheRefiller {
122    #[cfg(test)]
123    spawn_refill_task: Option<SpawnRefillTask>,
124
125    config: Arc<CacheRefillConfig>,
126    meta_refill_concurrency: Option<Arc<Semaphore>>,
127    concurrency: Arc<Semaphore>,
128    sstable_store: SstableStoreRef,
129}
130
131impl CacheRefiller {
132    pub(crate) fn new(config: CacheRefillConfig, sstable_store: SstableStoreRef) -> Self {
133        let config = Arc::new(config);
134        let concurrency = Arc::new(Semaphore::new(config.concurrency));
135        let meta_refill_concurrency = if config.meta_refill_concurrency == 0 {
136            None
137        } else {
138            Some(Arc::new(Semaphore::new(config.meta_refill_concurrency)))
139        };
140        Self {
141            #[cfg(test)]
142            spawn_refill_task: None,
143            config,
144            meta_refill_concurrency,
145            concurrency,
146            sstable_store,
147        }
148    }
149
150    #[cfg(test)]
151    pub(super) fn with_foyer_spawn_task(mut self, spawn: SpawnRefillTask) -> Self {
152        self.spawn_refill_task = Some(spawn);
153        self
154    }
155
156    pub(crate) fn start_cache_refill(
157        &self,
158        mut deltas: Vec<SstDeltaInfo>,
159        metadata: &TableCacheMetadata,
160    ) -> JoinHandle<()> {
161        for delta in &mut deltas {
162            let for_serving = metadata.role.for_serving();
163            // Writer-appended L0 SSTs are already warm on the streaming side. Their data refill
164            // may therefore only be needed by serving workers.
165            let for_streaming =
166                metadata.role.for_streaming() && !delta.delete_sst_object_ids.is_empty();
167
168            if !for_serving && !for_streaming {
169                delta.insert_sst_infos.clear();
170                continue;
171            }
172
173            // This is deliberately a whole-SST admission check before Meta load. The serving
174            // mapping key set identifies result tables, but bitmap contents remain for the exact
175            // block/vnode decision in DataCacheRefillTaskGenerator after Meta load. An SST stays
176            // when any contained table matches either the serving or streaming refill lane.
177            delta.insert_sst_infos.retain(|sst| {
178                sst.table_ids.iter().any(|table_id| {
179                    // A missing entry means there is no table override, not that the table is
180                    // absent. Preserve the configured legacy/default policy in that case.
181                    let policy = metadata
182                        .table_cache_refill_policies
183                        .get(table_id)
184                        .copied()
185                        .unwrap_or(metadata.default_policy);
186
187                    // Enabled preserves legacy full refill. For scoped policies, mapping keys are
188                    // only a whole-SST coarse gate; bitmap bits still filter blocks post-Meta.
189                    match policy {
190                        CacheRefillPolicy::Enabled => for_streaming || for_serving,
191                        CacheRefillPolicy::Disabled => false,
192                        CacheRefillPolicy::Streaming => {
193                            for_streaming
194                                && metadata
195                                    .streaming_table_vnode_mapping
196                                    .contains_key(table_id)
197                        }
198                        CacheRefillPolicy::Serving => {
199                            for_serving
200                                && metadata.serving_table_vnode_mapping.contains_key(table_id)
201                        }
202                        CacheRefillPolicy::Both => {
203                            (for_streaming
204                                && metadata
205                                    .streaming_table_vnode_mapping
206                                    .contains_key(table_id))
207                                || (for_serving
208                                    && metadata.serving_table_vnode_mapping.contains_key(table_id))
209                        }
210                    }
211                })
212            });
213        }
214        let context = self.new_cache_refill_context(&deltas, metadata);
215        #[cfg(test)]
216        if let Some(spawn) = &self.spawn_refill_task {
217            return spawn(deltas, context);
218        }
219        foyer::spawn_refill_task(deltas, context)
220    }
221
222    fn new_cache_refill_context(
223        &self,
224        deltas: &[SstDeltaInfo],
225        metadata: &TableCacheMetadata,
226    ) -> CacheRefillContext {
227        let table_ids = deltas.iter().flat_map(|delta| {
228            delta
229                .insert_sst_infos
230                .iter()
231                .flat_map(|sst| sst.table_ids.iter().copied())
232        });
233        CacheRefillContext {
234            config: self.config.clone(),
235            meta_refill_concurrency: self.meta_refill_concurrency.clone(),
236            concurrency: self.concurrency.clone(),
237            sstable_store: self.sstable_store.clone(),
238            table_cache_refill_context_map: Arc::new(Self::table_cache_refill_contexts(
239                metadata, table_ids,
240            )),
241        }
242    }
243
244    fn table_cache_refill_contexts(
245        metadata: &TableCacheMetadata,
246        table_ids: impl IntoIterator<Item = TableId>,
247    ) -> TableCacheRefillContextMap {
248        let for_streaming = metadata.role.for_streaming();
249        let for_serving = metadata.role.for_serving();
250        let mut contexts = HashMap::new();
251        for table_id in table_ids {
252            if for_serving
253                && !for_streaming
254                && !metadata.serving_table_vnode_mapping.contains_key(&table_id)
255            {
256                continue;
257            }
258            contexts.entry(table_id).or_insert_with(|| {
259                let policy = metadata
260                    .table_cache_refill_policies
261                    .get(&table_id)
262                    .copied()
263                    .unwrap_or(metadata.default_policy);
264                let streaming_vnode_bitmap = (for_streaming && policy.is_streaming_scoped())
265                    .then(|| {
266                        metadata
267                            .streaming_table_vnode_mapping
268                            .get(&table_id)
269                            .cloned()
270                    })
271                    .flatten();
272                // `Enabled` normally does not use bitmap filtering. The only exception is L0
273                // insert-only refill, where serving workers still need serving-locality evidence.
274                let serving_vnode_bitmap = (for_serving
275                    && (policy.is_serving_scoped() || policy.is_unscoped_enabled()))
276                .then(|| metadata.serving_table_vnode_mapping.get(&table_id).cloned())
277                .flatten();
278                TableCacheRefillContext {
279                    streaming_vnode_bitmap,
280                    serving_vnode_bitmap,
281                    policy,
282                }
283            });
284        }
285        contexts
286    }
287
288    pub(crate) fn table_cache_refill_monitor_snapshot(
289        metadata: &TableCacheMetadata,
290    ) -> TableCacheRefillMonitorSnapshot {
291        let table_ids = metadata
292            .table_cache_refill_policies
293            .keys()
294            .chain(metadata.streaming_table_vnode_mapping.keys())
295            .chain(metadata.serving_table_vnode_mapping.keys())
296            .copied();
297        TableCacheRefillMonitorSnapshot {
298            contexts: Self::table_cache_refill_contexts(metadata, table_ids),
299            policies: metadata.table_cache_refill_policies.clone(),
300            default_policy: metadata.default_policy,
301            streaming_table_vnode_mapping: metadata.streaming_table_vnode_mapping.clone(),
302            serving_table_vnode_mapping: metadata.serving_table_vnode_mapping.clone(),
303        }
304    }
305}
306
307#[derive(Clone)]
308pub(crate) struct CacheRefillContext {
309    config: Arc<CacheRefillConfig>,
310    meta_refill_concurrency: Option<Arc<Semaphore>>,
311    concurrency: Arc<Semaphore>,
312    sstable_store: SstableStoreRef,
313    table_cache_refill_context_map: Arc<TableCacheRefillContextMap>,
314}
risingwave_storage/hummock/event_handler/refiller.rs

risingwave_storage/hummock/event_handler/
refiller.rs
