replicators: Automatically stop replicating CREATE TABLE LIKE tables

When a table is created using the `CREATE TABLE LIKE` syntax, the
replicator will automatically add it to the set of non-replicated
relations.

Fixes: REA-6164
AI-Use: level:1
Change-Id: Ib50ffbd468947f0a65e9620af7edc8976a6a6964
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/10956
Tested-by: Buildkite CI
Reviewed-by: Jason Brown <jason.b@readyset.io>

Author: mvzink

readyset-sql/src/ast/create.rs

@@ -608,6 +608,15 @@ pub enum CreateTableLike {
     Parenthesized(Relation),
 }
 
+impl CreateTableLike {
+    pub fn as_relation_mut(&mut self) -> &mut Relation {
+        match self {
+            CreateTableLike::Plain(relation) => relation,
+            CreateTableLike::Parenthesized(relation) => relation,
+        }
+    }
+}
+
 impl TryFromDialect<sqlparser::ast::CreateTableLikeKind> for CreateTableLike {
     fn try_from_dialect(
         value: sqlparser::ast::CreateTableLikeKind,

replicators/src/mysql_connector/connector.rs

@@ -31,7 +31,7 @@ use tracing::{error, info, warn};
 
 use database_utils::UpstreamConfig;
 use readyset_client::metrics::recorded;
-use readyset_client::recipe::changelist::Change;
+use readyset_client::recipe::changelist::{Change, IntoChanges as _};
 use readyset_client::recipe::ChangeList;
 use readyset_client::{Modification, ReadySetHandle, TableOperation};
 use readyset_data::{Collation as RsCollation, DfValue, Dialect, TimestampTz};
@@ -953,7 +953,9 @@ impl MySqlBinlogConnector {
     /// `COMMIT` queries are issued for writes on non-transactional storage engines such as MyISAM.
     /// We report the position after the `COMMIT` query if necessary.
     ///
-    /// TRUNCATE statements are also parsed and handled here.
+    /// `TRUNCATE` statements and `CREATE TABLE new_table LIKE existing_table` statements are also
+    /// parsed and handled here, because we fall into this path when the binlog event doesn't have
+    /// the [`binlog::consts::StatusVarKey::UpdatedDbNames`] status variable.
     ///
     /// TODO: Transactions begin with `BEGIN` queries, but we do not currently support those.
     fn try_non_ddl_action_from_query(
@@ -971,13 +973,33 @@ impl MySqlBinlogConnector {
                 // MySQL only allows one table in the statement, or we would be in trouble.
                 let mut relation = truncate.tables[0].relation.clone();
                 if relation.schema.is_none() {
-                    relation.schema = Some(SqlIdentifier::from(q_event.schema()))
+                    relation.schema = Some(q_event.schema().into())
                 }
                 Ok(ReplicationAction::TableAction {
                     table: relation,
                     actions: vec![TableOperation::Truncate],
                 })
             }
+            Ok(SqlQuery::CreateTable(mut create)) => {
+                if let Some(like) = &mut create.like {
+                    let relation = like.as_relation_mut();
+                    if relation.schema.is_none() {
+                        relation.schema = Some(q_event.schema().into())
+                    }
+                } else {
+                    return Err(ReadySetError::ReplicationFailed(format!(
+                        "Unexpected CREATE TABLE without LIKE should have been handled earlier: {}",
+                        create.table.display_unquoted()
+                    )));
+                }
+                if create.table.schema.is_none() {
+                    create.table.schema = Some(q_event.schema().into())
+                }
+                Ok(ReplicationAction::DdlChange {
+                    schema: q_event.schema().into(),
+                    changes: create.into_changes(),
+                })
+            }
             _ => Err(ReadySetError::SkipEvent),
         }
     }

replicators/src/noria_adapter.rs

@@ -786,12 +786,13 @@ impl<'a> NoriaAdapter<'a> {
                 let keep = self
                     .table_filter
                     .should_be_processed(schema.as_str(), statement.table.name.as_str())
-                    && statement.body.is_ok();
+                    && statement.body.is_ok()
+                    && statement.like.is_none();
                 if !keep {
                     non_replicated_tables.push(Relation {
                         schema: Some(schema.clone().into()),
                         name: statement.table.name.clone(),
-                    })
+                    });
                 }
                 keep
             }
@@ -820,14 +821,18 @@ impl<'a> NoriaAdapter<'a> {
         });
 
         // Mark all tables that were filtered as non-replicated, too
-        changelist
-            .changes_mut()
-            .extend(non_replicated_tables.into_iter().map(|relation| {
-                Change::AddNonReplicatedRelation(NonReplicatedRelation {
+        for relation in non_replicated_tables {
+            if let Some(schema) = &relation.schema {
+                self.table_filter
+                    .deny_replication(schema, relation.name.as_str());
+            }
+            changelist
+                .changes_mut()
+                .push(Change::AddNonReplicatedRelation(NonReplicatedRelation {
                     name: relation,
                     reason: NotReplicatedReason::Configuration,
-                })
-            }));
+                }));
+        }
 
         // Resolve schemas and drop renamed tables; the new names will get snapshotted when we
         // return `ResnapshotNeeded` in the next `if` block

replicators/tests/tests.rs

@@ -1128,6 +1128,94 @@ async fn mysql_rename_swap_with_ignore() {
     shutdown_tx.shutdown().await;
 }
 
+#[tokio::test(flavor = "multi_thread")]
+#[tags(serial, slow, mysql_upstream)]
+async fn mysql_create_table_like_with_ignored() {
+    readyset_tracing::init_test_logging();
+    let url = &mysql_url();
+    let mut client = DbConnection::connect(url).await.unwrap();
+
+    // Create the initial table and populate it
+    client
+        .query(
+            "
+            DROP TABLE IF EXISTS original_table CASCADE;
+            DROP TABLE IF EXISTS like_table CASCADE;
+            CREATE TABLE original_table (id INT PRIMARY KEY, name VARCHAR(50));
+            INSERT INTO original_table VALUES (1, 'Alice'), (2, 'Bob'), (3, 'Charlie');
+            ",
+        )
+        .await
+        .unwrap();
+
+    // Start replication to snapshot the original table
+    let (mut ctx, shutdown_tx) = TestHandle::start_noria(
+        url.to_string(),
+        None, // No special config - replicate all tables
+    )
+    .await
+    .unwrap();
+
+    ctx.controller_rx
+        .as_mut()
+        .unwrap()
+        .snapshot_completed()
+        .await
+        .unwrap();
+
+    // Verify that original_table is replicated
+    check_results!(
+        ctx,
+        "original_table",
+        "create_table_like_before",
+        &[
+            &[DfValue::Int(1), tiny(b"Alice")],
+            &[DfValue::Int(2), tiny(b"Bob")],
+            &[DfValue::Int(3), tiny(b"Charlie")]
+        ]
+    );
+
+    // Create a new table using CREATE TABLE ... LIKE
+    client
+        .query("CREATE TABLE like_table LIKE original_table")
+        .await
+        .unwrap();
+
+    // Give the replicator a moment to process the DDL change
+    sleep(Duration::from_secs(2)).await;
+
+    // Verify that like_table is NOT replicated (added to ignored tables)
+    ctx.assert_table_missing("public", "like_table").await;
+
+    // Verify that original_table is still replicated and working
+    check_results!(
+        ctx,
+        "original_table",
+        "create_table_like_after",
+        &[
+            &[DfValue::Int(1), tiny(b"Alice")],
+            &[DfValue::Int(2), tiny(b"Bob")],
+            &[DfValue::Int(3), tiny(b"Charlie")]
+        ]
+    );
+
+    assert!(!ctx.noria.tables().await.unwrap().contains_key(&Relation {
+        schema: Some("public".into()),
+        name: "like_table".into(),
+    }));
+
+    let non_replicated_rels = ctx.noria.non_replicated_relations().await.unwrap();
+    let relation = Relation {
+        schema: Some("public".into()),
+        name: "like_table".into(),
+    };
+    assert!(non_replicated_rels.contains(&NonReplicatedRelation::new(relation.clone()),));
+
+    ctx.stop().await;
+    client.stop().await;
+    shutdown_tx.shutdown().await;
+}
+
 #[tokio::test(flavor = "multi_thread")]
 #[tags(serial, slow, mysql_upstream)]
 async fn mysql_replication_resnapshot() {