risingwave_frontend/handler/
alter_system.rs1use pgwire::pg_response::StatementType;
16use risingwave_common::session_config::SessionConfig;
17use risingwave_common::system_param::reader::SystemParamsRead;
18use risingwave_sqlparser::ast::{Ident, SetVariableValue};
19
20use super::variable::set_var_to_param_str;
21use super::{HandlerArgs, RwPgResponse};
22use crate::error::{ErrorCode, Result};
23
24const NOTICE_BARRIER_INTERVAL_MS: u32 = 300000;
26const NOTICE_CHECKPOINT_FREQUENCY: u64 = 60;
28
29pub async fn handle_alter_system(
30 handler_args: HandlerArgs,
31 param: Ident,
32 value: SetVariableValue,
33) -> Result<RwPgResponse> {
34 let value = set_var_to_param_str(&value);
35 let param_name = param.to_string();
36 let meta_client = handler_args.session.env().meta_client();
37 let mut builder = RwPgResponse::builder(StatementType::ALTER_SYSTEM);
38
39 if SessionConfig::contains_param(¶m_name) {
42 if SessionConfig::check_no_alter_sys(¶m_name).unwrap() {
43 return Err(ErrorCode::InternalError(format!(
44 "session param {} cannot be altered system wide",
45 param_name
46 ))
47 .into());
48 }
49 meta_client.set_session_param(param_name, value).await?;
50 } else {
51 let params = meta_client.set_system_param(param_name, value).await?;
52 if let Some(params) = params {
53 if params.barrier_interval_ms() >= NOTICE_BARRIER_INTERVAL_MS {
54 builder = builder.notice(
55 format!("Barrier interval is set to {} ms >= {} ms. This can hurt freshness and potentially cause OOM.",
56 params.barrier_interval_ms(), NOTICE_BARRIER_INTERVAL_MS));
57 }
58 if params.checkpoint_frequency() >= NOTICE_CHECKPOINT_FREQUENCY {
59 builder = builder.notice(
60 format!("Checkpoint frequency is set to {} >= {}. This can hurt freshness and potentially cause OOM.",
61 params.checkpoint_frequency(), NOTICE_CHECKPOINT_FREQUENCY));
62 }
63 }
64 }
65 Ok(builder.into())
66}