risingwave_common/config/streaming/
cache_refill.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::str::FromStr;
16
17use parse_display::Display;
18use risingwave_pb::meta::table_cache_refill_policies::table_cache_refill_policy::PbCacheRefillPolicy;
19use serde_with::{DeserializeFromStr, SerializeDisplay};
20
21#[derive(Copy, Debug, Clone, PartialEq, Eq, Display, SerializeDisplay, DeserializeFromStr)]
22#[display(style = "snake_case")]
23pub enum CacheRefillPolicy {
24    /// Enable normal cache refill for the table.
25    Enabled,
26    /// Disable cache refill for the table.
27    Disabled,
28    /// Enable cache refill optimized for streaming workloads for this table.
29    Streaming,
30    /// Enable cache refill optimized for serving workloads for this table.
31    Serving,
32    /// Enable cache refill optimized for both streaming and serving workloads for this table.
33    Both,
34}
35
36impl FromStr for CacheRefillPolicy {
37    type Err = &'static str;
38
39    fn from_str(s: &str) -> Result<Self, Self::Err> {
40        let s = s.to_ascii_lowercase().replace('-', "_");
41        match s.as_str() {
42            "enabled" => Ok(Self::Enabled),
43            "disabled" => Ok(Self::Disabled),
44            "streaming" => Ok(Self::Streaming),
45            "serving" => Ok(Self::Serving),
46            "both" => Ok(Self::Both),
47            _ => Err("expect one of [enabled, disabled, streaming, serving, both]"),
48        }
49    }
50}
51
52impl CacheRefillPolicy {
53    pub fn to_protobuf(self) -> PbCacheRefillPolicy {
54        match self {
55            Self::Enabled => PbCacheRefillPolicy::Enabled,
56            Self::Disabled => PbCacheRefillPolicy::Disabled,
57            Self::Streaming => PbCacheRefillPolicy::Streaming,
58            Self::Serving => PbCacheRefillPolicy::Serving,
59            Self::Both => PbCacheRefillPolicy::Both,
60        }
61    }
62
63    pub fn from_protobuf(pb: PbCacheRefillPolicy) -> Option<Self> {
64        match pb {
65            PbCacheRefillPolicy::Unspecified => None,
66            PbCacheRefillPolicy::Enabled => Some(Self::Enabled),
67            PbCacheRefillPolicy::Disabled => Some(Self::Disabled),
68            PbCacheRefillPolicy::Streaming => Some(Self::Streaming),
69            PbCacheRefillPolicy::Serving => Some(Self::Serving),
70            PbCacheRefillPolicy::Both => Some(Self::Both),
71        }
72    }
73
74    /// Legacy/default table-level allow policy. Normal refill is not scoped by
75    /// streaming or serving vnode ownership under this policy.
76    pub fn is_unscoped_enabled(self) -> bool {
77        matches!(self, Self::Enabled)
78    }
79
80    pub fn is_streaming_scoped(self) -> bool {
81        matches!(self, Self::Streaming | Self::Both)
82    }
83
84    pub fn is_serving_scoped(self) -> bool {
85        matches!(self, Self::Serving | Self::Both)
86    }
87}
88
89#[cfg(test)]
90mod tests {
91    use super::*;
92
93    #[test]
94    fn test_cache_refill_policy_contract() {
95        for (policy, name, unscoped, streaming, serving) in [
96            (CacheRefillPolicy::Enabled, "enabled", true, false, false),
97            (CacheRefillPolicy::Disabled, "disabled", false, false, false),
98            (
99                CacheRefillPolicy::Streaming,
100                "streaming",
101                false,
102                true,
103                false,
104            ),
105            (CacheRefillPolicy::Serving, "serving", false, false, true),
106            (CacheRefillPolicy::Both, "both", false, true, true),
107        ] {
108            assert_eq!(policy.to_string(), name);
109            assert_eq!(name.parse(), Ok(policy));
110            assert_eq!(name.to_ascii_uppercase().parse(), Ok(policy));
111            assert_eq!(
112                CacheRefillPolicy::from_protobuf(policy.to_protobuf()),
113                Some(policy)
114            );
115            assert_eq!(policy.is_unscoped_enabled(), unscoped);
116            assert_eq!(policy.is_streaming_scoped(), streaming);
117            assert_eq!(policy.is_serving_scoped(), serving);
118        }
119
120        assert_eq!(
121            CacheRefillPolicy::from_protobuf(PbCacheRefillPolicy::Unspecified),
122            None
123        );
124        assert!("unknown".parse::<CacheRefillPolicy>().is_err());
125    }
126}
risingwave_common/config/streaming/cache_refill.rs

risingwave_common/config/streaming/
cache_refill.rs
