readyset-adapter: Implement SHOW REPLAY PATHS backend handler

Add handler for SHOW REPLAY PATHS statement that queries the
controller's /replay_paths endpoint and formats results as a table
with columns: domain, tag, source, destination_index, target_index,
path_segments, trigger_type, trigger_index, trigger_source_options.

Fixes: REA-6176

Release-Note-Core: Added `SHOW REPLAY PATHS` command to display
  replay paths in the graph. This is helpful for detailed information
  about replays and evictions.

Change-Id: I362eecb3cd00abc35da7b56b61c6ffaf92c21d80
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/10965
Reviewed-by: Michael Zink <michael.z@readyset.io>
Tested-by: Buildkite CI

Author: altmannmarcelo

readyset-adapter/src/backend.rs

@@ -2786,6 +2786,52 @@ where
         ))
     }
 
+    /// Responds to a `SHOW REPLAY PATHS` query
+    /// Returns replay paths data as a result set with columns and rows
+    async fn show_replay_paths(&mut self) -> ReadySetResult<noria_connector::QueryResult<'static>> {
+        // Get replay paths from the controller (already flattened and sorted)
+        let replay_paths = self.noria.replay_paths().await?;
+
+        // Create schema with all columns
+        let schema = create_dummy_schema!(
+            "domain",
+            "tag",
+            "source",
+            "destination_index",
+            "target_index",
+            "path",
+            "trigger_type",
+            "trigger_index",
+            "trigger_source_options"
+        );
+
+        // Convert each ReplayPathInfo into a row
+        let rows: Vec<Vec<DfValue>> = replay_paths
+            .into_iter()
+            .map(|info| {
+                vec![
+                    info.domain.to_string().into(),
+                    info.tag.to_string().into(),
+                    info.source
+                        .map(|s| s.to_string())
+                        .unwrap_or_else(|| "None".to_string())
+                        .into(),
+                    info.destination_index.unwrap_or_default().into(),
+                    info.target_index.unwrap_or_default().into(),
+                    info.path_segments.join(" → ").into(),
+                    info.trigger_type.into(),
+                    info.trigger_index.unwrap_or_default().into(),
+                    info.trigger_source_options.into(),
+                ]
+            })
+            .collect();
+
+        Ok(noria_connector::QueryResult::from_owned(
+            schema,
+            vec![Results::new(rows)],
+        ))
+    }
+
     async fn query_readyset_extensions<'a>(
         &'a mut self,
         query: &'a SqlQuery,
@@ -3041,6 +3087,7 @@ where
                 )
                 .await
             }
+            SqlQuery::Show(ShowStatement::ReplayPaths) => self.show_replay_paths().await,
             SqlQuery::Show(ShowStatement::Rls(_maybe_table)) => {
                 unsupported!("SHOW RLS statement is not yet supported")
             }

readyset-adapter/src/backend/noria_connector.rs

@@ -899,6 +899,12 @@ impl NoriaConnector {
         Ok(QueryResult::from_owned(schema, vec![Results::new(data)]))
     }
 
+    pub(crate) async fn replay_paths(
+        &mut self,
+    ) -> ReadySetResult<Vec<readyset_client::replay_path::ReplayPathInfo>> {
+        noria_await!(self.inner, self.inner.noria.replay_paths())
+    }
+
     /// Set the schema search path
     pub fn set_schema_search_path(&mut self, search_path: Vec<SqlIdentifier>) {
         self.schema_search_path = search_path;