Implement non-blocking SHOW READYSET TABLES with more statuses

This patch switches away from the old SHOW READYSET TABLES behavior,
which would run a blocking query on all domains for their tables'
statuses, to the new state-machine-backed SHOW READYSET TABLES behavior,
which responds immediately with the statuses of all tables by reporting
what's presently in the state machine.

The old way could block for several hours when compacting many large
tables, which was unfortunate if you were interested in seeing what the
current status was.  The blocking could also happen after creating a
cache, as the cache creation might create an index on our copy of the
table, which again holds up our single-threaded domains for potentially
long periods of time.

Previously, the reported statuses were only:

- "Snapshotting"
- "Snapshotted"
- "Not Replicated" (with reason in the description column)

Now, the reported statuses are:

- "Initializing"
- "Snapshotting" (with percentage in description column)
- "Compacting" (with percentage in description column)
- "Creating index" (with percentage in description column)
- "Online"
- "Not replicated" (with reason in the description column)

Instead of having to read logs to determine the progress of these
operations, you can now find the progress reported as percentages
displayed in the description column for the table.

Fixes: REA-2463, REA-2907, REA-4863, REA-5751
Release-Note-Core: Overhaul SHOW READYSET TABLES to always respond
  immediately, report more table statuses such as compaction and index
  creation, and report progress as a percentage in the description
  column during snapshot, compaction, and index creation.
Change-Id: I094cf7e0e9aa75fb65ea036abd94db2ed5240a08
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/10207
Tested-by: Buildkite CI
Reviewed-by: Michael Zink <michael.z@readyset.io>

Author: davisjc

readyset-adapter/src/backend/noria_connector.rs

@@ -937,27 +937,14 @@ impl NoriaConnector {
         };
         let data = statuses
             .into_iter()
-            .map(|(tbl, status)| {
-                let replication_status_str = status.to_string();
-                let replication_split = replication_status_str
-                    .splitn(2, ": ")
-                    .collect::<Vec<&str>>();
-                let (replication_status, description) =
-                    if replication_split[0].starts_with("Not Replicated") {
-                        (
-                            replication_split[0].to_string(),
-                            replication_split.get(1).unwrap_or(&"").to_string(),
-                        )
-                    } else {
-                        (status.to_string(), "".to_string())
-                    };
+            .map(|(table, status)| {
                 vec![
-                    tbl.display(self.parse_dialect).to_string().into(),
-                    replication_status.into(),
-                    description.into(),
+                    table.display(self.parse_dialect).to_string().into(),
+                    status.to_string().into(),
+                    status.description().into(),
                 ]
             })
-            .collect::<Vec<_>>();
+            .collect();
 
         Ok(QueryResult::from_owned(schema, vec![Results::new(data)]))
     }

readyset-client/src/controller.rs

@@ -37,8 +37,7 @@ use crate::status::ReadySetControllerStatus;
 use crate::table::{PersistencePoint, Table, TableBuilder, TableRpc};
 use crate::view::{View, ViewBuilder, ViewRpc};
 use crate::{
-    ReplicationOffset, SingleKeyEviction, TableReplicationStatus, TableStatus, ViewCreateRequest,
-    ViewFilter, ViewRequest,
+    ReplicationOffset, SingleKeyEviction, TableStatus, ViewCreateRequest, ViewFilter, ViewRequest,
 };
 
 mod rpc;
@@ -519,9 +518,9 @@ impl ReadySetHandle {
     }
 
     simple_request!(
-        /// Query the status of all known tables, including those not replicated by ReadySet if
+        /// Query the status of all known tables, including those not replicated by Readyset if
         /// the `all` parameter is set to `true`.
-        table_statuses(all: bool) -> BTreeMap<Relation, TableReplicationStatus>
+        table_statuses(all: bool) -> BTreeMap<Relation, TableStatus>
     );
 
     simple_request!(

readyset-client/src/lib.rs

@@ -354,8 +354,7 @@ pub use crate::controller::{
 };
 pub use crate::table::{
     Modification, Operation, PacketData, PacketPayload, PacketTrace, PersistencePoint, Table,
-    TableOperation, TableReplicationStatus, TableRequest, TableStatus,
-    TABLE_STATUS_REPORT_INTERVAL,
+    TableOperation, TableRequest, TableStatus, TABLE_STATUS_REPORT_INTERVAL,
 };
 pub use crate::view::{
     KeyComparison, LookupResult, ReadQuery, ReadReply, ReadReplyBatch, ReadReplyStats, SchemaType,

readyset-client/src/table.rs

@@ -272,30 +272,6 @@ impl fmt::Debug for PacketData {
     }
 }
 
-/// The status of a single table with respect to replication
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
-pub enum TableReplicationStatus {
-    /// The table is not being replicated by ReadySet
-    NotReplicated(NotReplicatedReason),
-    /// The table is currently being snapshotted by ReadySet
-    Snapshotting,
-    /// The table has been successfully snapshotted by ReadySet
-    Snapshotted,
-}
-
-impl Display for TableReplicationStatus {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match self {
-            TableReplicationStatus::NotReplicated(reason) => {
-                write!(f, "Not Replicated : {}", reason.description())?;
-            }
-            TableReplicationStatus::Snapshotting => f.write_str("Snapshotting")?,
-            TableReplicationStatus::Snapshotted => f.write_str("Snapshotted")?,
-        }
-        Ok(())
-    }
-}
-
 /// The status of a table.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum TableStatus {

readyset-server/src/controller/inner.rs

@@ -432,11 +432,8 @@ impl Leader {
             }
             (&Method::POST, "/table_statuses") => {
                 let all: bool = bincode::deserialize(&body)?;
-                let res = {
-                    let ds = self.dataflow_state_handle.read().await;
-                    ds.table_statuses(all).await?
-                };
-                return_serialized!(res)
+                let table_statuses = self.table_statuses.get(all).await;
+                return_serialized!(table_statuses)
             }
             (&Method::POST, "/set_table_status") => {
                 let statuses: HashMap<Relation, TableStatus> = bincode::deserialize(&body)?;

readyset-server/src/controller/mod.rs

@@ -1415,9 +1415,7 @@ mod tests {
     use dataflow::DomainIndex;
     use readyset_client::debug::info::KeyCount;
     use readyset_client::recipe::changelist::{Change, ChangeList};
-    use readyset_client::{
-        PersistencePoint, TableOperation, TableReplicationStatus, ViewCreateRequest,
-    };
+    use readyset_client::{PersistencePoint, TableOperation, TableStatus, ViewCreateRequest};
     use readyset_data::Dialect as DataDialect;
     use readyset_sql::ast::{NonReplicatedRelation, NotReplicatedReason, Relation};
     use readyset_sql::Dialect;
@@ -1829,14 +1827,14 @@ mod tests {
                         schema: Some("s2".into()),
                         name: "snapshotting_t".into(),
                     },
-                    TableReplicationStatus::Snapshotting
+                    TableStatus::Snapshotting(None)
                 ),
                 (
                     Relation {
                         schema: Some("s2".into()),
                         name: "snapshotted_t".into(),
                     },
-                    TableReplicationStatus::Snapshotted
+                    TableStatus::Online
                 ),
             ])
         );
@@ -1850,21 +1848,21 @@ mod tests {
                         schema: Some("s1".into()),
                         name: "t".into(),
                     },
-                    TableReplicationStatus::NotReplicated(NotReplicatedReason::Configuration)
+                    TableStatus::NotReplicated(NotReplicatedReason::Configuration)
                 ),
                 (
                     Relation {
                         schema: Some("s2".into()),
                         name: "snapshotting_t".into(),
                     },
-                    TableReplicationStatus::Snapshotting
+                    TableStatus::Snapshotting(None)
                 ),
                 (
                     Relation {
                         schema: Some("s2".into()),
                         name: "snapshotted_t".into(),
                     },
-                    TableReplicationStatus::Snapshotted
+                    TableStatus::Online
                 ),
             ])
         );

readyset-server/src/controller/state.rs

@@ -47,8 +47,8 @@ use readyset_client::query::QueryId;
 use readyset_client::recipe::changelist::{Change, ChangeList};
 use readyset_client::recipe::{CacheExpr, ExtendRecipeSpec};
 use readyset_client::{
-    PersistencePoint, SingleKeyEviction, TableReplicationStatus, TableStatus, ViewCreateRequest,
-    ViewFilter, ViewRequest, ViewSchema,
+    PersistencePoint, SingleKeyEviction, TableStatus, ViewCreateRequest, ViewFilter, ViewRequest,
+    ViewSchema,
 };
 use readyset_data::{DfValue, Dialect};
 use readyset_errors::{
@@ -1054,36 +1054,6 @@ impl DfState {
             .collect()
     }
 
-    /// Query the status of all known tables, including those not replicated by ReadySet if `all`
-    /// is set to `true`.
-    pub(super) async fn table_statuses(
-        &self,
-        all: bool,
-    ) -> ReadySetResult<BTreeMap<Relation, TableReplicationStatus>> {
-        let known_tables = self.tables();
-        let snapshotting_tables = self.snapshotting_tables().await?;
-        let out = known_tables.into_keys().map(|tbl| {
-            let status = if snapshotting_tables.contains(&tbl) {
-                TableReplicationStatus::Snapshotting
-            } else {
-                TableReplicationStatus::Snapshotted
-            };
-            (tbl, status)
-        });
-        if all {
-            Ok(out
-                .chain(
-                    self.non_replicated_relations()
-                        .iter()
-                        .cloned()
-                        .map(|tbl| (tbl.name, TableReplicationStatus::NotReplicated(tbl.reason))),
-                )
-                .collect())
-        } else {
-            Ok(out.collect())
-        }
-    }
-
     /// Returns a list of all table names that are currently involved in snapshotting.
     pub(super) async fn snapshotting_tables(&self) -> ReadySetResult<HashSet<Relation>> {
         let domains = self.domains_with_base_tables().await?;

readyset-server/src/controller/table_status.rs

@@ -103,6 +103,30 @@ impl TableStatusState {
         *self.state.write().await = Some(state);
     }
 
+    /// Get all known table statuses.
+    pub async fn get(&self, all: bool) -> BTreeMap<Relation, TableStatus> {
+        let state = self.state.read().await;
+        let state = match state.as_ref() {
+            Some(state) => state,
+            None => return BTreeMap::new(),
+        };
+
+        if all {
+            state.clone()
+        } else {
+            state
+                .iter()
+                .filter_map(|(table, status)| {
+                    if matches!(status, TableStatus::NotReplicated(..)) {
+                        None
+                    } else {
+                        Some((table.clone(), status.clone()))
+                    }
+                })
+                .collect()
+        }
+    }
+
     /// Update the statuses for multiple tables.
     ///
     /// NOTE: Only call this if you know what you're doing.  Most users should instead feed updates