diff --git a/bench/latency_probe.py b/bench/latency_probe.py
new file mode 100644
index 00000000..fed67544
--- /dev/null
+++ b/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()
diff --git a/bench/run-tf-bench.sh b/bench/run-tf-bench.sh
new file mode 100755
index 00000000..96f65a5f
--- /dev/null
+++ b/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
diff --git a/src/bootstrap.rs b/src/bootstrap.rs
index 8ac3fc38..a850b973 100644
--- a/src/bootstrap.rs
+++ b/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)
             })
diff --git a/src/buffered_write_layer.rs b/src/buffered_write_layer.rs
index b2fd1fd4..9f060611 100644
--- a/src/buffered_write_layer.rs
+++ b/src/buffered_write_layer.rs
@@ -167,6 +167,91 @@ pub type DeltaWatermark = Vec<Option<walrus_rust::WalPosition>>;
 pub type DeltaWriteCallback =
     Arc<dyn Fn(String, String, Vec<RecordBatch>, DeltaWatermark) -> futures::future::BoxFuture<'static, anyhow::Result<Vec<String>>> + Send + Sync>;
 
+/// Accumulator used by `flush_completed_buckets` to fold every per-bucket
+/// `FlushableBucket` for one (project_id, table_name) into a single combined
+/// commit. Each Delta commit pays a fixed cost (log scan + commit log write +
+/// S3 RTT + tantivy build); coalescing turns N×O(commit) into 1×O(commit).
+#[derive(Default)]
+struct CoalescedGroup {
+    /// `Option` not `String` so the first-bucket sentinel doesn't collide with
+    /// the legitimate empty-project_id path (which falls back to "default" in
+    /// the buffered-layer but reaches this code as `""`). Using `is_empty` as
+    /// the sentinel previously let every subsequent bucket in such a group
+    /// silently re-overwrite project_id/table_name.
+    key:               Option<(String, String)>,
+    batches:           Vec<RecordBatch>,
+    row_count:         usize,
+    /// Sum of per-shard counts across all absorbed buckets — drives walrus
+    /// `advance_by_counts` once for the entire commit.
+    wal_shard_counts:  Vec<u64>,
+    /// Per-shard max position across all absorbed buckets — written into
+    /// Delta commit metadata for crash-mid-flush cursor recovery.
+    wal_positions:     Vec<Option<walrus_rust::WalPosition>>,
+    /// Source bucket_ids; drained from MemBuffer after the combined commit succeeds.
+    source_bucket_ids: Vec<i64>,
+}
+
+struct CombinedBucket {
+    combined:          crate::mem_buffer::FlushableBucket,
+    source_bucket_ids: Vec<i64>,
+}
+
+fn merge_wal_positions(a: Vec<Option<walrus_rust::WalPosition>>, b: Vec<Option<walrus_rust::WalPosition>>) -> Vec<Option<walrus_rust::WalPosition>> {
+    let len = a.len().max(b.len());
+    (0..len)
+        .map(|i| match (a.get(i).cloned().flatten(), b.get(i).cloned().flatten()) {
+            (Some(x), Some(y)) => Some(if (y.block_id, y.offset) > (x.block_id, x.offset) { y } else { x }),
+            (Some(x), None) => Some(x),
+            (None, Some(y)) => Some(y),
+            (None, None) => None,
+        })
+        .collect()
+}
+
+impl CoalescedGroup {
+    fn absorb(&mut self, b: crate::mem_buffer::FlushableBucket) {
+        self.key.get_or_insert_with(|| (b.project_id.clone(), b.table_name.clone()));
+        self.row_count += b.row_count;
+        self.batches.extend(b.batches);
+        // Sum per-shard counts (resizing to the wider length).
+        let len = self.wal_shard_counts.len().max(b.wal_shard_counts.len());
+        self.wal_shard_counts.resize(len, 0);
+        for (i, c) in b.wal_shard_counts.iter().enumerate() {
+            self.wal_shard_counts[i] += *c;
+        }
+        // Merge per-shard positions (max).
+        self.wal_positions = merge_wal_positions(std::mem::take(&mut self.wal_positions), b.wal_positions);
+        self.source_bucket_ids.push(b.bucket_id);
+    }
+
+    fn into_combined_bucket(self) -> CombinedBucket {
+        let CoalescedGroup {
+            key,
+            batches,
+            row_count,
+            wal_shard_counts,
+            wal_positions,
+            source_bucket_ids,
+        } = self;
+        // `absorb` is only called via `groups.entry(..).or_default().absorb(b)`
+        // so `key` is always set by the time we collapse the group.
+        let (project_id, table_name) = key.unwrap_or_default();
+        // Use the max source bucket_id as a stable identifier for tracing only —
+        // drain happens per source_bucket_id, not via this synthetic id.
+        let bucket_id = source_bucket_ids.iter().copied().max().unwrap_or(0);
+        let combined = crate::mem_buffer::FlushableBucket {
+            project_id,
+            table_name,
+            bucket_id,
+            batches,
+            row_count,
+            wal_shard_counts,
+            wal_positions,
+        };
+        CombinedBucket { combined, source_bucket_ids }
+    }
+}
+
 /// Optional callback invoked AFTER a successful Delta commit. Receives the
 /// `(project_id, table_name, batches, added_file_uris)` and is responsible
 /// for building and uploading any sidecar index. The `added_file_uris` are
@@ -202,6 +287,13 @@ pub struct BufferedWriteLayer {
     flush_failed_total:     AtomicU64,
     // Required for WAL replay of UPDATE/DELETE whose SQL references UDFs.
     function_registry:      Arc<crate::functions::FnRegistry>,
+    /// Caps concurrent detached tantivy sidecar builds so a fast flush cycle
+    /// (post-F4 — one build per (project, table) per cycle) can't fan out
+    /// past S3 connection / memory limits when many tables flush together.
+    /// FOLLOW-UP: handles aren't stored; graceful shutdown does not await
+    /// in-flight tantivy uploads. Acceptable for now because the sidecar is
+    /// best-effort and the index can be rebuilt from Delta on demand.
+    tantivy_spawn_sem:      Arc<tokio::sync::Semaphore>,
 }
 
 impl std::fmt::Debug for BufferedWriteLayer {
@@ -245,6 +337,11 @@ impl BufferedWriteLayer {
             flush_completed_total: AtomicU64::new(0),
             flush_failed_total: AtomicU64::new(0),
             function_registry,
+            // 16 is well above realistic per-cycle table fan-out for the
+            // monoscope workload (~5 distinct table names) while still
+            // bounding worst-case S3 / tantivy heap usage if more tables
+            // appear.
+            tantivy_spawn_sem: Arc::new(tokio::sync::Semaphore::new(16)),
         })
     }
 
@@ -357,23 +454,22 @@ impl BufferedWriteLayer {
 
     #[instrument(skip(self, batches), fields(project_id, table_name, batch_count))]
     pub async fn insert(&self, project_id: &str, table_name: &str, batches: Vec<RecordBatch>) -> anyhow::Result<()> {
-        // Check memory pressure and trigger early flush if needed.
-        // We use `flush_all_now` (not `flush_completed_buckets`) here because
-        // under memory pressure the data consuming the budget is almost
-        // always the *current* bucket — `flush_completed_buckets` only
-        // drains buckets older than `bucket_duration_secs`, which leaves
-        // the current-bucket pressure unrelieved and the next insert
-        // still fails. `flush_all_now` includes the current bucket and
-        // matches what an operator would do manually in this state.
+        // Memory pressure no longer triggers a synchronous flush_all_now in the
+        // insert path — that violated the "inserts return fast, Delta happens on
+        // a routine" invariant by stalling pgwire/gRPC threads on S3 commits
+        // (and worse, holding the global flush_lock so one slow tenant froze
+        // ingest for everyone). The safety nets are: (a) `try_reserve_memory`
+        // rejects inserts past the 120% hard limit, surfacing backpressure to
+        // the client; (b) the post-CAS `pressure_notify.notify_one()` already
+        // wakes the background flush task when reservations cross the
+        // configured pressure threshold.
         if self.is_memory_pressure() {
             warn!(
-                "Memory pressure detected ({}MB >= {}MB), triggering early flush_all_now",
+                "Memory pressure (used={}MB / max={}MB) — notifying background flush; insert path will not block on Delta",
                 self.effective_memory_bytes() / (1024 * 1024),
                 self.config.buffer.max_memory_mb()
             );
-            if let Err(e) = self.flush_all_now().await {
-                error!("Early flush due to memory pressure failed: {}", e);
-            }
+            self.pressure_notify.notify_one();
         }
 
         let row_count: usize = batches.iter().map(|b| b.num_rows()).sum();
@@ -739,39 +835,94 @@ impl BufferedWriteLayer {
             return Ok(());
         }
 
-        debug!("Flushing {} buckets to Delta", flushable.len());
+        // Coalesce per (project_id, table_name): one Delta commit per table per
+        // cycle instead of one per bucket. Each commit pays a fixed cost
+        // (log scan + JSON write + S3 RTT + tantivy build), so collapsing
+        // N per-table buckets into a single combined write turns N×O(commit)
+        // into 1×O(commit). Also lets dedup span all flushed time windows.
+        let mut groups: std::collections::HashMap<(String, String), CoalescedGroup> = std::collections::HashMap::new();
+        let bucket_count = flushable.len();
+        for bucket in flushable {
+            let entry = groups.entry((bucket.project_id.clone(), bucket.table_name.clone())).or_default();
+            entry.absorb(bucket);
+        }
+
+        debug!("Flushing {} bucket(s) → {} per-table commit(s)", bucket_count, groups.len());
 
-        // Flush buckets in parallel with bounded concurrency
+        // Flush groups in parallel with bounded concurrency. Per-(project,table)
+        // commits are independent — each Delta table has its own write lock
+        // inside `insert_records_batch`, so parallelism here = cross-table
+        // concurrency.
         let parallelism = self.config.buffer.flush_parallelism();
-        let flush_results: Vec<_> = stream::iter(flushable)
-            .map(|bucket| async move {
-                let result = self.flush_bucket(&bucket).await;
-                (bucket, result)
+        let flush_results: Vec<_> = stream::iter(groups.into_values())
+            .map(|group| async move {
+                let combined = group.into_combined_bucket();
+                let result = self.flush_bucket(&combined.combined).await;
+                (combined, result)
             })
             .buffer_unordered(parallelism)
             .collect()
             .await;
 
-        // Process results: checkpoint WAL and drain MemBuffer for successful flushes
+        // Process results: checkpoint WAL and drain MemBuffer for successful flushes.
+        //
+        // Counter semantics: `flush_completed_total`/`flush_failed_total` continue
+        // to count source bucket IDs (not coalesced groups). Pre-F4 each bucket
+        // was its own commit, so `count = buckets = commits`. Post-F4 it's
+        // `count = buckets ≠ commits`; the per-cycle commit count is
+        // `groups.len()` and is visible only in the `Flushing N → M commits`
+        // debug log. Dashboards thresholding on these counters keep their old
+        // numeric meaning (work units done), but a "commits per minute"
+        // dashboard derived from them now overstates real Delta commit rate.
         let mut any_ok = false;
-        for (bucket, result) in flush_results {
+        for (combined, result) in flush_results {
+            let CombinedBucket { combined, source_bucket_ids } = combined;
             match result {
                 Ok(()) => {
-                    self.checkpoint_and_drain(&bucket);
-                    any_ok = true;
-                    crate::metrics::record_flush(true);
-                    self.flush_completed_total.fetch_add(1, Ordering::Relaxed);
-                    debug!(
-                        "Flushed bucket: project={}, table={}, bucket_id={}, rows={}",
-                        bucket.project_id, bucket.table_name, bucket.bucket_id, bucket.row_count
-                    );
+                    // Critical ordering: only drain the buckets from MemBuffer
+                    // AFTER `advance_by_counts` succeeds. If we drained then
+                    // failed to advance, restart would not replay those rows
+                    // (gone from MemBuffer) and the WAL cursor would not
+                    // reflect that they were processed — silent data loss.
+                    // On advance failure we leave the buckets in MemBuffer;
+                    // the next flush cycle re-commits them. Delta's optimistic
+                    // concurrency + dedup_keys handle the duplicate commit
+                    // (last-write-wins on the per-table key set).
+                    match self.wal.advance_by_counts(&combined.project_id, &combined.table_name, &combined.wal_shard_counts) {
+                        Ok(()) => {
+                            for bucket_id in &source_bucket_ids {
+                                self.mem_buffer.drain_bucket(&combined.project_id, &combined.table_name, *bucket_id);
+                            }
+                            any_ok = true;
+                            crate::metrics::record_flush(true);
+                            self.flush_completed_total.fetch_add(source_bucket_ids.len() as u64, Ordering::Relaxed);
+                            debug!(
+                                "Flushed coalesced commit: project={}, table={}, buckets={}, rows={}",
+                                combined.project_id,
+                                combined.table_name,
+                                source_bucket_ids.len(),
+                                combined.row_count
+                            );
+                        }
+                        Err(e) => {
+                            // Treat as a flush failure for metrics purposes —
+                            // the next cycle will retry the commit and the
+                            // advance.
+                            crate::metrics::record_flush(false);
+                            self.flush_failed_total.fetch_add(source_bucket_ids.len() as u64, Ordering::Relaxed);
+                            error!(
+                                "WAL advance_by_counts failed after successful Delta commit (project={}, table={}, buckets={:?}); leaving buckets in MemBuffer for retry — next flush will re-commit (dedup_keys protect downstream): {}",
+                                combined.project_id, combined.table_name, source_bucket_ids, e
+                            );
+                        }
+                    }
                 }
                 Err(e) => {
                     crate::metrics::record_flush(false);
-                    self.flush_failed_total.fetch_add(1, Ordering::Relaxed);
+                    self.flush_failed_total.fetch_add(source_bucket_ids.len() as u64, Ordering::Relaxed);
                     error!(
-                        "Failed to flush bucket: project={}, table={}, bucket_id={}: {}",
-                        bucket.project_id, bucket.table_name, bucket.bucket_id, e
+                        "Failed to flush coalesced commit: project={}, table={}, buckets={:?}: {}",
+                        combined.project_id, combined.table_name, source_bucket_ids, e
                     );
                 }
             }
@@ -817,16 +968,32 @@ impl BufferedWriteLayer {
             Vec::new()
         };
         // Sidecar tantivy index — best-effort, never fails the flush.
-        // We still count the failure so ops can alert on accumulating index
-        // drift (silent UDF-fallback degradation is otherwise invisible).
-        if let Some(ref idx_cb) = self.tantivy_index_callback
-            && let Err(e) = idx_cb(bucket.project_id.clone(), bucket.table_name.clone(), batches, added_files).await
-        {
-            crate::metrics::record_tantivy_build_failure();
-            warn!(
-                "Tantivy index build failed (non-fatal): project={}, table={}, bucket_id={}: {}",
-                bucket.project_id, bucket.table_name, bucket.bucket_id, e
-            );
+        // Spawned as a detached task so the Delta commit critical path doesn't
+        // wait on tar.zst + S3 upload (a per-bucket cost that was dominating
+        // flush latency at prod scale). F4 already collapses N bucket flushes
+        // into one tantivy build per (project, table) per cycle; the semaphore
+        // bounds the worst-case fan-out (many tables flushing simultaneously)
+        // so concurrent uploads can't saturate S3 connections or grow tantivy
+        // writer heap unbounded.
+        if let Some(ref idx_cb) = self.tantivy_index_callback {
+            let cb = idx_cb.clone();
+            let pid = bucket.project_id.clone();
+            let tname = bucket.table_name.clone();
+            let bid = bucket.bucket_id;
+            let sem = self.tantivy_spawn_sem.clone();
+            tokio::spawn(async move {
+                let _permit = match sem.acquire_owned().await {
+                    Ok(p) => p,
+                    Err(_) => return, // semaphore closed — process is shutting down
+                };
+                if let Err(e) = cb(pid.clone(), tname.clone(), batches, added_files).await {
+                    crate::metrics::record_tantivy_build_failure();
+                    warn!(
+                        "Tantivy index build failed (non-fatal): project={}, table={}, bucket_id={}: {}",
+                        pid, tname, bid, e
+                    );
+                }
+            });
         }
         Ok(())
     }
diff --git a/src/database.rs b/src/database.rs
index 01a27f19..c4ff6de0 100644
--- a/src/database.rs
+++ b/src/database.rs
@@ -2309,9 +2309,14 @@ impl Database {
             use_queue = Empty,
         )
     )]
+    /// Insert batches and return the URIs of files newly added by this commit
+    /// (empty for the buffered-layer / batch-queue paths where the actual
+    /// Delta write happens later). Callers use the returned list to drive
+    /// cache warming and the tantivy sidecar without paying for a second
+    /// `update_state()` log scan.
     pub async fn insert_records_batch(
         &self, project_id: &str, table_name: &str, batches: Vec<RecordBatch>, skip_queue: bool, watermark: Option<&crate::buffered_write_layer::DeltaWatermark>,
-    ) -> Result<()> {
+    ) -> Result<Vec<String>> {
         let span = tracing::Span::current();
         // Normalize timezone-as-offset (`+00:00`) timestamp columns to the
         // IANA `"UTC"` form. Delta-rs Arrow→Delta schema conversion only
@@ -2340,10 +2345,12 @@ impl Database {
         // Use provided table_name or default to otel_logs_and_spans
         let table_name = if table_name.is_empty() { "otel_logs_and_spans".to_string() } else { table_name.to_string() };
 
-        // If buffered layer is configured and not skipping, use it (WAL → MemBuffer flow)
+        // If buffered layer is configured and not skipping, use it (WAL → MemBuffer flow).
+        // No files are written synchronously on this path; an empty URI list is correct.
         if !skip_queue && let Some(ref layer) = self.buffered_layer {
             span.record("use_queue", "buffered_layer");
-            return layer.insert(&project_id, &table_name, batches).await;
+            layer.insert(&project_id, &table_name, batches).await?;
+            return Ok(Vec::new());
         }
 
         // Fallback to legacy batch queue if configured
@@ -2358,7 +2365,7 @@ impl Database {
                     return Err(anyhow::anyhow!("Queue error: {}", e));
                 }
             }
-            return Ok(());
+            return Ok(Vec::new());
         }
 
         span.record("use_queue", false);
@@ -2395,6 +2402,12 @@ impl Database {
                 debug!("Failed to update table state before write (attempt {}): {}", retry_count + 1, e);
             }
 
+            // Snapshot the pre-write file URIs while the write lock is held and
+            // update_state has just run. Cheap in-memory walk of add_actions; the
+            // post-write diff replaces two redundant `list_file_uris` calls
+            // (each of which would re-acquire the write lock + re-run update_state).
+            let pre_uris: std::collections::HashSet<String> = table.get_file_uris().map(|it| it.collect()).unwrap_or_default();
+
             let write_span = tracing::trace_span!(parent: &span, "delta.write_operation", retry_attempt = retry_count + 1);
             let write_result = async {
                 // Schema evolution enabled: new columns will be automatically added to the table
@@ -2427,12 +2440,16 @@ impl Database {
 
                     *table = new_table;
 
+                    // Compute added files from the post-write snapshot while we still
+                    // hold the write lock — no extra log scan required.
+                    let added: Vec<String> = table.get_file_uris().map(|it| it.filter(|u| !pre_uris.contains(u)).collect()).unwrap_or_default();
+
                     // Invalidate statistics cache after successful write
                     drop(table); // Release write lock before async operation
                     self.statistics_extractor.invalidate(&project_id, &table_name).await;
                     debug!("Invalidated statistics cache after write to {}/{}", project_id, table_name);
 
-                    return Ok(());
+                    return Ok(added);
                 }
                 Err(e) => {
                     let error_str = e.to_string();
@@ -3331,6 +3348,28 @@ impl ProjectRoutingTable {
         filters.iter().find_map(crate::optimizers::extract_project_id_from_expr)
     }
 
+    /// pgwire-INSERT fast path. Skips `DataSinkExec` + `ValuesExec` entirely:
+    /// caller (the plan_cache hook) has already materialized the incoming
+    /// VALUES into a RecordBatch from substituted literals, so we just run
+    /// the per-batch fixups (`convert_variant_columns`, project-id routing,
+    /// `normalize_timestamp_tz` is run inside `insert_records_batch`) and
+    /// hand straight to `insert_records_batch` → `BufferedWriteLayer.insert`.
+    /// Returns the inserted row count.
+    pub async fn fast_insert_batch(&self, batch: RecordBatch) -> DFResult<u64> {
+        let total_rows = batch.num_rows() as u64;
+        if total_rows == 0 {
+            return Ok(0);
+        }
+        let project_id = extract_project_id(&batch).unwrap_or_else(|| self.default_project.clone());
+        let target_schema = self.real_schema();
+        let converted = convert_variant_columns(batch, &target_schema)?;
+        self.database
+            .insert_records_batch(&project_id, &self.table_name, vec![converted], false, None)
+            .await
+            .map_err(|e| DataFusionError::Execution(format!("fast_insert_batch for project {} table {}: {}", project_id, self.table_name, e)))?;
+        Ok(total_rows)
+    }
+
     fn schema(&self) -> SchemaRef {
         // Present Variant cols as Utf8View at the table-provider boundary so the SQL planner's
         // INSERT VALUES type check accepts JSON string literals (arrow has no Utf8→Struct cast).
diff --git a/src/dml.rs b/src/dml.rs
index 0d114902..80c4716b 100644
--- a/src/dml.rs
+++ b/src/dml.rs
@@ -307,6 +307,7 @@ fn find_table_scan_name(plan: &LogicalPlan) -> Option<String> {
 /// Given a `Join` and the target table name, decide which child is the target
 /// (the side that scans the target table) and extract equi-join key pairs in
 /// `(target_col_name, source_col_name)` order.
+#[allow(clippy::type_complexity)] // Tuple shape is the natural result of "(target, source, key_pairs)" and a named type would be one-shot.
 fn identify_target_side<'a>(join: &'a Join, target_table_name: &str) -> Result<(&'a LogicalPlan, &'a LogicalPlan, Vec<(String, String)>)> {
     let left_scan = find_table_scan_name(&join.left);
     let right_scan = find_table_scan_name(&join.right);
diff --git a/src/grpc_handlers.rs b/src/grpc_handlers.rs
index a56589f9..af1645f9 100644
--- a/src/grpc_handlers.rs
+++ b/src/grpc_handlers.rs
@@ -148,7 +148,10 @@ async fn process_one(db: &Database, msg: WriteBatch) -> WriteAck {
     let result = decode_and_insert(&msg.arrow_ipc, |batch| {
         let project_id = project_id.clone();
         let table_name = table_name.clone();
-        async move { db.insert_records_batch(&project_id, &table_name, vec![batch], false, None).await }
+        async move {
+            db.insert_records_batch(&project_id, &table_name, vec![batch], false, None).await?;
+            Ok(())
+        }
     })
     .await;
 
diff --git a/src/main.rs b/src/main.rs
index 31ba4664..06a69021 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -85,18 +85,13 @@ async fn async_main(cfg: &'static AppConfig) -> anyhow::Result<()> {
               wal_watermark: timefusion::buffered_write_layer::DeltaWatermark| {
             let db = db_for_callback.clone();
             Box::pin(async move {
-                // Capture pre-state file URIs so we can derive the post-write delta.
-                let pre = db.list_file_uris(&project_id, &table_name).await.unwrap_or_default();
-                // skip_queue=true to write directly to Delta. Watermark goes into
-                // Delta commit metadata for crash-mid-flush recovery.
-                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();
+                // insert_records_batch returns the URIs of files newly added by this
+                // commit, derived from the post-write snapshot under the same write
+                // lock — no second log scan. Watermark goes into Delta commit metadata
+                // for crash-mid-flush recovery.
+                let added = db.insert_records_batch(&project_id, &table_name, batches, true, Some(&wal_watermark)).await?;
                 // Warm the just-flushed files into the cache so the first
                 // dashboard read of this partition hits Foyer, not S3.
-                // Fire-and-forget (spawns internally); clone since `added` is
-                // also returned for the sidecar tantivy indexer.
                 db.warm_cache_for_table(&project_id, &table_name, added.clone());
                 if !added.is_empty() {
                     db.mark_delta_has_files(&project_id, &table_name);
diff --git a/src/plan_cache.rs b/src/plan_cache.rs
index a9bc0d83..b8b49a0e 100644
--- a/src/plan_cache.rs
+++ b/src/plan_cache.rs
@@ -28,21 +28,264 @@
 //! also gain a schema-version token in the key (e.g. an `Arc<AtomicU64>`
 //! bumped on each reload) or a full flush on reload.
 
+use std::sync::Arc;
+
 use async_trait::async_trait;
 use datafusion::{
-    logical_expr::LogicalPlan,
+    arrow::{array::UInt64Array, datatypes::Schema, record_batch::RecordBatch},
+    common::{
+        ParamValues,
+        tree_node::{Transformed, TreeNode},
+    },
+    logical_expr::{Cast, Expr, LogicalPlan, dml::WriteOp},
     prelude::SessionContext,
+    scalar::ScalarValue,
     sql::{parser::Statement as DfStatement, sqlparser::ast::Statement},
 };
 use datafusion_postgres::{
     hooks::{HookClient, QueryHook},
     pgwire::{
-        api::{ClientInfo, results::Response},
+        api::{
+            ClientInfo,
+            results::{Response, Tag},
+        },
         error::{PgWireError, PgWireResult},
     },
 };
 use tracing::debug;
 
+/// Walk a plan and replace every `CAST(Literal(v), T)` with `Literal(cast(v, T))`.
+///
+/// After `replace_params_with_values` substitutes `$N → literal`, the `CAST`
+/// wrappers `insert_coerce` puts around every placeholder turn into per-cell
+/// `CAST(Literal, T)` exprs inside `ValuesExec`. Executing those casts at
+/// query time, once per (row, column), is responsible for ~9–10 ms/row of
+/// pgwire-INSERT overhead at the 88-col schema (measured). The cast values
+/// are constant so we can fold them once, at substitution time, and let
+/// `ValuesExec` see plain literals.
+fn fold_literal_casts(plan: LogicalPlan) -> datafusion::error::Result<LogicalPlan> {
+    plan.transform_up(|node| {
+        let new_exprs: Vec<Expr> = node
+            .expressions()
+            .into_iter()
+            .map(|expr| {
+                expr.transform_up(|e| {
+                    if let Expr::Cast(Cast { expr, data_type }) = &e
+                        && let Expr::Literal(value, metadata) = expr.as_ref()
+                    {
+                        return match value.cast_to(data_type) {
+                            Ok(folded) => Ok(Transformed::yes(Expr::Literal(folded, metadata.clone()))),
+                            // If a particular literal can't be cast (e.g. lossy
+                            // string-→-number), leave it alone — the executor's
+                            // cast will surface a clear error. Trace-level so a
+                            // run-time spike is visible without polluting the
+                            // hot path.
+                            Err(err) => {
+                                tracing::trace!(target: "plan_cache", %err, ?value, ?data_type, "fold_literal_casts: cast_to failed, leaving CAST for executor");
+                                Ok(Transformed::no(e))
+                            }
+                        };
+                    }
+                    Ok(Transformed::no(e))
+                })
+                .map(|t| t.data)
+            })
+            .collect::<datafusion::error::Result<_>>()?;
+        let rebuilt = node.with_new_exprs(new_exprs, node.inputs().into_iter().cloned().collect())?;
+        Ok(Transformed::yes(rebuilt))
+    })
+    .map(|t| t.data)
+}
+
+/// pgwire-INSERT bypass: recognise `Dml(Insert) → [Projection →] Values(literals)`
+/// and short-circuit the whole DataFusion executor by building the RecordBatch
+/// directly from the literals and calling `ProjectRoutingTable.fast_insert_batch`.
+/// Skips `ValuesExec`, `DataSinkExec`, and the per-row `replace_params_with_values`
+/// walk that together account for ~5-6 ms/row of overhead at the 88-col schema.
+///
+/// Returns `Ok(Some(rows))` on success, `Ok(None)` if the plan shape isn't
+/// the supported fast-path INSERT (caller should fall back to the regular
+/// `execute_logical_plan` path).
+async fn try_fast_path_insert(plan: &LogicalPlan, session_context: &SessionContext) -> datafusion::error::Result<Option<u64>> {
+    use datafusion::logical_expr::dml::DmlStatement;
+
+    let LogicalPlan::Dml(DmlStatement {
+        table_name,
+        op: WriteOp::Insert(_),
+        input,
+        ..
+    }) = plan
+    else {
+        return Ok(None);
+    };
+
+    // Input is either `Projection → Values` (the common case for INSERT INTO
+    // t (cols) VALUES …) or `Values` directly. We use the Values schema for
+    // building the RecordBatch (row width must match cell count), and where
+    // there is a Projection above, we require each projection expr to be a
+    // simple column rename: `column1 AS target_col`. After `fold_literal_casts`
+    // this is true for parameterised pgwire INSERTs where `insert_coerce`
+    // already pushed the target type down into each placeholder. Anything
+    // more complex (default expressions, computed columns) falls back to the
+    // executor.
+    // For Projection above Values, each output column is either `Column(..)`
+    // or `Alias(Column(..), name)`. We capture (values_input_idx, output_name)
+    // per projection position so we can reorder Values columns to match the
+    // table's expected layout. Anything more complex (computed cols, casts
+    // we didn't fold, sub-exprs) falls back to the executor.
+    // Each output column maps either to a Values column position (column refs
+    // — the common case) or to a constant the optimizer folded into the
+    // projection (NULL defaults for unspecified columns, etc.). We record
+    // (source, name) per output column so the build step can pull from the
+    // right place.
+    enum ColumnSource {
+        Values(usize),
+        Constant(ScalarValue),
+    }
+    let (column_plan, values): (Option<Vec<(ColumnSource, String)>>, &datafusion::logical_expr::Values) = match input.as_ref() {
+        LogicalPlan::Projection(p) => {
+            let LogicalPlan::Values(v) = p.input.as_ref() else {
+                return Ok(None);
+            };
+            let mut plan: Vec<(ColumnSource, String)> = Vec::with_capacity(p.expr.len());
+            for (i, e) in p.expr.iter().enumerate() {
+                let (inner, name) = match e {
+                    Expr::Alias(a) => (a.expr.as_ref(), a.name.clone()),
+                    other => (other, p.schema.field(i).name().to_string()),
+                };
+                match inner {
+                    Expr::Column(c) => {
+                        let Some(input_idx) = v.schema.fields().iter().position(|f| f.name() == &c.name) else {
+                            return Ok(None);
+                        };
+                        plan.push((ColumnSource::Values(input_idx), name));
+                    }
+                    Expr::Literal(val, _) => {
+                        plan.push((ColumnSource::Constant(val.clone()), name));
+                    }
+                    _ => return Ok(None),
+                }
+            }
+            (Some(plan), v)
+        }
+        LogicalPlan::Values(v) => (None, v),
+        _ => return Ok(None),
+    };
+    let target_schema = values.schema.clone();
+
+    // Every cell must be a literal — possibly wrapped in an Alias from the
+    // pgwire `$N` placeholder name retained after substitution. Anything else
+    // legitimately needs the full executor (subqueries, function calls,
+    // correlated refs, etc.).
+    fn cell_as_literal(e: &Expr) -> Option<&ScalarValue> {
+        match e {
+            Expr::Literal(v, _) => Some(v),
+            Expr::Alias(a) => cell_as_literal(&a.expr),
+            _ => None,
+        }
+    }
+    if !values.values.iter().all(|row| row.iter().all(|e| cell_as_literal(e).is_some())) {
+        return Ok(None);
+    }
+
+    let values_schema: Arc<Schema> = Arc::new(target_schema.as_arrow().clone());
+    let num_values_cols = values_schema.fields().len();
+    let num_rows = values.values.len();
+
+    // Build one array per Values column (in Values' native order).
+    let mut values_columns: Vec<datafusion::arrow::array::ArrayRef> = Vec::with_capacity(num_values_cols);
+    for col_idx in 0..num_values_cols {
+        let target_ty = values_schema.field(col_idx).data_type();
+        if num_rows == 0 {
+            values_columns.push(datafusion::arrow::array::new_empty_array(target_ty));
+            continue;
+        }
+        let scalars: Vec<ScalarValue> = values
+            .values
+            .iter()
+            .map(|row| {
+                cell_as_literal(&row[col_idx])
+                    .cloned()
+                    .ok_or_else(|| datafusion::error::DataFusionError::Internal(format!("try_fast_path_insert: expected Literal, got {:?}", row[col_idx])))
+            })
+            .collect::<datafusion::error::Result<_>>()?;
+        let arr = ScalarValue::iter_to_array(scalars)?;
+        // `iter_to_array` may return a different concrete type than the
+        // Values column declares (e.g. all-NULL columns come back as Null).
+        // Cast back to target so the downstream MemBuffer schema check sees
+        // exactly what the table expects.
+        let arr = if arr.data_type() == target_ty {
+            arr
+        } else {
+            datafusion::arrow::compute::cast(&arr, target_ty).map_err(|e| datafusion::error::DataFusionError::ArrowError(Box::new(e), None))?
+        };
+        values_columns.push(arr);
+    }
+
+    // Apply the projection: pull Values columns by index, or materialize a
+    // constant array for projection cells the optimizer folded to a literal
+    // (NULL defaults for columns not in the INSERT, etc.).
+    let (final_schema, columns) = if let Some(plan) = column_plan {
+        let n = num_rows.max(1);
+        let mut fields: Vec<Arc<datafusion::arrow::datatypes::Field>> = Vec::with_capacity(plan.len());
+        let mut cols: Vec<datafusion::arrow::array::ArrayRef> = Vec::with_capacity(plan.len());
+        for (src, name) in &plan {
+            match src {
+                ColumnSource::Values(idx) => {
+                    let f = values_schema.field(*idx);
+                    fields.push(Arc::new(datafusion::arrow::datatypes::Field::new(name, f.data_type().clone(), f.is_nullable())));
+                    cols.push(values_columns[*idx].clone());
+                }
+                ColumnSource::Constant(val) => {
+                    let arr = val.to_array_of_size(n)?;
+                    fields.push(Arc::new(datafusion::arrow::datatypes::Field::new(name, arr.data_type().clone(), true)));
+                    cols.push(arr);
+                }
+            }
+        }
+        (Arc::new(Schema::new(fields)), cols)
+    } else {
+        (values_schema, values_columns)
+    };
+    let batch = RecordBatch::try_new(final_schema, columns).map_err(|e| datafusion::error::DataFusionError::ArrowError(Box::new(e), None))?;
+
+    let provider = session_context.table_provider(table_name.clone()).await?;
+    let Some(routing) = provider.as_any().downcast_ref::<crate::database::ProjectRoutingTable>() else {
+        return Ok(None);
+    };
+    let rows = routing.fast_insert_batch(batch).await?;
+    Ok(Some(rows))
+}
+
+/// Mirror of the vendored `datafusion_postgres::handlers::dml_completion`,
+/// which is `pub(super)` and so unreachable from outside the crate.
+///
+/// **Re-vendor checklist.** When bumping the vendored `datafusion-postgres`,
+/// diff `vendored handlers.rs::dml_completion` against this implementation —
+/// upstream changes to the tag format ("INSERT 0 N" oid + count), the
+/// `count` column name, or the count column's Arrow type are silent
+/// divergence here (no compile error, wrong wire response). Search for the
+/// `RE-VENDOR-DML-COMPLETION` marker below and confirm parity.
+// RE-VENDOR-DML-COMPLETION: keep in sync with vendor/datafusion-postgres/src/handlers.rs.
+async fn dml_completion(df: datafusion::dataframe::DataFrame) -> PgWireResult<Option<Response>> {
+    let tag = match df.logical_plan() {
+        LogicalPlan::Dml(d) => match d.op {
+            WriteOp::Insert(_) => Tag::new("INSERT").with_oid(0),
+            WriteOp::Update => Tag::new("UPDATE"),
+            WriteOp::Delete => Tag::new("DELETE"),
+            _ => return Ok(None),
+        },
+        _ => return Ok(None),
+    };
+    let batches = df.collect().await.map_err(|e| PgWireError::ApiError(Box::new(e)))?;
+    let rows = batches
+        .first()
+        .and_then(|b| b.column_by_name("count"))
+        .and_then(|c| c.as_any().downcast_ref::<UInt64Array>())
+        .map_or(0, |a| a.value(0) as usize);
+    Ok(Some(Response::Execution(tag.with_rows(rows))))
+}
+
 const DEFAULT_PLAN_CACHE_CAPACITY: usize = 256;
 
 /// Singleton handle so `timefusion_stats` can read the same cache the
@@ -245,10 +488,42 @@ impl QueryHook for PlanCacheHook {
     }
 
     async fn handle_extended_query(
-        &self, _statement: &Statement, _logical_plan: &LogicalPlan, _params: &datafusion::common::ParamValues, _session_context: &SessionContext,
-        _client: &mut dyn HookClient,
+        &self, _statement: &Statement, logical_plan: &LogicalPlan, params: &ParamValues, session_context: &SessionContext, _client: &mut dyn HookClient,
     ) -> Option<PgWireResult<Response>> {
-        None
+        // Only intercept DML — for SELECTs the vendored path is fine.
+        // The win here is post-substitution constant folding of `CAST(Literal, T)`
+        // exprs that `insert_coerce` puts around every placeholder; folding them
+        // before `ValuesExec` evaluates the plan saves ~9–10 ms per inserted
+        // row on the 88-col schema (measured).
+        if !matches!(logical_plan, LogicalPlan::Dml(_)) {
+            return None;
+        }
+        let substituted = match logical_plan.clone().replace_params_with_values(params) {
+            Ok(p) => p,
+            Err(e) => return Some(Err(PgWireError::ApiError(Box::new(e)))),
+        };
+        let folded = match fold_literal_casts(substituted) {
+            Ok(p) => p,
+            Err(e) => return Some(Err(PgWireError::ApiError(Box::new(e)))),
+        };
+        // Fast-path: `Dml(Insert) → [Projection →] Values(literals)` skips the
+        // executor entirely and writes the batch straight into the buffered
+        // layer. Saves the ~5-6 ms/row that `ValuesExec` + `DataSinkExec`
+        // were costing at the 88-col schema.
+        match try_fast_path_insert(&folded, session_context).await {
+            Ok(Some(rows)) => return Some(Ok(Response::Execution(Tag::new("INSERT").with_oid(0).with_rows(rows as usize)))),
+            Ok(None) => {} // fall through to the normal executor
+            Err(e) => return Some(Err(PgWireError::ApiError(Box::new(e)))),
+        }
+        let df = match session_context.execute_logical_plan(folded).await {
+            Ok(df) => df,
+            Err(e) => return Some(Err(PgWireError::ApiError(Box::new(e)))),
+        };
+        Some(match dml_completion(df).await {
+            Ok(Some(resp)) => Ok(resp),
+            Ok(None) => Err(PgWireError::ApiError("internal error: DML plan returned non-DML completion".to_string().into())),
+            Err(e) => Err(e),
+        })
     }
 
     /// Signal to the do_query path that any plan we returned is already