1use itertools::Itertools;
16use risingwave_pb::expr::expr_node::{self, RexNode};
17use risingwave_pb::expr::{ExprNode, FunctionCall, UserDefinedFunction};
18use risingwave_pb::plan_common::PbColumnDesc;
19use risingwave_sqlparser::ast::{
20 Array, CdcTableInfo, CreateSink, CreateSinkStatement, CreateSourceStatement,
21 CreateSubscriptionStatement, Distinct, Expr, Function, FunctionArg, FunctionArgExpr,
22 FunctionArgList, Ident, ObjectName, Query, SelectItem, SetExpr, Statement, TableAlias,
23 TableFactor, TableWithJoins, Window,
24};
25use risingwave_sqlparser::parser::Parser;
26
27pub fn alter_relation_rename(definition: &str, new_name: &str) -> String {
32 if definition.is_empty() {
35 tracing::warn!("found empty definition when renaming relation, ignored.");
36 return definition.into();
37 }
38 let ast = Parser::parse_sql(definition).expect("failed to parse relation definition");
39 let mut stmt =
40 Itertools::exactly_one(ast.into_iter()).expect("should contains only one statement");
41
42 match &mut stmt {
43 Statement::CreateTable { name, .. }
44 | Statement::CreateView { name, .. }
45 | Statement::CreateIndex { name, .. }
46 | Statement::CreateSource {
47 stmt: CreateSourceStatement {
48 source_name: name, ..
49 },
50 }
51 | Statement::CreateSubscription {
52 stmt:
53 CreateSubscriptionStatement {
54 subscription_name: name,
55 ..
56 },
57 }
58 | Statement::CreateSink {
59 stmt: CreateSinkStatement {
60 sink_name: name, ..
61 },
62 } => replace_table_name(name, new_name),
63 _ => unreachable!(),
64 };
65
66 stmt.to_string()
67}
68
69pub fn alter_relation_rename_refs(definition: &str, from: &str, to: &str) -> String {
72 let ast = Parser::parse_sql(definition).expect("failed to parse relation definition");
73 let mut stmt =
74 Itertools::exactly_one(ast.into_iter()).expect("should contains only one statement");
75
76 match &mut stmt {
77 Statement::CreateTable {
78 query: Some(query), ..
79 }
80 | Statement::CreateView { query, .. }
81 | Statement::Query(query) | Statement::CreateSink {
83 stmt:
84 CreateSinkStatement {
85 sink_from: CreateSink::AsQuery(query),
86 into_table_name: None,
87 ..
88 },
89 } => {
90 QueryRewriter::rewrite_query(query, from, to);
91 }
92 Statement::CreateIndex { table_name, .. }
93 | Statement::CreateSink {
94 stmt:
95 CreateSinkStatement {
96 sink_from: CreateSink::From(table_name),
97 into_table_name: None,
98 ..
99 },
100 }| Statement::CreateSubscription {
101 stmt:
102 CreateSubscriptionStatement {
103 subscription_from: table_name,
104 ..
105 },
106 } | Statement::CreateTable {
107 cdc_table_info:
108 Some(CdcTableInfo {
109 source_name: table_name,
110 ..
111 }),
112 ..
113 } | Statement::CreateSource {
114 stmt: CreateSourceStatement {
115 cdc_table_info:
116 Some(CdcTableInfo {
117 source_name: table_name,
118 ..
119 }),
120 ..
121 },
122 } => replace_table_name(table_name, to),
123 Statement::CreateSink {
124 stmt: CreateSinkStatement {
125 sink_from,
126 into_table_name: Some(table_name),
127 ..
128 }
129 } => {
130 let idx = table_name.0.len() - 1;
131 if table_name.0[idx].real_value() == from {
132 table_name.0[idx] = Ident::from_real_value(to);
133 } else {
134 match sink_from {
135 CreateSink::From(table_name) => replace_table_name(table_name, to),
136 CreateSink::AsQuery(query) => QueryRewriter::rewrite_query(query, from, to),
137 }
138 }
139 }
140 _ => unreachable!(),
141 };
142 stmt.to_string()
143}
144
145fn replace_table_name(table_name: &mut ObjectName, to: &str) {
148 let idx = table_name.0.len() - 1;
149 table_name.0[idx] = Ident::from_real_value(to);
150}
151
152struct QueryRewriter<'a> {
155 from: &'a str,
156 to: &'a str,
157}
158
159impl QueryRewriter<'_> {
160 fn rewrite_query(query: &mut Query, from: &str, to: &str) {
161 let rewriter = QueryRewriter { from, to };
162 rewriter.visit_query(query)
163 }
164
165 fn visit_query(&self, query: &mut Query) {
167 if let Some(with) = &mut query.with {
168 for cte_table in &mut with.cte_tables {
169 match &mut cte_table.cte_inner {
170 risingwave_sqlparser::ast::CteInner::Query(query) => self.visit_query(query),
171 risingwave_sqlparser::ast::CteInner::ChangeLog(name) => {
172 let idx = name.0.len() - 1;
173 if name.0[idx].real_value() == self.from {
174 replace_table_name(name, self.to);
175 }
176 }
177 }
178 }
179 }
180 self.visit_set_expr(&mut query.body);
181 for expr in &mut query.order_by {
182 self.visit_expr(&mut expr.expr);
183 }
184 }
185
186 fn visit_table_factor(&self, table_factor: &mut TableFactor) {
198 match table_factor {
199 TableFactor::Table { name, alias, .. } => {
200 let idx = name.0.len() - 1;
201 if name.0[idx].real_value() == self.from {
202 if alias.is_none() {
203 *alias = Some(TableAlias {
204 name: Ident::from_real_value(self.from),
205 columns: vec![],
206 });
207 }
208 name.0[idx] = Ident::from_real_value(self.to);
209 }
210 }
211 TableFactor::Derived { subquery, .. } => self.visit_query(subquery),
212 TableFactor::TableFunction { args, .. } => {
213 for arg in args {
214 self.visit_function_arg(arg);
215 }
216 }
217 TableFactor::NestedJoin(table_with_joins) => {
218 self.visit_table_with_joins(table_with_joins);
219 }
220 TableFactor::MatchRecognize { table, .. } => {
221 self.visit_table_factor(table);
225 }
226 }
227 }
228
229 fn visit_table_with_joins(&self, table_with_joins: &mut TableWithJoins) {
231 self.visit_table_factor(&mut table_with_joins.relation);
232 for join in &mut table_with_joins.joins {
233 self.visit_table_factor(&mut join.relation);
234 }
235 }
236
237 fn visit_set_expr(&self, set_expr: &mut SetExpr) {
239 match set_expr {
240 SetExpr::Select(select) => {
241 if let Distinct::DistinctOn(exprs) = &mut select.distinct {
242 for expr in exprs {
243 self.visit_expr(expr);
244 }
245 }
246 for select_item in &mut select.projection {
247 self.visit_select_item(select_item);
248 }
249 for from_item in &mut select.from {
250 self.visit_table_with_joins(from_item);
251 }
252 if let Some(where_clause) = &mut select.selection {
253 self.visit_expr(where_clause);
254 }
255 for expr in &mut select.group_by {
256 self.visit_expr(expr);
257 }
258 if let Some(having) = &mut select.having {
259 self.visit_expr(having);
260 }
261 for named_window in &mut select.window {
262 for expr in &mut named_window.window_spec.partition_by {
263 self.visit_expr(expr);
264 }
265 for expr in &mut named_window.window_spec.order_by {
266 self.visit_expr(&mut expr.expr);
267 }
268 }
269 }
270 SetExpr::Query(query) => self.visit_query(query),
271 SetExpr::SetOperation { left, right, .. } => {
272 self.visit_set_expr(left);
273 self.visit_set_expr(right);
274 }
275 SetExpr::Values(_) => {}
276 }
277 }
278
279 fn visit_function_arg(&self, function_arg: &mut FunctionArg) {
281 match function_arg {
282 FunctionArg::Unnamed(arg) | FunctionArg::Named { arg, .. } => match arg {
283 FunctionArgExpr::Expr(expr) | FunctionArgExpr::ExprQualifiedWildcard(expr, _) => {
284 self.visit_expr(expr)
285 }
286 FunctionArgExpr::QualifiedWildcard(_, None) | FunctionArgExpr::Wildcard(None) => {}
287 FunctionArgExpr::QualifiedWildcard(_, Some(exprs))
288 | FunctionArgExpr::Wildcard(Some(exprs)) => {
289 for expr in exprs {
290 self.visit_expr(expr);
291 }
292 }
293 FunctionArgExpr::SecretRef(_) => {}
294 },
295 }
296 }
297
298 fn visit_function_arg_list(&self, arg_list: &mut FunctionArgList) {
299 for arg in &mut arg_list.args {
300 self.visit_function_arg(arg);
301 }
302 for expr in &mut arg_list.order_by {
303 self.visit_expr(&mut expr.expr)
304 }
305 }
306
307 fn visit_function(&self, function: &mut Function) {
309 self.visit_function_arg_list(&mut function.arg_list);
310 if let Some(over) = &mut function.over {
311 match over {
312 Window::Spec(window) => {
313 for expr in &mut window.partition_by {
314 self.visit_expr(expr);
315 }
316 for expr in &mut window.order_by {
317 self.visit_expr(&mut expr.expr);
318 }
319 }
320 Window::Name(_) => {
321 }
323 }
324 }
325 }
326
327 fn visit_expr(&self, expr: &mut Expr) {
329 match expr {
330 Expr::FieldIdentifier(expr, ..)
331 | Expr::IsNull(expr)
332 | Expr::IsNotNull(expr)
333 | Expr::IsTrue(expr)
334 | Expr::IsNotTrue(expr)
335 | Expr::IsFalse(expr)
336 | Expr::IsNotFalse(expr)
337 | Expr::IsUnknown(expr)
338 | Expr::IsNotUnknown(expr)
339 | Expr::IsJson { expr, .. }
340 | Expr::InList { expr, .. }
341 | Expr::SomeOp(expr)
342 | Expr::AllOp(expr)
343 | Expr::UnaryOp { expr, .. }
344 | Expr::Cast { expr, .. }
345 | Expr::TryCast { expr, .. }
346 | Expr::AtTimeZone {
347 timestamp: expr, ..
348 }
349 | Expr::Extract { expr, .. }
350 | Expr::Substring { expr, .. }
351 | Expr::Overlay { expr, .. }
352 | Expr::Trim { expr, .. }
353 | Expr::Nested(expr)
354 | Expr::Index { obj: expr, .. }
355 | Expr::ArrayRangeIndex { obj: expr, .. } => self.visit_expr(expr),
356
357 Expr::Position { substring, string } => {
358 self.visit_expr(substring);
359 self.visit_expr(string);
360 }
361
362 Expr::InSubquery { expr, subquery, .. } => {
363 self.visit_expr(expr);
364 self.visit_query(subquery);
365 }
366 Expr::Between {
367 expr, low, high, ..
368 } => {
369 self.visit_expr(expr);
370 self.visit_expr(low);
371 self.visit_expr(high);
372 }
373 Expr::Like {
374 expr, pattern: pat, ..
375 } => {
376 self.visit_expr(expr);
377 self.visit_expr(pat);
378 }
379 Expr::ILike {
380 expr, pattern: pat, ..
381 } => {
382 self.visit_expr(expr);
383 self.visit_expr(pat);
384 }
385 Expr::SimilarTo {
386 expr, pattern: pat, ..
387 } => {
388 self.visit_expr(expr);
389 self.visit_expr(pat);
390 }
391
392 Expr::IsDistinctFrom(expr1, expr2)
393 | Expr::IsNotDistinctFrom(expr1, expr2)
394 | Expr::BinaryOp {
395 left: expr1,
396 right: expr2,
397 ..
398 } => {
399 self.visit_expr(expr1);
400 self.visit_expr(expr2);
401 }
402 Expr::Function(function) => self.visit_function(function),
403 Expr::Exists(query) | Expr::Subquery(query) | Expr::ArraySubquery(query) => {
404 self.visit_query(query)
405 }
406
407 Expr::GroupingSets(exprs_vec) | Expr::Cube(exprs_vec) | Expr::Rollup(exprs_vec) => {
408 for exprs in exprs_vec {
409 for expr in exprs {
410 self.visit_expr(expr);
411 }
412 }
413 }
414
415 Expr::Row(exprs) | Expr::Array(Array { elem: exprs, .. }) => {
416 for expr in exprs {
417 self.visit_expr(expr);
418 }
419 }
420 Expr::Map { entries } => {
421 for (key, value) in entries {
422 self.visit_expr(key);
423 self.visit_expr(value);
424 }
425 }
426
427 Expr::LambdaFunction { body, args: _ } => self.visit_expr(body),
428
429 Expr::Identifier(_)
431 | Expr::CompoundIdentifier(_)
432 | Expr::Collate { .. }
433 | Expr::Value(_)
434 | Expr::Parameter { .. }
435 | Expr::TypedString { .. }
436 | Expr::Case { .. } => {}
437 }
438 }
439
440 fn visit_select_item(&self, select_item: &mut SelectItem) {
442 match select_item {
443 SelectItem::UnnamedExpr(expr)
444 | SelectItem::ExprQualifiedWildcard(expr, _)
445 | SelectItem::ExprWithAlias { expr, .. } => self.visit_expr(expr),
446 SelectItem::QualifiedWildcard(_, None) | SelectItem::Wildcard(None) => {}
447 SelectItem::QualifiedWildcard(_, Some(exprs)) | SelectItem::Wildcard(Some(exprs)) => {
448 for expr in exprs {
449 self.visit_expr(expr);
450 }
451 }
452 }
453 }
454}
455
456pub struct IndexItemRewriter {
459 pub original_columns: Vec<PbColumnDesc>,
460 pub new_columns: Vec<PbColumnDesc>,
461}
462
463impl IndexItemRewriter {
464 pub fn rewrite_expr(&self, expr: &mut ExprNode) {
465 let rex_node = expr.rex_node.as_mut().unwrap();
466 match rex_node {
467 RexNode::InputRef(idx) => {
468 let old_idx = *idx as usize;
469 let original_column = &self.original_columns[old_idx];
470 let (new_idx, new_column) = self
471 .new_columns
472 .iter()
473 .find_position(|c| c.column_id == original_column.column_id)
474 .expect("should already checked index referencing column still exists");
475 *idx = new_idx as u32;
476
477 if new_column.column_type != original_column.column_type {
481 let old_type = original_column.column_type.clone().unwrap();
482 let new_type = new_column.column_type.clone().unwrap();
483
484 assert_eq!(&old_type, expr.return_type.as_ref().unwrap());
485 expr.return_type = Some(new_type); let new_expr_node = ExprNode {
488 function_type: expr_node::Type::CompositeCast as _,
489 return_type: Some(old_type),
490 rex_node: RexNode::FuncCall(FunctionCall {
491 children: vec![expr.clone()],
492 })
493 .into(),
494 };
495
496 *expr = new_expr_node;
497 }
498 }
499 RexNode::Constant(_) => {}
500 RexNode::Udf(udf) => self.rewrite_udf(udf),
501 RexNode::FuncCall(function_call) => self.rewrite_function_call(function_call),
502 RexNode::Now(_) | RexNode::SecretRef(_) => {}
503 }
504 }
505
506 fn rewrite_udf(&self, udf: &mut UserDefinedFunction) {
507 udf.children
508 .iter_mut()
509 .for_each(|expr| self.rewrite_expr(expr));
510 }
511
512 fn rewrite_function_call(&self, function_call: &mut FunctionCall) {
513 function_call
514 .children
515 .iter_mut()
516 .for_each(|expr| self.rewrite_expr(expr));
517 }
518}
519
520#[cfg(test)]
521mod tests {
522 use super::*;
523
524 #[test]
525 fn test_alter_table_rename() {
526 let definition = "CREATE TABLE foo (a int, b int)";
527 let new_name = "bar";
528 let expected = "CREATE TABLE bar (a INT, b INT)";
529 let actual = alter_relation_rename(definition, new_name);
530 assert_eq!(expected, actual);
531 }
532
533 #[test]
534 fn test_rename_index_refs() {
535 let definition = "CREATE INDEX idx1 ON foo(v1 DESC, v2)";
536 let from = "foo";
537 let to = "bar";
538 let expected = "CREATE INDEX idx1 ON bar(v1 DESC, v2)";
539 let actual = alter_relation_rename_refs(definition, from, to);
540 assert_eq!(expected, actual);
541 }
542
543 #[test]
544 fn test_rename_sink_refs() {
545 let definition =
546 "CREATE SINK sink_t FROM foo WITH (connector = 'kafka', format = 'append_only')";
547 let from = "foo";
548 let to = "bar";
549 let expected =
550 "CREATE SINK sink_t FROM bar WITH (connector = 'kafka', format = 'append_only')";
551 let actual = alter_relation_rename_refs(definition, from, to);
552 assert_eq!(expected, actual);
553 }
554
555 #[test]
556 fn test_rename_with_alias_refs() {
557 let definition =
558 "CREATE MATERIALIZED VIEW mv1 AS SELECT foo.v1 AS m1v, foo.v2 AS m2v FROM foo";
559 let from = "foo";
560 let to = "bar";
561 let expected =
562 "CREATE MATERIALIZED VIEW mv1 AS SELECT foo.v1 AS m1v, foo.v2 AS m2v FROM bar AS foo";
563 let actual = alter_relation_rename_refs(definition, from, to);
564 assert_eq!(expected, actual);
565
566 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";
567 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";
568 let actual = alter_relation_rename_refs(definition, from, to);
569 assert_eq!(expected, actual);
570
571 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";
572 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";
573 let actual = alter_relation_rename_refs(definition, from, to);
574 assert_eq!(expected, actual);
575 }
576
577 #[test]
578 fn test_rename_with_complex_funcs() {
579 let definition = "CREATE MATERIALIZED VIEW mv1 AS SELECT \
580 agg1(\
581 foo.v1, func2(foo.v2) \
582 ORDER BY \
583 (SELECT foo.v3 FROM foo), \
584 (SELECT first_value(foo.v4) OVER (PARTITION BY (SELECT foo.v5 FROM foo) ORDER BY (SELECT foo.v6 FROM foo)) FROM foo)\
585 ) \
586 FROM foo";
587 let from = "foo";
588 let to = "bar";
589 let expected = "CREATE MATERIALIZED VIEW mv1 AS SELECT \
590 agg1(\
591 foo.v1, func2(foo.v2) \
592 ORDER BY \
593 (SELECT foo.v3 FROM bar AS foo), \
594 (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)\
595 ) \
596 FROM bar AS foo";
597 let actual = alter_relation_rename_refs(definition, from, to);
598 assert_eq!(expected, actual);
599 }
600}