feat: add support for EXPLAIN logging of slow queries

- Introduced a new static variable `SLOW_QUERY_EXPLAIN` to enable capturing EXPLAIN output for slow queries.
- Implemented `set_slow_query_explain` and `slow_query_explain_enabled` functions to manage this feature.
- Updated the query logging to include EXPLAIN output when slow query logging is enabled, enhancing debugging and performance analysis capabilities.

Author: jordcodes

src/bin/rillflow.rs

@@ -146,6 +146,9 @@ struct Cli {
     /// Slow query logging threshold in milliseconds (default 500)
     #[arg(long, default_value_t = 500)]
     slow_query_ms: u64,
+    /// Also capture EXPLAIN (FORMAT TEXT) for slow queries
+    #[arg(long, default_value_t = false)]
+    slow_query_explain: bool,
 
     #[command(subcommand)]
     command: Commands,
@@ -534,6 +537,7 @@ async fn main() -> rillflow::Result<()> {
     rillflow::metrics::set_slow_query_threshold(std::time::Duration::from_millis(
         cli.slow_query_ms,
     ));
+    rillflow::metrics::set_slow_query_explain(cli.slow_query_explain);
     let mgr = store.schema();
 
     let tenant_helper = TenantHelper::new(

src/metrics.rs

@@ -55,6 +55,7 @@ static METRICS: OnceLock<Metrics> = OnceLock::new();
 static TENANT_METRICS: OnceLock<Mutex<HashMap<String, TenantCounters>>> = OnceLock::new();
 static QUERY_DURATIONS: OnceLock<Mutex<HashMap<String, (u64, u64)>>> = OnceLock::new();
 static SLOW_QUERY_THRESHOLD_MS: OnceLock<std::sync::atomic::AtomicU64> = OnceLock::new();
+static SLOW_QUERY_EXPLAIN: OnceLock<std::sync::atomic::AtomicBool> = OnceLock::new();
 static PROJECTION_STATS: OnceLock<Mutex<HashMap<String, (u64, u64, u64)>>> = OnceLock::new();
 
 #[derive(Default, Clone)]
@@ -409,6 +410,16 @@ pub fn slow_query_threshold() -> Duration {
     Duration::from_millis(atom.load(std::sync::atomic::Ordering::Relaxed))
 }
 
+pub fn set_slow_query_explain(enabled: bool) {
+    let atom = SLOW_QUERY_EXPLAIN.get_or_init(|| std::sync::atomic::AtomicBool::new(false));
+    atom.store(enabled, std::sync::atomic::Ordering::Relaxed);
+}
+
+pub fn slow_query_explain_enabled() -> bool {
+    let atom = SLOW_QUERY_EXPLAIN.get_or_init(|| std::sync::atomic::AtomicBool::new(false));
+    atom.load(std::sync::atomic::Ordering::Relaxed)
+}
+
 pub fn record_projection_batch(name: &str, events_processed: u64, dur: Duration) {
     if let Ok(mut map) = PROJECTION_STATS.get_or_init(Default::default).lock() {
         let entry = map.entry(name.to_string()).or_insert((0, 0, 0));

src/query/mod.rs

@@ -1100,6 +1100,18 @@ where
         // slow query logging
         if start.elapsed() > crate::metrics::slow_query_threshold() {
             tracing::warn!(target: "rillflow::slow_query", elapsed_ms = start.elapsed().as_millis() as u64, sql = %sql_captured, "slow document query");
+            if crate::metrics::slow_query_explain_enabled() {
+                // Best-effort EXPLAIN (without executing)
+                let explain = format!("EXPLAIN (VERBOSE, FORMAT TEXT) {}", sql_captured);
+                let rows = sqlx::query_scalar::<_, String>(&explain)
+                    .fetch_all(&pool)
+                    .await;
+                if let Ok(lines) = rows {
+                    for line in lines {
+                        tracing::warn!(target: "rillflow::slow_query", plan = %line);
+                    }
+                }
+            }
         }
         rows.into_iter()
             .map(|(value,)| serde_json::from_value(value).map_err(Into::into))
@@ -1115,6 +1127,17 @@ where
         crate::metrics::record_query_duration("docs_fetch_optional", start.elapsed());
         if start.elapsed() > crate::metrics::slow_query_threshold() {
             tracing::warn!(target: "rillflow::slow_query", elapsed_ms = start.elapsed().as_millis() as u64, sql = %sql_captured, "slow document query");
+            if crate::metrics::slow_query_explain_enabled() {
+                let explain = format!("EXPLAIN (VERBOSE, FORMAT TEXT) {}", sql_captured);
+                let rows = sqlx::query_scalar::<_, String>(&explain)
+                    .fetch_all(&pool)
+                    .await;
+                if let Ok(lines) = rows {
+                    for line in lines {
+                        tracing::warn!(target: "rillflow::slow_query", plan = %line);
+                    }
+                }
+            }
         }
         match row {
             Some((value,)) => Ok(Some(serde_json::from_value(value)?)),
@@ -1131,6 +1154,17 @@ where
         crate::metrics::record_query_duration("docs_fetch_one", start.elapsed());
         if start.elapsed() > crate::metrics::slow_query_threshold() {
             tracing::warn!(target: "rillflow::slow_query", elapsed_ms = start.elapsed().as_millis() as u64, sql = %sql_captured, "slow document query");
+            if crate::metrics::slow_query_explain_enabled() {
+                let explain = format!("EXPLAIN (VERBOSE, FORMAT TEXT) {}", sql_captured);
+                let rows = sqlx::query_scalar::<_, String>(&explain)
+                    .fetch_all(&pool)
+                    .await;
+                if let Ok(lines) = rows {
+                    for line in lines {
+                        tracing::warn!(target: "rillflow::slow_query", plan = %line);
+                    }
+                }
+            }
         }
         Ok(serde_json::from_value(value)?)
     }