risingwave_frontend/handler/
alter_streaming_config.rs

1// Copyright 2025 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;
16
17use anyhow::Context;
18use pgwire::pg_response::StatementType;
19use risingwave_sqlparser::ast::{ObjectName, SqlOption, SqlOptionValue, Value as AstValue};
20use toml::Value as TomlValue;
21use toml::map::Map as TomlMap;
22
23use crate::error::{Result, bail_invalid_input_syntax};
24use crate::handler::alter_utils::resolve_streaming_job_id_for_alter;
25use crate::handler::{HandlerArgs, RwPgResponse};
26
27/// A diff of a TOML map. `None` means the key should be removed.
28type TomlMapDiff = TomlMap<String, Option<TomlValue>>;
29
30const STREAMING_CACHE_REFILL_POLICY_CONFIG_PATH: &str = "streaming.developer.cache_refill_policy";
31const ALTER_CONFIG_RECOVER_NOTICE: &str =
32    "ALTER CONFIG requires a RECOVER on the specified streaming job to take effect.";
33
34fn alter_config_requires_recover(map_diff: &TomlMapDiff) -> bool {
35    map_diff
36        .keys()
37        .any(|key| key != STREAMING_CACHE_REFILL_POLICY_CONFIG_PATH)
38}
39
40fn collect_options(entries: Vec<SqlOption>) -> Result<TomlMapDiff> {
41    let mut map = TomlMap::new();
42
43    for SqlOption { name, value } in entries {
44        let name = name.real_value();
45        if !name.starts_with("streaming.") {
46            bail_invalid_input_syntax!(
47                "ALTER CONFIG only accepts options starting with `streaming.`"
48            );
49        }
50        let SqlOptionValue::Value(value) = value else {
51            bail_invalid_input_syntax!("ALTER CONFIG only accepts value options");
52        };
53
54        let value = match value {
55            AstValue::Number(n) => {
56                let n: TomlValue = n.parse().context("Invalid number for ALTER CONFIG")?;
57                Some(n)
58            }
59            AstValue::SingleQuotedString(s) | AstValue::DoubleQuotedString(s) => {
60                Some(TomlValue::String(s))
61            }
62            AstValue::Boolean(b) => Some(TomlValue::Boolean(b)),
63            AstValue::Null => None,
64            _ => bail_invalid_input_syntax!("Unsupported value for ALTER CONFIG: {}", value),
65        };
66
67        let old = map.insert(name.clone(), value);
68        if old.is_some() {
69            bail_invalid_input_syntax!("Duplicate option for ALTER CONFIG: {}", name);
70        }
71    }
72
73    Ok(map)
74}
75
76pub async fn handle_alter_streaming_set_config(
77    handler_args: HandlerArgs,
78    obj_name: ObjectName,
79    entries: Vec<SqlOption>,
80    stmt_type: StatementType,
81) -> Result<RwPgResponse> {
82    let session = handler_args.session;
83
84    let job_id = resolve_streaming_job_id_for_alter(&session, obj_name, stmt_type, "config")?;
85    let map_diff = collect_options(entries)?;
86    let requires_recover = alter_config_requires_recover(&map_diff);
87
88    let mut entries_to_add = HashMap::new();
89    let mut keys_to_remove = Vec::new();
90
91    for (k, v) in map_diff {
92        if let Some(v) = v {
93            entries_to_add.insert(k, v.to_string());
94        } else {
95            keys_to_remove.push(k);
96        }
97    }
98
99    let catalog_writer = session.catalog_writer()?;
100    catalog_writer
101        .alter_config(job_id, entries_to_add, keys_to_remove)
102        .await?;
103
104    let mut builder = RwPgResponse::builder(stmt_type);
105    if requires_recover {
106        builder = builder.notice(ALTER_CONFIG_RECOVER_NOTICE);
107    }
108    Ok(builder.into())
109}
110
111pub async fn handle_alter_streaming_reset_config(
112    handler_args: HandlerArgs,
113    obj_name: ObjectName,
114    keys: Vec<ObjectName>,
115    stmt_type: StatementType,
116) -> Result<RwPgResponse> {
117    let entries = keys
118        .into_iter()
119        .map(|k| SqlOption {
120            name: k,
121            value: SqlOptionValue::null(),
122        })
123        .collect();
124
125    // Simply delegate to `handle_alter_streaming_set_config` with all values set to `NULL`.
126    handle_alter_streaming_set_config(handler_args, obj_name, entries, stmt_type).await
127}
128
129#[cfg(test)]
130mod tests {
131    use toml::Value as TomlValue;
132
133    use super::{
134        STREAMING_CACHE_REFILL_POLICY_CONFIG_PATH, TomlMapDiff, alter_config_requires_recover,
135    };
136
137    #[test]
138    fn test_cache_refill_policy_config_does_not_require_recover() {
139        let mut map_diff = TomlMapDiff::new();
140        map_diff.insert(
141            STREAMING_CACHE_REFILL_POLICY_CONFIG_PATH.to_owned(),
142            Some(TomlValue::String("both".to_owned())),
143        );
144        assert!(!alter_config_requires_recover(&map_diff));
145
146        map_diff.insert(STREAMING_CACHE_REFILL_POLICY_CONFIG_PATH.to_owned(), None);
147        assert!(!alter_config_requires_recover(&map_diff));
148    }
149
150    #[test]
151    fn test_other_streaming_config_still_requires_recover() {
152        let mut map_diff = TomlMapDiff::new();
153        map_diff.insert(
154            "streaming.developer.some_other_config".to_owned(),
155            Some(TomlValue::Boolean(true)),
156        );
157        assert!(alter_config_requires_recover(&map_diff));
158    }
159
160    #[test]
161    fn test_mixed_streaming_config_requires_recover() {
162        let mut map_diff = TomlMapDiff::new();
163        map_diff.insert(
164            STREAMING_CACHE_REFILL_POLICY_CONFIG_PATH.to_owned(),
165            Some(TomlValue::String("both".to_owned())),
166        );
167        map_diff.insert(
168            "streaming.developer.some_other_config".to_owned(),
169            Some(TomlValue::Boolean(true)),
170        );
171        assert!(alter_config_requires_recover(&map_diff));
172    }
173}
risingwave_frontend/handler/alter_streaming_config.rs

risingwave_frontend/handler/
alter_streaming_config.rs
