refactor: non-blocking TRUNCATE via barrier queue (#61)

## Summary

- TRUNCATE now uses the same non-blocking `PendingDdl` barrier queue as
DDL sync (ADD/DROP/RENAME COLUMN), instead of blocking the WAL consumer
for up to 30s via `drain_and_wait_table()`
- The flush thread autonomously drains pending changes then executes
`DELETE FROM` on the target — the WAL consumer enqueues the barrier and
continues immediately
- For truncate-only barriers the DuckDB worker stays alive (schema
unchanged), avoiding unnecessary reconnection overhead
- Removes the now-dead `drain_and_wait_table()` method (shutdown uses
the separate `drain_and_wait_all()`)

## Test plan

- [x] `make installcheck` — all 42 regression tests pass, including
`truncate` and `fan_in` which exercise TRUNCATE propagation, fan-in
scoped DELETE, and INSERT-after-TRUNCATE semantics
- [x] `cargo check` — clean compilation (no new warnings)

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

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: YuweiXiao

doc/CODE_WALKTHROUGH.md

@@ -517,7 +517,7 @@ Fairness: threads take a ticket on arrival and wait in strict FIFO order — no
 - `ensure_queue(target_key, mapping_id, attnames, key_attrs, atttypes)` — create queue + spawn flush thread if new (or respawn if dead); passes `Arc<FlushGate>` to thread
 - `push_change(target_key, change)` — lock queue, push change, notify condvar, increment backpressure counter
 - `is_backpressured() -> bool` — true when total_queued >= max_queued
-- `drain_and_wait_table(target_key)` — per-table synchronous drain (used for TRUNCATE)
+- `get_meta(mapping_id) -> Option<QueueMeta>` — clone current queue metadata (used to build TRUNCATE barriers)
 - `drain_and_wait_all() -> Vec<FlushThreadResult>` — synchronous barrier (retained for shutdown only)
 - `collect_results() -> Vec<FlushThreadResult>` — non-blocking drain of mpsc result channel
 - `set_max_concurrent_flushes(n)` — dynamically update gate limit (for runtime GUC changes)
@@ -643,7 +643,7 @@ Iterates over `(lsn, data)` tuples, parsing each pgoutput binary message:
 | `'D'` (DELETE) | Decode old key values, push `Delete` change |
 | `'B'` (BEGIN) | Parse but no action (LSN tracked from COMMIT) |
 | `'C'` (COMMIT) | Update `group.pending_lsn = end_lsn` |
-| `'T'` (TRUNCATE) | Per-table drain (`drain_and_wait_table`), then `DELETE FROM` each target table |
+| `'T'` (TRUNCATE) | Enqueue barrier (`DdlCommand::Truncate`); flush thread drains then executes `DELETE FROM` |
 
 After the message loop:
 1. `coordinator.collect_results()` — non-blocking drain of flush results for error logging (no synchronous barrier)
@@ -660,7 +660,7 @@ After the message loop:
 - On failure: `FlushWorker` is dropped (lazily recreated), flush thread records error in PG, transitions to ERRORED after 3 consecutive failures
 - Per-table error isolation: one table failing doesn't block others
 - Backpressure: WAL consumer pauses when total queued changes exceed `max_queued_changes`
-- TRUNCATE: uses `drain_and_wait_table()` for per-table synchronous drain before DELETE
+- TRUNCATE: non-blocking barrier queue (same as DDL); flush thread drains pending changes then executes DELETE
 - Crash safety: replication slot only advances past what all tables have durably flushed (confirmed_lsn = min(applied_lsn) read from PG)
 
 ### 5.1. Schema DDL Sync

duckpipe-core/src/flush_coordinator.rs

@@ -5,8 +5,8 @@
 //! - Handle PG metadata updates (applied_lsn, metrics, error state) independently
 //! - Backpressure via AtomicI64 total_queued prevents unbounded memory growth
 //!
-//! `drain_and_wait_all()` is retained for shutdown. `drain_and_wait_table()` is
-//! used for TRUNCATE (must flush before DELETE).
+//! `drain_and_wait_all()` is retained for shutdown. TRUNCATE uses the same
+//! non-blocking barrier queue as DDL (see `DdlCommand::Truncate`).
 
 use std::collections::{HashMap, HashSet};
 use std::sync::atomic::{AtomicBool, AtomicI64, Ordering};
@@ -55,7 +55,7 @@ pub struct QueueMeta {
     pub sync_mode: String,
 }
 
-/// DDL command to apply to the DuckLake target table.
+/// Barrier command: DDL (ADD/DROP/RENAME COLUMN) or TRUNCATE (DELETE by source).
 #[derive(Debug, Clone)]
 pub enum DdlCommand {
     AddColumn {
@@ -74,10 +74,13 @@ pub enum DdlCommand {
         old_type: String,
         new_type: String,
     },
+    Truncate {
+        source_label: String,
+    },
 }
 
-/// DDL barrier: set by the WAL consumer when a schema change is detected.
-/// The flush thread drains old-schema changes, flushes, applies DDL, then continues.
+/// Barrier: set by the WAL consumer when a schema change or TRUNCATE is detected.
+/// The flush thread drains old changes, flushes, applies commands, then continues.
 pub struct PendingDdl {
     pub commands: Vec<DdlCommand>,
     pub new_meta: QueueMeta,
@@ -604,6 +607,14 @@ impl FlushCoordinator {
         }
     }
 
+    /// Clone the current QueueMeta for a table (used to construct barriers
+    /// where the schema is unchanged, e.g. TRUNCATE).
+    pub fn get_meta(&self, mapping_id: i32) -> Option<QueueMeta> {
+        self.threads
+            .get(&mapping_id)
+            .map(|entry| entry.queue_handle.inner.lock().unwrap().meta.clone())
+    }
+
     /// Check if backpressure should pause WAL consumption.
     /// Excludes changes queued for paused (SNAPSHOT) tables so that buffered
     /// snapshot WAL changes don't block streaming for other tables.
@@ -800,28 +811,6 @@ impl FlushCoordinator {
             .retain(|id, _| active_mapping_ids.contains(id));
     }
 
-    /// Per-table synchronous drain: signal one flush thread to drain, wait for completion.
-    /// Used for TRUNCATE — must flush pending changes before DELETE.
-    pub fn drain_and_wait_table(&mut self, mapping_id: i32) {
-        if let Some(entry) = self.threads.get(&mapping_id) {
-            // Reset drain_complete flag
-            {
-                let mut done = entry.drain_complete.0.lock().unwrap();
-                *done = false;
-            }
-            entry.control.drain_requested.store(true, Ordering::Release);
-            entry.queue_handle.condvar.notify_one();
-
-            // Wait for completion
-            let (lock, cvar) = &*entry.drain_complete;
-            let guard = lock.lock().unwrap();
-            let _guard = cvar
-                .wait_timeout_while(guard, Duration::from_secs(30), |done| !*done)
-                .unwrap()
-                .0;
-        }
-    }
-
     /// Synchronous barrier: signal all flush threads to drain their queues,
     /// wait for all to complete, then collect results.
     /// Retained for shutdown.
@@ -1208,7 +1197,7 @@ fn flush_thread_main(
                 guard.meta = new_meta.clone();
                 drop(guard);
 
-                // 1. Flush old-schema buffer if any data is buffered
+                // 1. Flush old buffer if any data is buffered
                 if buffered_count > 0 {
                     if let Some(meta) = buffered_meta.as_ref() {
                         do_flush_buffer(
@@ -1229,20 +1218,17 @@ fn flush_thread_main(
                     pending_local.store(0, Ordering::Relaxed);
                 }
 
-                // 2. Apply DDL via pg_ducklake ALTER TABLE.
-                //    The DDL must go through PG so that both the PG catalog
-                //    (pg_attribute) and the DuckLake catalog are updated.
-                //
-                worker = None; // DETACH DuckLake before PG DDL
+                // 2. Apply barrier commands via PG.
+                worker = None;
                 if let Err(e) = rt.block_on(apply_ddl_commands(
                     &commands, target_oid, pg_connstr, group_name,
                 )) {
                     tracing::error!(
-                        "pg_duckpipe: DDL sync failed for {}: {}",
+                        "pg_duckpipe: barrier command failed for {}: {}",
                         new_meta.target_key,
                         e
                     );
-                    let error_msg = format!("DDL sync failed: {}", e);
+                    let error_msg = format!("barrier command failed: {}", e);
                     // For unsupported DDL (e.g. ALTER COLUMN TYPE), transition to
                     // ERRORED immediately (threshold=1) instead of waiting for 3 failures.
                     let threshold = if commands
@@ -1275,7 +1261,7 @@ fn flush_thread_main(
                 next_seq = 0;
                 buffered_lsn = 0;
                 last_flush = Instant::now();
-                continue; // process new-schema changes on next iteration
+                continue; // process post_changes on next iteration
             } else {
                 drop(guard);
             }
@@ -1552,12 +1538,19 @@ async fn apply_ddl_commands(
                     new_name.replace('"', "\"\""),
                 )
             }
+            DdlCommand::Truncate { source_label } => {
+                format!(
+                    "DELETE FROM {} WHERE \"_duckpipe_source\" = '{}'",
+                    target_ref,
+                    source_label.replace('\'', "''"),
+                )
+            }
         };
-        tracing::info!("pg_duckpipe: DDL sync (PG): {}", sql);
+        tracing::info!("pg_duckpipe: barrier exec (PG): {}", sql);
         client
             .execute(&sql, &[])
             .await
-            .map_err(|e| format!("DDL exec '{}': {}", sql, e))?;
+            .map_err(|e| format!("barrier exec '{}': {}", sql, e))?;
     }
 
     drop(client);

duckpipe-core/src/service.rs

@@ -684,7 +684,8 @@ async fn process_one_wal_message(
             return Ok(true);
         }
         'T' => {
-            // TRUNCATE — flush per-table pending queues, then DELETE all from targets.
+            // TRUNCATE — enqueue a barrier so the flush thread drains pending
+            // changes, then executes DELETE FROM on the target (non-blocking).
             // Uses DELETE FROM instead of TRUNCATE because DuckLake tables
             // (via pg_duckdb) silently ignore TRUNCATE — it succeeds but
             // doesn't actually remove rows.
@@ -704,23 +705,24 @@ async fn process_one_wal_message(
                                     mapping.source_schema,
                                     mapping.source_table
                                 );
+                            } else if let Some(meta) = coordinator.get_meta(mapping.id) {
+                                coordinator.set_pending_ddl(
+                                    mapping.id,
+                                    PendingDdl {
+                                        commands: vec![DdlCommand::Truncate {
+                                            source_label: mapping.source_label.clone(),
+                                        }],
+                                        new_meta: meta,
+                                        target_oid: mapping.target_oid,
+                                        post_changes: Vec::new(),
+                                    },
+                                );
                             } else {
-                                coordinator.drain_and_wait_table(mapping.id);
-                                // Scope DELETE by _duckpipe_source (always set)
-                                let delete_sql = format!(
-                                    "DELETE FROM \"{}\".\"{}\" WHERE \"_duckpipe_source\" = '{}'",
-                                    mapping.target_schema.replace('"', "\"\""),
-                                    mapping.target_table.replace('"', "\"\""),
-                                    mapping.source_label.replace('\'', "''")
+                                tracing::warn!(
+                                    "pg_duckpipe: TRUNCATE on {}.{} — no flush queue yet, skipping",
+                                    mapping.source_schema,
+                                    mapping.source_table
                                 );
-                                if let Err(e) = client.execute(&delete_sql, &[]).await {
-                                    tracing::error!(
-                                        "pg_duckpipe: failed to clear target table {}.{}: {}",
-                                        mapping.target_schema,
-                                        mapping.target_table,
-                                        e
-                                    );
-                                }
                             }
                         }
                     }