readyset-adapter: Improve duplicate cache creation handling

Previously, before creating a named cache, we would first drop any
existing cache with that same name before creation.  In Iad22f58, we
added a similar check for unnamed caches, comparing query_id to first
drop a duplicate before creation.

It's important to first drop a duplicate cache to prevent the graph
metadata in SqlToMirConverter from erroneously being updated with node
index values that are then never persisted to the graph itself, which
can happen if we decide to abandon the in-progress graph creation to
instead reuse the previous cache.

Unfortunately, we could still run into this inconsistency between
SqlToMirConverter and the graph when creating duplicate caches with
different names.

Now, instead of these targeted checks that made decisions based on what
was in the adapter's view name cache, we instead query the controller
for any existing caches that match either the same query_id or requested
name.  If any such cache exists, drop it before creation.

Fixes: REA-5765
Release-Note-Core: Fix a bug that led to errors when attempting to
  create duplicate caches.
Change-Id: I4e5aa25d19e1ea4488c986ee65f4a85ec8a801a5
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/9655
Tested-by: Buildkite CI
Reviewed-by: Jason Brown <jason.b@readyset.io>

Author: davisjc

readyset-adapter/src/backend.rs

@@ -84,6 +84,7 @@ use mysql_common::row::convert::{FromRow, FromRowError};
 use readyset_adapter_types::{DeallocateId, ParsedCommand, PreparedStatementType};
 use readyset_client::consensus::{Authority, AuthorityControl, CacheDDLRequest};
 use readyset_client::query::*;
+use readyset_client::recipe::CacheExpr;
 use readyset_client::results::Results;
 use readyset_client::{ColumnSchema, PlaceholderIdx, ViewCreateRequest};
 pub use readyset_client_metrics::QueryDestination;
@@ -1922,22 +1923,36 @@ where
         concurrently: bool,
     ) -> ReadySetResult<noria_connector::QueryResult<'static>> {
         adapter_rewrites::process_query(&mut stmt, self.noria.rewrite_params())?;
+        let query_id = QueryId::from_select(&stmt, self.noria.schema_search_path());
+        let requested_name = name.clone();
         let name = match name {
             Some(name) => &*name,
             None => {
-                *name = Some(QueryId::from_select(&stmt, self.noria.schema_search_path()).into());
+                *name = Some(query_id.into());
                 name.as_ref().unwrap()
             }
         };
-        // If we have another query with the same name, drop that query first
-        if let Some(view_request) = self.noria.view_create_request_from_name(name).await {
+
+        // If we have existing caches with the same query_id or name, drop them first.
+        for CacheExpr {
+            name,
+            statement,
+            query_id,
+            ..
+        } in self
+            .noria
+            .verbose_views(Some(query_id), requested_name.as_ref())
+            .await?
+        {
             warn!(
-                statement = %Sensitive(&view_request.statement.display(self.settings.dialect)),
-                name = %name.display(readyset_sql::Dialect::MySQL),
+                query_id = %query_id,
+                name = %name.display(DB::SQL_DIALECT),
+                statement = %Sensitive(&statement.display(self.settings.dialect)),
                 "Dropping previously cached query",
             );
-            self.drop_cached_query(name).await?;
+            self.drop_cached_query(&name).await?;
         }
+
         // Now migrate the new query
         let migration_state = match self
             .noria

readyset-mysql/tests/integration.rs

@@ -3005,7 +3005,8 @@ async fn test_default_value_not_null_for_replication() {
 }
 
 #[tokio::test(flavor = "multi_thread")]
-async fn create_duplicate_caches() {
+#[tags(serial)]
+async fn create_duplicate_unnamed_caches() {
     let (opts, _handle, shutdown_tx) = setup().await;
     let mut conn = Conn::new(opts).await.unwrap();
 
@@ -3030,6 +3031,61 @@ async fn create_duplicate_caches() {
     shutdown_tx.shutdown().await;
 }
 
+#[tokio::test(flavor = "multi_thread")]
+#[tags(serial)]
+async fn create_duplicate_named_caches() {
+    let (opts, _handle, shutdown_tx) = setup().await;
+    let mut conn = Conn::new(opts).await.unwrap();
+
+    conn.query_drop("CREATE TABLE foo (a INT, b INT)")
+        .await
+        .unwrap();
+
+    conn.query_drop("CREATE CACHE name1 FROM SELECT b FROM foo WHERE a = ?")
+        .await
+        .unwrap();
+    conn.query_drop("CREATE CACHE name2 FROM SELECT b FROM foo WHERE a IN (?, ?, ?)")
+        .await
+        .unwrap();
+    conn.query_drop("CREATE CACHE name2 FROM SELECT b FROM foo WHERE a IN (?, ?, ?)")
+        .await
+        .unwrap();
+
+    let caches: Vec<(String, String, String, String)> = conn.query("SHOW CACHES").await.unwrap();
+    assert_eq!(caches.len(), 1, "unexpected caches: {caches:?}");
+
+    shutdown_tx.shutdown().await;
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[tags(serial)]
+async fn create_duplicate_query_id_and_name_caches() {
+    let (opts, _handle, shutdown_tx) = setup().await;
+    let mut conn = Conn::new(opts).await.unwrap();
+
+    conn.query_drop("CREATE TABLE foo (a INT, b INT)")
+        .await
+        .unwrap();
+
+    conn.query_drop("CREATE CACHE name1 FROM SELECT b FROM foo")
+        .await
+        .unwrap();
+    conn.query_drop("CREATE CACHE name2 FROM SELECT b FROM foo WHERE a = ?")
+        .await
+        .unwrap();
+    conn.query_drop("CREATE CACHE name1 FROM SELECT b FROM foo WHERE a IN (?, ?, ?)")
+        .await
+        .unwrap();
+    conn.query_drop("CREATE CACHE name1 FROM SELECT b FROM foo WHERE a IN (?, ?, ?)")
+        .await
+        .unwrap();
+
+    let caches: Vec<(String, String, String, String)> = conn.query("SHOW CACHES").await.unwrap();
+    assert_eq!(caches.len(), 1, "unexpected caches: {caches:?}");
+
+    shutdown_tx.shutdown().await;
+}
+
 #[tokio::test(flavor = "multi_thread")]
 #[tags(serial, mysql8_upstream)]
 #[ignore]