risingwave_frontend/handler/
alter_source_props.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 risingwave_common::id::{ConnectionId, SourceId};
16use risingwave_pb::catalog::connection::Info::ConnectionParams;
17
18use super::RwPgResponse;
19use super::create_source::validate_heartbeat_interval;
20use crate::catalog::catalog_service::CatalogReadGuard;
21use crate::catalog::root_catalog::SchemaPath;
22use crate::error::{ErrorCode, Result};
23use crate::handler::{HandlerArgs, ObjectName, SqlOption, StatementType};
24use crate::session::SessionImpl;
25use crate::utils::resolve_connection_ref_and_secret_ref;
26use crate::{Binder, WithOptions};
27
28pub async fn handle_alter_table_connector_props(
29    handler_args: HandlerArgs,
30    table_name: ObjectName,
31    alter_props: Vec<SqlOption>,
32) -> Result<RwPgResponse> {
33    let session = handler_args.session;
34    let db_name = &session.database();
35    let (schema_name, real_table_name) =
36        Binder::resolve_schema_qualified_name(db_name, &table_name)?;
37    let search_path = session.config().search_path();
38    let user_name = &session.user_name();
39    let schema_path = SchemaPath::new(schema_name.as_deref(), &search_path, user_name);
40
41    let source_id = {
42        let reader = session.env().catalog_reader().read_guard();
43        let (table, schema_name) =
44            reader.get_any_table_by_name(db_name, schema_path, &real_table_name)?;
45        let Some(associate_source_id) = table.associated_source_id else {
46            return Err(ErrorCode::InvalidInputSyntax(
47                "Only table with connector can use ALTER TABLE CONNECTOR syntax.".to_owned(),
48            )
49            .into());
50        };
51
52        session.check_privilege_for_drop_alter(schema_name, &**table)?;
53        let (source_catalog, _) =
54            reader.get_source_by_id(db_name, schema_path, associate_source_id)?;
55
56        ensure_alter_props_not_set_by_connection(
57            &reader,
58            db_name,
59            source_catalog.connection_id,
60            &alter_props,
61        )?;
62
63        tracing::info!(
64            "handle_alter_table_connector_props: table_name: {}, table id: {}, source_id: {}",
65            real_table_name,
66            table.id,
67            associate_source_id
68        );
69
70        associate_source_id
71    };
72
73    handle_alter_source_props_inner(&session, alter_props, source_id).await?;
74
75    Ok(RwPgResponse::empty_result(StatementType::ALTER_TABLE))
76}
77
78async fn handle_alter_source_props_inner(
79    session: &SessionImpl,
80    alter_props: Vec<SqlOption>,
81    source_id: SourceId,
82) -> Result<()> {
83    let meta_client = session.env().meta_client();
84    let (resolved_with_options, _, connector_conn_ref) = resolve_connection_ref_and_secret_ref(
85        WithOptions::try_from(alter_props.as_ref() as &[SqlOption])?,
86        session,
87        None,
88    )?;
89    let (changed_props, changed_secret_refs) = resolved_with_options.into_parts();
90    if connector_conn_ref.is_some() {
91        return Err(ErrorCode::InvalidInputSyntax(
92            "ALTER SOURCE CONNECTOR does not support CONNECTION".to_owned(),
93        )
94        .into());
95    }
96
97    // Validate cdc.source.wait.streaming.start.timeout if present
98    if let Some(timeout_value) = changed_props.get("cdc.source.wait.streaming.start.timeout")
99        && timeout_value.parse::<u32>().is_err()
100    {
101        return Err(ErrorCode::InvalidConfigValue {
102            config_entry: "cdc.source.wait.streaming.start.timeout".to_owned(),
103            config_value: timeout_value.to_owned(),
104        }
105        .into());
106    }
107
108    // Validate debezium.max.queue.size if present
109    if let Some(queue_size_value) = changed_props.get("debezium.max.queue.size")
110        && queue_size_value.parse::<u32>().is_err()
111    {
112        return Err(ErrorCode::InvalidConfigValue {
113            config_entry: "debezium.max.queue.size".to_owned(),
114            config_value: queue_size_value.to_owned(),
115        }
116        .into());
117    }
118
119    validate_heartbeat_interval(&changed_props)?;
120
121    meta_client
122        .alter_source_connector_props(
123            source_id,
124            changed_props,
125            changed_secret_refs,
126            connector_conn_ref, // always None, keep the interface for future extension
127        )
128        .await?;
129    Ok(())
130}
131
132pub async fn handle_alter_source_connector_props(
133    handler_args: HandlerArgs,
134    source_name: ObjectName,
135    alter_props: Vec<SqlOption>,
136) -> Result<RwPgResponse> {
137    let session = handler_args.session;
138    let db_name = &session.database();
139    let (schema_name, real_source_name) =
140        Binder::resolve_schema_qualified_name(db_name, &source_name)?;
141    let search_path = session.config().search_path();
142    let user_name = &session.user_name();
143    let schema_path = SchemaPath::new(schema_name.as_deref(), &search_path, user_name);
144
145    let source_id = {
146        let reader = session.env().catalog_reader().read_guard();
147        let (source, schema_name) =
148            reader.get_source_by_name(db_name, schema_path, &real_source_name)?;
149
150        // For `CREATE TABLE WITH (connector = '...')`, users should call `ALTER TABLE` instead.
151        if source.associated_table_id.is_some() {
152            return Err(ErrorCode::InvalidInputSyntax(
153                "Use `ALTER TABLE` to alter a table with connector.".to_owned(),
154            )
155            .into());
156        }
157        if source.is_cdc_table_source() {
158            return Err(ErrorCode::NotSupported(
159                "altering connector properties of a CDC table source is not supported".to_owned(),
160                "Drop and recreate the CDC table source".to_owned(),
161            )
162            .into());
163        }
164
165        session.check_privilege_for_drop_alter(schema_name, &**source)?;
166
167        ensure_alter_props_not_set_by_connection(
168            &reader,
169            db_name,
170            source.connection_id,
171            &alter_props,
172        )?;
173
174        source.id
175    };
176
177    handle_alter_source_props_inner(&session, alter_props, source_id).await?;
178
179    Ok(RwPgResponse::empty_result(StatementType::ALTER_SOURCE))
180}
181
182/// Validates that the properties being altered don't conflict with properties set by a CONNECTION.
183pub(crate) fn ensure_alter_props_not_set_by_connection(
184    reader: &CatalogReadGuard,
185    db_name: &str,
186    connection_id: Option<ConnectionId>,
187    alter_props: &[SqlOption],
188) -> Result<()> {
189    if let Some(conn_id) = connection_id {
190        let conn = reader.get_connection_by_id(db_name, conn_id)?;
191        if let ConnectionParams(params) = &conn.info {
192            for prop in alter_props {
193                let prop_key = prop.name.real_value();
194                if params.properties.contains_key(&prop_key)
195                    || params.secret_refs.contains_key(&prop_key)
196                {
197                    return Err(ErrorCode::InvalidInputSyntax(
198                        "Cannot alter connector properties that are set by CONNECTION..".to_owned(),
199                    )
200                    .into());
201                }
202            }
203        }
204    }
205    Ok(())
206}
risingwave_frontend/handler/alter_source_props.rs

risingwave_frontend/handler/
alter_source_props.rs
