feat: schema DDL sync — ADD/DROP/RENAME COLUMN, block ALTER TYPE (#39)

## Summary

- **Automatic DDL propagation** from source to DuckLake target tables
for ADD COLUMN, DROP COLUMN, and RENAME COLUMN
- **Detection via RELATION message diffing** — pgoutput sends a RELATION
message before the first DML after a schema change; comparing with the
cached entry detects column additions, removals, and renames
- **Non-blocking DDL barrier queue** — DDL is a `PendingDdl` in a
per-table `VecDeque`; the WAL consumer sets the barrier and continues
immediately while the flush thread drains old-schema data, applies ALTER
TABLE via a short-lived PG connection, then resumes with the new schema.
Multiple barriers are queued (not merged) so each batch is processed
with the correct column layout.
- **OID-based target resolution** — `target_oid` stored in
`table_mappings`; `apply_ddl_commands()` resolves the current target
name from `pg_class` so the pipeline survives user-initiated target
table renames
- **Source rename does NOT rename target** — source table renames update
`source_schema`/`source_table` metadata only; target table name stays
unchanged
- **ALTER COLUMN TYPE blocked** — same-name-different-OID columns are
detected and the table is transitioned to ERRORED immediately
(threshold=1) with a resync hint, preventing silent data corruption from
stale column types
- **Fix: preserve error_message in ERRORED state** —
`clear_error_on_success()` now skips tables in ERRORED state so a
subsequent successful flush doesn't wipe the diagnostic message
- **New types**: `DdlCommand` enum (with `UnsupportedAlterColumnType`
variant), `PendingDdl` struct
- **Metadata helpers**: `get_column_type()`, `update_source_name()`,
`update_target_oid()`, `pg_oid_to_type_name()`

## Test plan

- [x] `make check-regression TEST=ddl_sync` — ADD COLUMN, DROP COLUMN,
RENAME COLUMN propagate correctly; multi-DDL barrier queuing works;
ALTER COLUMN TYPE errors the pipeline
- [x] `make check-regression TEST=rename_table` — source rename does NOT
rename target, metadata updated, UPDATE/DELETE work after rename
- [x] `make installcheck` — all 42 tests pass

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

---------

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

Author: YuweiXiao

PROGRESS.md

@@ -16,8 +16,9 @@
 - [ ] Mixed DML replication lag 50-100x append — WAL amplification + Parquet-scan DELETE phase
 
 ### Features
+- [x] Schema DDL sync — ALTER TABLE ADD/DROP/RENAME COLUMN propagation (OID-based target resolution)
+- [ ] TRUNCATE as DDL barrier — route TRUNCATE through the DDL barrier queue (DELETE FROM target), removing drain logic from the WAL consumer side
 - [ ] Per-table config JSONB column on `table_mappings` — consolidate `routing_enabled` and future per-table settings into a single `config JSONB` column (like `sync_groups.config`), avoid schema sprawl
-- [ ] Schema DDL sync — ALTER TABLE ADD/DROP COLUMN propagation
 - [ ] Staged storage — durable delta layer between WAL and DuckLake to decouple CDC from file proliferation
 - [ ] Explicit `Value` variants for more PG types (date, timestamp, uuid, numeric, interval, json)
 - [x] Sync tables with no PK (append mode; upsert requires PK)

doc/CODE_WALKTHROUGH.md

@@ -637,7 +637,7 @@ Iterates over `(lsn, data)` tuples, parsing each pgoutput binary message:
 
 | Message | Action |
 |---------|--------|
-| `'R'` (RELATION) | Cache schema info in `rel_cache` |
+| `'R'` (RELATION) | Cache schema info in `rel_cache`; detect DDL changes via diffing (see §5.1) |
 | `'I'` (INSERT) | Decode typed values, resolve mapping, skip if disabled/ERRORED/CATCHUP, push to queue |
 | `'U'` (UPDATE) | Decode old+new, if TOAST unchanged → `Update` change, else → `Delete` + `Insert` |
 | `'D'` (DELETE) | Decode old key values, push `Delete` change |
@@ -663,6 +663,28 @@ After the message loop:
 - TRUNCATE: uses `drain_and_wait_table()` for per-table synchronous drain before 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
+
+Schema changes (ADD/DROP/RENAME COLUMN) are automatically propagated from source to DuckLake target tables. Source table renames are tracked (metadata updated) but do **not** rename the target — the target table name is stable and independent of the source name.
+
+**Detection** — pgoutput sends a RELATION message before the first DML after a schema change. The `'R'` handler in `process_one_wal_message()` compares the new RELATION with the cached entry via `detect_schema_changes()`:
+
+- **ADD COLUMN**: column names in new but not in old → query `pg_attribute` for type
+- **DROP COLUMN**: column names in old but not in new
+- **RENAME COLUMN**: same position, different name, same type OID
+
+**Target resolution** — `apply_ddl_commands()` resolves the current target table name from `target_oid` via `pg_class` rather than using a stored name string. This means the pipeline survives user-initiated renames of the target table. For old mappings without `target_oid`, it falls back to the name stored in metadata.
+
+**Propagation** — DDL is treated as a non-blocking barrier event in the per-table queue (`PendingDdl`). The WAL consumer sets the barrier and continues immediately; the flush thread handles it autonomously:
+
+1. WAL consumer detects schema diff → calls `coordinator.set_pending_ddl()` with `DdlCommand` list and new `QueueMeta`
+2. While barrier is set, `push_change()` routes new-schema changes to `pending_after_ddl`
+3. Flush thread drains and flushes old-schema changes from the buffer
+4. Flush thread applies `ALTER TABLE` commands to the DuckLake target via a short-lived PG connection (`apply_ddl_commands()`)
+5. Flush thread resets `FlushWorker`, merges `pending_after_ddl` into main queue with updated `QueueMeta`, continues normally
+
+The barrier ensures old-schema data is flushed before ALTER TABLE, preventing column mismatch errors or data loss.
+
 ---
 
 ## 6. duckpipe-pg: The PostgreSQL Extension

duckpipe-core/src/flush_coordinator.rs

@@ -45,21 +45,58 @@ pub struct TableMetrics {
 
 /// Cloneable table schema info used by the flush thread for DuckDB buffer operations.
 #[derive(Clone, Debug)]
-struct QueueMeta {
-    target_key: String,
-    mapping_id: i32,
-    attnames: Vec<String>,
-    key_attrs: Vec<usize>,
-    atttypes: Vec<u32>,
-    source_label: String,
-    sync_mode: String,
+pub struct QueueMeta {
+    pub target_key: String,
+    pub mapping_id: i32,
+    pub attnames: Vec<String>,
+    pub key_attrs: Vec<usize>,
+    pub atttypes: Vec<u32>,
+    pub source_label: String,
+    pub sync_mode: String,
+}
+
+/// DDL command to apply to the DuckLake target table.
+#[derive(Debug, Clone)]
+pub enum DdlCommand {
+    AddColumn {
+        col_name: String,
+        col_type: String,
+    },
+    DropColumn {
+        col_name: String,
+    },
+    RenameColumn {
+        old_name: String,
+        new_name: String,
+    },
+    UnsupportedAlterColumnType {
+        col_name: String,
+        old_type: String,
+        new_type: 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.
+pub struct PendingDdl {
+    pub commands: Vec<DdlCommand>,
+    pub new_meta: QueueMeta,
+    /// Target table OID for resolving current name via pg_class.
+    pub target_oid: i64,
+    /// Changes received after this DDL but before the next DDL (or present).
+    /// Populated by `push_change()` routing and by `set_pending_ddl()` when
+    /// a subsequent barrier arrives.
+    pub post_changes: Vec<Change>,
 }
 
 /// Shared queue data protected by Mutex.
 struct SharedTableQueue {
     meta: QueueMeta,
     changes: Vec<Change>,
     last_lsn: u64,
+    /// DDL barrier queue. When non-empty, push_change routes to the last
+    /// barrier's `post_changes`. Processed FIFO by the flush thread.
+    pending_ddls: std::collections::VecDeque<PendingDdl>,
 }
 
 /// Arc-shared handle between producer (main thread) and consumer (flush thread).
@@ -455,6 +492,7 @@ impl FlushCoordinator {
                 meta: meta.clone(),
                 changes: Vec::new(),
                 last_lsn: 0,
+                pending_ddls: std::collections::VecDeque::new(),
             }),
             condvar: Condvar::new(),
         });
@@ -525,13 +563,20 @@ impl FlushCoordinator {
     ///
     /// The flush thread wakes on its own via `drain_poll_ms` condvar timeout,
     /// so no explicit notification is needed after pushing changes.
+    ///
+    /// When a DDL barrier is set, changes are routed to the last barrier's
+    /// `post_changes` so the flush thread can drain old-schema changes first.
     pub fn push_change(&self, mapping_id: i32, change: Change) {
         if let Some(entry) = self.threads.get(&mapping_id) {
             let mut guard = entry.queue_handle.inner.lock().unwrap();
             if change.lsn > guard.last_lsn {
                 guard.last_lsn = change.lsn;
             }
-            guard.changes.push(change);
+            if let Some(last_ddl) = guard.pending_ddls.back_mut() {
+                last_ddl.post_changes.push(change);
+            } else {
+                guard.changes.push(change);
+            }
             self.backpressure
                 .total_queued
                 .fetch_add(1, Ordering::Relaxed);
@@ -543,6 +588,22 @@ impl FlushCoordinator {
         }
     }
 
+    /// Set a DDL barrier on the queue for a target table.
+    ///
+    /// The WAL consumer calls this when it detects a schema change from a RELATION message.
+    /// New changes after this call are routed to the latest barrier's `post_changes`.
+    /// The flush thread processes barriers FIFO: drain old-schema changes, flush, apply
+    /// DDL, move post_changes to main queue, reset worker, then check for next barrier.
+    ///
+    /// Multiple barriers are queued (not merged) so that each batch of post_changes is
+    /// processed with the correct schema after its corresponding DDL is applied.
+    pub fn set_pending_ddl(&self, mapping_id: i32, ddl: PendingDdl) {
+        if let Some(entry) = self.threads.get(&mapping_id) {
+            let mut guard = entry.queue_handle.inner.lock().unwrap();
+            guard.pending_ddls.push_back(ddl);
+        }
+    }
+
     /// 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.
@@ -692,6 +753,11 @@ impl FlushCoordinator {
                 let shared = {
                     let guard = entry.queue_handle.inner.lock().unwrap();
                     guard.changes.len() as i64
+                        + guard
+                            .pending_ddls
+                            .iter()
+                            .map(|d| d.post_changes.len() as i64)
+                            .sum::<i64>()
                 };
                 let local = entry.pending_local.load(Ordering::Relaxed);
                 let abandoned = shared + local;
@@ -980,8 +1046,9 @@ fn flush_thread_main(
         {
             let mut guard = queue_handle.inner.lock().unwrap();
 
-            // Wait with timeout for new changes, drain request, or shutdown
+            // Wait with timeout for new changes, DDL barrier, drain request, or shutdown
             if guard.changes.is_empty()
+                && guard.pending_ddls.is_empty()
                 && !control.shutdown.load(Ordering::Acquire)
                 && !control.drain_requested.load(Ordering::Acquire)
                 && wait_timeout > Duration::ZERO
@@ -1126,6 +1193,89 @@ fn flush_thread_main(
                     }
                 }
                 // Local Vec<Change> is dropped here — Rust memory freed.
+            } else if !guard.pending_ddls.is_empty() {
+                // All old-schema changes drained, DDL barrier(s) present.
+                // Process the first barrier only — its post_changes become the
+                // new main queue. If more barriers remain, they'll be processed
+                // on subsequent iterations (each with its own schema).
+                let PendingDdl {
+                    commands,
+                    new_meta,
+                    target_oid,
+                    post_changes,
+                } = guard.pending_ddls.pop_front().unwrap();
+                guard.changes = post_changes;
+                guard.meta = new_meta.clone();
+                drop(guard);
+
+                // 1. Flush old-schema buffer if any data is buffered
+                if buffered_count > 0 {
+                    if let Some(meta) = buffered_meta.as_ref() {
+                        do_flush_buffer(
+                            &mut worker,
+                            buffered_count,
+                            buffered_bytes,
+                            buffered_lsn,
+                            meta,
+                            pg_connstr,
+                            group_name,
+                            &result_tx,
+                            &rt,
+                        );
+                    }
+                    backpressure
+                        .total_queued
+                        .fetch_sub(buffered_count, Ordering::Relaxed);
+                    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
+                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 {}: {}",
+                        new_meta.target_key,
+                        e
+                    );
+                    let error_msg = format!("DDL sync 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
+                        .iter()
+                        .any(|c| matches!(c, DdlCommand::UnsupportedAlterColumnType { .. }))
+                    {
+                        1
+                    } else {
+                        ERRORED_THRESHOLD
+                    };
+                    let _ = result_tx.send(FlushThreadResult::Error {
+                        target_key: new_meta.target_key.clone(),
+                        mapping_id: new_meta.mapping_id,
+                        error: error_msg.clone(),
+                    });
+                    let _ = rt.block_on(flush_worker::update_error_state(
+                        pg_connstr,
+                        new_meta.mapping_id,
+                        &error_msg,
+                        threshold,
+                        group_name,
+                    ));
+                }
+
+                // 3. Reset FlushWorker (forces re-discovery of lake_info)
+                worker = None;
+                buffered_meta = None;
+                buffered_count = 0;
+                buffered_bytes = 0;
+                next_seq = 0;
+                buffered_lsn = 0;
+                last_flush = Instant::now();
+                continue; // process new-schema changes on next iteration
             } else {
                 drop(guard);
             }
@@ -1329,3 +1479,89 @@ fn signal_drain_complete(
     control.drain_requested.store(false, Ordering::Release);
     cvar.notify_one();
 }
+
+/// Apply DDL commands to the DuckLake target table via a short-lived PG connection.
+///
+/// Resolves the current target table name from `target_oid` via `pg_class`, so the
+/// pipeline survives user-initiated renames of the target table.
+async fn apply_ddl_commands(
+    commands: &[DdlCommand],
+    target_oid: i64,
+    pg_connstr: &str,
+    group_name: &str,
+) -> Result<(), String> {
+    let app_name = crate::connstr::app_name(group_name, "ddl");
+    let (client, conn_handle) = crate::connstr::pg_connect_with_app_name(pg_connstr, &app_name)
+        .await
+        .map_err(|e| format!("DDL connect: {}", e))?;
+
+    // Resolve current target table name from OID
+    let rows = client
+        .query(
+            "SELECT n.nspname, c.relname FROM pg_class c \
+             JOIN pg_namespace n ON n.oid = c.relnamespace \
+             WHERE c.oid = $1::bigint::oid",
+            &[&target_oid],
+        )
+        .await
+        .map_err(|e| format!("OID resolve: {}", e))?;
+    if rows.is_empty() {
+        return Err(format!("target OID {} not found in pg_class", target_oid));
+    }
+    let schema: String = rows[0].get(0);
+    let table: String = rows[0].get(1);
+    let target_ref = format!(
+        "\"{}\".\"{}\"",
+        schema.replace('"', "\"\""),
+        table.replace('"', "\"\""),
+    );
+
+    for cmd in commands {
+        let sql = match cmd {
+            DdlCommand::UnsupportedAlterColumnType {
+                col_name,
+                old_type,
+                new_type,
+            } => {
+                return Err(format!(
+                    "ALTER COLUMN TYPE on \"{}\" ({} → {}) is not supported by DDL sync. \
+                     Run SELECT duckpipe.resync('schema.table') to re-snapshot the table.",
+                    col_name, old_type, new_type
+                ));
+            }
+            DdlCommand::AddColumn { col_name, col_type } => {
+                format!(
+                    "ALTER TABLE {} ADD COLUMN \"{}\" {}",
+                    target_ref,
+                    col_name.replace('"', "\"\""),
+                    col_type,
+                )
+            }
+            DdlCommand::DropColumn { col_name } => {
+                format!(
+                    "ALTER TABLE {} DROP COLUMN \"{}\"",
+                    target_ref,
+                    col_name.replace('"', "\"\""),
+                )
+            }
+            DdlCommand::RenameColumn { old_name, new_name } => {
+                format!(
+                    "ALTER TABLE {} RENAME COLUMN \"{}\" TO \"{}\"",
+                    target_ref,
+                    old_name.replace('"', "\"\""),
+                    new_name.replace('"', "\"\""),
+                )
+            }
+        };
+        tracing::info!("pg_duckpipe: DDL sync (PG): {}", sql);
+        client
+            .execute(&sql, &[])
+            .await
+            .map_err(|e| format!("DDL exec '{}': {}", sql, e))?;
+    }
+
+    drop(client);
+    let _ = conn_handle.await;
+
+    Ok(())
+}

duckpipe-core/src/flush_worker.rs

@@ -73,6 +73,8 @@ pub async fn update_error_state(
 }
 
 /// Clear error state on successful flush: reset consecutive_failures and error_message.
+/// Skips tables in ERRORED state — those retain their error_message for user visibility
+/// and are recovered only through the auto-retry mechanism or manual resync.
 pub async fn clear_error_on_success(
     connstr: &str,
     mapping_id: i32,
@@ -83,9 +85,15 @@ pub async fn clear_error_on_success(
         .await
         .map_err(|e| format!("clear_error connect: {}", e))?;
 
-    let meta = MetadataClient::new(&client);
-    let _ = meta.clear_consecutive_failures(mapping_id).await;
-    let _ = meta.record_error_message(mapping_id, "").await;
+    client
+        .execute(
+            "UPDATE duckpipe.table_mappings \
+             SET consecutive_failures = 0, error_message = '' \
+             WHERE id = $1 AND state != 'ERRORED'",
+            &[&mapping_id],
+        )
+        .await
+        .map_err(|e| format!("clear_error: {}", e))?;
 
     drop(client);
     let _ = conn_handle.await;