Drop MySQL tables that are not part of replication filters

This commit add a functionality to drop tables that have being
removed from replication filters. This happens if a table is either
removed from --replication-tables or added to
--replication-tables-ignore.

Release-Note-Core: Added functionality to remove leftover tables that
  have being excluded from replication filters.

Fixes: REA-4379
Closes: #1397
Change-Id: I8bc0035c5a1d3bd7247b3448c6a29806fb7aa8bb
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/8367
Tested-by: Buildkite CI
Reviewed-by: Michael Zink <michael.z@readyset.io>

Author: altmannmarcelo

replicators/src/mysql_connector/snapshot.rs

@@ -32,6 +32,7 @@ use crate::db_util::DatabaseSchemas;
 use crate::mysql_connector::snapshot_type::SnapshotType;
 use crate::table_filter::TableFilter;
 use crate::TablesSnapshottingGaugeGuard;
+use std::collections::HashSet;
 
 const RS_BATCH_SIZE: usize = 1000; // How many queries to buffer before pushing to ReadySet
 
@@ -178,7 +179,7 @@ impl<'a> MySqlReplicator<'a> {
                 self.table_filter
                     .should_be_processed(schema.as_str(), table.as_str())
             });
-
+        self.drop_leftover_tables(noria, &replicated_tables).await?;
         noria
             .extend_recipe_no_leader_ready(ChangeList::from_changes(
                 non_replicated_tables
@@ -749,6 +750,48 @@ impl<'a> MySqlReplicator<'a> {
             .await?;
         Ok(())
     }
+
+    /// This functions drops all tables that we have internally and have not been selected as
+    /// part of snapshot, normally because table filters have been changed.
+    ///
+    /// # Arguments
+    ///
+    /// * `noria`: The target ReadySet deployment
+    /// * `snapshot_tables`: The list of tables that are part of the snapshot
+    async fn drop_leftover_tables(
+        &mut self,
+        noria: &mut readyset_client::ReadySetHandle,
+        snapshot_tables: &[(String, String)],
+    ) -> ReadySetResult<()> {
+        let mut changes = Vec::new();
+        let current_tables = noria.tables().await?;
+        let snapshot_tables_set: HashSet<Relation> = snapshot_tables
+            .iter()
+            .map(|(schema, name)| Relation {
+                schema: Some(schema.clone().into()),
+                name: name.clone().into(),
+            })
+            .collect();
+
+        for table in current_tables.keys() {
+            if !snapshot_tables_set.contains(table) {
+                warn!(table = %table.display(nom_sql::Dialect::MySQL), "Dropping table that is not part of snapshot");
+                changes.push(Change::Drop {
+                    name: table.clone(),
+                    if_exists: false,
+                });
+            }
+        }
+        if !changes.is_empty() {
+            noria
+                .extend_recipe_no_leader_ready(ChangeList::from_changes(
+                    changes,
+                    Dialect::DEFAULT_MYSQL,
+                ))
+                .await?;
+        }
+        Ok(())
+    }
 }
 
 /// Convert each entry in a row to a ReadySet type that can be inserted into the base tables