From 5119d3da2a82be9ad0e6fd256e4b6f39dea95480 Mon Sep 17 00:00:00 2001
From: Anthony Alaribe <anthonyalaribe@gmail.com>
Date: Tue, 9 Jun 2026 17:34:56 +0200
Subject: [PATCH 1/3] perf: speed up pgwire INSERT path and flush throughput
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Five changes targeting the production symptom of TF being unable to
keep up with consumer ingest rate. Validated with bench/concurrent_load.py
and bench/latency_probe.py against a local MinIO.

1. buffered_write_layer.rs — insert path no longer awaits flush_all_now
   under memory pressure. The previous behaviour stalled pgwire/gRPC
   threads on S3 commits and held the global flush_lock so one slow
   tenant froze ingest for everyone. The safety net is the existing
   hard-limit reject in try_reserve_memory + pressure_notify waking the
   background flush task.

2. database.rs — insert_records_batch now returns the URIs of files
   newly added by the commit, derived from the post-write snapshot under
   the same write lock. Callers (main.rs, bootstrap.rs) drop their pre/
   post list_file_uris calls. Eliminates two redundant update_state log
   scans per flushed bucket — at prod scale (~2,700 add_files) that's
   the ~200-300ms per-bucket tax that was dominating flush latency.

3. buffered_write_layer.rs — flush_completed_buckets coalesces per
   (project_id, table_name): one Delta commit per (project, table) per
   cycle instead of one per bucket. Each commit pays a fixed
   log-scan + JSON + S3 RTT, so collapsing N micro-commits into one
   turns N×O(commit) into 1×O(commit). Also eliminates read-write
   contention measured as p95 dropping from 122-151ms to 70-85ms on the
   8-writer + 4-reader bench.

4. buffered_write_layer.rs — tantivy index build spawned off the flush
   critical path. The Delta commit no longer waits on tar.zst + S3
   upload; F4's coalescing naturally bounds the per-cycle tantivy queue
   depth to num_tables.

5. plan_cache.rs — new handle_extended_query for DML that substitutes
   params, folds CAST(Literal, T) into Literal_T, then executes. The
   insert_coerce pass wraps every $N in CAST so pgwire infers
   placeholder types; after substitution those CAST(literal, T) exprs
   stayed in every ValuesExec cell and were re-evaluated per (row, col)
   — at 88 cols x 30+ rows this was ~9-10ms/row of pure pipeline
   overhead, measured. Folding them up front drops single-conn
   throughput from 100 r/s to 161 r/s (+61%) and 16-writer throughput
   from ~440 r/s to 607 r/s (+38%).
---
 bench/latency_probe.py      |  48 +++++++++
 bench/run-tf-bench.sh       |  42 ++++++++
 src/bootstrap.rs            |   8 +-
 src/buffered_write_layer.rs | 194 +++++++++++++++++++++++++++++-------
 src/database.rs             |  30 +++++-
 src/grpc_handlers.rs        |   5 +-
 src/main.rs                 |  17 ++--
 src/plan_cache.rs           | 102 ++++++++++++++++++-
 8 files changed, 385 insertions(+), 61 deletions(-)
 create mode 100644 bench/latency_probe.py
 create mode 100755 bench/run-tf-bench.sh

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..a8b3d54a 100644
--- a/src/bootstrap.rs
+++ b/src/bootstrap.rs
@@ -49,11 +49,9 @@ 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..9e0d1b6a 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 {
+    project_id:       String,
+    table_name:       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) {
+        if self.project_id.is_empty() {
+            self.project_id = b.project_id.clone();
+            self.table_name = 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 {
+            project_id,
+            table_name,
+            batches,
+            row_count,
+            wal_shard_counts,
+            wal_positions,
+            source_bucket_ids,
+        } = self;
+        // 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
@@ -357,23 +442,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,14 +823,32 @@ 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_insert_with(CoalescedGroup::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()
@@ -754,24 +856,35 @@ impl BufferedWriteLayer {
 
         // Process results: checkpoint WAL and drain MemBuffer for successful flushes
         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);
+                    // advance_by_counts uses the merged shard counts (sums across
+                    // source buckets); drain each source bucket_id from MemBuffer.
+                    if let Err(e) = self.wal.advance_by_counts(&combined.project_id, &combined.table_name, &combined.wal_shard_counts) {
+                        warn!("WAL advance_by_counts failed: {}", e);
+                    }
+                    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(1, Ordering::Relaxed);
+                    self.flush_completed_total.fetch_add(source_bucket_ids.len() as u64, Ordering::Relaxed);
                     debug!(
-                        "Flushed bucket: project={}, table={}, bucket_id={}, rows={}",
-                        bucket.project_id, bucket.table_name, bucket.bucket_id, bucket.row_count
+                        "Flushed coalesced commit: project={}, table={}, buckets={}, rows={}",
+                        combined.project_id,
+                        combined.table_name,
+                        source_bucket_ids.len(),
+                        combined.row_count
                     );
                 }
                 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 +930,25 @@ 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). The callback is responsible for any
+        // internal bounding; flush throughput is naturally bounded by F4
+        // (one tantivy build per (project, table) per cycle, not per bucket).
+        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;
+            tokio::spawn(async move {
+                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..281970cb 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,19 @@ 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();
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..92e314fc 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -85,18 +85,15 @@ 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..46b1af2b 100644
--- a/src/plan_cache.rs
+++ b/src/plan_cache.rs
@@ -30,19 +30,87 @@
 
 use async_trait::async_trait;
 use datafusion::{
-    logical_expr::LogicalPlan,
+    arrow::array::UInt64Array,
+    common::{
+        ParamValues,
+        tree_node::{Transformed, TreeNode},
+    },
+    logical_expr::{Cast, Expr, LogicalPlan, dml::WriteOp},
     prelude::SessionContext,
     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.
+                            Err(_) => 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)
+}
+
+/// Mirror of the vendored `datafusion_postgres::handlers::dml_completion` which
+/// is private. Drives the Insert/Update/Delete dataframe to completion and
+/// builds a `Tag::new(<verb>).with_rows(N)` Execution response.
+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 +313,36 @@ impl QueryHook for PlanCacheHook {
     }
 
     async fn handle_extended_query(
-        &self, _statement: &Statement, _logical_plan: &LogicalPlan, _params: &datafusion::common::ParamValues, _session_context: &SessionContext,
+        &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)))),
+        };
+        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

From 39aa9930dd6e9ca3990a18dc144aa976323d0836 Mon Sep 17 00:00:00 2001
From: Anthony Alaribe <anthonyalaribe@gmail.com>
Date: Tue, 9 Jun 2026 18:19:38 +0200
Subject: [PATCH 2/3] perf: pgwire INSERT fast path bypasses DataFusion
 executor
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Extends the plan_cache hook to short-circuit `Dml(Insert) → [Projection →]
Values(literals)` after `replace_params_with_values`. Builds the
RecordBatch from the substituted literals (with column reorder + rename
to match the projection), looks up the target ProjectRoutingTable, and
calls a new `fast_insert_batch` that does just the per-batch fixups
(`convert_variant_columns`, project-id routing) before
`insert_records_batch`. Skips ValuesExec, DataSinkExec, the
write_all stream loop, and the per-cell expression evaluation that was
the residual ~5-6 ms/row cost after the CAST(Literal) fold.

Handles two non-trivial projection shapes that the optimizer produces:
- column reorder (Projection emits columns in a different order than Values)
- constant-folded NULL defaults (`Alias(Literal, name)` for cols absent
  from the INSERT)

Non-matching plans (subqueries in VALUES, UDFs, etc.) fall back to the
existing fold→execute path. SELECTs still go to the vendored handler.

Measured on bench/latency_probe.py + bench/concurrent_load.py with the
88-col `otel_logs_and_spans` schema (debug build, local MinIO):

  scenario          master      after PR45     after this PR
  1 conn b=100      100 r/s     161 r/s        359 r/s   (3.6x)
  1 conn b=500      102 r/s     162 r/s        366 r/s
  8 writers         430 r/s     364 r/s        1,197 r/s (2.8x)
  16 writers        ~440 r/s    607 r/s        1,218 r/s (2.8x)
  32 writers        188 r/s     137 r/s        340 r/s   (1.8x)

Per-row latency dropped 9.8 ms → 2.8 ms at batch=100.

Fast-path hit-rate in the bench was 100% (2,919/2,919 inserts), zero
fallback skips for the all-placeholder 88-col INSERT shape monoscope
sends.
---
 src/database.rs   |  24 +++++
 src/plan_cache.rs | 243 +++++++++++++++++++++++++++++++++++++++++++++-
 2 files changed, 264 insertions(+), 3 deletions(-)

diff --git a/src/database.rs b/src/database.rs
index 281970cb..ba9e7538 100644
--- a/src/database.rs
+++ b/src/database.rs
@@ -3351,6 +3351,30 @@ 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/plan_cache.rs b/src/plan_cache.rs
index 46b1af2b..c4e02a4e 100644
--- a/src/plan_cache.rs
+++ b/src/plan_cache.rs
@@ -28,16 +28,24 @@
 //! 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::{
-    arrow::array::UInt64Array,
+    arrow::{array::UInt64Array, datatypes::Schema, record_batch::RecordBatch},
+    catalog::MemTable,
     common::{
         ParamValues,
         tree_node::{Transformed, TreeNode},
     },
-    logical_expr::{Cast, Expr, LogicalPlan, dml::WriteOp},
+    logical_expr::{Cast, Expr, LogicalPlan, LogicalPlanBuilder, dml::WriteOp},
     prelude::SessionContext,
-    sql::{parser::Statement as DfStatement, sqlparser::ast::Statement},
+    scalar::ScalarValue,
+    sql::{
+        TableReference,
+        parser::Statement as DfStatement,
+        sqlparser::ast::Statement,
+    },
 };
 use datafusion_postgres::{
     hooks::{HookClient, QueryHook},
@@ -89,6 +97,226 @@ fn fold_literal_casts(plan: LogicalPlan) -> datafusion::error::Result<LogicalPla
     .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))
+}
+
+/// Replace every `LogicalPlan::Values` whose rows are entirely literal with a
+/// `TableScan` over an in-memory `MemTable` carrying a precomputed
+/// `RecordBatch`. Currently unused — kept for reference; the `try_fast_path_insert`
+/// bypass is cleaner because it also avoids the `Projection`/`DataSinkExec`
+/// stream overhead.
+#[allow(dead_code)]
+fn values_to_memtable(plan: LogicalPlan) -> datafusion::error::Result<LogicalPlan> {
+    plan.transform_up(|node| {
+        let LogicalPlan::Values(values) = &node else {
+            return Ok(Transformed::no(node));
+        };
+        // Only literal-only Values qualify for the shortcut. Anything else
+        // legitimately needs ValuesExec to evaluate.
+        let all_literal = values.values.iter().all(|row| row.iter().all(|e| matches!(e, Expr::Literal(_, _))));
+        if !all_literal {
+            return Ok(Transformed::no(node));
+        }
+        let arrow_schema: Arc<Schema> = Arc::new(values.schema.as_arrow().clone());
+        let num_cols = arrow_schema.fields().len();
+        let num_rows = values.values.len();
+        // Build column-by-column: collect each column's scalar values, then
+        // `ScalarValue::iter_to_array` packs them into an `ArrayRef` in one go.
+        let mut columns: Vec<datafusion::arrow::array::ArrayRef> = Vec::with_capacity(num_cols);
+        for col_idx in 0..num_cols {
+            let target_ty = arrow_schema.field(col_idx).data_type();
+            if num_rows == 0 {
+                columns.push(datafusion::arrow::array::new_empty_array(target_ty));
+                continue;
+            }
+            let scalars: Vec<ScalarValue> = values
+                .values
+                .iter()
+                .map(|row| match &row[col_idx] {
+                    Expr::Literal(v, _) => Ok(v.clone()),
+                    other => Err(datafusion::error::DataFusionError::Internal(format!(
+                        "values_to_memtable: expected Literal, got {other:?}"
+                    ))),
+                })
+                .collect::<datafusion::error::Result<_>>()?;
+            let arr = ScalarValue::iter_to_array(scalars)?;
+            // ScalarValue::iter_to_array can return an array whose type doesn't
+            // exactly match the Values column type (e.g. all-NULL columns come
+            // back as Null arrays). Cast back to the declared type so the
+            // upstream Projection's column refs type-check.
+            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))?
+            };
+            columns.push(arr);
+        }
+        let batch = RecordBatch::try_new(arrow_schema.clone(), columns)
+            .map_err(|e| datafusion::error::DataFusionError::ArrowError(Box::new(e), None))?;
+        let mem_table = MemTable::try_new(arrow_schema, vec![vec![batch]])?;
+        let scan = LogicalPlanBuilder::scan(TableReference::bare("__pgwire_values"), datafusion::datasource::provider_as_source(Arc::new(mem_table)), None)?.build()?;
+        Ok(Transformed::yes(scan))
+    })
+    .map(|t| t.data)
+}
+
 /// Mirror of the vendored `datafusion_postgres::handlers::dml_completion` which
 /// is private. Drives the Insert/Update/Delete dataframe to completion and
 /// builds a `Tag::new(<verb>).with_rows(N)` Execution response.
@@ -332,6 +560,15 @@ impl QueryHook for PlanCacheHook {
             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)))),

From d0cfae41fd68125dc4058c25eed50a5be3afccca Mon Sep 17 00:00:00 2001
From: Anthony Alaribe <anthonyalaribe@gmail.com>
Date: Tue, 9 Jun 2026 19:23:17 +0200
Subject: [PATCH 3/3] review: fix WAL-advance drain order + safer coalesce
 sentinel + cleanup
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Addresses PR review feedback:

- CRITICAL (#1): drain MemBuffer ONLY after WAL advance_by_counts
  succeeds. The previous flow committed to Delta, ignored WAL advance
  failures, and drained anyway — restart would not replay those rows
  (gone from MemBuffer) and the WAL cursor wouldn't reflect them, so
  rows could be lost. Now on advance failure we leave the buckets in
  MemBuffer for the next cycle to retry; Delta dedup_keys make the
  re-commit idempotent on the downstream side.

- #2 fragile sentinel: CoalescedGroup used `project_id.is_empty()` as
  the first-bucket marker, which silently overwrote on every absorb if
  the project_id was ever "". Switched to `Option<(project_id, table)>`
  and `get_or_insert_with`. Also satisfies clippy `unwrap_or_default`
  via `.entry(..).or_default()`.

- #4 unbounded tantivy spawning: added a 16-permit semaphore so a
  cycle where many tables flush together can't fan out beyond S3 /
  tantivy-heap limits. Permit is acquired inside the spawned task so
  the Delta critical path still returns immediately. Graceful-shutdown
  await of in-flight tantivy builds is a follow-up (documented in the
  struct field).

- #5 dml_completion: explicit re-vendor marker so a future
  datafusion-postgres bump can't silently diverge from upstream
  (INSERT tag format / count column name / count Arrow type).

- #6 dead code: deleted the unused values_to_memtable function and
  trimmed its now-unused imports.

- #9 metrics semantics: counter still counts source bucket IDs (work
  units done) — documented the change in meaning so dashboards
  computing "commits per minute" from it don't silently overstate the
  real Delta commit rate.

- nit: trace! log on fold_literal_casts cast misses so unexpected
  fall-throughs are diagnosable without polluting the hot path.

Also fixes pre-existing clippy warnings blocking this PR's CI:
- src/dml.rs:310 `identify_target_side` complex tuple — added a scoped
  `#[allow(clippy::type_complexity)]` with rationale.

Plus `cargo fmt` over the touched files.

Bench numbers unchanged (1-conn batch=100 ~380 r/s, matches the
prior 359 r/s within noise; per-row 2.6 ms/row).
---
 src/bootstrap.rs            |   4 +-
 src/buffered_write_layer.rs | 125 ++++++++++++++++++++++++------------
 src/database.rs             |   9 +--
 src/dml.rs                  |   1 +
 src/main.rs                 |   4 +-
 src/plan_cache.rs           | 122 ++++++++++-------------------------
 6 files changed, 123 insertions(+), 142 deletions(-)

diff --git a/src/bootstrap.rs b/src/bootstrap.rs
index a8b3d54a..a850b973 100644
--- a/src/bootstrap.rs
+++ b/src/bootstrap.rs
@@ -49,9 +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 added = db
-                    .insert_records_batch(&project_id, &table_name, batches, true, Some(&wal_watermark))
-                    .await?;
+                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 9e0d1b6a..9f060611 100644
--- a/src/buffered_write_layer.rs
+++ b/src/buffered_write_layer.rs
@@ -173,16 +173,20 @@ pub type DeltaWriteCallback =
 /// S3 RTT + tantivy build); coalescing turns N×O(commit) into 1×O(commit).
 #[derive(Default)]
 struct CoalescedGroup {
-    project_id:       String,
-    table_name:       String,
-    batches:          Vec<RecordBatch>,
-    row_count:        usize,
+    /// `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>,
+    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>>,
+    wal_positions:     Vec<Option<walrus_rust::WalPosition>>,
     /// Source bucket_ids; drained from MemBuffer after the combined commit succeeds.
     source_bucket_ids: Vec<i64>,
 }
@@ -192,10 +196,7 @@ struct CombinedBucket {
     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>> {
+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()) {
@@ -209,10 +210,7 @@ fn merge_wal_positions(
 
 impl CoalescedGroup {
     fn absorb(&mut self, b: crate::mem_buffer::FlushableBucket) {
-        if self.project_id.is_empty() {
-            self.project_id = b.project_id.clone();
-            self.table_name = b.table_name.clone();
-        }
+        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).
@@ -228,14 +226,16 @@ impl CoalescedGroup {
 
     fn into_combined_bucket(self) -> CombinedBucket {
         let CoalescedGroup {
-            project_id,
-            table_name,
+            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);
@@ -287,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 {
@@ -330,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)),
         })
     }
 
@@ -831,9 +843,7 @@ impl BufferedWriteLayer {
         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_insert_with(CoalescedGroup::default);
+            let entry = groups.entry((bucket.project_id.clone(), bucket.table_name.clone())).or_default();
             entry.absorb(bucket);
         }
 
@@ -854,30 +864,58 @@ impl BufferedWriteLayer {
             .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 (combined, result) in flush_results {
             let CombinedBucket { combined, source_bucket_ids } = combined;
             match result {
                 Ok(()) => {
-                    // advance_by_counts uses the merged shard counts (sums across
-                    // source buckets); drain each source bucket_id from MemBuffer.
-                    if let Err(e) = self.wal.advance_by_counts(&combined.project_id, &combined.table_name, &combined.wal_shard_counts) {
-                        warn!("WAL advance_by_counts failed: {}", e);
-                    }
-                    for bucket_id in &source_bucket_ids {
-                        self.mem_buffer.drain_bucket(&combined.project_id, &combined.table_name, *bucket_id);
+                    // 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
+                            );
+                        }
                     }
-                    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) => {
                     crate::metrics::record_flush(false);
@@ -932,15 +970,22 @@ impl BufferedWriteLayer {
         // Sidecar tantivy index — best-effort, never fails the flush.
         // 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). The callback is responsible for any
-        // internal bounding; flush throughput is naturally bounded by F4
-        // (one tantivy build per (project, table) per cycle, not per bucket).
+        // 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!(
diff --git a/src/database.rs b/src/database.rs
index ba9e7538..c4ff6de0 100644
--- a/src/database.rs
+++ b/src/database.rs
@@ -2442,10 +2442,7 @@ impl Database {
 
                     // 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();
+                    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
@@ -3369,9 +3366,7 @@ impl ProjectRoutingTable {
         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))
-            })?;
+            .map_err(|e| DataFusionError::Execution(format!("fast_insert_batch for project {} table {}: {}", project_id, self.table_name, e)))?;
         Ok(total_rows)
     }
 
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/main.rs b/src/main.rs
index 92e314fc..06a69021 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -89,9 +89,7 @@ async fn async_main(cfg: &'static AppConfig) -> anyhow::Result<()> {
                 // 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?;
+                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.
                 db.warm_cache_for_table(&project_id, &table_name, added.clone());
diff --git a/src/plan_cache.rs b/src/plan_cache.rs
index c4e02a4e..b8b49a0e 100644
--- a/src/plan_cache.rs
+++ b/src/plan_cache.rs
@@ -33,19 +33,14 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use datafusion::{
     arrow::{array::UInt64Array, datatypes::Schema, record_batch::RecordBatch},
-    catalog::MemTable,
     common::{
         ParamValues,
         tree_node::{Transformed, TreeNode},
     },
-    logical_expr::{Cast, Expr, LogicalPlan, LogicalPlanBuilder, dml::WriteOp},
+    logical_expr::{Cast, Expr, LogicalPlan, dml::WriteOp},
     prelude::SessionContext,
     scalar::ScalarValue,
-    sql::{
-        TableReference,
-        parser::Statement as DfStatement,
-        sqlparser::ast::Statement,
-    },
+    sql::{parser::Statement as DfStatement, sqlparser::ast::Statement},
 };
 use datafusion_postgres::{
     hooks::{HookClient, QueryHook},
@@ -82,8 +77,13 @@ fn fold_literal_casts(plan: LogicalPlan) -> datafusion::error::Result<LogicalPla
                             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.
-                            Err(_) => Ok(Transformed::no(e)),
+                            // 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))
@@ -109,7 +109,13 @@ fn fold_literal_casts(plan: LogicalPlan) -> datafusion::error::Result<LogicalPla
 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 {
+    let LogicalPlan::Dml(DmlStatement {
+        table_name,
+        op: WriteOp::Insert(_),
+        input,
+        ..
+    }) = plan
+    else {
         return Ok(None);
     };
 
@@ -198,12 +204,9 @@ async fn try_fast_path_insert(plan: &LogicalPlan, session_context: &SessionConte
             .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]
-                    ))
-                })
+                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)?;
@@ -214,8 +217,7 @@ async fn try_fast_path_insert(plan: &LogicalPlan, session_context: &SessionConte
         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))?
+            datafusion::arrow::compute::cast(&arr, target_ty).map_err(|e| datafusion::error::DataFusionError::ArrowError(Box::new(e), None))?
         };
         values_columns.push(arr);
     }
@@ -245,8 +247,7 @@ async fn try_fast_path_insert(plan: &LogicalPlan, session_context: &SessionConte
     } 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 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 {
@@ -256,70 +257,16 @@ async fn try_fast_path_insert(plan: &LogicalPlan, session_context: &SessionConte
     Ok(Some(rows))
 }
 
-/// Replace every `LogicalPlan::Values` whose rows are entirely literal with a
-/// `TableScan` over an in-memory `MemTable` carrying a precomputed
-/// `RecordBatch`. Currently unused — kept for reference; the `try_fast_path_insert`
-/// bypass is cleaner because it also avoids the `Projection`/`DataSinkExec`
-/// stream overhead.
-#[allow(dead_code)]
-fn values_to_memtable(plan: LogicalPlan) -> datafusion::error::Result<LogicalPlan> {
-    plan.transform_up(|node| {
-        let LogicalPlan::Values(values) = &node else {
-            return Ok(Transformed::no(node));
-        };
-        // Only literal-only Values qualify for the shortcut. Anything else
-        // legitimately needs ValuesExec to evaluate.
-        let all_literal = values.values.iter().all(|row| row.iter().all(|e| matches!(e, Expr::Literal(_, _))));
-        if !all_literal {
-            return Ok(Transformed::no(node));
-        }
-        let arrow_schema: Arc<Schema> = Arc::new(values.schema.as_arrow().clone());
-        let num_cols = arrow_schema.fields().len();
-        let num_rows = values.values.len();
-        // Build column-by-column: collect each column's scalar values, then
-        // `ScalarValue::iter_to_array` packs them into an `ArrayRef` in one go.
-        let mut columns: Vec<datafusion::arrow::array::ArrayRef> = Vec::with_capacity(num_cols);
-        for col_idx in 0..num_cols {
-            let target_ty = arrow_schema.field(col_idx).data_type();
-            if num_rows == 0 {
-                columns.push(datafusion::arrow::array::new_empty_array(target_ty));
-                continue;
-            }
-            let scalars: Vec<ScalarValue> = values
-                .values
-                .iter()
-                .map(|row| match &row[col_idx] {
-                    Expr::Literal(v, _) => Ok(v.clone()),
-                    other => Err(datafusion::error::DataFusionError::Internal(format!(
-                        "values_to_memtable: expected Literal, got {other:?}"
-                    ))),
-                })
-                .collect::<datafusion::error::Result<_>>()?;
-            let arr = ScalarValue::iter_to_array(scalars)?;
-            // ScalarValue::iter_to_array can return an array whose type doesn't
-            // exactly match the Values column type (e.g. all-NULL columns come
-            // back as Null arrays). Cast back to the declared type so the
-            // upstream Projection's column refs type-check.
-            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))?
-            };
-            columns.push(arr);
-        }
-        let batch = RecordBatch::try_new(arrow_schema.clone(), columns)
-            .map_err(|e| datafusion::error::DataFusionError::ArrowError(Box::new(e), None))?;
-        let mem_table = MemTable::try_new(arrow_schema, vec![vec![batch]])?;
-        let scan = LogicalPlanBuilder::scan(TableReference::bare("__pgwire_values"), datafusion::datasource::provider_as_source(Arc::new(mem_table)), None)?.build()?;
-        Ok(Transformed::yes(scan))
-    })
-    .map(|t| t.data)
-}
-
-/// Mirror of the vendored `datafusion_postgres::handlers::dml_completion` which
-/// is private. Drives the Insert/Update/Delete dataframe to completion and
-/// builds a `Tag::new(<verb>).with_rows(N)` Execution response.
+/// 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 {
@@ -541,8 +488,7 @@ impl QueryHook for PlanCacheHook {
     }
 
     async fn handle_extended_query(
-        &self, _statement: &Statement, logical_plan: &LogicalPlan, params: &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>> {
         // Only intercept DML — for SELECTs the vendored path is fine.
         // The win here is post-substitution constant folding of `CAST(Literal, T)`
@@ -575,9 +521,7 @@ impl QueryHook for PlanCacheHook {
         };
         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(),
-            )),
+            Ok(None) => Err(PgWireError::ApiError("internal error: DML plan returned non-DML completion".to_string().into())),
             Err(e) => Err(e),
         })
     }