test(resharding): add benchmarks for copy_data & replication (#1013)

Add benchmarks scripts (with hyperfine inside) for pure copy_data and
replication to measure its performance.

Author: meskill

benches/README.md

@@ -0,0 +1,138 @@
+# Benchmarks
+
+Performance benchmarks for PgDog. Results are stored in
+`target/pgdog_benches/` and compared automatically between runs.
+
+## Structure
+
+```
+benches/
+  bench.sh          shared runner library
+  compare.py        result comparison table
+  resharding/       resharding benchmark suite
+    pgdog.toml      config: 3 source shards (pgdog1-3) → 4 destination shards (shard_0-3)
+    users.toml
+    setup.sql       schema + tables + publication (run per source shard)
+    fill.sql        synthetic data generator (BENCH_SCALE env var or \set scale)
+    prepare.sh      hyperfine --prepare: drop destination schema, run schema-sync
+    copy_data/
+      run.sh        entry point: setup sources, then benchmark data-sync --sync-only
+```
+
+## Running
+
+```sh
+# Setup sources and benchmark (compares against the previous run automatically)
+bash benches/resharding/copy_data/run.sh
+
+# Save as a named baseline
+bash benches/resharding/copy_data/run.sh --save-baseline main
+
+# Compare against a named baseline (errors immediately if it does not exist)
+bash benches/resharding/copy_data/run.sh --baseline main
+
+# Fewer runs for a quick check
+RUNS=1 WARMUP=0 bash benches/resharding/copy_data/run.sh
+```
+
+### Flow
+
+1. **Setup** (once per invocation): drops and recreates `bench_copy` schema on each
+   source shard (`pgdog1`, `pgdog2`, `pgdog3`), then fills each shard with its own
+   slice of rows using a stepped `generate_series` keyed on `tenant_id`, so the source
+   data is pre-distributed before the copy benchmark runs.
+   Set `BENCH_SCALE=N` before running, or edit `\set scale` in `resharding/fill.sql` directly.
+
+2. **Benchmark** (timed, N runs): hyperfine calls `prepare.sh` before each iteration
+   (drops destination schema, runs `schema-sync`), then times `data-sync --sync-only`.
+   This measures pure data-copy throughput with no WAL replication involved.
+
+On first run there is no history yet:
+
+```
+No previous run found -- run again to see comparison.
+```
+
+On subsequent runs a comparison table is printed automatically:
+
+```
+copy_data
+
+                  latest         current      change
+  ----------  ----------  -------------  ----------
+        mean    24.904 s       22.194 s     -10.88%
+      median    24.100 s       21.900 s      -9.13%
+         min    24.904 s       22.194 s     -10.88%
+         max    24.904 s       22.194 s     -10.88%
+      stddev      1.200 s        0.900 s     -25.00%
+        user     2.226 s       11.234 s    +404.63%
+      system   962.045ms        1.419 s     +47.50%
+     max_mem      48.3 MB        46.1 MB      -4.55%
+
+  Performance has improved.
+```
+
+`latest` is always the previous run. Named baselines persist in
+`target/pgdog_benches/` until manually removed.
+
+If a run is interrupted, the result file is left empty. The next run detects
+this, skips the snapshot, and prints a notice instead of crashing. The run
+after that resumes normal comparison automatically.
+
+## Default approach: bench.sh + compare.py
+
+`bench.sh` is a sourceable shell library that provides a single function:
+
+```sh
+bench_run NAME COMMAND [--prepare CMD] [--save-baseline NAME] [--baseline NAME]
+```
+
+It handles:
+- Building pgdog in release mode (sets `PGDOG_BIN`, skipped if already set)
+- Setting `RUST_LOG=error` to suppress log I/O overhead
+- Running the command under [hyperfine](https://github.com/sharkdp/hyperfine)
+- Passing `--prepare CMD` to hyperfine when provided (runs before each timed iteration)
+- Saving results as JSON to `target/pgdog_benches/<name>.json`
+- Snapshotting the previous result to `<name>.prev.json` before each run
+- Calling `compare.py` to render the comparison table
+
+`compare.py` reads two hyperfine JSON files and prints a Criterion-style table
+with mean, median, min, max, stddev, user time, system time, and peak memory
+(`max_mem` — peak of `memory_usage_byte` samples), relative change per field, and an overall pass/regression verdict. It is independent of `bench.sh`
+and can be called directly:
+
+```sh
+python3 benches/compare.py target/pgdog_benches/before.json \
+                            target/pgdog_benches/after.json \
+                            --prev-label before --curr-label after
+```
+
+Benchmarks are not required to use this approach. A test can drive hyperfine
+differently, use a different timing tool entirely, or produce its own output —
+`bench.sh` is a convenience, not a contract.
+
+## Adding a new benchmark
+
+Create `benches/<suite>/run.sh`, source `bench.sh`, and call `bench_run`:
+
+```sh
+#!/usr/bin/env bash
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+BENCH_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)"
+
+source "${BENCH_DIR}/bench.sh"
+
+bench_run "<name>" "<command to benchmark>" "$@"
+```
+
+To reset state between hyperfine iterations, pass `--prepare`:
+
+```sh
+bench_run "<name>" "<command>" --prepare "<reset command>" "$@"
+```
+
+The benchmarked script reads these variables from the environment at runtime,
+so the same env vars control both the benchmark harness and the integration
+test when run directly — the integration test simply uses its own defaults.

benches/bench.sh

@@ -0,0 +1,106 @@
+#!/usr/bin/env bash
+# Shared benchmark runner. Source this file and call bench_run.
+#
+# Usage in a test script:
+#   source "${ROOT_DIR}/benches/bench.sh"
+#   bench_run "test_name" "command to run" "$@"
+#
+# Flags forwarded from the caller:
+#   --prepare CMD          command hyperfine runs before each timed run (reset state)
+#   --save-baseline NAME   save results as a named baseline
+#   --baseline NAME        compare against a named baseline (default: previous temp)
+#
+# Environment:
+#   RUNS    hyperfine run count  (default: 3)
+#   WARMUP  hyperfine warmup     (default: 1)
+
+BENCH_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+BENCH_COMPARE="${BENCH_DIR}/compare.py"
+
+bench_run() {
+    local name="$1" cmd="$2"
+    shift 2
+
+    # -- Parse flags ---------------------------------------------------------
+    local prepare_cmd="" save_baseline="" baseline=""
+    while [[ $# -gt 0 ]]; do
+        case $1 in
+            --prepare)       prepare_cmd="$2";    shift 2 ;;
+            --save-baseline) save_baseline="$2"; shift 2 ;;
+            --baseline)      baseline="$2";      shift 2 ;;
+            *) echo "bench_run: unknown flag '$1'" >&2; return 2 ;;
+        esac
+    done
+
+    # -- Logging ------------------------------------------------------------
+    # Suppress all log output so it does not skew timing measurements.
+    export RUST_LOG=error
+
+    # -- Build ---------------------------------------------------------------
+    if [ -z "${PGDOG_BIN:-}" ]; then
+        local root_dir
+        root_dir="$(cd "${BENCH_DIR}/.." && pwd)"
+        echo "Building pgdog (release)..."
+        cargo build --release --manifest-path "${root_dir}/Cargo.toml"
+        export PGDOG_BIN="${root_dir}/target/release/pgdog"
+    fi
+    echo "Binary: $(${PGDOG_BIN} --version 2>&1 || true)"
+    echo ""
+
+    # -- Paths ---------------------------------------------------------------
+    local results_dir="${BENCH_DIR}/../target/pgdog_benches"
+    mkdir -p "${results_dir}"
+
+    local latest_file="${results_dir}/${name}.json"
+    local prev_file prev_label
+
+    if [ -n "${baseline}" ]; then
+        # Named baseline: verify it exists before proceeding.
+        prev_file="${results_dir}/${name}.${baseline}.json"
+        prev_label="${baseline}"
+        if [ ! -f "${prev_file}" ]; then
+            echo "ERROR: baseline '${baseline}' not found (${prev_file})"
+            exit 1
+        fi
+    else
+        # Auto baseline: snapshot current latest before overwriting it.
+        # Skip if the file is empty — an interrupted run leaves a zero-byte file.
+        prev_file="${results_dir}/${name}.prev.json"
+        prev_label="latest"
+        [ -s "${latest_file}" ] && cp "${latest_file}" "${prev_file}"
+    fi
+
+    # -- Run ----------------------------------------------------------------
+    local hyperfine_args=(
+        --runs "${RUNS:-3}"
+        --warmup "${WARMUP:-1}"
+        --show-output
+        --export-json "${latest_file}"
+        --command-name "${name}"
+    )
+    if [ -n "${prepare_cmd}" ]; then
+        hyperfine_args+=(--prepare "${prepare_cmd}")
+    fi
+    hyperfine "${hyperfine_args[@]}" "${cmd}"
+
+    # -- Save named baseline ------------------------------------------------
+    if [ -n "${save_baseline}" ]; then
+        local named_file="${results_dir}/${name}.${save_baseline}.json"
+        if [ -s "${latest_file}" ]; then
+            cp "${latest_file}" "${named_file}"
+            echo "Saved baseline '${save_baseline}' -> ${named_file}"
+        else
+            echo "WARNING: run did not complete cleanly, baseline '${save_baseline}' not saved."
+        fi
+    fi
+
+    # -- Compare ------------------------------------------------------------
+    echo ""
+    if [ -s "${prev_file}" ]; then
+        python3 "${BENCH_COMPARE}" "${prev_file}" "${latest_file}" \
+            --prev-label "${prev_label}" --curr-label "current"
+    else
+        echo "No previous run found -- run again to see comparison."
+        echo ""
+    fi
+}

benches/compare.py

@@ -0,0 +1,143 @@
+#!/usr/bin/env python3
+"""
+Compare two hyperfine JSON results and print a Criterion-style table.
+
+Usage:
+    compare.py PREVIOUS.json CURRENT.json [--prev-label NAME] [--curr-label NAME]
+"""
+
+import json
+import sys
+import argparse
+
+GREEN  = "\033[32m"
+RED    = "\033[31m"
+YELLOW = "\033[33m"
+BOLD   = "\033[1m"
+RESET  = "\033[0m"
+
+
+def supports_color() -> bool:
+    return hasattr(sys.stdout, "isatty") and sys.stdout.isatty()
+
+
+def color(text: str, code: str) -> str:
+    return f"{code}{text}{RESET}" if supports_color() else text
+
+
+def fmt_time(seconds: float | None) -> str:
+    if seconds is None:
+        return "N/A"
+    if seconds >= 60:
+        return f"{seconds / 60:.2f} min"
+    if seconds >= 1:
+        return f"{seconds:.3f} s"
+    if seconds >= 0.001:
+        return f"{seconds * 1000:.3f} ms"
+    return f"{seconds * 1e6:.3f} us"
+
+
+def fmt_bytes(n: int | float | None) -> str:
+    if n is None:
+        return "N/A"
+    for unit, threshold in (("GB", 1 << 30), ("MB", 1 << 20), ("KB", 1 << 10)):
+        if n >= threshold:
+            return f"{n / threshold:.1f} {unit}"
+    return f"{int(n)} B"
+
+
+def fmt_pct(prev: float | None, curr: float | None) -> tuple[str, float | None]:
+    """Returns (formatted string, raw delta %) or ('N/A', None)."""
+    if prev is None or curr is None or prev == 0:
+        return "N/A", None
+    delta = (curr - prev) / prev * 100
+    sign = "+" if delta >= 0 else ""
+    return f"{sign}{delta:.2f}%", delta
+
+
+def load(path: str) -> dict:
+    try:
+        with open(path) as f:
+            data = json.load(f)
+        return data["results"][0]
+    except (json.JSONDecodeError, KeyError, IndexError, OSError) as e:
+        print(f"(skipping comparison: {path} is empty or incomplete — {e})")
+        sys.exit(0)
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("prev", help="previous baseline JSON")
+    ap.add_argument("curr", help="current result JSON")
+    ap.add_argument("--prev-label", default="previous")
+    ap.add_argument("--curr-label", default="current")
+    args = ap.parse_args()
+
+    prev = load(args.prev)
+    curr = load(args.curr)
+
+    name = curr.get("command", "benchmark")
+
+    def peak_mem(result: dict) -> float | None:
+        samples = result.get("memory_usage_byte") or []
+        return max(samples) if samples else None
+
+    rows = [
+        ("mean",    prev.get("mean"),    curr.get("mean")),
+        ("median",  prev.get("median"),  curr.get("median")),
+        ("min",     prev.get("min"),     curr.get("min")),
+        ("max",     prev.get("max"),     curr.get("max")),
+        ("stddev",  prev.get("stddev"),  curr.get("stddev")),
+        ("user",    prev.get("user"),    curr.get("user")),
+        ("system",  prev.get("system"),  curr.get("system")),
+        ("max_mem", peak_mem(prev),       peak_mem(curr)),
+    ]
+    w_label  = 10
+    w_time   = 14
+    w_change = 10
+
+    header_prev = args.prev_label[:w_time].rjust(w_time)
+    header_curr = args.curr_label[:w_time].rjust(w_time)
+
+    print(f"\n{color(name, BOLD)}\n")
+    print(f"  {'':>{w_label}}  {header_prev}  {header_curr}  {'change':>{w_change}}")
+    print(f"  {'-' * w_label}  {'-' * w_time}  {'-' * w_time}  {'-' * w_change}")
+
+    mean_delta = None
+    for label, pv, cv in rows:
+        pct_str, delta = fmt_pct(pv, cv)
+
+        if label == "mean":
+            mean_delta = delta
+
+        if delta is None or abs(delta) < 1.0:
+            pct_colored = pct_str
+        elif delta < 0:
+            # lower is better for time; higher is worse for memory
+            pct_colored = color(pct_str, RED if label == "max_mem" else GREEN)
+        else:
+            pct_colored = color(pct_str, GREEN if label == "max_mem" else RED)
+
+        fmt = fmt_bytes if label == "max_mem" else fmt_time
+        print(
+            f"  {label:>{w_label}}  "
+            f"{fmt(pv):>{w_time}}  "
+            f"{fmt(cv):>{w_time}}  "
+            f"{pct_colored:>{w_change}}"
+        )
+
+    # Verdict
+    print()
+    if mean_delta is None:
+        pass
+    elif abs(mean_delta) < 1.0:
+        print(f"  No change (<1%).")
+    elif mean_delta < 0:
+        print(f"  {color('Performance has improved.', GREEN)}")
+    else:
+        print(f"  {color('Performance has regressed.', RED)}")
+    print()
+
+
+if __name__ == "__main__":
+    main()