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

risingwave_frontend/handler/
alter_table_drop_connector.rs
