graph, node, store: Add progress reporting for dump and restore

Add DumpReporter and RestoreReporter callback traits following the
established PruneReporter pattern. The store layer calls trait methods
at key moments (table start/finish, batch completion, schema creation,
finalization) while the CLI provides concrete implementations using
indicatif spinners that show per-table progress with row counts and
elapsed time.

Author: lutter

Cargo.lock

@@ -46,6 +46,7 @@ chrono = "0.4.43"
 bs58 = "0.5.1"
 clap = { version = "4.5.4", features = ["derive", "env", "wrap_help"] }
 clap_complete = "4"
+console = "0.16"
 derive_more = { version = "2.1.1", default-features = false }
 diesel = { version = "2.2.7", features = [
     "postgres",
@@ -70,6 +71,7 @@ graphman-server = { path = "./server/graphman" }
 graphman = { path = "./core/graphman" }
 graphman-store = { path = "./core/graphman_store" }
 graphql-tools = "0.5.1"
+indicatif = "0.18"
 Inflector = "0.11.3"
 itertools = "0.14.0"
 lazy_static = "1.5.0"

Cargo.toml

@@ -57,9 +57,9 @@ reqwest = { workspace = true }
 thiserror = { workspace = true }
 
 # Console output
-indicatif = "0.18"
-console = "0.16"
 similar = "2"
+indicatif = { workspace = true }
+console = { workspace = true }
 
 # Code generation
 graphql-tools = { workspace = true }

gnd/Cargo.toml

@@ -978,6 +978,69 @@ pub trait PruneReporter: Send + 'static {
     fn finish(&mut self) {}
 }
 
+/// Callbacks for `SubgraphStore.dump` so that callers can report progress
+/// of the dump procedure to users
+#[allow(unused_variables)]
+pub trait DumpReporter: Send + 'static {
+    /// Called at the start with deployment hash and total entity table count.
+    fn start(&mut self, deployment: &str, table_count: usize) {}
+
+    /// Called before dumping an entity table. `rows_approx` is estimated
+    /// from the vid range (max_vid - min_vid + 1), 0 if empty.
+    fn start_table(&mut self, table: &str, rows_approx: usize) {}
+
+    /// Called after a batch of rows has been written to Parquet.
+    fn batch_dumped(&mut self, table: &str, rows: usize) {}
+
+    /// Called after an entity table has been fully dumped.
+    fn finish_table(&mut self, table: &str, rows: usize) {}
+
+    /// Called before dumping data_sources$.
+    fn start_data_sources(&mut self) {}
+
+    /// Called after data_sources$ has been dumped.
+    fn finish_data_sources(&mut self, rows: usize) {}
+
+    /// Called when the entire dump has completed.
+    fn finish(&mut self) {}
+}
+
+/// Callbacks for `SubgraphStore.restore` so that callers can report
+/// progress of the restore procedure to users
+#[allow(unused_variables)]
+pub trait RestoreReporter: Send + 'static {
+    /// Called at the start with deployment hash and total entity table count.
+    fn start(&mut self, deployment: &str, table_count: usize) {}
+
+    /// Called when creating the schema.
+    fn start_create_schema(&mut self) {}
+    fn finish_create_schema(&mut self) {}
+
+    /// Called before importing an entity table. `total_rows` is the sum
+    /// of row counts across all chunks for this table.
+    fn start_table(&mut self, table: &str, total_rows: usize) {}
+
+    /// Called after a batch of rows has been inserted.
+    fn batch_imported(&mut self, table: &str, rows: usize) {}
+
+    /// Called when a table is skipped because it was already restored.
+    fn skip_table(&mut self, table: &str) {}
+
+    /// Called after an entity table has been fully imported.
+    fn finish_table(&mut self, table: &str, rows: usize) {}
+
+    /// Called before/after importing data_sources$.
+    fn start_data_sources(&mut self, total_rows: usize) {}
+    fn finish_data_sources(&mut self, rows: usize) {}
+
+    /// Called during finalization (sequence resets, head block set).
+    fn start_finalize(&mut self) {}
+    fn finish_finalize(&mut self) {}
+
+    /// Called when the entire restore has completed.
+    fn finish(&mut self) {}
+}
+
 /// Select how pruning should be done
 #[derive(Clone, Copy, Debug, Display, PartialEq)]
 pub enum PruningStrategy {

graph/src/components/store/mod.rs

@@ -40,6 +40,8 @@ diesel = { workspace = true }
 diesel-async = { workspace = true }
 prometheus = { version = "0.14.0", features = ["push"] }
 json-structural-diff = { version = "0.2", features = ["colorize"] }
+indicatif = { workspace = true }
+console = { workspace = true }
 
 # Dependencies related to Amp subgraphs
 tokio-util.workspace = true

node/Cargo.toml

@@ -1,12 +1,117 @@
 use std::fs;
 use std::sync::Arc;
+use std::time::Instant;
 
+use console::style;
+use graph::components::store::DumpReporter;
 use graph::prelude::anyhow::Result;
+use indicatif::ProgressBar;
 
 use graph_store_postgres::{ConnectionPool, SubgraphStore};
 
 use crate::manager::deployment::DeploymentSearch;
 
+struct DumpProgress {
+    spinner: ProgressBar,
+    start: Instant,
+    table_start: Instant,
+    table_rows: usize,
+}
+
+impl DumpProgress {
+    fn new() -> Self {
+        let spinner = ProgressBar::new_spinner();
+        spinner.enable_steady_tick(std::time::Duration::from_millis(80));
+        Self {
+            spinner,
+            start: Instant::now(),
+            table_start: Instant::now(),
+            table_rows: 0,
+        }
+    }
+}
+
+impl DumpReporter for DumpProgress {
+    fn start(&mut self, deployment: &str, table_count: usize) {
+        self.start = Instant::now();
+        self.spinner.set_message(format!(
+            "Dumping deployment {} ({} tables)",
+            style(deployment).cyan(),
+            table_count,
+        ));
+    }
+
+    fn start_table(&mut self, table: &str, rows_approx: usize) {
+        self.table_start = Instant::now();
+        self.table_rows = 0;
+        self.spinner
+            .set_message(format!("{:<32} ~{} rows", table, format_count(rows_approx),));
+    }
+
+    fn batch_dumped(&mut self, table: &str, rows: usize) {
+        self.table_rows += rows;
+        self.spinner.set_message(format!(
+            "{:<32} {} rows",
+            table,
+            format_count(self.table_rows),
+        ));
+    }
+
+    fn finish_table(&mut self, table: &str, rows: usize) {
+        let elapsed = self.table_start.elapsed().as_secs();
+        let line = format!(
+            "  {} {:<32} {:>10} rows {:>4}s",
+            style("\u{2714}").green(),
+            table,
+            format_count(rows),
+            elapsed,
+        );
+        self.spinner.suspend(|| println!("{line}"));
+    }
+
+    fn start_data_sources(&mut self) {
+        self.table_start = Instant::now();
+        self.table_rows = 0;
+        self.spinner.set_message("data_sources$".to_string());
+    }
+
+    fn finish_data_sources(&mut self, rows: usize) {
+        let elapsed = self.table_start.elapsed().as_secs();
+        let line = format!(
+            "  {} {:<32} {:>10} rows {:>4}s",
+            style("\u{2714}").green(),
+            "data_sources$",
+            format_count(rows),
+            elapsed,
+        );
+        self.spinner.suspend(|| println!("{line}"));
+    }
+
+    fn finish(&mut self) {
+        let elapsed = self.start.elapsed().as_secs();
+        self.spinner.finish_with_message(format!(
+            "{} Dump complete ({}s)",
+            style("\u{2714}").green(),
+            elapsed
+        ));
+    }
+}
+
+fn format_count(n: usize) -> String {
+    if n == 0 {
+        return "0".to_string();
+    }
+    let s = n.to_string();
+    let mut result = String::new();
+    for (i, c) in s.chars().rev().enumerate() {
+        if i > 0 && i % 3 == 0 {
+            result.push(',');
+        }
+        result.push(c);
+    }
+    result.chars().rev().collect()
+}
+
 pub async fn run(
     subgraph_store: Arc<SubgraphStore>,
     primary_pool: ConnectionPool,
@@ -18,6 +123,7 @@ pub async fn run(
 
     let loc = deployment.locate_unique(&primary_pool).await?;
 
-    subgraph_store.dump(&loc, directory).await?;
+    let reporter = Box::new(DumpProgress::new());
+    subgraph_store.dump(&loc, directory, reporter).await?;
     Ok(())
 }