risingwave_frontend/handler/
alter_table_drop_connector.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::HashSet;
16use std::sync::{Arc, LazyLock};
17
18use risingwave_connector::parser::additional_columns::gen_default_addition_col_name;
19use risingwave_connector::sink::decouple_checkpoint_log_sink::COMMIT_CHECKPOINT_INTERVAL;
20use risingwave_pb::ddl_service::TableJobType;
21use risingwave_sqlparser::ast::{ColumnDef, Ident};
22
23use crate::catalog::root_catalog::SchemaPath;
24use crate::catalog::source_catalog::SourceCatalog;
25use crate::error::{ErrorCode, Result};
26use crate::handler::create_source::SqlColumnStrategy;
27use crate::handler::{
28    HandlerArgs, ObjectName, PgResponse, RwPgResponse, Statement, StatementType,
29    get_replace_table_plan,
30};
31use crate::session::SessionImpl;
32use crate::utils::data_type::DataTypeToAst;
33use crate::utils::options::RETENTION_SECONDS;
34use crate::{Binder, TableCatalog, bind_data_type};
35
36// allowed in with clause but irrelevant to connector
37static TABLE_PROPS: LazyLock<HashSet<&str>> =
38    LazyLock::new(|| HashSet::from([COMMIT_CHECKPOINT_INTERVAL, RETENTION_SECONDS]));
39
40fn fetch_schema_info(
41    session: &Arc<SessionImpl>,
42    table_name: ObjectName,
43) -> Result<(Arc<TableCatalog>, Arc<SourceCatalog>)> {
44    let db_name = session.database();
45    let (schema_name, real_table_name) =
46        Binder::resolve_schema_qualified_name(db_name.as_str(), &table_name)?;
47    let search_path = session.config().search_path();
48    let user_name = &session.auth_context().user_name;
49
50    let schema_path = SchemaPath::new(schema_name.as_deref(), &search_path, user_name);
51    let reader = session.env().catalog_reader().read_guard();
52
53    let (table_def, schema_name) =
54        reader.get_any_table_by_name(db_name.as_str(), schema_path, &real_table_name)?;
55    session.check_privilege_for_drop_alter(schema_name, &**table_def)?;
56
57    let Some(source_id) = table_def.associated_source_id else {
58        return Err(ErrorCode::ProtocolError(format!(
59            "Table {} is not associated with a connector",
60            real_table_name
61        ))
62        .into());
63    };
64    let (source_def, _) = reader.get_source_by_id(db_name.as_str(), schema_path, source_id)?;
65    Ok((table_def.clone(), source_def.clone()))
66}
67
68fn rewrite_table_definition(
69    original_table_def: &Arc<TableCatalog>,
70    original_source_def: &Arc<SourceCatalog>,
71    original_statement: Statement,
72) -> Result<Statement> {
73    let Statement::CreateTable {
74        mut columns,
75        include_column_options,
76        or_replace,
77        temporary,
78        if_not_exists,
79        name,
80        wildcard_idx,
81        constraints,
82        mut with_options,
83        source_watermarks,
84        append_only,
85        on_conflict,
86        with_version_columns,
87        query,
88        engine,
89        ..
90    } = original_statement
91    else {
92        panic!("unexpected statement: {:?}", original_statement);
93    };
94
95    // identical logic with func `handle_addition_columns`, reverse the order to keep the original order of additional columns
96    for item in include_column_options.iter().rev() {
97        let col_name = if let Some(col_alias) = &item.column_alias {
98            col_alias.real_value()
99        } else {
100            let data_type = if let Some(dt) = &item.header_inner_expect_type {
101                Some(bind_data_type(dt)?)
102            } else {
103                None
104            };
105            gen_default_addition_col_name(
106                original_source_def.connector_name().as_str(),
107                item.column_type.real_value().as_str(),
108                item.inner_field.as_deref(),
109                data_type.as_ref(),
110            )
111        };
112        // find the column def in the catalog
113        if let Some(col_def) = original_table_def
114            .columns
115            .iter()
116            .find(|col_def| col_def.name() == col_name)
117        {
118            columns.push(ColumnDef {
119                name: Ident::from(col_name.as_str()),
120                data_type: Some(col_def.data_type().to_ast()),
121                collation: None,
122                options: vec![],
123            });
124        }
125    }
126
127    let new_statement = Statement::CreateTable {
128        or_replace,
129        temporary,
130        if_not_exists,
131        name,
132        columns: columns.clone(),
133        wildcard_idx,
134        constraints,
135        with_options: {
136            with_options.retain(|item| {
137                TABLE_PROPS.contains(item.name.real_value().to_lowercase().as_str())
138            });
139            with_options
140        },
141        format_encode: None,
142        // The watermark belongs to the table: it also filters DML and sink-into-table rows, and
143        // downstream jobs rely on it.
144        source_watermarks,
145        append_only,
146        on_conflict,
147        with_version_columns,
148        query,
149        cdc_table_info: None,
150        include_column_options: vec![],
151        webhook_info: None,
152        engine,
153    };
154    Ok(new_statement)
155}
156
157pub async fn handle_alter_table_drop_connector(
158    handler_args: HandlerArgs,
159    table_name: ObjectName,
160) -> Result<RwPgResponse> {
161    let session = handler_args.session;
162    let (table_def, source_def) = fetch_schema_info(&session, table_name.clone())?;
163    let original_definition = table_def.create_sql_ast_purified()?;
164
165    let new_statement = rewrite_table_definition(&table_def, &source_def, original_definition)?;
166    let (_, table, graph, _) = Box::pin(get_replace_table_plan(
167        &session,
168        table_name,
169        new_statement,
170        &table_def,
171        SqlColumnStrategy::FollowUnchecked,
172    ))
173    .await?;
174
175    let catalog_writer = session.catalog_writer()?;
176    catalog_writer
177        .replace_table(None, table.to_prost(), graph, TableJobType::General as _)
178        .await?;
179
180    Ok(PgResponse::empty_result(StatementType::ALTER_TABLE))
181}
risingwave_frontend/handler/alter_table_drop_connector.rs

risingwave_frontend/handler/
alter_table_drop_connector.rs
