perf: speed up pgwire INSERT path and flush throughput

bench/latency_probe.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+"""Measure per-INSERT latency at the server with the same row shape as
+concurrent_load.py — but synchronously, one batch at a time, to pin
+TF's per-query cost vs client-side overhead."""
+import gzip, json, os, sys, time, uuid
+from datetime import datetime, timezone, timedelta
+from pathlib import Path
+import psycopg
+
+ROOT = Path(__file__).resolve().parent
+DUMP = ROOT / "data" / "sample.jsonl.gz"
+URL = "host=127.0.0.1 port=12345 user=postgres password=postgres dbname=postgres"
+
+# Same columns as concurrent_load.py
+sys.path.insert(0, str(ROOT))
+from concurrent_load import COLUMNS, JSON_COLS, LIST_COLS, _to_pg, load_rows, shift_for_project
+
+def main():
+    rows = load_rows(2000)
+    shift = datetime.now(timezone.utc) - max(datetime.fromisoformat(r["timestamp"]) for r in rows)
+    pid = f"latprobe-{uuid.uuid4().hex[:6]}"
+    rows = shift_for_project(rows, pid, shift)
+
+    placeholders = "(" + ", ".join(["%s"] * len(COLUMNS)) + ")"
+    base = f"INSERT INTO otel_logs_and_spans ({', '.join(COLUMNS)}) VALUES "
+
+    for batch_size in (1, 10, 30, 100, 500, 1000):
+        with psycopg.connect(URL, autocommit=True) as c, c.cursor() as cur:
+            times = []
+            for i in range(8):
+                batch = rows[i * batch_size : (i + 1) * batch_size]
+                if len(batch) < batch_size: break
+                flat = []
+                for r in batch:
+                    for col in COLUMNS:
+                        flat.append(_to_pg(col, r.get(col)))
+                t = time.perf_counter()
+                cur.execute(base + ", ".join([placeholders] * len(batch)), flat)
+                times.append((time.perf_counter() - t) * 1000)
+            if not times: continue
+            warm = times[2:]  # drop first 2 cold runs
+            avg = sum(warm) / len(warm) if warm else times[-1]
+            per_row = avg / batch_size
+            r_s = 1000 / per_row if per_row > 0 else 0
+            print(f"batch={batch_size:>4}  warm_avg={avg:7.1f}ms  per_row={per_row:6.3f}ms  → {r_s:7.0f} r/s (1 conn)")
+
+if __name__ == "__main__":
+    main()

bench/run-tf-bench.sh

@@ -0,0 +1,42 @@
+#!/usr/bin/env bash
+# Restart TF for a clean benchmark iteration.
+# Usage: ./bench/run-tf-bench.sh <label>
+set -euo pipefail
+label="${1:-bench}"
+data_dir="./data/bench-${label}"
+log="/tmp/tf-${label}.log"
+
+pkill -f 'target/(debug|release)/timefusion' 2>/dev/null || true
+sleep 1
+rm -rf "$data_dir"
+mkdir -p "$data_dir"
+AWS_ACCESS_KEY_ID=minioadmin AWS_SECRET_ACCESS_KEY=minioadmin \
+  aws --endpoint-url http://127.0.0.1:9000 s3 rm s3://timefusion-bench --recursive >/dev/null 2>&1 || true
+
+set -a; source .env; set +a
+# Bench-specific overrides
+export AWS_S3_BUCKET=timefusion-bench
+export TIMEFUSION_DATA_DIR="$data_dir"
+export TIMEFUSION_TABLE_PREFIX=bench
+export RUST_LOG=warn,timefusion=info
+export TIMEFUSION_BUFFER_FLUSH_INTERVAL_SECS=60
+export TIMEFUSION_BUFFER_MAX_MEMORY_MB=2048
+export TIMEFUSION_ALLOW_INSECURE_AUTH=true
+export MAX_PG_CONNECTIONS=64
+# Production-shaped batches; this is the prod symptom — many small inserts.
+export TIMEFUSION_BATCH_QUEUE_CAPACITY=10000
+# Silence prod OTel export — it floods the log with oversized-batch errors
+# and skews the bench. Pointing at a sink that doesn't exist:
+unset OTEL_EXPORTER_OTLP_ENDPOINT
+export OTEL_SDK_DISABLED=true
+
+nohup ./target/debug/timefusion >"$log" 2>&1 &
+echo $! > /tmp/tf.pid
+for i in $(seq 1 60); do
+  if nc -z 127.0.0.1 12345 2>/dev/null; then
+    echo "TF[$label] up after ${i}s (pid=$(cat /tmp/tf.pid))"
+    exit 0
+  fi
+  sleep 1
+done
+echo "TF[$label] failed to start"; tail -20 "$log"; exit 1

src/bootstrap.rs

@@ -49,11 +49,7 @@ pub async fn bootstrap(cfg: Arc<AppConfig>) -> Result<Bootstrapped> {
         move |project_id: String, table_name: String, batches: Vec<RecordBatch>, wal_watermark: DeltaWatermark| {
             let db = db_for_callback.clone();
             Box::pin(async move {
-                let pre = db.list_file_uris(&project_id, &table_name).await.unwrap_or_default();
-                db.insert_records_batch(&project_id, &table_name, batches, true, Some(&wal_watermark)).await?;
-                let post = db.list_file_uris(&project_id, &table_name).await.unwrap_or_default();
-                let pre_set: std::collections::HashSet<String> = pre.into_iter().collect();
-                let added: Vec<String> = post.into_iter().filter(|u| !pre_set.contains(u)).collect();
+                let added = db.insert_records_batch(&project_id, &table_name, batches, true, Some(&wal_watermark)).await?;
                 db.warm_cache_for_table(&project_id, &table_name, added.clone());
                 Ok(added)
             })