risingwave_frontend/handler/
alter_utils.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 pgwire::pg_response::StatementType;
16use risingwave_common::bail;
17use risingwave_pb::id::JobId;
18use risingwave_sqlparser::ast::ObjectName;
19
20use crate::Binder;
21use crate::catalog::CatalogError;
22use crate::catalog::root_catalog::SchemaPath;
23use crate::catalog::table_catalog::{TableCatalog, TableType};
24use crate::error::{Result, bail_invalid_input_syntax};
25use crate::session::SessionImpl;
26
27pub(super) fn validate_table_type_for_alter(
28    table: &TableCatalog,
29    alter_stmt_type: StatementType,
30    alter_target: &str,
31) -> Result<()> {
32    match (table.table_type(), alter_stmt_type) {
33        (TableType::Internal, _) => {
34            // We treat internal tables as not found because they are not user-addressable objects.
35            Err(CatalogError::not_found("table", table.name()).into())
36        }
37        (TableType::Table, StatementType::ALTER_TABLE)
38        | (TableType::MaterializedView, StatementType::ALTER_MATERIALIZED_VIEW)
39        | (TableType::Index, StatementType::ALTER_INDEX) => Ok(()),
40        _ => {
41            bail_invalid_input_syntax!(
42                "cannot alter {alter_target} of {} {} by {}",
43                table.table_type().to_prost().as_str_name(),
44                table.name(),
45                alter_stmt_type,
46            );
47        }
48    }
49}
50
51/// Resolve the **streaming** job id for alter operations.
52///
53/// This function will decide which catalog to lookup based on the given statement type, which should
54/// be one of `ALTER TABLE`, `ALTER MATERIALIZED VIEW`, `ALTER SOURCE`, `ALTER SINK`, `ALTER INDEX`.
55pub(super) fn resolve_streaming_job_id_for_alter(
56    session: &SessionImpl,
57    obj_name: ObjectName,
58    alter_stmt_type: StatementType,
59    alter_target: &str,
60) -> Result<JobId> {
61    let db_name = &session.database();
62    let (schema_name, real_table_name) = Binder::resolve_schema_qualified_name(db_name, &obj_name)?;
63    let search_path = session.config().search_path();
64    let user_name = &session.user_name();
65    let schema_path = SchemaPath::new(schema_name.as_deref(), &search_path, user_name);
66    let reader = session.env().catalog_reader().read_guard();
67
68    let job_id = match alter_stmt_type {
69        StatementType::ALTER_TABLE
70        | StatementType::ALTER_MATERIALIZED_VIEW
71        | StatementType::ALTER_INDEX => {
72            let (table, schema_name) =
73                reader.get_table_by_name(db_name, schema_path, &real_table_name, true)?;
74
75            validate_table_type_for_alter(table, alter_stmt_type, alter_target)?;
76
77            session.check_privilege_for_drop_alter(schema_name, &**table)?;
78            table.id.as_job_id()
79        }
80        StatementType::ALTER_SOURCE => {
81            let (source, schema_name) =
82                reader.get_source_by_name(db_name, schema_path, &real_table_name)?;
83
84            if !source.info.is_shared() {
85                bail_invalid_input_syntax!(
86                    "cannot alter {alter_target} of non-shared source.\n\
87                     Use `ALTER MATERIALIZED VIEW` to alter the materialized view using the source instead."
88                );
89            }
90
91            session.check_privilege_for_drop_alter(schema_name, &**source)?;
92            source.id.as_share_source_job_id()
93        }
94        StatementType::ALTER_SINK => {
95            let (sink, schema_name) =
96                reader.get_any_sink_by_name(db_name, schema_path, &real_table_name)?;
97
98            session.check_privilege_for_drop_alter(schema_name, &**sink)?;
99            sink.id.as_job_id()
100        }
101        _ => bail!(
102            "invalid statement type for alter {alter_target}: {:?}",
103            alter_stmt_type
104        ),
105    };
106
107    Ok(job_id)
108}
risingwave_frontend/handler/alter_utils.rs

risingwave_frontend/handler/
alter_utils.rs
