controller: remove_query return whether any expressions were removed

Modify the remove_query RPC to return 1 if one or more expressions were
removed, and 0 otherwise, and use this value to populate a DELETE result
from DROP CACHE. This feedback is useful e.g. if the user tries to drop
a non-existent cache name, or if the cache was already dropped
elsewhere.

Later, we may want to return the number of expressions (and aliases?)
removed.

Release-Note-Core: `DROP CACHE` now indicates whether a cache was
  removed via a DELETE result with a row count of either 1 or 0.

Change-Id: I620564a4146c72a4683e57ccd917ff45a651090f
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/5770
Tested-by: Buildkite CI
Reviewed-by: Dan Wilbanks <dan@readyset.io>

readyset-adapter/src/backend.rs

@@ -1673,7 +1673,7 @@ where
         name: &Relation,
     ) -> ReadySetResult<noria_connector::QueryResult<'static>> {
         let maybe_view_request = self.noria.view_create_request_from_name(name).await;
-        self.noria.drop_view(name).await?;
+        let result = self.noria.drop_view(name).await?;
         if let Some(view_request) = maybe_view_request {
             // drop_query() should not be called if we have no view for this query.
             self.state.query_status_cache.drop_query(&view_request);
@@ -1682,7 +1682,9 @@ where
                 .always_attempt_readyset(&view_request, false);
             self.invalidate_prepared_statements_cache(&view_request);
         }
-        Ok(noria_connector::QueryResult::Empty)
+        Ok(noria_connector::QueryResult::Delete {
+            num_rows_deleted: result,
+        })
     }
 
     /// Forwards a `DROP ALL CACHES` request to noria

readyset-adapter/src/backend/noria_connector.rs

@@ -1136,13 +1136,13 @@ impl NoriaConnector {
 
     /// Make a request to ReadySet to drop the query with the given name, and remove it from all
     /// internal state.
-    pub async fn drop_view(&mut self, name: &Relation) -> ReadySetResult<()> {
-        noria_await!(
+    pub async fn drop_view(&mut self, name: &Relation) -> ReadySetResult<u64> {
+        let result = noria_await!(
             self.inner.get_mut()?,
             self.inner.get_mut()?.noria.remove_query(name)
         )?;
         self.view_cache.remove_statement(name).await;
-        Ok(())
+        Ok(result)
     }
 
     /// Make a request to ReadySet to drop all cached queries, and empty all internal state

readyset-client/src/controller.rs

@@ -740,7 +740,7 @@ impl ReadySetHandle {
     pub fn remove_query(
         &mut self,
         name: &Relation,
-    ) -> impl Future<Output = ReadySetResult<()>> + '_ {
+    ) -> impl Future<Output = ReadySetResult<u64>> + '_ {
         self.rpc("remove_query", name, self.migration_timeout)
     }
 

readyset-server/src/controller/inner.rs

@@ -555,9 +555,9 @@ impl Leader {
                 require_leader_ready()?;
                 let query_name = bincode::deserialize(&body)?;
                 let mut writer = self.dataflow_state_handle.write().await;
-                writer.as_mut().remove_query(&query_name).await?;
+                let result = writer.as_mut().remove_query(&query_name).await?;
                 self.dataflow_state_handle.commit(writer, authority).await?;
-                return_serialized!(());
+                return_serialized!(result);
             }
             (&Method::POST, "/remove_all_queries") => {
                 require_leader_ready()?;

readyset-server/src/controller/state.rs

@@ -1470,9 +1470,11 @@ impl DfState {
         }
     }
 
-    pub(super) async fn remove_query(&mut self, query_name: &Relation) -> ReadySetResult<()> {
+    /// Return 1 if one or more expressions were removed, else return 0.
+    /// Someday we may want to return # expressions (and aliases?) dropped.
+    pub(super) async fn remove_query(&mut self, query_name: &Relation) -> ReadySetResult<u64> {
         let name = match self.recipe.resolve_alias(query_name) {
-            None => return Ok(()),
+            None => return Ok(0),
             Some(name) => name,
         };
 
@@ -1489,7 +1491,7 @@ impl DfState {
             return Err(error);
         }
 
-        Ok(())
+        Ok(1)
     }
 
     pub(super) async fn remove_all_queries(&mut self) -> ReadySetResult<()> {