adapter: WIP, fix some out-of-order bugs in QueryTracker

Author: aljoscha

src/adapter/src/coord/peek.rs

@@ -54,8 +54,8 @@ use uuid::Uuid;
 
 use crate::active_compute_sink::{ActiveComputeSink, ActiveCopyTo};
 use crate::coord::timestamp_selection::TimestampDetermination;
-use crate::query_tracker::QueryTrackerCmd;
 use crate::optimize::OptimizerError;
+use crate::query_tracker::QueryTrackerCmd;
 use crate::statement_logging::WatchSetCreation;
 use crate::statement_logging::{StatementEndedExecutionReason, StatementExecutionStrategy};
 use crate::util::ResultExt;
@@ -841,6 +841,7 @@ impl crate::coord::Coordinator {
             mz_compute_types::dyncfgs::PEEK_RESPONSE_STASH_READ_MEMORY_BUDGET_BYTES
                 .get(self.catalog().system_config().dyncfgs());
 
+        let cluster_name = self.catalog().get_cluster(compute_instance).name.clone();
         let peek_response_stream = Self::create_peek_response_stream(
             rows_rx,
             finishing,
@@ -850,6 +851,10 @@ impl crate::coord::Coordinator {
             persist_client,
             peek_stash_read_batch_size_bytes,
             peek_stash_read_memory_budget_bytes,
+            Some(format!(
+                "query could not complete because cluster \"{}\" was dropped",
+                cluster_name
+            )),
         );
 
         Ok(crate::ExecuteResponse::SendingRowsStreaming {
@@ -872,14 +877,19 @@ impl crate::coord::Coordinator {
         mut persist_client: mz_persist_client::PersistClient,
         peek_stash_read_batch_size_bytes: usize,
         peek_stash_read_memory_budget_bytes: usize,
+        rows_rx_err: Option<String>,
     ) -> impl futures::Stream<Item = PeekResponseUnary> {
         async_stream::stream!({
             let result = rows_rx.await;
 
             let rows = match result {
                 Ok(rows) => rows,
                 Err(e) => {
-                    yield PeekResponseUnary::Error(e.to_string());
+                    if let Some(msg) = rows_rx_err {
+                        yield PeekResponseUnary::Error(msg);
+                    } else {
+                        yield PeekResponseUnary::Error(e.to_string());
+                    }
                     return;
                 }
             };

src/adapter/src/peek_client.rs

@@ -32,12 +32,12 @@ use crate::catalog::Catalog;
 use crate::command::{CatalogSnapshot, Command};
 use crate::coord::Coordinator;
 use crate::coord::peek::FastPathPlan;
+use crate::query_tracker::QueryTrackerCmd;
 use crate::statement_logging::WatchSetCreation;
 use crate::statement_logging::{
     FrontendStatementLoggingEvent, PreparedStatementEvent, StatementLoggingFrontend,
     StatementLoggingId,
 };
-use crate::query_tracker::QueryTrackerCmd;
 use crate::{AdapterError, Client, CollectionIdBundle, ReadHolds, statement_logging};
 
 /// Storage collections trait alias we need to consult for since/frontiers.
@@ -390,7 +390,8 @@ impl PeekClient {
         if let Err(err) = peek_result {
             // Clean up the registered peek since the peek failed to issue.
             // The frontend will handle statement logging for the error.
-            self.query_tracker.send(QueryTrackerCmd::UntrackPeek { uuid });
+            self.query_tracker
+                .send(QueryTrackerCmd::UntrackPeek { uuid });
             return Err(AdapterError::concurrent_dependency_drop_from_peek_error(
                 err,
                 compute_instance,
@@ -406,6 +407,7 @@ impl PeekClient {
             self.persist_client.clone(),
             peek_stash_read_batch_size_bytes,
             peek_stash_read_memory_budget_bytes,
+            None,
         );
 
         Ok(crate::ExecuteResponse::SendingRowsStreaming {