adapter: Fix views synchronizer bugs

readyset-adapter/Cargo.toml

@@ -51,6 +51,7 @@ quanta = { version = "0.11", default-features = false }
 lru = "0.12.0"
 crossbeam-skiplist = "0.1.1"
 slab = "0.4"
+xxhash-rust = { version = "0.8.10", features = ["xxh3"] }
 
 readyset-adapter-types = { path = "../readyset-adapter-types/" }
 readyset-alloc = { path = "../readyset-alloc/" }
@@ -81,6 +82,10 @@ path = "src/lib.rs"
 name = "parse"
 harness = false
 
+[[bench]]
+name = "hash"
+harness = false
+
 [features]
 ryw = []
 failure_injection = ["fail/failpoints"]

readyset-adapter/src/migration_handler.rs

@@ -344,8 +344,18 @@ impl MigrationHandler {
         match controller.dry_run(changelist).await {
             Ok(_) => {
                 self.start_time.remove(view_request);
+
+                // It's possible that the ViewsSynchronizer found an existing view for this query
+                // on the server while we were performing a dry run, in which case it would have
+                // updated the query's status to "successful". In this situation, we don't want to
+                // overwrite the "successful" status, so we only write the new "dry run succeeded"
+                // status if the query's status is still "pending"
                 self.query_status_cache
-                    .update_query_migration_state(view_request, MigrationState::DryRunSucceeded);
+                    .with_mut_migration_state(view_request, |status| {
+                        if status.is_pending() {
+                            *status = MigrationState::DryRunSucceeded;
+                        }
+                    });
             }
             Err(e) if e.caused_by_unsupported() => {
                 self.start_time.remove(view_request);

readyset-adapter/src/query_status_cache.rs

@@ -560,6 +560,24 @@ impl QueryStatusCache {
         }
     }
 
+    /// Yields to the given function `f` a mutable reference to the migration state of the query
+    /// `q`. The primary purpose of this method is allow for atomic reads and writes of the
+    /// migration state of a query.
+    pub fn with_mut_migration_state<Q, F>(&self, q: &Q, f: F) -> bool
+    where
+        Q: QueryStatusKey,
+        F: Fn(&mut MigrationState),
+    {
+        q.with_mut_status(self, |maybe_query_status| {
+            if let Some(query_status) = maybe_query_status {
+                f(&mut query_status.migration_state);
+                true
+            } else {
+                false
+            }
+        })
+    }
+
     /// Marks a query as dropped by the user.
     ///
     /// NOTE: this should only be called after we successfully remove a View for this query. This is
@@ -785,6 +803,22 @@ impl QueryStatusCache {
             .into()
     }
 
+    /// Returns a list of queries whose migration states match `states`.
+    pub fn queries_with_statuses(&self, states: &[MigrationState]) -> QueryList {
+        let statuses = self.persistent_handle.statuses.read();
+        statuses
+            .iter()
+            .filter_map(|(_query_id, (query, status))| {
+                if states.contains(&status.migration_state) {
+                    Some((query.clone(), status.clone()))
+                } else {
+                    None
+                }
+            })
+            .collect::<Vec<(Query, QueryStatus)>>()
+            .into()
+    }
+
     /// Returns a list of queries that have a state of [`QueryState::Successful`].
     pub fn allow_list(&self) -> Vec<(QueryId, Arc<ViewCreateRequest>, QueryStatus)> {
         self.persistent_handle.allow_list()

readyset-adapter/src/views_synchronizer.rs

@@ -1,3 +1,4 @@
+use std::collections::HashSet;
 use std::sync::Arc;
 
 use dataflow_expression::Dialect;
@@ -8,6 +9,7 @@ use readyset_util::shared_cache::LocalCache;
 use readyset_util::shutdown::ShutdownReceiver;
 use tokio::select;
 use tracing::{debug, info, instrument, trace, warn};
+use xxhash_rust::xxh3;
 
 use crate::query_status_cache::QueryStatusCache;
 
@@ -22,6 +24,13 @@ pub struct ViewsSynchronizer {
     dialect: Dialect,
     /// Global and thread-local cache of view endpoints and prepared statements.
     view_name_cache: LocalCache<ViewCreateRequest, Relation>,
+    /// A cache to keep track of the queries for which we've already checked the server for
+    /// existing views. Note that this cache is *not* updated (i.e. a query is not removed) when a
+    /// "dry run succeeded" query is migrated.
+    ///
+    /// This HashSet stores 128-bit hashes computed via xxHash in an attempt to minimize the amount
+    /// of data we need to store to keep track of the queries we've already seen.
+    views_checked: HashSet<u128>,
 }
 
 impl ViewsSynchronizer {
@@ -38,6 +47,7 @@ impl ViewsSynchronizer {
             poll_interval,
             dialect,
             view_name_cache,
+            views_checked: HashSet::new(),
         }
     }
 
@@ -69,24 +79,34 @@ impl ViewsSynchronizer {
 
     async fn poll(&mut self) {
         debug!("Views synchronizer polling");
-        let queries = self
+        let (queries, hashes): (Vec<_>, Vec<_>) = self
             .query_status_cache
-            .pending_migration()
+            .queries_with_statuses(&[MigrationState::DryRunSucceeded, MigrationState::Pending])
             .into_iter()
             .filter_map(|(q, _)| {
-                q.into_parsed()
-                    // once arc_unwrap_or_clone is stabilized, we can use that cleaner syntax
-                    .map(|p| Arc::try_unwrap(p).unwrap_or_else(|arc| (*arc).clone()))
+                q.into_parsed().and_then(|p| {
+                    let hash = xxh3::xxh3_128(&bincode::serialize(&*p).unwrap());
+
+                    if self.views_checked.contains(&hash) {
+                        // once arc_unwrap_or_clone is stabilized, we can use that cleaner syntax
+                        Some((
+                            Arc::try_unwrap(p).unwrap_or_else(|arc| (*arc).clone()),
+                            hash,
+                        ))
+                    } else {
+                        None
+                    }
+                })
             })
-            .collect::<Vec<_>>();
+            .unzip();
 
         match self
             .controller
             .view_names(queries.clone(), self.dialect)
             .await
         {
             Ok(statuses) => {
-                for (query, name) in queries.into_iter().zip(statuses) {
+                for ((query, name), hash) in queries.into_iter().zip(statuses).zip(hashes) {
                     trace!(
                         // FIXME(REA-2168): Use correct dialect.
                         query = %query.statement.display(nom_sql::Dialect::MySQL),
@@ -96,7 +116,8 @@ impl ViewsSynchronizer {
                     if let Some(name) = name {
                         self.view_name_cache.insert(query.clone(), name).await;
                         self.query_status_cache
-                            .update_query_migration_state(&query, MigrationState::Successful)
+                            .update_query_migration_state(&query, MigrationState::Successful);
+                        self.views_checked.insert(hash);
                     }
                 }
             }

readyset-clustertest/src/readyset_mysql.rs

@@ -1599,7 +1599,7 @@ async fn views_synchronize_between_deployments() {
 
     // Eventually it should show up in adapter 1 too
     eventually! {
-        adapter_1.as_mysql_conn().unwrap().query_drop("SELECT * FROM t1;");
+        adapter_1.as_mysql_conn().unwrap().query_drop("SELECT * FROM t1;").await.unwrap();
         last_statement_destination(adapter_1.as_mysql_conn().unwrap()).await == QueryDestination::Readyset
     }