replicators: Add support for RENAME TABLE

Implements support for `RENAME TABLE` (MySQL) and somewhat improves
`ALTER TABLE ... RENAME TO ...` handling (MySQL and PostgreSQL).

This is achieved by adding a new `AlterTableDefinition::RenameTable`
variant and, when constructing a `Changelist` for the server, converting
MySQL `RenameTableStatement` into a series of `AlterTableStatement`s.

The resulting `Change`s are handled in the replicator by triggering a
resnapshot of each affected table. This is not very efficient, but
actually "renaming" already-existing base tables would break a lot of
assumptions and require coordinating with the worker/domain to rename
the RocksDB directories safely. So this is the safest option, and it
should be good enough for now.

Release-Note-Core: Added support for `RENAME TABLE` and `ALTER TABLE ...
  RENAME TO ...` DDL statements. ReadySet will now detect table renames
  and automatically resnapshot the affected tables to maintain cache
  consistency.
Project: rename
AI-Use: level:3
Change-Id: I275c12bfbf0dfb94fd3a88f55ea3dd896a6a6964
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/10887
Reviewed-by: Jason Brown <jason.b@readyset.io>
Tested-by: Buildkite CI

Author: mvzink

readyset-adapter/src/backend.rs

@@ -3515,6 +3515,7 @@ where
                     | SqlQuery::DropTable(_)
                     | SqlQuery::DropView(_)
                     | SqlQuery::AlterTable(_)
+                    | SqlQuery::RenameTable(_)
                     | SqlQuery::Truncate(_)
                     | SqlQuery::Use(_)
                     | SqlQuery::CreateIndex(_) => {
@@ -3524,9 +3525,6 @@ where
                             .await
                             .map(|r| QueryResult::Upstream(r, None, None))
                     }
-                    SqlQuery::RenameTable(_) => {
-                        unsupported!("{} not yet supported", query.query_type());
-                    }
                     SqlQuery::Set(_)
                     | SqlQuery::CompoundSelect(_)
                     | SqlQuery::Show(_)

readyset-client/src/recipe/changelist.rs

@@ -41,7 +41,8 @@ use readyset_errors::{internal, unsupported, ReadySetError, ReadySetResult};
 use readyset_sql::ast::{
     AlterTableDefinition, AlterTableStatement, CacheInner, CreateCacheStatement,
     CreateTableStatement, CreateViewStatement, DropTableStatement, DropViewStatement,
-    NonReplicatedRelation, Relation, SelectStatement, SqlIdentifier, SqlQuery, TableKey,
+    NonReplicatedRelation, Relation, RenameTableStatement, SelectStatement, SqlIdentifier,
+    SqlQuery, TableKey,
 };
 use readyset_sql_parsing::{parse_query_with_config, ParsingConfig, ParsingPreset};
 use readyset_sql_passes::adapter_rewrites::{self, AdapterRewriteParams};
@@ -113,6 +114,25 @@ impl IntoChanges for AlterTableStatement {
     }
 }
 
+impl IntoChanges for RenameTableStatement {
+    /// For simplicity, we map MySQL `RENAME TABLE` statements to multiple `ALTER TABLE` statements.
+    /// This way, Postgres (no `RENAME`) and MySQL don't diverge, and we can handle this the same
+    /// way and in the same place as `ALTER TABLE RENAME`.
+    fn into_changes(self) -> Vec<Change> {
+        self.ops
+            .into_iter()
+            .map(|op| {
+                let alter_statement = AlterTableStatement {
+                    table: op.from,
+                    only: false, // RENAME TABLE doesn't have an ONLY keyword
+                    definitions: Ok(vec![AlterTableDefinition::RenameTable { new_name: op.to }]),
+                };
+                Change::AlterTable(alter_statement)
+            })
+            .collect()
+    }
+}
+
 impl IntoChanges for Vec<Change> {
     fn into_changes(self) -> Vec<Change> {
         self
@@ -211,6 +231,9 @@ impl ChangeList {
                     name: dcs.name,
                     if_exists: false,
                 }),
+                SqlQuery::RenameTable(rts) => {
+                    changes.extend(rts.into_changes());
+                }
                 _ => unsupported!(
                     "Only DDL statements supported in ChangeList (got {})",
                     parsed.query_type()
@@ -400,6 +423,7 @@ impl Change {
                         | AlterTableDefinition::DropColumn { .. }
                         | AlterTableDefinition::ChangeColumn { .. }
                         | AlterTableDefinition::RenameColumn { .. }
+                        | AlterTableDefinition::RenameTable { .. }
                         | AlterTableDefinition::AddKey(TableKey::PrimaryKey { .. })
                         | AlterTableDefinition::AddKey(TableKey::UniqueKey { .. })
                         | AlterTableDefinition::DropForeignKey { .. }

readyset-server/src/controller/sql/mod.rs

@@ -1062,6 +1062,7 @@ impl SqlIncorporator {
                             | AlterTableDefinition::DropColumn { .. }
                             | AlterTableDefinition::ChangeColumn { .. }
                             | AlterTableDefinition::RenameColumn { .. }
+                            | AlterTableDefinition::RenameTable { .. }
                             | AlterTableDefinition::DropConstraint { .. }
                             | AlterTableDefinition::ReplicaIdentity(_)
                             | AlterTableDefinition::DropForeignKey { .. } => {

readyset-sql-parsing/tests/parity.rs

@@ -890,8 +890,21 @@ fn where_not_in_not_between() {
 #[test]
 fn rename_table() {
     check_parse_both!("RENAME TABLE tb1 TO tb2");
-    check_parse_both!("ALTER TABLE tb1 RENAME TO tb2");
-    check_parse_both!("ALTER TABLE tb1 RENAME AS tb2");
+    check_parse_fails!(
+        Dialect::PostgreSQL,
+        "ALTER TABLE tb1 RENAME TO tb2",
+        "nom-sql AST differs"
+    );
+    check_parse_fails!(
+        Dialect::MySQL,
+        "ALTER TABLE tb1 RENAME TO tb2",
+        "nom-sql AST differs"
+    );
+    check_parse_fails!(
+        Dialect::MySQL,
+        "ALTER TABLE tb1 RENAME AS tb2",
+        "nom-sql AST differs"
+    );
 }
 
 #[test]

readyset-sql-parsing/tests/roundtrip.rs

@@ -116,3 +116,10 @@ fn create_index() {
         "found: SPATIAL"
     );
 }
+
+#[test]
+fn test_alter_table_rename() {
+    check_rt_mysql_sqlparser!("ALTER TABLE tb1 RENAME TO tb2");
+    check_rt_postgres_sqlparser!("ALTER TABLE tb1 RENAME TO tb2");
+    check_rt_mysql_sqlparser!("ALTER TABLE tb1 RENAME AS tb2");
+}

readyset-sql/src/analysis/visit.rs

@@ -1115,6 +1115,7 @@ pub fn walk_alter_table_definition<'a, V: Visitor<'a>>(
             visitor.visit_sql_identifier(name)?;
             visitor.visit_sql_identifier(new_name)
         }
+        AlterTableDefinition::RenameTable { new_name } => visitor.visit_table(new_name),
         AlterTableDefinition::ReplicaIdentity(_)
         | AlterTableDefinition::Algorithm { .. }
         | AlterTableDefinition::Lock { .. } => Ok(()),

readyset-sql/src/analysis/visit_mut.rs

@@ -1154,6 +1154,7 @@ pub fn walk_alter_table_definition<'a, V: VisitorMut<'a>>(
             visitor.visit_sql_identifier(name)?;
             visitor.visit_sql_identifier(new_name)
         }
+        AlterTableDefinition::RenameTable { new_name } => visitor.visit_table(new_name),
         AlterTableDefinition::ReplicaIdentity(_)
         | AlterTableDefinition::Algorithm { .. }
         | AlterTableDefinition::Lock { .. } => Ok(()),

readyset-sql/src/ast/alter.rs

@@ -7,6 +7,7 @@ use proptest::{
 };
 use readyset_util::fmt::fmt_with;
 use serde::{Deserialize, Serialize};
+use sqlparser::ast::RenameTableNameKind;
 use test_strategy::Arbitrary;
 
 use crate::{
@@ -172,6 +173,9 @@ pub enum AlterTableDefinition {
         name: SqlIdentifier,
         new_name: SqlIdentifier,
     },
+    RenameTable {
+        new_name: Relation,
+    },
     DropConstraint {
         name: SqlIdentifier,
         drop_behavior: Option<DropBehavior>,
@@ -235,6 +239,12 @@ impl TryFromDialect<sqlparser::ast::AlterTableOperation> for AlterTableDefinitio
                 name: old_column_name.into_dialect(dialect),
                 new_name: new_column_name.into_dialect(dialect),
             }),
+            sqlparser::ast::AlterTableOperation::RenameTable {
+                table_name:
+                    RenameTableNameKind::As(object_name) | RenameTableNameKind::To(object_name),
+            } => Ok(Self::RenameTable {
+                new_name: object_name.try_into_dialect(dialect)?,
+            }),
             ChangeColumn {
                 old_name,
                 new_name,
@@ -310,6 +320,9 @@ impl DialectDisplay for AlterTableDefinition {
                     dialect.quote_identifier(new_name)
                 )
             }
+            Self::RenameTable { new_name } => {
+                write!(f, "RENAME TO {}", new_name.display(dialect))
+            }
             Self::DropConstraint {
                 name,
                 drop_behavior,

replicators/src/noria_adapter.rs

@@ -795,9 +795,27 @@ impl<'a> NoriaAdapter<'a> {
                 }
                 keep
             }
-            Change::AlterTable(stmt) => self
-                .table_filter
-                .should_be_processed(schema.as_str(), stmt.table.name.as_str()),
+            Change::AlterTable(stmt) => {
+                // If this alter table is a rename, we also check whether the new name is filtered
+                // out. Only if both are ignored will this return false and this event get skipped.
+                self.table_filter
+                    .should_be_processed(schema.as_str(), stmt.table.name.as_str())
+                    || if let Ok(defs) = &stmt.definitions {
+                        defs.iter().any(|def| {
+                            if let readyset_sql::ast::AlterTableDefinition::RenameTable {
+                                new_name,
+                            } = def
+                            {
+                                self.table_filter
+                                    .should_be_processed(schema.as_str(), new_name.name.as_str())
+                            } else {
+                                false
+                            }
+                        })
+                    } else {
+                        false
+                    }
+            }
             _ => true,
         });
 
@@ -811,6 +829,35 @@ impl<'a> NoriaAdapter<'a> {
                 })
             }));
 
+        // Resolve schemas and drop renamed tables; the new names will get snapshotted when we
+        // return `ResnapshotNeeded` in the next `if` block
+        for change in changelist.changes_mut() {
+            if let Change::AlterTable(stmt) = change {
+                if stmt.table.schema.is_none() {
+                    stmt.table.schema = Some(schema.clone().into());
+                }
+
+                if let Ok(definitions) = &mut stmt.definitions {
+                    for def in definitions.iter_mut() {
+                        if let readyset_sql::ast::AlterTableDefinition::RenameTable { new_name } =
+                            def
+                        {
+                            if new_name.schema.is_none() {
+                                new_name.schema = Some(schema.clone().into());
+                            }
+
+                            info!(
+                                old_table = %stmt.table.display_unquoted(),
+                                new_table = %new_name.display_unquoted(),
+                                "Table rename detected, triggering resnapshot for table"
+                            );
+                            self.drop_table_for_resnapshot(stmt.table.clone()).await?;
+                        }
+                    }
+                }
+            }
+        }
+
         if self.supports_resnapshot && changelist.changes().any(Change::requires_resnapshot) {
             // In case we detect a DDL change that requires a full schema resnapshot exit the loop
             // with the proper status