From e445dc462ff5071488e64c0055c42bf44f85d5d9 Mon Sep 17 00:00:00 2001 From: August Date: Tue, 30 Jan 2024 20:10:30 +0800 Subject: [PATCH 1/2] fix: fix rename of target table for sink into table --- src/meta/src/controller/catalog.rs | 16 ++++++++++++++++ src/meta/src/manager/catalog/mod.rs | 4 +++- src/meta/src/manager/catalog/utils.rs | 19 +++++++++++++++++++ 3 files changed, 38 insertions(+), 1 deletion(-) diff --git a/src/meta/src/controller/catalog.rs b/src/meta/src/controller/catalog.rs index d4f67c6ad1ee..75ef08c92278 100644 --- a/src/meta/src/controller/catalog.rs +++ b/src/meta/src/controller/catalog.rs @@ -1803,6 +1803,22 @@ impl CatalogController { }}; } let objs = get_referring_objects(object_id, &txn).await?; + // TODO: For sink into table. when sink into table is ready. + // if object_type == ObjectType::Table { + // let incoming_sinks: Vec<_> = Table::find_by_id(object_id) + // .select_only() + // .column(table::Column::IncomingSinks) + // .into_tuple() + // .one(&txn) + // .await? + // .ok_or_else(|| MetaError::catalog_id_not_found("table", object_id))?; + // objs.extend(incoming_sinks.into_iter().map(|id| PartialObject { + // oid: id as _, + // obj_type: ObjectType::Sink, + // ..Default::default() + // })); + // } + for obj in objs { match obj.obj_type { ObjectType::Table => rename_relation_ref!(Table, table, table_id, obj.oid), diff --git a/src/meta/src/manager/catalog/mod.rs b/src/meta/src/manager/catalog/mod.rs index 89294fdcd43d..4d0476f5d488 100644 --- a/src/meta/src/manager/catalog/mod.rs +++ b/src/meta/src/manager/catalog/mod.rs @@ -1749,7 +1749,9 @@ impl CatalogManager { } for sink in database_mgr.sinks.values() { - if sink.dependent_relations.contains(&relation_id) { + if sink.dependent_relations.contains(&relation_id) + || sink.target_table == Some(relation_id) + { let mut sink = sink.clone(); sink.definition = alter_relation_rename_refs(&sink.definition, from, to); to_update_sinks.push(sink); diff --git a/src/meta/src/manager/catalog/utils.rs b/src/meta/src/manager/catalog/utils.rs index b9272b533daa..69592180aa60 100644 --- a/src/meta/src/manager/catalog/utils.rs +++ b/src/meta/src/manager/catalog/utils.rs @@ -107,6 +107,7 @@ pub fn alter_relation_rename_refs(definition: &str, from: &str, to: &str) -> Str stmt: CreateSinkStatement { sink_from: CreateSink::AsQuery(query), + into_table_name: None, .. }, } => { @@ -117,9 +118,27 @@ pub fn alter_relation_rename_refs(definition: &str, from: &str, to: &str) -> Str stmt: CreateSinkStatement { sink_from: CreateSink::From(table_name), + into_table_name: None, .. }, } => replace_table_name(table_name, to), + Statement::CreateSink { + stmt: CreateSinkStatement { + sink_from, + into_table_name: Some(table_name), + .. + } + } => { + let idx = table_name.0.len() - 1; + if table_name.0[idx].real_value() == from { + table_name.0[idx] = Ident::new_unchecked(to); + } else { + match sink_from { + CreateSink::From(table_name) => replace_table_name(table_name, to), + CreateSink::AsQuery(query) => QueryRewriter::rewrite_query(query, from, to), + } + } + } _ => unreachable!(), }; stmt.to_string() From 67d35af913fbb24a8dd63bf65400f83652bc73a4 Mon Sep 17 00:00:00 2001 From: August Date: Tue, 30 Jan 2024 20:24:06 +0800 Subject: [PATCH 2/2] fix rename tests --- e2e_test/sink/sink_into_table/rename.slt | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/e2e_test/sink/sink_into_table/rename.slt b/e2e_test/sink/sink_into_table/rename.slt index 19b24c9c20b4..f518349c01c2 100644 --- a/e2e_test/sink/sink_into_table/rename.slt +++ b/e2e_test/sink/sink_into_table/rename.slt @@ -19,7 +19,7 @@ ALTER SINK s RENAME TO s1; query TT SELECT name, definition from rw_sinks; ---- -s1 CREATE SINK s1 INTO target AS SELECT * FROM source WITH (type = 'append-only') +s1 CREATE SINK s1 INTO target AS SELECT * FROM source WITH (type = 'append-only') statement ok ALTER TABLE source RENAME TO src; @@ -36,7 +36,7 @@ tar CREATE TABLE tar (v INT) query TT SELECT name, definition from rw_sinks; ---- -s1 CREATE SINK s1 INTO target AS SELECT * FROM src AS source WITH (type = 'append-only') +s1 CREATE SINK s1 INTO tar AS SELECT * FROM src AS source WITH (type = 'append-only') statement ok drop sink s1;