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

risingwave_frontend/handler/
alter_utils.rs
