risingwave_meta/controller/
rename.rs

1// Copyright 2023 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 anyhow::anyhow;
16use itertools::Itertools;
17use risingwave_pb::expr::expr_node::{self, RexNode};
18use risingwave_pb::expr::{ExprNode, FunctionCall, UserDefinedFunction};
19use risingwave_pb::plan_common::PbColumnDesc;
20use risingwave_sqlparser::ast::{
21    Array, CdcTableInfo, CreateSink, CreateSinkStatement, CreateSourceStatement,
22    CreateSubscriptionStatement, Distinct, Expr, Function, FunctionArg, FunctionArgExpr,
23    FunctionArgList, Ident, ObjectName, Query, SelectItem, SetExpr, Statement, TableAlias,
24    TableFactor, TableWithJoins, Window,
25};
26use risingwave_sqlparser::parser::Parser;
27
28use crate::MetaResult;
29
30/// A stored definition can come from an older version, so it may not parse anymore. The error leaves
31/// out the definition and the parser's message, which quotes it, because the connector options in a
32/// definition can have secrets.
33fn parse_definition(definition: &str) -> MetaResult<Statement> {
34    let ast = Parser::parse_sql(definition)
35        .map_err(|_| anyhow!("failed to parse the stored definition"))?;
36    Ok(Itertools::exactly_one(ast.into_iter()).expect("should contains only one statement"))
37}
38
39/// `alter_relation_rename` renames a relation to a new name in its `Create` statement, and returns
40/// the updated definition raw sql. Note that the `definition` must be a `Create` statement and the
41/// `new_name` must be a valid identifier, it should be validated before calling this function. To
42/// update all relations that depend on the renamed one, use `alter_relation_rename_refs`.
43pub fn alter_relation_rename(definition: &str, new_name: &str) -> MetaResult<String> {
44    // This happens when we try to rename a table that's created by `CREATE TABLE AS`. Remove it
45    // when we support `SHOW CREATE TABLE` for `CREATE TABLE AS`.
46    if definition.is_empty() {
47        tracing::warn!("found empty definition when renaming relation, ignored.");
48        return Ok(definition.into());
49    }
50    let mut stmt = parse_definition(definition)?;
51
52    match &mut stmt {
53        Statement::CreateTable { name, .. }
54        | Statement::CreateView { name, .. }
55        | Statement::CreateIndex { name, .. }
56        | Statement::CreateSource {
57            stmt: CreateSourceStatement {
58                source_name: name, ..
59            },
60        }
61        | Statement::CreateSubscription {
62            stmt:
63                CreateSubscriptionStatement {
64                    subscription_name: name,
65                    ..
66                },
67        }
68        | Statement::CreateSink {
69            stmt: CreateSinkStatement {
70                sink_name: name, ..
71            },
72        } => replace_table_name(name, new_name),
73        _ => unreachable!(),
74    };
75
76    Ok(stmt.to_string())
77}
78
79/// `alter_relation_rename_refs` updates all references of renamed-relation in the definition of
80/// target relation's `Create` statement.
81pub fn alter_relation_rename_refs(definition: &str, from: &str, to: &str) -> MetaResult<String> {
82    let mut stmt = parse_definition(definition)?;
83
84    match &mut stmt {
85        Statement::CreateTable {
86            query: Some(query), ..
87        }
88        | Statement::CreateView { query, .. }
89        | Statement::Query(query) // Used by view, actually we store a query as the definition of view.
90        | Statement::CreateSink {
91            stmt:
92            CreateSinkStatement {
93                sink_from: CreateSink::AsQuery(query),
94                into_table_name: None,
95                ..
96            },
97        } => {
98            QueryRewriter::rewrite_query(query, from, to);
99        }
100        Statement::CreateIndex { table_name, .. }
101        | Statement::CreateSink {
102            stmt:
103            CreateSinkStatement {
104                sink_from: CreateSink::From(table_name),
105                into_table_name: None,
106                ..
107            },
108        }| Statement::CreateSubscription {
109            stmt:
110            CreateSubscriptionStatement {
111                subscription_from: table_name,
112                ..
113            },
114        } | Statement::CreateTable {
115            cdc_table_info:
116            Some(CdcTableInfo {
117                source_name: table_name,
118                ..
119            }),
120            ..
121        } | Statement::CreateSource {
122            stmt: CreateSourceStatement {
123                cdc_table_info:
124                    Some(CdcTableInfo {
125                        source_name: table_name,
126                        ..
127                    }),
128                ..
129            },
130        } => replace_table_name(table_name, to),
131        Statement::CreateSink {
132            stmt: CreateSinkStatement {
133                sink_from,
134                into_table_name: Some(table_name),
135                ..
136            }
137        } => {
138            let idx = table_name.0.len() - 1;
139            if table_name.0[idx].real_value() == from {
140                table_name.0[idx] = Ident::from_real_value(to);
141            } else {
142                match sink_from {
143                    CreateSink::From(table_name) => replace_table_name(table_name, to),
144                    CreateSink::AsQuery(query) => QueryRewriter::rewrite_query(query, from, to),
145                }
146            }
147        }
148        _ => unreachable!(),
149    };
150    Ok(stmt.to_string())
151}
152
153/// Replace the last ident in the `table_name` with the given name, the object name is ensured to be
154/// non-empty. e.g. `schema.table` or `database.schema.table`.
155fn replace_table_name(table_name: &mut ObjectName, to: &str) {
156    let idx = table_name.0.len() - 1;
157    table_name.0[idx] = Ident::from_real_value(to);
158}
159
160/// `QueryRewriter` is a visitor that updates all references of relation named `from` to `to` in the
161/// given query, which is the part of create statement of `relation`.
162struct QueryRewriter<'a> {
163    from: &'a str,
164    to: &'a str,
165}
166
167impl QueryRewriter<'_> {
168    fn rewrite_query(query: &mut Query, from: &str, to: &str) {
169        let rewriter = QueryRewriter { from, to };
170        rewriter.visit_query(query)
171    }
172
173    /// Visit the query and update all references of relation named `from` to `to`.
174    fn visit_query(&self, query: &mut Query) {
175        if let Some(with) = &mut query.with {
176            for cte_table in &mut with.cte_tables {
177                match &mut cte_table.cte_inner {
178                    risingwave_sqlparser::ast::CteInner::Query(query) => self.visit_query(query),
179                    risingwave_sqlparser::ast::CteInner::ChangeLog { from: name, .. } => {
180                        let idx = name.0.len() - 1;
181                        if name.0[idx].real_value() == self.from {
182                            replace_table_name(name, self.to);
183                        }
184                    }
185                }
186            }
187        }
188        self.visit_set_expr(&mut query.body);
189        for expr in &mut query.order_by {
190            self.visit_expr(&mut expr.expr);
191        }
192    }
193
194    /// Visit table factor and update all references of relation named `from` to `to`.
195    /// Rewrite idents(i.e. `schema.table`, `table`) that contains the old name in the
196    /// following pattern:
197    /// 1. `FROM a` to `FROM new_a AS a`
198    /// 2. `FROM a AS b` to `FROM new_a AS b`
199    ///
200    /// So that we DON'T have to:
201    /// 1. rewrite the select and expr part like `schema.table.column`, `table.column`,
202    ///    `alias.column` etc.
203    /// 2. handle the case that the old name is used as alias.
204    /// 3. handle the case that the new name is used as alias.
205    fn visit_table_factor(&self, table_factor: &mut TableFactor) {
206        match table_factor {
207            TableFactor::Table { name, alias, .. } => {
208                let idx = name.0.len() - 1;
209                if name.0[idx].real_value() == self.from {
210                    if alias.is_none() {
211                        *alias = Some(TableAlias {
212                            name: Ident::from_real_value(self.from),
213                            columns: vec![],
214                        });
215                    }
216                    name.0[idx] = Ident::from_real_value(self.to);
217                }
218            }
219            TableFactor::Derived { subquery, .. } => self.visit_query(subquery),
220            TableFactor::TableFunction { args, .. } => {
221                for arg in args {
222                    self.visit_function_arg(arg);
223                }
224            }
225            TableFactor::NestedJoin(table_with_joins) => {
226                self.visit_table_with_joins(table_with_joins);
227            }
228            TableFactor::MatchRecognize { table, .. } => {
229                // Only the input table can reference a relation: the binder rejects subqueries in
230                // both DEFINE and MEASURES (they have no representation in the executor's scalar
231                // expressions), so the remaining clauses contain no rename targets.
232                self.visit_table_factor(table);
233            }
234        }
235    }
236
237    /// Visit table with joins and update all references of relation named `from` to `to`.
238    fn visit_table_with_joins(&self, table_with_joins: &mut TableWithJoins) {
239        self.visit_table_factor(&mut table_with_joins.relation);
240        for join in &mut table_with_joins.joins {
241            self.visit_table_factor(&mut join.relation);
242        }
243    }
244
245    /// Visit query body expression and update all references.
246    fn visit_set_expr(&self, set_expr: &mut SetExpr) {
247        match set_expr {
248            SetExpr::Select(select) => {
249                if let Distinct::DistinctOn(exprs) = &mut select.distinct {
250                    for expr in exprs {
251                        self.visit_expr(expr);
252                    }
253                }
254                for select_item in &mut select.projection {
255                    self.visit_select_item(select_item);
256                }
257                for from_item in &mut select.from {
258                    self.visit_table_with_joins(from_item);
259                }
260                if let Some(where_clause) = &mut select.selection {
261                    self.visit_expr(where_clause);
262                }
263                for expr in &mut select.group_by {
264                    self.visit_expr(expr);
265                }
266                if let Some(having) = &mut select.having {
267                    self.visit_expr(having);
268                }
269                for named_window in &mut select.window {
270                    for expr in &mut named_window.window_spec.partition_by {
271                        self.visit_expr(expr);
272                    }
273                    for expr in &mut named_window.window_spec.order_by {
274                        self.visit_expr(&mut expr.expr);
275                    }
276                }
277            }
278            SetExpr::Query(query) => self.visit_query(query),
279            SetExpr::SetOperation { left, right, .. } => {
280                self.visit_set_expr(left);
281                self.visit_set_expr(right);
282            }
283            SetExpr::Values(_) => {}
284        }
285    }
286
287    /// Visit function arguments and update all references.
288    fn visit_function_arg(&self, function_arg: &mut FunctionArg) {
289        match function_arg {
290            FunctionArg::Unnamed(arg) | FunctionArg::Named { arg, .. } => match arg {
291                FunctionArgExpr::Expr(expr) | FunctionArgExpr::ExprQualifiedWildcard(expr, _) => {
292                    self.visit_expr(expr)
293                }
294                FunctionArgExpr::QualifiedWildcard(_, None) | FunctionArgExpr::Wildcard(None) => {}
295                FunctionArgExpr::QualifiedWildcard(_, Some(exprs))
296                | FunctionArgExpr::Wildcard(Some(exprs)) => {
297                    for expr in exprs {
298                        self.visit_expr(expr);
299                    }
300                }
301                FunctionArgExpr::SecretRef(_) => {}
302            },
303        }
304    }
305
306    fn visit_function_arg_list(&self, arg_list: &mut FunctionArgList) {
307        for arg in &mut arg_list.args {
308            self.visit_function_arg(arg);
309        }
310        for expr in &mut arg_list.order_by {
311            self.visit_expr(&mut expr.expr)
312        }
313    }
314
315    /// Visit function and update all references.
316    fn visit_function(&self, function: &mut Function) {
317        self.visit_function_arg_list(&mut function.arg_list);
318        if let Some(over) = &mut function.over {
319            match over {
320                Window::Spec(window) => {
321                    for expr in &mut window.partition_by {
322                        self.visit_expr(expr);
323                    }
324                    for expr in &mut window.order_by {
325                        self.visit_expr(&mut expr.expr);
326                    }
327                }
328                Window::Name(_) => {
329                    // Named window references don't contain expressions to rewrite
330                }
331            }
332        }
333    }
334
335    /// Visit expression and update all references.
336    fn visit_expr(&self, expr: &mut Expr) {
337        match expr {
338            Expr::FieldIdentifier(expr, ..)
339            | Expr::IsNull(expr)
340            | Expr::IsNotNull(expr)
341            | Expr::IsTrue(expr)
342            | Expr::IsNotTrue(expr)
343            | Expr::IsFalse(expr)
344            | Expr::IsNotFalse(expr)
345            | Expr::IsUnknown(expr)
346            | Expr::IsNotUnknown(expr)
347            | Expr::IsJson { expr, .. }
348            | Expr::InList { expr, .. }
349            | Expr::SomeOp(expr)
350            | Expr::AllOp(expr)
351            | Expr::UnaryOp { expr, .. }
352            | Expr::Cast { expr, .. }
353            | Expr::TryCast { expr, .. }
354            | Expr::AtTimeZone {
355                timestamp: expr, ..
356            }
357            | Expr::Extract { expr, .. }
358            | Expr::Substring { expr, .. }
359            | Expr::Overlay { expr, .. }
360            | Expr::Trim { expr, .. }
361            | Expr::Nested(expr)
362            | Expr::Index { obj: expr, .. }
363            | Expr::ArrayRangeIndex { obj: expr, .. } => self.visit_expr(expr),
364
365            Expr::Position { substring, string } => {
366                self.visit_expr(substring);
367                self.visit_expr(string);
368            }
369
370            Expr::InSubquery { expr, subquery, .. } => {
371                self.visit_expr(expr);
372                self.visit_query(subquery);
373            }
374            Expr::Between {
375                expr, low, high, ..
376            } => {
377                self.visit_expr(expr);
378                self.visit_expr(low);
379                self.visit_expr(high);
380            }
381            Expr::Like {
382                expr, pattern: pat, ..
383            } => {
384                self.visit_expr(expr);
385                self.visit_expr(pat);
386            }
387            Expr::ILike {
388                expr, pattern: pat, ..
389            } => {
390                self.visit_expr(expr);
391                self.visit_expr(pat);
392            }
393            Expr::SimilarTo {
394                expr, pattern: pat, ..
395            } => {
396                self.visit_expr(expr);
397                self.visit_expr(pat);
398            }
399
400            Expr::IsDistinctFrom(expr1, expr2)
401            | Expr::IsNotDistinctFrom(expr1, expr2)
402            | Expr::BinaryOp {
403                left: expr1,
404                right: expr2,
405                ..
406            } => {
407                self.visit_expr(expr1);
408                self.visit_expr(expr2);
409            }
410            Expr::Function(function) => self.visit_function(function),
411            Expr::Exists(query) | Expr::Subquery(query) | Expr::ArraySubquery(query) => {
412                self.visit_query(query)
413            }
414
415            Expr::GroupingSets(exprs_vec) | Expr::Cube(exprs_vec) | Expr::Rollup(exprs_vec) => {
416                for exprs in exprs_vec {
417                    for expr in exprs {
418                        self.visit_expr(expr);
419                    }
420                }
421            }
422
423            Expr::Row(exprs) | Expr::Array(Array { elem: exprs, .. }) => {
424                for expr in exprs {
425                    self.visit_expr(expr);
426                }
427            }
428            Expr::Map { entries } => {
429                for (key, value) in entries {
430                    self.visit_expr(key);
431                    self.visit_expr(value);
432                }
433            }
434
435            Expr::LambdaFunction { body, args: _ } => self.visit_expr(body),
436
437            // No need to visit.
438            Expr::Identifier(_)
439            | Expr::CompoundIdentifier(_)
440            | Expr::Collate { .. }
441            | Expr::Value(_)
442            | Expr::Parameter { .. }
443            | Expr::TypedString { .. }
444            | Expr::Case { .. } => {}
445        }
446    }
447
448    /// Visit select item and update all references.
449    fn visit_select_item(&self, select_item: &mut SelectItem) {
450        match select_item {
451            SelectItem::UnnamedExpr(expr)
452            | SelectItem::ExprQualifiedWildcard(expr, _)
453            | SelectItem::ExprWithAlias { expr, .. } => self.visit_expr(expr),
454            SelectItem::QualifiedWildcard(_, None) | SelectItem::Wildcard(None) => {}
455            SelectItem::QualifiedWildcard(_, Some(exprs)) | SelectItem::Wildcard(Some(exprs)) => {
456                for expr in exprs {
457                    self.visit_expr(expr);
458                }
459            }
460        }
461    }
462}
463
464/// Rewrite the expression in index item after there's a schema change on the primary table.
465// TODO: move this out of `rename.rs`, this has nothing to do with renaming.
466pub struct IndexItemRewriter {
467    pub original_columns: Vec<PbColumnDesc>,
468    pub new_columns: Vec<PbColumnDesc>,
469}
470
471impl IndexItemRewriter {
472    pub fn rewrite_expr(&self, expr: &mut ExprNode) {
473        let rex_node = expr.rex_node.as_mut().unwrap();
474        match rex_node {
475            RexNode::InputRef(idx) => {
476                let old_idx = *idx as usize;
477                let original_column = &self.original_columns[old_idx];
478                let (new_idx, new_column) = self
479                    .new_columns
480                    .iter()
481                    .find_position(|c| c.column_id == original_column.column_id)
482                    .expect("should already checked index referencing column still exists");
483                *idx = new_idx as u32;
484
485                // If there's a type change, we need to wrap it with an internal `CompositeCast` to
486                // maintain the correct return type. It cannot execute and will be eliminated in
487                // the frontend when rebuilding the index items.
488                if new_column.column_type != original_column.column_type {
489                    let old_type = original_column.column_type.clone().unwrap();
490                    let new_type = new_column.column_type.clone().unwrap();
491
492                    assert_eq!(&old_type, expr.return_type.as_ref().unwrap());
493                    expr.return_type = Some(new_type); // update return type of `InputRef`
494
495                    let new_expr_node = ExprNode {
496                        function_type: expr_node::Type::CompositeCast as _,
497                        return_type: Some(old_type),
498                        rex_node: RexNode::FuncCall(FunctionCall {
499                            children: vec![expr.clone()],
500                        })
501                        .into(),
502                    };
503
504                    *expr = new_expr_node;
505                }
506            }
507            RexNode::Constant(_) => {}
508            RexNode::Udf(udf) => self.rewrite_udf(udf),
509            RexNode::FuncCall(function_call) => self.rewrite_function_call(function_call),
510            RexNode::Now(_) | RexNode::SecretRef(_) => {}
511        }
512    }
513
514    fn rewrite_udf(&self, udf: &mut UserDefinedFunction) {
515        udf.children
516            .iter_mut()
517            .for_each(|expr| self.rewrite_expr(expr));
518    }
519
520    fn rewrite_function_call(&self, function_call: &mut FunctionCall) {
521        function_call
522            .children
523            .iter_mut()
524            .for_each(|expr| self.rewrite_expr(expr));
525    }
526}
527
528#[cfg(test)]
529mod tests {
530    use thiserror_ext::AsReport;
531
532    use super::*;
533
534    #[test]
535    fn test_alter_table_rename() {
536        let definition = "CREATE TABLE foo (a int, b int)";
537        let new_name = "bar";
538        let expected = "CREATE TABLE bar (a INT, b INT)";
539        let actual = alter_relation_rename(definition, new_name).unwrap();
540        assert_eq!(expected, actual);
541    }
542
543    #[test]
544    fn test_rename_with_unparsable_definition() {
545        let definition = "CREATE SINK s FROM t WITH (password = 'secret') AND";
546        for result in [
547            alter_relation_rename(definition, "s2"),
548            alter_relation_rename_refs(definition, "t", "t2"),
549        ] {
550            let error = result.unwrap_err().to_report_string();
551            assert!(!error.contains("secret"), "{error}");
552        }
553    }
554
555    #[test]
556    fn test_rename_index_refs() {
557        let definition = "CREATE INDEX idx1 ON foo(v1 DESC, v2)";
558        let from = "foo";
559        let to = "bar";
560        let expected = "CREATE INDEX idx1 ON bar(v1 DESC, v2)";
561        let actual = alter_relation_rename_refs(definition, from, to).unwrap();
562        assert_eq!(expected, actual);
563    }
564
565    #[test]
566    fn test_rename_sink_refs() {
567        let definition =
568            "CREATE SINK sink_t FROM foo WITH (connector = 'kafka', format = 'append_only')";
569        let from = "foo";
570        let to = "bar";
571        let expected =
572            "CREATE SINK sink_t FROM bar WITH (connector = 'kafka', format = 'append_only')";
573        let actual = alter_relation_rename_refs(definition, from, to).unwrap();
574        assert_eq!(expected, actual);
575    }
576
577    #[test]
578    fn test_rename_with_alias_refs() {
579        let definition =
580            "CREATE MATERIALIZED VIEW mv1 AS SELECT foo.v1 AS m1v, foo.v2 AS m2v FROM foo";
581        let from = "foo";
582        let to = "bar";
583        let expected =
584            "CREATE MATERIALIZED VIEW mv1 AS SELECT foo.v1 AS m1v, foo.v2 AS m2v FROM bar AS foo";
585        let actual = alter_relation_rename_refs(definition, from, to).unwrap();
586        assert_eq!(expected, actual);
587
588        let definition = "CREATE MATERIALIZED VIEW mv1 AS SELECT foo.v1 AS m1v, (foo.v2).v3 AS m2v FROM foo WHERE foo.v1 = 1 AND (foo.v2).v3 IS TRUE";
589        let expected = "CREATE MATERIALIZED VIEW mv1 AS SELECT foo.v1 AS m1v, (foo.v2).v3 AS m2v FROM bar AS foo WHERE foo.v1 = 1 AND (foo.v2).v3 IS TRUE";
590        let actual = alter_relation_rename_refs(definition, from, to).unwrap();
591        assert_eq!(expected, actual);
592
593        let definition = "CREATE MATERIALIZED VIEW mv1 AS SELECT bar.v1 AS m1v, (bar.v2).v3 AS m2v FROM foo AS bar WHERE bar.v1 = 1";
594        let expected = "CREATE MATERIALIZED VIEW mv1 AS SELECT bar.v1 AS m1v, (bar.v2).v3 AS m2v FROM bar AS bar WHERE bar.v1 = 1";
595        let actual = alter_relation_rename_refs(definition, from, to).unwrap();
596        assert_eq!(expected, actual);
597    }
598
599    #[test]
600    fn test_rename_with_complex_funcs() {
601        let definition = "CREATE MATERIALIZED VIEW mv1 AS SELECT \
602                            agg1(\
603                              foo.v1, func2(foo.v2) \
604                              ORDER BY \
605                              (SELECT foo.v3 FROM foo), \
606                              (SELECT first_value(foo.v4) OVER (PARTITION BY (SELECT foo.v5 FROM foo) ORDER BY (SELECT foo.v6 FROM foo)) FROM foo)\
607                            ) \
608                          FROM foo";
609        let from = "foo";
610        let to = "bar";
611        let expected = "CREATE MATERIALIZED VIEW mv1 AS SELECT \
612                          agg1(\
613                            foo.v1, func2(foo.v2) \
614                            ORDER BY \
615                            (SELECT foo.v3 FROM bar AS foo), \
616                            (SELECT first_value(foo.v4) OVER (PARTITION BY (SELECT foo.v5 FROM bar AS foo) ORDER BY (SELECT foo.v6 FROM bar AS foo)) FROM bar AS foo)\
617                          ) \
618                        FROM bar AS foo";
619        let actual = alter_relation_rename_refs(definition, from, to).unwrap();
620        assert_eq!(expected, actual);
621    }
622}
risingwave_meta/controller/rename.rs

risingwave_meta/controller/
rename.rs
