feat: expose snapshot timing in duckpipe.status() (#9)

## Summary
- Add `snapshot_duration_ms` and `snapshot_rows` columns to
`table_mappings` and `duckpipe.status()` so users can query snapshot
performance directly via SQL instead of parsing PG logs
- Snapshot duration is always measured unconditionally (not gated on the
`timing` flag), stored on SNAPSHOT→CATCHUP transition
- `resync_table()` resets both columns to NULL for clean re-snapshot
semantics

## Test plan
- [x] All 24 regression tests pass (`make installcheck`)
- [x] `monitoring` test verifies new columns appear in
`duckpipe.status()` with correct NULL defaults
- [x] Code reviewed for type consistency across the full stack
(snapshot.rs → metadata.rs → service.rs → api.rs → bootstrap.sql)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: YuweiXiao

PROGRESS.md

@@ -64,7 +64,7 @@
 ### Monitoring / Observability
 - [ ] `applied_lsn` stays NULL during SNAPSHOT/CATCHUP — should be set to `snapshot_lsn` after snapshot completes
 - [ ] `worker_state` not updated during snapshot processing — stale metrics while snapshots run
-- [ ] Expose snapshot timing in `duckpipe.status()` — currently snapshot duration/rate is only logged to PG log (`DuckPipe timing: action=snapshot_table`), not queryable via SQL. Add columns like `snapshot_duration_ms` and `snapshot_rows` to `table_mappings` so users and benchmark scripts can read them directly. Would also make `analyze_results.py` Summary consistent with per-table timings (no need to parse PG logs).
+- [x] Expose snapshot timing in `duckpipe.status()` — added `snapshot_duration_ms` and `snapshot_rows` columns to `table_mappings` and `duckpipe.status()`. Snapshot duration is always measured (not gated on `timing` flag) and stored on CATCHUP transition. `resync_table()` resets both columns.
 - [x] Benchmark suite (`bench_suite.sh`) — 4 scenarios (single/multi × insert/mixed) with automated analysis report (`analyze_results.py`)
 
 ### Bugs

README.md

@@ -72,10 +72,10 @@ Sysbench on Apple M1 Pro, 100k rows/table, 30s OLTP phase:
 
 | Scenario | Snapshot | OLTP TPS | Avg Lag | Consistency |
 |----------|----------|----------|---------|-------------|
-| 1 table, `oltp_insert` | 14,569 rows/s | 9,261 | 2.7 MB | PASS |
-| 4 tables, `oltp_insert` | 26,353 rows/s | 8,076 | 55 MB | PASS |
-| 1 table, `oltp_read_write` | 6,020 rows/s | 614 | 154 MB | PASS |
-| 4 tables, `oltp_read_write` | 38,858 rows/s | 444 | 332 MB | PASS |
+| 1 table, `oltp_insert` | 136,240 rows/s | 10,102 | 3.3 MB | PASS |
+| 4 tables, `oltp_insert` | 163,599 rows/s | 9,413 | 64.6 MB | PASS |
+| 1 table, `oltp_read_write` | 132,450 rows/s | 627 | 170.8 MB | PASS |
+| 4 tables, `oltp_read_write` | 150,830 rows/s | 450 | 369.3 MB | PASS |
 
 Full breakdown (flush latency, phase timing, snapshot per-table, WAL cycles): [benchmark/results/report.md](benchmark/results/report.md)
 

benchmark/analyze_results.py

@@ -104,6 +104,16 @@ def parse_benchmark_log(path):
     elif "Consistency mismatches" in text:
         result.consistency = "FAIL"
 
+    # Per-table snapshot lines: Snapshot [public.sbtest1]  : 100,000 rows in 800ms (125,000 rows/s)
+    for m in re.finditer(
+        r"Snapshot \[(\S+)\]\s*:\s*([\d,]+)\s*rows in\s*([\d,]+)ms",
+        text,
+    ):
+        table = m.group(1)
+        rows = int(m.group(2).replace(",", ""))
+        ms = float(m.group(3).replace(",", ""))
+        result.snapshot_timings.append((table, rows, ms))
+
     return result
 
 
@@ -152,13 +162,6 @@ def parse_pg_log(result, path):
             "total": float(m.group(11)),
         })
 
-    # DuckPipe timing: action=snapshot_table source=... rows=N elapsed_ms=X
-    for m in re.finditer(
-        r"DuckPipe timing: action=snapshot_table\s+source=(\S+)\s+target=\S+\s+rows=(\d+)\s+elapsed_ms=([\d.]+)",
-        text,
-    ):
-        result.snapshot_timings.append((m.group(1), int(m.group(2)), float(m.group(3))))
-
     # DuckPipe timing: action=process_sync_group_streaming ... elapsed_ms=X
     for m in re.finditer(
         r"DuckPipe timing: action=process_sync_group_streaming\s+.*?elapsed_ms=([\d.]+)",

benchmark/results/report.md

@@ -4,70 +4,70 @@
 
 | Metric | Single-table append (1T, oltp_insert) | Multi-table append (4T, oltp_insert) | Single-table mixed (1T, oltp_read_write) | Multi-table mixed (4T, oltp_read_write) |
 |--------|---|---|---|---|
-| Snapshot (rows/s) | 14,569 | 26,353 | 6,020 | 38,858 |
-| OLTP TPS | 9,261.0 | 8,075.7 | 613.6 | 444.4 |
-| Avg Lag (MB) | 2.7 | 55.0 | 154 | 332.1 |
-| Peak Lag (MB) | 4.0 | 126 | 171.7 | 357.3 |
-| Catch-up (s) | 2.2 | 2.5 | 65.4 | 68.5 |
-| Catch-up (rows/s) | - | 109 | - | - |
+| Snapshot (rows/s) | 136,240 | 163,599 | 132,450 | 150,830 |
+| OLTP TPS | 10,102.3 | 9,412.7 | 626.6 | 450.3 |
+| Avg Lag (MB) | 3.3 | 64.6 | 170.8 | 369.3 |
+| Peak Lag (MB) | 5.3 | 125.5 | 189.2 | 390.9 |
+| Catch-up (s) | 2.2 | 2.4 | 65.6 | 69.0 |
+| Catch-up (rows/s) | - | - | - | - |
 | Consistency | PASS | PASS | PASS | PASS |
 
 ## Flush Performance
 
 | Metric | Single-table append (1T, oltp_insert) | Multi-table append (4T, oltp_insert) | Single-table mixed (1T, oltp_read_write) | Multi-table mixed (4T, oltp_read_write) |
 |--------|---|---|---|---|
-| Flush count | 29 | 119 | 30 | 120 |
-| Avg latency (ms) | 41.3 | 24.9 | 35.0 | 33.9 |
-| P50 latency (ms) | 33.8 | 16.8 | 33.2 | 23.9 |
-| P99 latency (ms) | 229.2 | 240.9 | 109.5 | 169.4 |
-| Avg rows/flush | 9,580.6 | 2,036.0 | 3,682.2 | 666.6 |
+| Flush count | 31 | 121 | 30 | 122 |
+| Avg latency (ms) | 35.1 | 17.6 | 31.0 | 25.6 |
+| P50 latency (ms) | 33.6 | 15.2 | 30.3 | 22.8 |
+| P99 latency (ms) | 63.6 | 107 | 48.4 | 122.1 |
+| Avg rows/flush | 9,776.6 | 2,333.8 | 3,760 | 664.4 |
 
 ### Flush Phase Breakdown (avg ms)
 
 | Phase | Single-table append (1T, oltp_insert) | Multi-table append (4T, oltp_insert) | Single-table mixed (1T, oltp_read_write) | Multi-table mixed (4T, oltp_read_write) |
 |-------|---|---|---|---|
-| discover | 0.6 | 0.5 | - | - |
-| buf_create | 0.5 | 0.3 | 0.3 | 0.4 |
-| load | 18.2 | 4.4 | 6.9 | 1.7 |
-| compact | 6.6 | 3.8 | 3.9 | 3.5 |
+| discover | 0.5 | 0.5 | - | - |
+| buf_create | 0.4 | 0.3 | 0.3 | 0.3 |
+| load | 16.9 | 4.7 | 6.8 | 1.6 |
+| compact | 4.9 | 3.5 | 3.8 | 3.2 |
 | begin | 0.1 | 0.1 | 0.1 | 0.1 |
-| delete | 2.1 | 0.2 | 14.8 | 9.2 |
-| insert | 9.9 | 5.9 | 2.8 | 1.6 |
-| commit | 2.0 | 9.0 | 5.5 | 16.9 |
-| cleanup | 0.5 | 0.4 | 0.4 | 0.4 |
+| delete | 0.3 | 0.3 | 14.8 | 8.3 |
+| insert | 9.3 | 4.0 | 2.0 | 1.5 |
+| commit | 2.0 | 3.6 | 2.7 | 10.0 |
+| cleanup | 0.4 | 0.4 | 0.4 | 0.4 |
 
 ## Snapshot Timings
 
 **Single-table append (1T, oltp_insert)**
 
-- `public.sbtest1`: 100,000 rows in 799.7ms (125046 rows/s)
+- `public.sbtest1`: 100,000 rows in 734.0ms (136240 rows/s)
 
 **Multi-table append (4T, oltp_insert)**
 
-- `public.sbtest1`: 100,000 rows in 2903.2ms (34445 rows/s)
-- `public.sbtest2`: 100,000 rows in 3150.2ms (31744 rows/s)
-- `public.sbtest3`: 100,000 rows in 3162.5ms (31621 rows/s)
-- `public.sbtest4`: 100,000 rows in 3187.5ms (31373 rows/s)
+- `public.sbtest1`: 100,000 rows in 2423.0ms (41271 rows/s)
+- `public.sbtest2`: 100,000 rows in 2405.0ms (41580 rows/s)
+- `public.sbtest3`: 100,000 rows in 2235.0ms (44743 rows/s)
+- `public.sbtest4`: 100,000 rows in 2445.0ms (40900 rows/s)
 
 **Single-table mixed (1T, oltp_read_write)**
 
-- `public.sbtest1`: 100,000 rows in 1037.5ms (96383 rows/s)
+- `public.sbtest1`: 100,000 rows in 755.0ms (132450 rows/s)
 
 **Multi-table mixed (4T, oltp_read_write)**
 
-- `public.sbtest1`: 100,000 rows in 4152.2ms (24084 rows/s)
-- `public.sbtest3`: 100,000 rows in 4231.0ms (23635 rows/s)
-- `public.sbtest4`: 100,000 rows in 4369.0ms (22889 rows/s)
-- `public.sbtest2`: 100,000 rows in 4550.1ms (21978 rows/s)
+- `public.sbtest1`: 100,000 rows in 2545.0ms (39293 rows/s)
+- `public.sbtest2`: 100,000 rows in 2627.0ms (38066 rows/s)
+- `public.sbtest3`: 100,000 rows in 2642.0ms (37850 rows/s)
+- `public.sbtest4`: 100,000 rows in 2652.0ms (37707 rows/s)
 
 
 ## WAL Processing Cycle Times
 
 | Metric | Single-table append (1T, oltp_insert) | Multi-table append (4T, oltp_insert) | Single-table mixed (1T, oltp_read_write) | Multi-table mixed (4T, oltp_read_write) |
 |--------|---|---|---|---|
-| Cycles | 61 | 70 | 2,631 | 3,059 |
-| Avg (ms) | 596.8 | 581.6 | 19.4 | 14.6 |
-| P99 (ms) | 5,004.0 | 5,004.6 | 17.0 | 10.5 |
+| Cycles | 35 | 51 | 2,671 | 3,184 |
+| Avg (ms) | 1,049.3 | 804.7 | 19.1 | 13.9 |
+| P99 (ms) | 5,003.3 | 5,005.2 | 18.9 | 9.3 |
 
 ## Issues & Observations
 

benchmark/run_sysbench.py

@@ -104,6 +104,28 @@ def get_total_queued_changes(db_params):
     )
     return parse_int(res, 0)
 
+def get_snapshot_metrics(db_params, num_tables):
+    """Query duckpipe.status() for per-table snapshot timing from SQL-exposed metrics."""
+    table_filter = ", ".join([f"'public.sbtest{i}'" for i in range(1, num_tables + 1)])
+    res = run_sql(
+        db_params,
+        f"SELECT source_table, snapshot_duration_ms, snapshot_rows "
+        f"FROM duckpipe.status() "
+        f"WHERE source_table IN ({table_filter}) "
+        f"AND snapshot_duration_ms IS NOT NULL",
+    )
+    metrics = []
+    if res:
+        for line in res.strip().split("\n"):
+            parts = line.split("|")
+            if len(parts) == 3:
+                table = parts[0].strip()
+                duration_ms = float(parts[1].strip())
+                rows = int(parts[2].strip())
+                metrics.append((table, rows, duration_ms))
+    return metrics
+
+
 def get_benchmark_rows_synced(args, db_params):
     table_filter = ", ".join([f"'public.sbtest{i}'" for i in range(1, args.tables + 1)])
     res = run_sql(
@@ -286,9 +308,21 @@ def benchmark_snapshot(args, db_params):
 
         if target_count >= total_rows and not_streaming == 0:
             elapsed = time.time() - start_time
-            snapshot_rate = target_count / elapsed if elapsed > 0 else 0
+            wall_clock_rate = target_count / elapsed if elapsed > 0 else 0
             print(f"\n[+] Snapshot complete!")
-            return snapshot_rate
+
+            # Fetch per-table snapshot metrics from duckpipe.status()
+            snapshot_metrics = get_snapshot_metrics(db_params, args.tables)
+            if snapshot_metrics:
+                # Concurrent snapshots overlap; effective rate = total rows / max duration
+                max_duration_ms = max(ms for _, _, ms in snapshot_metrics)
+                total_snapshot_rows = sum(r for _, r, _ in snapshot_metrics)
+                snapshot_rate = total_snapshot_rows / max_duration_ms * 1000 if max_duration_ms > 0 else wall_clock_rate
+            else:
+                snapshot_rate = wall_clock_rate
+                snapshot_metrics = []
+
+            return snapshot_rate, snapshot_metrics
 
         # If all rows are physically copied but tables are stuck in CATCHUP (no SNAPSHOT
         # remaining), nudge the WAL consumer by emitting a transactional logical message.
@@ -579,7 +613,7 @@ def main():
     if not args.skip_prepare:
         prepare_env(args, db_params)
 
-    snapshot_rate = benchmark_snapshot(args, db_params)
+    snapshot_rate, snapshot_metrics = benchmark_snapshot(args, db_params)
     (streaming_tps, avg_lag, max_lag, final_lag,
      catchup_elapsed, catchup_throughput,
      actual_final_count, expected_final_count,
@@ -588,6 +622,9 @@ def main():
     print("\n===========================================================")
     print(" Final Results")
     print("===========================================================")
+    for table, rows, ms in snapshot_metrics:
+        rate = rows / ms * 1000 if ms > 0 else 0
+        print(f" Snapshot [{table}]  : {rows:,} rows in {ms:.0f}ms ({rate:,.0f} rows/s)")
     print(f" Snapshot Throughput  : {snapshot_rate:.0f} rows/sec")
     print(f" OLTP Throughput      : {streaming_tps:.2f} TPS")
     print(f" Avg Replication Lag  : {avg_lag/1024/1024:.1f} MB  [during OLTP]")

doc/CODE_WALKTHROUGH.md

@@ -361,7 +361,7 @@ Async operations against `duckpipe.sync_groups` and `duckpipe.table_mappings` vi
 - `update_applied_lsn(id, lsn)` — per-table flush checkpoint
 
 **State transitions:**
-- `set_catchup_state(id, snapshot_lsn)` — SNAPSHOT → CATCHUP
+- `set_catchup_state(id, snapshot_lsn, duration_ms, snapshot_rows)` — SNAPSHOT → CATCHUP, stores snapshot timing
 - `transition_catchup_to_streaming(group_id, pending_lsn)` — bulk promote CATCHUP tables where `snapshot_lsn <= pending_lsn`
 
 **Error tracking:**
@@ -589,7 +589,7 @@ The top-level function called by both the bgworker and daemon. One complete cycl
 4. **Disconnect**
 5. Return whether any work was done (caller uses this to decide sleep vs. immediate re-poll)
 
-**`process_snapshots(meta, group, connstr, timing)`** — spawns all SNAPSHOT tables as concurrent tokio tasks. Each runs `snapshot::process_snapshot_task()` independently. On success: `set_catchup_state()`. On failure: `record_error_message()` (table stays in SNAPSHOT for retry).
+**`process_snapshots(meta, group, connstr, timing)`** — spawns all SNAPSHOT tables as concurrent tokio tasks. Each runs `snapshot::process_snapshot_task()` independently. On success: `set_catchup_state()` stores the snapshot_lsn, duration_ms, and rows_copied. On failure: `record_error_message()` (table stays in SNAPSHOT for retry).
 
 **`process_sync_group_streaming(client, meta, group, config, consumer, coordinator)`** — streaming WAL consumption:
 1. Check backpressure — if flush threads are lagging, skip this poll round (but still read `min(applied_lsn)` from PG and advance slot)

doc/USAGE.md

@@ -96,6 +96,8 @@ FROM duckpipe.status();
 | `consecutive_failures` | Number of flush failures since last success (ERRORED triggers at 3) |
 | `retry_at` | Scheduled auto-retry time when in ERRORED state |
 | `error_message` | Last error message (empty when healthy) |
+| `snapshot_duration_ms` | Time taken by the initial snapshot (NULL before snapshot completes) |
+| `snapshot_rows` | Number of rows copied during the initial snapshot |
 
 ### Group Overview
 

duckpipe-core/src/metadata.rs

@@ -560,18 +560,24 @@ impl<'a> MetadataClient<'a> {
             .collect())
     }
 
-    /// Update table state to CATCHUP with snapshot_lsn.
+    /// Update table state to CATCHUP with snapshot_lsn, duration, and row count.
     pub async fn set_catchup_state(
         &self,
         mapping_id: i32,
         snapshot_lsn: u64,
+        duration_ms: u64,
+        snapshot_rows: u64,
     ) -> Result<(), tokio_postgres::Error> {
         let lsn_str = format_lsn(snapshot_lsn);
+        let duration_ms_i64 = duration_ms as i64;
+        let snapshot_rows_i64 = snapshot_rows as i64;
         self.client
             .execute(
                 "UPDATE duckpipe.table_mappings SET state = 'CATCHUP', \
-                 snapshot_lsn = $1::text::pg_lsn WHERE id = $2",
-                &[&lsn_str, &mapping_id],
+                 snapshot_lsn = $1::text::pg_lsn, \
+                 snapshot_duration_ms = $2, snapshot_rows = $3 \
+                 WHERE id = $4",
+                &[&lsn_str, &duration_ms_i64, &snapshot_rows_i64, &mapping_id],
             )
             .await?;
         Ok(())

duckpipe-core/src/service.rs

@@ -828,8 +828,8 @@ pub async fn run_sync_cycle(
     let snap_results = snapshot_manager.collect_results();
     for snap in &snap_results {
         match &snap.result {
-            Ok((snapshot_lsn, rows_copied)) => {
-                meta.set_catchup_state(snap.task_id, *snapshot_lsn)
+            Ok((snapshot_lsn, rows_copied, duration_ms)) => {
+                meta.set_catchup_state(snap.task_id, *snapshot_lsn, *duration_ms, *rows_copied)
                     .await
                     .map_err(|e| format!("set_catchup_state: {}", e))?;
                 if *rows_copied > 0 {

duckpipe-core/src/snapshot.rs

@@ -40,16 +40,16 @@ pub struct SnapshotResult {
     pub source_table: String,
     pub target_schema: String,
     pub target_table: String,
-    /// Ok((consistent_point_lsn, rows_copied)) on success, Err(message) on failure.
-    pub result: Result<(u64, u64), String>,
+    /// Ok((consistent_point_lsn, rows_copied, duration_ms)) on success, Err(message) on failure.
+    pub result: Result<(u64, u64, u64), String>,
 }
 
 /// Process snapshot for a single table using a chunked COPY pipeline.
 ///
 /// Creates a temporary logical replication slot via SQL to get a consistent_point,
 /// then streams COPY TO STDOUT through CSV chunk files into DuckDB via read_csv.
 ///
-/// Returns `(consistent_point_lsn, rows_copied)` on success.
+/// Returns `(consistent_point_lsn, rows_copied, duration_ms)` on success.
 pub async fn process_snapshot_task(
     source_schema: &str,
     source_table: &str,
@@ -60,8 +60,8 @@ pub async fn process_snapshot_task(
     ducklake_schema: &str,
     timing: bool,
     task_id: i32,
-) -> Result<(u64, u64), String> {
-    let table_start = if timing { Some(Instant::now()) } else { None };
+) -> Result<(u64, u64, u64), String> {
+    let table_start = Instant::now();
 
     // Step 1: Open control connection — creates temp slot and runs COPY.
     let (ctrl_client, ctrl_connection) =
@@ -201,6 +201,8 @@ pub async fn process_snapshot_task(
 
     let rows_copied = copy_result?;
 
+    let elapsed_ms = table_start.elapsed().as_millis() as u64;
+
     tracing::info!(
         "DuckPipe: Snapshot complete for {}.{}: {} rows copied (consistent_point={})",
         source_schema,
@@ -209,19 +211,19 @@ pub async fn process_snapshot_task(
         format_lsn(consistent_point)
     );
 
-    if let Some(start) = table_start {
+    if timing {
         tracing::info!(
-            "DuckPipe timing: action=snapshot_table source={}.{} target={}.{} rows={} elapsed_ms={:.3}",
+            "DuckPipe timing: action=snapshot_table source={}.{} target={}.{} rows={} elapsed_ms={}",
             source_schema,
             source_table,
             target_schema,
             target_table,
             rows_copied,
-            start.elapsed().as_secs_f64() * 1000.0
+            elapsed_ms
         );
     }
 
-    Ok((consistent_point, rows_copied))
+    Ok((consistent_point, rows_copied, elapsed_ms))
 }
 
 /// CSV producer: streams COPY TO STDOUT bytes into chunk files with headers.