feat: Add plan compression and CLI decode support

Cargo.toml

@@ -59,6 +59,7 @@ insta = { version = "1.46.0", features = ["filters"], optional = true }
 parquet = { version = "58", optional = true }
 arrow = { version = "58", optional = true, features = ["test_utils"] }
 hyper-util = { version = "0.1.16", optional = true }
+base64 = "0.23.0"
 
 [features]
 default = ["grpc"]

console/Cargo.toml

@@ -15,6 +15,7 @@ tonic = "0.14.2"
 datafusion-distributed = { path = "..", features = ["integration", "system-metrics"] }
 url = "2.5.7"
 tokio-stream = "0.1.18"
+base64 = "0.23.0"
 
 [dev-dependencies]
 arrow = "58"

console/src/main.rs

@@ -5,6 +5,8 @@ mod ui;
 mod worker;
 
 use app::App;
+use base64::engine::general_purpose::STANDARD;
+use base64::engine::Engine;
 use crossterm::event::{self, Event};
 use ratatui::DefaultTerminal;
 use std::time::{Duration, Instant};
@@ -24,6 +26,10 @@ struct Args {
     /// Polling interval in milliseconds
     #[structopt(long = "poll-interval", default_value = "100")]
     poll_interval: u64,
+
+    /// Decode and print a base64-encoded plan string produced by explain_analyze.
+    #[structopt(long = "encoded-plan")]
+    encoded_plan: Option<String>,
 }
 
 #[tokio::main]
@@ -32,6 +38,14 @@ async fn main() -> color_eyre::Result<()> {
 
     let args = Args::from_args();
 
+    if let Some(encoded) = args.encoded_plan {
+        let decoded = Engine::decode(&STANDARD, &encoded)
+            .map(|b| String::from_utf8_lossy(&b).into_owned())
+            .unwrap_or(encoded);
+        println!("{decoded}");
+        return Ok(());
+    }
+
     let seed_url = Url::parse(&format!("http://localhost:{}", args.port)).expect("valid URL");
 
     let poll_interval = Duration::from_millis(args.poll_interval);

src/coordinator/distributed.rs

@@ -1,3 +1,5 @@
+use base64::engine::general_purpose::STANDARD;
+use base64::engine::Engine;
 use crate::common::require_one_child;
 use crate::coordinator::metrics_store::MetricsStore;
 use crate::coordinator::prepare_dynamic_plan::prepare_dynamic_plan;
@@ -134,6 +136,12 @@ impl DistributedExec {
             .clone()
             .ok_or_else(|| internal_datafusion_err!("No head stage found. Was execute() called?"))
     }
+    /// Decodes a base64-encoded plan string produced by [`explain_analyze`](crate::explain_analyze).
+    pub fn extract_encoded_plan(&self, encoded_plan: &str) -> String {
+        Engine::decode(&STANDARD, encoded_plan)
+        .map(|bytes| String::from_utf8_lossy(&bytes).into_owned())
+        .unwrap_or_else(|_| encoded_plan.to_string())
+    }
 }
 
 impl DisplayAs for DistributedExec {

src/stage.rs

@@ -1,4 +1,7 @@
+use base64::engine::general_purpose::STANDARD;
+use base64::engine::Engine;
 use crate::coordinator::{DistributedExec, MetricsStore};
+
 use crate::execution_plans::{DistributedLeafExec, NetworkCoalesceExec};
 use crate::metrics::DISTRIBUTED_DATAFUSION_TASK_ID_LABEL;
 use datafusion::common::{HashMap, Statistics, config_err};
@@ -252,7 +255,12 @@ pub async fn explain_analyze(
             .to_string()),
         Some(_) => {
             let executed = rewrite_distributed_plan_with_metrics(executed.clone(), format).await?;
-            Ok(display_plan_ascii(executed.as_ref(), true))
+            let display_string = display_plan_ascii(executed.as_ref(), true);
+            if display_string.len() >= 10_000 {
+                Ok(Engine::encode(&STANDARD, display_string.as_bytes()))
+            } else {
+                Ok(display_string)
+            }
         }
     }
 }