readyset-adapter: SHOW READYSET STATUS shows enabled feature flags

Add a row to the output of SHOW READYSET STATUS that lists
the enabled core dataflow feature flags.

Release-Note-Core: `SHOW READYSET STATUS` now lists the enabled
  feature flags like post-lookups, topk, and straddled-joins.

Fixes: REA-4766
Change-Id: I12de7e9ba6b11f2fba5b3a606921a45b78446b18
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/10335
Reviewed-by: Johnathan Davis <jcd@readyset.io>
Tested-by: Buildkite CI

Author: MohamedAbdeen-rs

readyset-adapter/src/status_reporter.rs

@@ -22,6 +22,7 @@ pub struct ReadySetStatus {
     pub upstream_reachable: Option<bool>,
     pub connection_count: usize,
     pub persistent_stats: Option<PersistentStats>,
+    pub enabled_features: Vec<String>,
 }
 
 impl ReadySetStatus {
@@ -73,6 +74,15 @@ impl ReadySetStatus {
             }
         }
 
+        if self.enabled_features.is_empty() {
+            status.push(("Enabled Features".to_string(), "None".to_string()));
+        } else {
+            status.push((
+                "Enabled Features".to_string(),
+                self.enabled_features.join(", "),
+            ));
+        }
+
         QueryResult::MetaVariables(status.into_iter().map(MetaVariable::from).collect())
     }
 }
@@ -110,12 +120,14 @@ where
         rs_handle: Option<ReadySetHandle>,
         connections: Arc<SkipSet<SocketAddr>>,
         authority: Arc<Authority>,
+        enabled_features: Vec<String>,
     ) -> Self {
         let inner = Arc::new(Mutex::new(ReadySetStatusReporterInner {
             upstream: upstream_config.into(),
             rs_handle,
             connections,
             authority,
+            enabled_features,
         }));
         Self { inner }
     }
@@ -138,6 +150,8 @@ struct ReadySetStatusReporterInner<U> {
     pub(crate) connections: Arc<SkipSet<SocketAddr>>,
     /// A shared handle to the Authority, used for reading persistent_stats for /readyset_status
     pub(crate) authority: Arc<Authority>,
+    /// Enabled features to display in status
+    pub(crate) enabled_features: Vec<String>,
 }
 
 impl<U> ReadySetStatusReporterInner<U>
@@ -156,6 +170,7 @@ where
             upstream_reachable: self.upstream_reachable().await,
             connection_count: self.connections.len(),
             persistent_stats,
+            enabled_features: self.enabled_features.clone(),
         }
     }
 

readyset-client-test-helpers/src/lib.rs

@@ -416,6 +416,7 @@ impl TestBuilder {
                         Some(rh),
                         Default::default(),
                         authority.clone(),
+                        Vec::new(),
                     );
                     let backend = backend_builder
                         .dialect(A::DIALECT)

readyset-logictest/src/runner.rs

@@ -702,6 +702,7 @@ impl TestScript {
                         Some(rh),
                         Default::default(),
                         authority.clone(),
+                        Vec::new(),
                     );
                     BackendBuilder::new()
                         .require_authentication(false)

readyset-mysql/tests/integration.rs

@@ -1911,7 +1911,7 @@ async fn show_readyset_status() {
 
     // NOTE: If this readyset extension has changed, verify the new behavior is correct then update
     // the expected values below
-    assert_eq!(ret.len(), 8);
+    assert_eq!(ret.len(), 9);
     let row = ret.remove(0);
     assert_eq!(row.get::<String, _>(0).unwrap(), "Database Connection");
     assert_eq!(row.get::<String, _>(1).unwrap(), "Connected");
@@ -1945,6 +1945,9 @@ async fn show_readyset_status() {
     let row = ret.remove(0);
     assert_eq!(row.get::<String, _>(0).unwrap(), "Last started replication");
     assert!(valid_timestamp(row.get::<String, _>(1).unwrap()));
+    let row = ret.remove(0);
+    assert_eq!(row.get::<String, _>(0).unwrap(), "Enabled Features");
+    assert_eq!(row.get::<String, _>(1).unwrap(), "None");
     readyset_maintenance_mode(&mut conn).await;
     shutdown_tx.shutdown().await;
 }
@@ -1955,7 +1958,7 @@ async fn readyset_maintenance_mode(conn: &mut Conn) {
         .unwrap();
     sleep().await;
     let ret: Vec<Row> = conn.query("SHOW READYSET STATUS;").await.unwrap();
-    assert_eq!(ret.len(), 8);
+    assert_eq!(ret.len(), 9);
     // find the row with "Status"
     let row = ret
         .iter()
@@ -1970,7 +1973,7 @@ async fn readyset_maintenance_mode(conn: &mut Conn) {
         .unwrap();
     sleep().await;
     let ret: Vec<Row> = conn.query("SHOW READYSET STATUS;").await.unwrap();
-    assert_eq!(ret.len(), 8);
+    assert_eq!(ret.len(), 9);
     let row = ret
         .iter()
         .find(|r| r.get::<String, _>(0).unwrap() == "Status")

readyset-server/src/lib.rs

@@ -760,6 +760,32 @@ impl WorkerOptions {
 
         None
     }
+
+    /// Get a list of enabled feature flags for status display
+    pub fn enabled_features(&self) -> Vec<String> {
+        let mut enabled = Vec::new();
+
+        if self.feature_materialization_persistence {
+            enabled.push("Materialization Persistence".to_string());
+        }
+        if self.feature_mixed_comparisons {
+            enabled.push("Mixed Comparisons".to_string());
+        }
+        if self.feature_pagination {
+            enabled.push("Pagination".to_string());
+        }
+        if self.feature_post_lookup {
+            enabled.push("Post-Lookup".to_string());
+        }
+        if self.feature_straddled_joins {
+            enabled.push("Straddled Joins".to_string());
+        }
+        if self.feature_topk {
+            enabled.push("Top K".to_string());
+        }
+
+        enabled
+    }
 }
 
 // TODO(justin): Change VolumeId type when we know this fixed size.
@@ -820,4 +846,19 @@ mod tests {
         let worker_opts = Wrapper::parse_from(["test", "--storage-dir", storage_dir]).worker_opts;
         assert_eq!(Some(PathBuf::from(storage_dir)), worker_opts.storage_dir());
     }
+
+    #[test]
+    fn enabled_features_default() {
+        let worker_opts = Wrapper::parse_from(["test"]).worker_opts;
+        let enabled = worker_opts.enabled_features();
+        assert!(enabled.is_empty());
+    }
+
+    #[test]
+    fn enabled_features_multiple() {
+        let worker_opts =
+            Wrapper::parse_from(["test", "--feature-pagination", "--feature-topk"]).worker_opts;
+        let enabled = worker_opts.enabled_features();
+        assert_eq!(enabled, vec!["Pagination", "Top K"]);
+    }
 }

readyset/src/lib.rs

@@ -875,6 +875,7 @@ where
             Some(rh.clone()),
             connections.clone(),
             adapter_authority.clone(),
+            options.server_worker_options.enabled_features(),
         );
         let ctrlc = tokio::signal::ctrl_c();
         let mut sigterm = {