replicators: Handle MySQL TRUNCATE statements

Since MySQL binlogs `TRUNCATE` as a statement (`QUERY_EVENT`), but it's
not DDL and doesn't have a corresponding recipe `Change`, we were just
ignoring it. Now the MySQL replicator parses it, emitting the
`TableOperation::Truncate` we had already added for Postgres.

Fixes: REA-4325
Closes: #1221
Release-Note-Core: Added support for `TRUNCATE TABLE` statements for MySQL.
Change-Id: Ia40551e40fa70598973587f5b26e8662419e9853
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/7488
Tested-by: Buildkite CI
Reviewed-by: Marcelo Altmann <marcelo@readyset.io>

logictests/truncate.test

@@ -0,0 +1,12 @@
+statement ok
+create table t (x int);
+
+statement ok
+insert into t (x) values (1);
+
+statement ok
+truncate t;
+
+query
+select * from t where x = 1;
+----

nom-sql/src/analysis/visit.rs

@@ -15,6 +15,7 @@ use crate::rename::{RenameTableOperation, RenameTableStatement};
 use crate::select::LimitClause;
 use crate::set::Variable;
 use crate::transaction::{CommitStatement, RollbackStatement, StartTransactionStatement};
+use crate::truncate::TruncateStatement;
 use crate::{
     AlterColumnOperation, AlterTableDefinition, AlterTableStatement, CacheInner, CaseWhenBranch,
     Column, ColumnConstraint, ColumnSpecification, CommentStatement, CommonTableExpr,
@@ -422,6 +423,13 @@ pub trait Visitor<'ast>: Sized {
         Ok(())
     }
 
+    fn visit_truncate_statement(
+        &mut self,
+        truncate_statement: &'ast TruncateStatement,
+    ) -> Result<(), Self::Error> {
+        walk_truncate_statement(self, truncate_statement)
+    }
+
     fn visit_sql_query(&mut self, sql_query: &'ast SqlQuery) -> Result<(), Self::Error> {
         walk_sql_query(self, sql_query)
     }
@@ -1123,6 +1131,17 @@ pub fn walk_drop_view_statement<'a, V: Visitor<'a>>(
     Ok(())
 }
 
+pub fn walk_truncate_statement<'a, V: Visitor<'a>>(
+    visitor: &mut V,
+    truncate_statement: &'a TruncateStatement,
+) -> Result<(), V::Error> {
+    for table in &truncate_statement.tables {
+        visitor.visit_table(&table.relation)?;
+    }
+
+    Ok(())
+}
+
 pub fn walk_sql_query<'a, V: Visitor<'a>>(
     visitor: &mut V,
     sql_query: &'a SqlQuery,
@@ -1156,6 +1175,7 @@ pub fn walk_sql_query<'a, V: Visitor<'a>>(
         SqlQuery::Explain(statement) => visitor.visit_explain_statement(statement),
         SqlQuery::Comment(statement) => visitor.visit_comment_statement(statement),
         SqlQuery::Deallocate(statement) => visitor.visit_deallocate_statement(statement),
+        SqlQuery::Truncate(statement) => visitor.visit_truncate_statement(statement),
     }
 }
 

nom-sql/src/analysis/visit_mut.rs

@@ -15,6 +15,7 @@ use crate::rename::{RenameTableOperation, RenameTableStatement};
 use crate::select::{LimitClause, LimitValue};
 use crate::set::Variable;
 use crate::transaction::{CommitStatement, RollbackStatement, StartTransactionStatement};
+use crate::truncate::TruncateStatement;
 use crate::{
     AlterColumnOperation, AlterTableDefinition, AlterTableStatement, CacheInner, CaseWhenBranch,
     Column, ColumnConstraint, ColumnSpecification, CommentStatement, CommonTableExpr,
@@ -437,6 +438,13 @@ pub trait VisitorMut<'ast>: Sized {
         Ok(())
     }
 
+    fn visit_truncate_statement(
+        &mut self,
+        truncate_statement: &'ast mut TruncateStatement,
+    ) -> Result<(), Self::Error> {
+        walk_truncate_statement(self, truncate_statement)
+    }
+
     fn visit_sql_query(&mut self, sql_query: &'ast mut SqlQuery) -> Result<(), Self::Error> {
         walk_sql_query(self, sql_query)
     }
@@ -1143,6 +1151,17 @@ pub fn walk_drop_view_statement<'a, V: VisitorMut<'a>>(
     Ok(())
 }
 
+pub fn walk_truncate_statement<'a, V: VisitorMut<'a>>(
+    visitor: &mut V,
+    truncate_statement: &'a mut TruncateStatement,
+) -> Result<(), V::Error> {
+    for table in &mut truncate_statement.tables {
+        visitor.visit_table(&mut table.relation)?;
+    }
+
+    Ok(())
+}
+
 pub fn walk_sql_query<'a, V: VisitorMut<'a>>(
     visitor: &mut V,
     sql_query: &'a mut SqlQuery,
@@ -1176,6 +1195,7 @@ pub fn walk_sql_query<'a, V: VisitorMut<'a>>(
         SqlQuery::Explain(statement) => visitor.visit_explain_statement(statement),
         SqlQuery::Comment(statement) => visitor.visit_comment_statement(statement),
         SqlQuery::Deallocate(statement) => visitor.visit_deallocate_statement(statement),
+        SqlQuery::Truncate(statement) => visitor.visit_truncate_statement(statement),
     }
 }
 

nom-sql/src/lib.rs

@@ -60,6 +60,7 @@ pub use self::table::{
     TableExprInner,
 };
 pub use self::transaction::StartTransactionStatement;
+pub use self::truncate::TruncateStatement;
 pub use self::update::UpdateStatement;
 pub use self::use_statement::UseStatement;
 
@@ -96,6 +97,7 @@ mod sql_identifier;
 mod sql_type;
 mod table;
 mod transaction;
+mod truncate;
 mod update;
 mod use_statement;
 pub mod whitespace;

nom-sql/src/parser.rs

@@ -34,6 +34,7 @@ use crate::transaction::{
     commit, rollback, start_transaction, CommitStatement, RollbackStatement,
     StartTransactionStatement,
 };
+use crate::truncate::{truncate, TruncateStatement};
 use crate::update::{updating, UpdateStatement};
 use crate::use_statement::{use_statement, UseStatement};
 use crate::whitespace::whitespace0;
@@ -69,6 +70,7 @@ pub enum SqlQuery {
     Explain(ExplainStatement),
     Comment(CommentStatement),
     Deallocate(DeallocateStatement),
+    Truncate(TruncateStatement),
 }
 
 impl DialectDisplay for SqlQuery {
@@ -98,6 +100,7 @@ impl DialectDisplay for SqlQuery {
             Self::Comment(c) => write!(f, "{}", c.display(dialect)),
             Self::DropAllProxiedQueries(drop) => write!(f, "{}", drop.display(dialect)),
             Self::Deallocate(dealloc) => write!(f, "{}", dealloc.display(dialect)),
+            Self::Truncate(truncate) => write!(f, "{}", truncate.display(dialect)),
         })
     }
 }
@@ -147,6 +150,7 @@ impl SqlQuery {
             Self::Explain(_) => "EXPLAIN",
             Self::Comment(_) => "COMMENT",
             Self::Deallocate(_) => "DEALLOCATE",
+            Self::Truncate(_) => "TRUNCATE",
         }
     }
 
@@ -191,6 +195,7 @@ impl SqlQuery {
             | SqlQuery::Rollback(_)
             | SqlQuery::RenameTable(_)
             | SqlQuery::Use(_)
+            | SqlQuery::Truncate(_)
             | SqlQuery::Comment(_) => false,
         }
     }
@@ -244,6 +249,7 @@ fn sql_query_part2(
 ) -> impl Fn(LocatedSpan<&[u8]>) -> NomSqlResult<&[u8], SqlQuery> {
     move |i| {
         alt((
+            map(truncate(dialect), SqlQuery::Truncate),
             // This does a more expensive clone of `i`, so process it last.
             map(create_cached_query(dialect), SqlQuery::CreateCache),
             map(comment(dialect), SqlQuery::Comment),