diff --git a/crates/adapters/src/integrated/postgres/cdc_input.rs b/crates/adapters/src/integrated/postgres/cdc_input.rs index a3efd62a23c..9c966de2aed 100644 --- a/crates/adapters/src/integrated/postgres/cdc_input.rs +++ b/crates/adapters/src/integrated/postgres/cdc_input.rs @@ -54,7 +54,7 @@ //! # Lifecycle and errors //! //! Pausing the Feldera pipeline stops the destination from accepting new etl batches; already -//! queued records can still drain. [`TableErrorMonitor`] turns non-retriable per-table etl errors, +//! queued records can still drain. [`TableStateMonitor`] turns non-retriable source-table errors, //! such as unsupported source schema changes, into connector errors instead of allowing the input //! to stall silently. Terminating or dropping the connector shuts down the etl pipeline and its //! completion watcher. @@ -67,9 +67,25 @@ //! replication position does not include it and the write must be replayed. On startup, //! `discard_shutdown_errors` therefore defaults to `true` and rolls back only this dropped-ack //! error. Other persisted table errors remain intact unless `discard_table_errors` is enabled. +//! +//! # Transactions +//! +//! With `transaction_mode: snapshot`, initial synchronization is split into two Feldera +//! transactions: one for the PostgreSQL table copy and one for the WAL catchup. etl cannot start +//! WAL catchup until the table copy has been marked durable. With fault tolerance enabled, this +//! means the copy transaction must first be committed and checkpointed; the catchup transaction +//! can then run and become durable independently. +//! +//! The copy transaction commits at etl's terminal copy barrier, and the catchup transaction commits +//! with its terminal durable write. Sometimes there are no WAL records to catch up: the table-sync +//! worker is already at the handoff LSN, so etl moves directly to `SyncDone` without calling the +//! destination. No catchup transaction is opened in that case; the table state monitor only retires +//! the pending catchup phase. Writes from the main apply worker are not grouped into +//! connector-managed transactions, so steady-state CDC continues normally. use crate::transport::{ - InputEndpoint, InputQueue, InputReaderCommand, IntegratedInputEndpoint, NonFtInputReaderCommand, + InputEndpoint, InputQueue, InputQueueEntry, InputReaderCommand, IntegratedInputEndpoint, + NonFtInputReaderCommand, }; use crate::{ControllerError, InputConsumer, InputReader, PipelineState, RecordFormat}; use anyhow::{Result as AnyResult, anyhow}; @@ -88,15 +104,17 @@ use etl::error::{ErrorKind, EtlResult}; use etl::etl_error; use etl::event::Event; use etl::pipeline::{Pipeline, ShutdownTx}; -use etl::schema::ReplicatedTableSchema; -use etl::store::{PostgresStore, StateStore, TableRetryPolicy, TableState}; +use etl::schema::{ReplicatedTableSchema, TableId}; +use etl::store::{PostgresStore, SchemaStore, StateStore, TableRetryPolicy, TableState}; use feldera_adapterlib::catalog::{DeCollectionStream, InputCollectionHandle}; -use feldera_adapterlib::format::ParseError; +use feldera_adapterlib::format::{InputBuffer, ParseError}; use feldera_adapterlib::transport::{Resume, Watermark}; use feldera_types::config::FtModel; use feldera_types::coordination::Completion; use feldera_types::format::json::JsonFlavor; -use feldera_types::transport::postgres::{PostgresCdcReaderConfig, PostgresTlsConfig}; +use feldera_types::transport::postgres::{ + PostgresCdcReaderConfig, PostgresCdcTransactionMode, PostgresTlsConfig, +}; use serde_json::{Value, json}; use std::collections::BTreeSet; use std::future::pending; @@ -117,6 +135,7 @@ use super::tls::make_etl_tls_config; const DROPPED_DESTINATION_ACK_ERROR: &str = "[DestinationError] Async result channel closed before sending"; const MAX_ERROR_ROLLBACKS_PER_TABLE: usize = 32; +const MAX_QUEUED_BUFFER_BYTES: usize = 2 * 1024 * 1024; /// An etl write acknowledgment deferred until Feldera has processed the data it covers. /// The ack is tagged by write type because snapshot and stream acks follow different rules. @@ -547,7 +566,6 @@ impl PostgresCdcInputInner { } else { (None, None) }; - Self { endpoint_name: endpoint_name.to_string(), config, @@ -638,33 +656,51 @@ impl PostgresCdcInputInner { } }; - let discard_errors_result = if self.config.discard_table_errors { - self.discard_table_errors(&store).await - } else if self.config.discard_shutdown_errors { - self.discard_shutdown_errors(&store).await - } else { - Ok(()) - }; - if let Err(e) = discard_errors_result { - let _ = init_status_sender.send(Err(e)); - return; + let discard_all_errors = self.config.discard_table_errors; + if discard_all_errors || self.config.discard_shutdown_errors { + let discard_result = async { + store.load_table_states().await?; + store.load_table_schemas().await?; + discard_matching_table_errors( + &self.endpoint_name, + &self.config.source_table, + &store, + discard_all_errors, + ) + .await + } + .await; + if let Err(e) = discard_result { + let _ = init_status_sender.send(Err(ControllerError::input_transport_error( + &self.endpoint_name, + true, + anyhow!("failed to discard persisted etl table errors: {e}"), + ))); + return; + } } - let destination = FelderaDestination { - input_stream: Arc::new(Mutex::new(input_stream)), - queue: Arc::clone(&self.queue), - source_table: self.config.source_table.clone(), - endpoint_name: self.endpoint_name.clone(), + let destination = FelderaDestination::new( + input_stream, + self.endpoint_name.clone(), + self.config.source_table.clone(), + self.config.transaction_mode, + self.pipeline_id, + Arc::clone(&self.queue), + store.clone(), feldera_required_columns, - completion_task_tx: self.completion_task_tx.clone(), - queued_acks: Arc::clone(&self.queued_acks), - pipeline_state_rx: receiver.clone(), - }; + self.completion_task_tx.clone(), + Arc::clone(&self.queued_acks), + receiver.clone(), + ); + let snapshot_transactions = destination.snapshot_transactions(); - let table_error_monitor = TableErrorMonitor { + let table_state_monitor = TableStateMonitor { endpoint_name: self.endpoint_name.clone(), + source_table: self.config.source_table.clone(), consumer: self.consumer.clone(), store: store.clone(), + snapshot_transactions, }; let mut pipeline = Pipeline::new(pipeline_config, store, destination); self.set_etl_shutdown_tx(pipeline.shutdown_tx()); @@ -704,7 +740,7 @@ impl PostgresCdcInputInner { _ => None, }; - // Run the pipeline alongside a watcher for non-retriable per-table + // Run the pipeline alongside a watcher for non-retriable source-table // errors. etl marks a table errored (e.g. on a source schema change) // without failing the whole pipeline, so `pipeline.wait` would block // forever while the input silently stalls; the watcher reports such an @@ -722,7 +758,7 @@ impl PostgresCdcInputInner { abort_completion_watcher(&mut completion_handle).await; (pipeline_wait.as_mut().await, false) } - _ = table_error_monitor.run() => { + _ = table_state_monitor.run() => { self.shutdown_etl_pipeline(); abort_completion_watcher(&mut completion_handle).await; (pipeline_wait.as_mut().await, false) @@ -758,143 +794,76 @@ impl PostgresCdcInputInner { let _ = shutdown_tx.shutdown(); } } +} - /// Roll back persisted errors caused by a destination acknowledgment being - /// dropped by a previous connector run. - async fn discard_shutdown_errors(&self, store: &PostgresStore) -> Result<(), ControllerError> { - self.discard_matching_table_errors(store, false).await - } - - /// Roll back every persisted `Errored` table state so etl retries those - /// tables on this run (the `discard_table_errors` config option). - async fn discard_table_errors(&self, store: &PostgresStore) -> Result<(), ControllerError> { - self.discard_matching_table_errors(store, true).await +/// Roll back matching persisted `Errored` states for the configured source +/// table before starting etl. Errored states can stack, so keep rolling back +/// until another state surfaces. If rollback fails or reaches the rollback +/// limit, reset the table to `Init`. +async fn discard_matching_table_errors( + endpoint_name: &str, + source_table: &str, + store: &PostgresStore, + discard_all: bool, +) -> EtlResult<()> { + let Some(table_id) = target_table_id(store, source_table).await? else { + return Ok(()); + }; + let Some(state) = store.get_table_state(table_id).await? else { + return Ok(()); + }; + if !should_discard_table_error(&state, discard_all) { + return Ok(()); } - /// Roll back matching persisted `Errored` table states. - /// - /// Errored states can stack, so keep rolling back until something else - /// surfaces. If a rollback fails, reset the table to `Init` instead, - /// which re-copies it from scratch. - async fn discard_matching_table_errors( - &self, - store: &PostgresStore, - discard_all: bool, - ) -> Result<(), ControllerError> { - store.load_table_states().await.map_err(|e| { - ControllerError::input_transport_error( - &self.endpoint_name, - true, - anyhow!("failed to load etl table states before discarding errors: {e}"), - ) - })?; - - let states = store.get_table_states().await.map_err(|e| { - ControllerError::input_transport_error( - &self.endpoint_name, - true, - anyhow!("failed to read etl table states before discarding errors: {e}"), - ) - })?; - - let errored_table_ids: Vec<_> = states - .iter() - .filter_map(|(table_id, state)| { - should_discard_table_error(state, discard_all).then_some(*table_id) - }) - .collect(); + warn!( + "postgres_cdc {}: discarding persisted etl {} error for table {table_id} before startup", + endpoint_name, + if discard_all { "table" } else { "shutdown" }, + ); - if errored_table_ids.is_empty() { - return Ok(()); + let mut discarded = 0usize; + let mut state = state; + while should_discard_table_error(&state, discard_all) { + if discarded == MAX_ERROR_ROLLBACKS_PER_TABLE { + warn!( + "postgres_cdc {}: table {table_id} still has a matching etl error after \ + {MAX_ERROR_ROLLBACKS_PER_TABLE} rollbacks; resetting table state to init", + endpoint_name + ); + store.update_table_state(table_id, TableState::Init).await?; + break; } - warn!( - "postgres_cdc {}: discarding {} persisted etl {} error(s) before startup", - &self.endpoint_name, - errored_table_ids.len(), - if discard_all { "table" } else { "shutdown" }, - ); - - for table_id in errored_table_ids { - let mut discarded = 0usize; - loop { - let Some(state) = store.get_table_state(table_id).await.map_err(|e| { - ControllerError::input_transport_error( - &self.endpoint_name, - true, - anyhow!("failed to read etl table state for table {table_id}: {e}"), - ) - })? - else { - break; - }; - if !should_discard_table_error(&state, discard_all) { - break; - } - - if discarded == MAX_ERROR_ROLLBACKS_PER_TABLE { - warn!( - "postgres_cdc {}: table {table_id} still has a matching etl error after \ - {MAX_ERROR_ROLLBACKS_PER_TABLE} rollbacks; resetting table state to init", - &self.endpoint_name - ); - store - .update_table_state(table_id, TableState::Init) - .await - .map_err(|e| { - ControllerError::input_transport_error( - &self.endpoint_name, - true, - anyhow!( - "failed to reset etl table state for table {table_id} after \ - reaching the rollback limit: {e}" - ), - ) - })?; - break; - } - - match store.rollback_table_state(table_id).await { - Ok(restored_state) => { - discarded += 1; - info!( - "postgres_cdc {}: discarded etl table error for table {table_id}, \ + match store.rollback_table_state(table_id).await { + Ok(restored_state) => { + discarded += 1; + info!( + "postgres_cdc {}: discarded etl table error for table {table_id}, \ restored previous state {restored_state}", - &self.endpoint_name - ); - } - Err(e) => { - warn!( - "postgres_cdc {}: failed to roll back etl table error for table \ + endpoint_name + ); + state = restored_state; + } + Err(e) => { + warn!( + "postgres_cdc {}: failed to roll back etl table error for table \ {table_id}: {e}; resetting table state to init", - &self.endpoint_name - ); - store - .update_table_state(table_id, TableState::Init) - .await - .map_err(|e| { - ControllerError::input_transport_error( - &self.endpoint_name, - true, - anyhow!( - "failed to reset etl table state for table {table_id} \ - after discarding error: {e}" - ), - ) - })?; - discarded += 1; - } - } + endpoint_name + ); + store.update_table_state(table_id, TableState::Init).await?; + discarded += 1; + break; } - - debug!( - "postgres_cdc {}: discarded {discarded} etl table error state(s) for table {table_id}", - &self.endpoint_name - ); } - - Ok(()) } + + debug!( + "postgres_cdc {}: discarded {discarded} etl table error state(s) for table {table_id}", + endpoint_name + ); + + Ok(()) } fn should_discard_table_error(state: &TableState, discard_all: bool) -> bool { @@ -912,16 +881,18 @@ impl Drop for PostgresCdcInputInner { } } -/// Monitor that turns etl table-state failures into Feldera connector failures. -struct TableErrorMonitor { +/// Monitor the configured source table's etl state. +struct TableStateMonitor { endpoint_name: String, + source_table: String, consumer: Box, store: PostgresStore, + snapshot_transactions: Arc, } -impl TableErrorMonitor { - /// Surface a non-retriable per-table replication error as a fatal endpoint - /// error. +impl TableStateMonitor { + /// Retire an idle catchup phase and surface non-retriable source-table + /// errors as fatal endpoint errors. /// /// When etl cannot continue replicating a table — most notably after a /// source schema change, which Feldera does not support — it marks the @@ -932,6 +903,10 @@ impl TableErrorMonitor { /// consumer so the controller fails the endpoint. `TimedRetry` errors are /// left alone: etl retries them and, once retries are exhausted, the apply /// worker propagates the failure through `pipeline.wait`. + /// + /// etl can enter catchup with its current LSN already at the target. It then + /// moves to `SyncDone` with an empty batch, so there is no destination write + /// to retire the pending catchup phase. async fn run(self) { const POLL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1); @@ -949,39 +924,280 @@ impl TableErrorMonitor { } }; - for (table_id, state) in states.iter() { - let TableState::Errored { - reason, - solution, - retry_policy, - .. - } = state - else { - continue; - }; - - // A timed retry clears on its own; leave it to etl. - if matches!(retry_policy, TableRetryPolicy::TimedRetry { .. }) { + let table_id = match target_table_id(&self.store, &self.source_table).await { + Ok(Some(table_id)) => table_id, + Ok(None) => continue, + Err(e) => { + debug!( + "postgres_cdc {}: failed to resolve source table state: {e}", + &self.endpoint_name + ); continue; } + }; + let Some(state) = states.get(&table_id) else { + continue; + }; - let detail = match solution { - Some(solution) => format!("{reason} ({solution})"), - None => reason.clone(), - }; - error!( - "postgres_cdc {}: table {table_id} replication errored: {detail}", - &self.endpoint_name - ); - self.consumer.error( - true, - anyhow!("postgres replication error on table {table_id}: {detail}"), - None, - ); - return; + if matches!(state, TableState::SyncDone { .. } | TableState::Ready) { + self.snapshot_transactions + .finish_catchup_after_etl_completion(table_id); + } + + let TableState::Errored { + reason, + solution, + retry_policy, + .. + } = state + else { + continue; + }; + + // A timed retry clears on its own; leave it to etl. + if matches!(retry_policy, TableRetryPolicy::TimedRetry { .. }) { + continue; } + + let detail = match solution { + Some(solution) => format!("{reason} ({solution})"), + None => reason.clone(), + }; + error!( + "postgres_cdc {}: table {table_id} replication errored: {detail}", + &self.endpoint_name + ); + self.consumer.error( + true, + anyhow!("postgres replication error on table {table_id}: {detail}"), + None, + ); + return; + } + } +} + +enum SnapshotPhase { + /// Phase is unknown until startup reconciles etl's durable table state. + Uninitialized, + /// Transactions are disabled, or initial sync is already complete. + Inactive, + /// Initial table COPY. + Copy { + started: bool, + table_id: Option, + }, + /// WAL catchup after a durable COPY. + Catchup { started: bool, table_id: TableId }, +} + +impl SnapshotPhase { + fn start_copy(&mut self, table_id: TableId) -> bool { + let Self::Copy { + started, + table_id: copy_table_id, + } = self + else { + return false; + }; + + *copy_table_id = Some(table_id); + if *started { + false + } else { + *started = true; + true } } + + fn finish_copy(&mut self, table_id: TableId) -> bool { + let Self::Copy { + started, + table_id: copy_table_id, + } = self + else { + return false; + }; + + debug_assert!(copy_table_id.is_none_or(|id| id == table_id)); + let commit = *started; + *self = Self::Catchup { + started: false, + table_id, + }; + commit + } + + fn start_catchup(&mut self, table_id: Option, has_data: bool) -> bool { + let Self::Catchup { + started, + table_id: catchup_table_id, + } = self + else { + return false; + }; + + if table_id != Some(*catchup_table_id) || !has_data || *started { + false + } else { + *started = true; + true + } + } + + fn finish_catchup(&mut self) -> bool { + let Self::Catchup { started, .. } = self else { + return false; + }; + + let commit = *started; + *self = Self::Inactive; + commit + } +} + +/// Shares snapshot transaction state between etl workers and the table-state +/// monitor. +/// +/// The COPY transaction starts when the first row buffer is queued, so an empty +/// table does not create a transaction. On restart, `Destination::startup` +/// reads etl's table state, which is persisted in the source Postgres database, +/// to restore whether this connector is copying, catching up, or already live. +struct SnapshotTransactions { + source_table: String, + transaction_mode: PostgresCdcTransactionMode, + pipeline_id: u64, + queue: Arc, + startup_store: Mutex>, + phase: Mutex, +} + +impl SnapshotTransactions { + fn new( + source_table: String, + transaction_mode: PostgresCdcTransactionMode, + pipeline_id: u64, + queue: Arc, + store: PostgresStore, + ) -> Self { + Self { + source_table, + transaction_mode, + pipeline_id, + queue, + startup_store: Mutex::new(Some(store)), + phase: Mutex::new(SnapshotPhase::Uninitialized), + } + } + + async fn initialize(&self) -> EtlResult<()> { + let store = self.startup_store.lock().unwrap().take().ok_or_else(|| { + etl_error!( + ErrorKind::DestinationError, + "Postgres CDC destination initialized more than once" + ) + })?; + let phase = self.initial_phase(&store).await?; + *self.phase.lock().unwrap() = phase; + Ok(()) + } + + /// Restore the transaction phase after a connector restart. This is the + /// only path that can enter catchup without finishing COPY in this process. + async fn initial_phase(&self, store: &PostgresStore) -> EtlResult { + if self.transaction_mode == PostgresCdcTransactionMode::None { + return Ok(SnapshotPhase::Inactive); + } + + let table_id = target_table_id(store, &self.source_table).await?; + let Some(table_id) = table_id else { + // On the first run etl has not stored the source schema yet. The + // first write_table_rows call supplies the table id. + return Ok(SnapshotPhase::Copy { + started: false, + table_id: None, + }); + }; + + let state = store.get_table_state(table_id).await?; + Ok(match state { + None | Some(TableState::Init | TableState::DataSync) => SnapshotPhase::Copy { + started: false, + table_id: Some(table_id), + }, + Some( + TableState::FinishedCopy | TableState::SyncWait { .. } | TableState::Catchup { .. }, + ) => SnapshotPhase::Catchup { + started: false, + table_id, + }, + Some(TableState::SyncDone { .. } | TableState::Ready | TableState::Errored { .. }) => { + SnapshotPhase::Inactive + } + }) + } + + fn lock_phase(&self) -> std::sync::MutexGuard<'_, SnapshotPhase> { + let phase = self.phase.lock().unwrap(); + assert!( + !matches!(&*phase, SnapshotPhase::Uninitialized), + "etl calls destination startup before submitting writes" + ); + phase + } + + /// Start the COPY transaction on its first queued row buffer. + fn start_copy(&self, table_id: TableId) -> Option> { + self.lock_phase() + .start_copy(table_id) + .then(|| Some(self.transaction_label("copy"))) + } + + fn finish_copy(&self, table_id: TableId) -> bool { + self.lock_phase().finish_copy(table_id) + } + + fn start_catchup(&self, table_id: Option, has_data: bool) -> Option> { + self.lock_phase() + .start_catchup(table_id, has_data) + .then(|| Some(self.transaction_label("catchup"))) + } + + fn finish_catchup(&self) -> bool { + self.lock_phase().finish_catchup() + } + + /// Finish catchup when etl reaches `SyncDone` without sending a terminal + /// destination write. + /// + /// etl can take this path when idle progress reaches the catchup target: + /// + fn finish_catchup_after_etl_completion(&self, table_id: TableId) { + let mut phase = self.phase.lock().unwrap(); + let should_finish = match &*phase { + SnapshotPhase::Catchup { + table_id: catchup_table_id, + .. + } => *catchup_table_id == table_id, + _ => false, + }; + let commit = should_finish && phase.finish_catchup(); + if commit { + // Earlier catchup writes opened a transaction, but etl reached + // SyncDone without a terminal write that could close it. + self.queue.push_entry( + InputQueueEntry::new_with_aux(Utc::now(), DeferredSenders::new()) + .with_commit_transaction(true), + Vec::new(), + ); + } + } + + /// Labels are intentionally stable across connector restarts: each + /// pipeline has at most one copy and one catchup transaction. + fn transaction_label(&self, phase: &str) -> String { + format!("snapshot-{phase}-{}", self.pipeline_id) + } } /// etl Destination implementation that pushes data into a Feldera DeCollectionStream. @@ -991,6 +1207,7 @@ struct FelderaDestination { queue: Arc, source_table: String, endpoint_name: String, + snapshot_transactions: Arc, /// Canonical names of the non-nullable Feldera columns. Each must be present /// (by name) in the target Postgres table schema etl passes with each target /// batch/event. Nullable and extra columns need not match. @@ -1011,6 +1228,10 @@ impl Destination for FelderaDestination { "feldera" } + async fn startup(&self) -> EtlResult<()> { + self.snapshot_transactions.initialize().await + } + async fn drop_table_for_copy( &self, replicated_table_schema: &ReplicatedTableSchema, @@ -1069,11 +1290,14 @@ impl Destination for FelderaDestination { } bytes += json_str.len(); - if bytes >= 2 * 1024 * 1024 { - self.queue.push_with_aux( + if bytes >= MAX_QUEUED_BUFFER_BYTES { + self.push_queue_entry( (stream.take_all(), errors), timestamp, DeferredSenders::new(), + self.snapshot_transactions + .start_copy(replicated_table_schema.id()), + false, ); queued_data = true; bytes = 0; @@ -1082,15 +1306,22 @@ impl Destination for FelderaDestination { } if bytes > 0 || !errors.is_empty() { - self.queue.push_with_aux( + self.push_queue_entry( (stream.take_all(), errors), timestamp, DeferredSenders::new(), + self.snapshot_transactions + .start_copy(replicated_table_schema.id()), + false, ); queued_data = true; } - self.ack_snapshot_copy(queued_data, async_result); + let commit_transaction = !queued_data + && self + .snapshot_transactions + .finish_copy(replicated_table_schema.id()); + self.ack_snapshot_copy(queued_data, commit_transaction, async_result); Ok(()) } @@ -1102,6 +1333,7 @@ impl Destination for FelderaDestination { ) -> EtlResult<()> { self.wait_unpaused().await?; + let target_table_id = self.target_table_id_for_events(&events); let mut stream = self.input_stream.lock().unwrap(); let mut bytes = 0; let mut errors = Vec::new(); @@ -1209,11 +1441,12 @@ impl Destination for FelderaDestination { // Relation events carry only schema, no row data. etl detects // schema changes upstream (it refuses to forward a Relation // whose schema differs from the resolved one) and marks the - // table errored; we surface that via `TableErrorMonitor`. - Event::Relation(_) | Event::Begin(_) | Event::Commit(_) | Event::Unsupported => {} + // table errored; we surface that via `TableStateMonitor`. + Event::Relation(_) => {} + Event::Begin(_) | Event::Commit(_) | Event::Unsupported => {} } - if bytes >= 2 * 1024 * 1024 { + if bytes >= MAX_QUEUED_BUFFER_BYTES { queue_entries.push((stream.take_all(), errors)); bytes = 0; errors = Vec::new(); @@ -1223,29 +1456,38 @@ impl Destination for FelderaDestination { if bytes > 0 || !errors.is_empty() { queue_entries.push((stream.take_all(), errors)); } - let ack = DeferredAck::Stream { result: async_result, durability, }; + let terminal = durability == WriteEventsDurability::RequireDurable; + let mut start_transaction = self + .snapshot_transactions + .start_catchup(target_table_id, !queue_entries.is_empty()); + let commit_transaction = terminal && self.snapshot_transactions.finish_catchup(); + if let Some(last_entry) = queue_entries.pop() { for entry in queue_entries { - self.queue - .push_with_aux(entry, timestamp, DeferredSenders::new()); - } - if self.completion_task_tx.is_some() { - // Increment before pushing so the count cannot underflow if - // the reader flushes and decrements before this lands. - self.queued_acks.fetch_add(1, Ordering::Release); - self.queue.push_with_aux(last_entry, timestamp, vec![ack]); - // etl waits for this ack before sending more events. Keep - // taking steps until the queue entry carrying it is flushed. - self.queue.consumer.request_step(); - } else { - self.queue - .push_with_aux(last_entry, timestamp, DeferredSenders::new()); - ack.complete(); + self.push_queue_entry( + entry, + timestamp, + DeferredSenders::new(), + start_transaction.take(), + false, + ); } + self.queue_stream_ack( + last_entry, + timestamp, + ack, + start_transaction, + commit_transaction, + ); + } else if commit_transaction { + // Catchup can end with only BEGIN/COMMIT events and no row buffer. + // Queue an empty entry so the Feldera transaction still closes + // after all preceding catchup data. + self.queue_stream_ack((None, Vec::new()), timestamp, ack, None, true); } else { self.ack_no_row_stream_write(ack); } @@ -1255,39 +1497,169 @@ impl Destination for FelderaDestination { } impl FelderaDestination { + #[allow(clippy::too_many_arguments)] + fn new( + input_stream: Box, + endpoint_name: String, + source_table: String, + transaction_mode: PostgresCdcTransactionMode, + pipeline_id: u64, + queue: Arc, + store: PostgresStore, + feldera_required_columns: Vec, + completion_task_tx: Option>, + queued_acks: Arc, + pipeline_state_rx: Receiver, + ) -> Self { + let snapshot_transactions = Arc::new(SnapshotTransactions::new( + source_table.clone(), + transaction_mode, + pipeline_id, + Arc::clone(&queue), + store, + )); + Self { + input_stream: Arc::new(Mutex::new(input_stream)), + queue, + source_table, + endpoint_name, + snapshot_transactions, + feldera_required_columns, + completion_task_tx, + queued_acks, + pipeline_state_rx, + } + } + + fn snapshot_transactions(&self) -> Arc { + Arc::clone(&self.snapshot_transactions) + } + + fn target_table_id_for_events(&self, events: &[Event]) -> Option { + events.iter().find_map(|event| match event { + Event::Insert(insert) => { + self.target_table_id_for_schema(&insert.replicated_table_schema) + } + Event::Update(update) => { + self.target_table_id_for_schema(&update.replicated_table_schema) + } + Event::Delete(delete) => { + self.target_table_id_for_schema(&delete.replicated_table_schema) + } + Event::Relation(relation) => { + self.target_table_id_for_schema(&relation.replicated_table_schema) + } + Event::Truncate(truncate) => truncate + .truncated_tables + .iter() + .find_map(|schema| self.target_table_id_for_schema(schema)), + Event::Begin(_) | Event::Commit(_) | Event::Unsupported => None, + }) + } + + fn target_table_id_for_schema(&self, schema: &ReplicatedTableSchema) -> Option { + let name = schema.name(); + self.is_target_table(&name.schema, &name.name) + .then(|| schema.id()) + } + + fn push_queue_entry( + &self, + (buffer, errors): (Option>, Vec), + timestamp: chrono::DateTime, + senders: DeferredSenders, + start_transaction: Option>, + commit_transaction: bool, + ) { + self.queue.push_entry( + InputQueueEntry::new_with_aux(timestamp, senders) + .with_buffer(buffer) + .with_start_transaction(start_transaction) + .with_commit_transaction(commit_transaction), + errors, + ); + } + + fn queue_stream_ack( + &self, + entry: (Option>, Vec), + timestamp: chrono::DateTime, + ack: DeferredAck, + start_transaction: Option>, + commit_transaction: bool, + ) { + if self.completion_task_tx.is_some() { + self.queued_acks.fetch_add(1, Ordering::Release); + self.push_queue_entry( + entry, + timestamp, + vec![ack], + start_transaction, + commit_transaction, + ); + // etl waits for this ack before sending more events. Keep taking + // steps until the queue entry carrying it is flushed. + self.queue.consumer.request_step(); + } else { + self.push_queue_entry( + entry, + timestamp, + DeferredSenders::new(), + start_transaction, + commit_transaction, + ); + ack.complete(); + } + } + /// Answer a table-copy write. /// - /// Without completion tracking there is nothing to wait for, so every - /// write is `Durable` immediately. - /// - /// With tracking, a batch that queued rows is answered `Accepted` right - /// away: Feldera owns the rows, and etl keeps copying while they wait for - /// the completion frontier. + /// A batch that queued rows is answered `Accepted` right away: Feldera owns + /// the rows, and etl keeps copying. This also forces etl to send its terminal + /// empty write, including when completion tracking is unavailable, so the + /// copy transaction always gets its commit barrier. /// /// The one empty write etl sends at the end of the copy (an empty table - /// sends only this) is the table's durability barrier: answering it - /// `Durable` tells etl the whole snapshot is safe. So instead of answering - /// now, queue it behind all the snapshot data and let the completion - /// watcher answer once its step passes the frontier. - fn ack_snapshot_copy(&self, queued_data: bool, async_result: WriteTableRowsResult) { - if self.completion_task_tx.is_none() { - async_result.send(Ok(DestinationWriteStatus::Durable)); - } else if queued_data { - async_result.send(Ok(DestinationWriteStatus::Accepted)); - } else { - // At most one barrier can be live at a time: only the configured - // source table's copy reaches this branch. Every other publication - // table fails the column match in `write_table_rows` and is acked - // Durable immediately. - // - // Increment before pushing so the count cannot underflow if the - // reader flushes and decrements before this increment lands. - self.queued_acks.fetch_add(1, Ordering::Release); - self.queue.push_with_aux( - (None, Vec::new()), - Utc::now(), - vec![DeferredAck::SnapshotCopyBarrier(async_result)], - ); + /// sends only this) is the table's durability barrier. With tracking, queue + /// it behind all snapshot data until the watcher passes its step. Without + /// tracking, queue an empty commit entry when it closes a copy transaction, + /// then answer `Durable` immediately. + fn ack_snapshot_copy( + &self, + queued_data: bool, + commit_transaction: bool, + async_result: WriteTableRowsResult, + ) { + let tracking = self.completion_task_tx.is_some(); + match immediate_snapshot_copy_status(tracking, queued_data) { + Some(DestinationWriteStatus::Accepted) => { + async_result.send(Ok(DestinationWriteStatus::Accepted)); + } + Some(DestinationWriteStatus::Durable) => { + if commit_transaction { + self.push_queue_entry( + (None, Vec::new()), + Utc::now(), + DeferredSenders::new(), + None, + true, + ); + } + async_result.send(Ok(DestinationWriteStatus::Durable)); + } + None => { + // We only copy the configured source table. write_table_rows + // ignores every other table in the publication and immediately + // marks its batches durable. + self.queued_acks.fetch_add(1, Ordering::Release); + self.push_queue_entry( + (None, Vec::new()), + Utc::now(), + vec![DeferredAck::SnapshotCopyBarrier(async_result)], + None, + commit_transaction, + ); + } } } @@ -1327,10 +1699,7 @@ impl FelderaDestination { } fn is_target_table(&self, schema_name: &str, table_name: &str) -> bool { - let qualified = format!("{schema_name}.{table_name}"); - self.source_table == qualified - || self.source_table == table_name - || self.source_table == format!("\"{schema_name}\".\"{table_name}\"") + is_target_table(&self.source_table, schema_name, table_name) } /// Resolve the replicated column names for `schema`. @@ -1384,6 +1753,34 @@ impl FelderaDestination { } } +/// Return the COPY status that can be reported immediately. +fn immediate_snapshot_copy_status( + tracking: bool, + queued_data: bool, +) -> Option { + match (tracking, queued_data) { + (_, true) => Some(DestinationWriteStatus::Accepted), + (false, false) => Some(DestinationWriteStatus::Durable), + (true, false) => None, + } +} + +async fn target_table_id(store: &PostgresStore, source_table: &str) -> EtlResult> { + Ok(store + .get_table_schemas() + .await? + .iter() + .find(|schema| is_target_table(source_table, &schema.name.schema, &schema.name.name)) + .map(|schema| schema.id)) +} + +fn is_target_table(source_table: &str, schema_name: &str, table_name: &str) -> bool { + let qualified = format!("{schema_name}.{table_name}"); + source_table == qualified + || source_table == table_name + || source_table == format!("\"{schema_name}\".\"{table_name}\"") +} + /// Replicated column names of `schema`, in the order etl emits cell values for /// a row. fn replicated_column_names(schema: &ReplicatedTableSchema) -> Vec { @@ -1925,6 +2322,95 @@ mod tests { assert!(should_discard_table_error(&replication_error, true)); } + #[test] + fn snapshot_transaction_moves_from_copy_to_catchup() { + let table_id = TableId::new(42); + let mut phase = SnapshotPhase::Copy { + started: false, + table_id: None, + }; + + assert!(phase.start_copy(table_id)); + assert!(!phase.start_copy(table_id)); + assert!(phase.finish_copy(table_id)); + assert!(matches!( + &phase, + SnapshotPhase::Catchup { + started, + table_id: id, + } if !*started && *id == table_id + )); + + assert!(phase.start_catchup(Some(table_id), true)); + assert!(!phase.start_catchup(Some(table_id), true)); + assert!(phase.finish_catchup()); + assert!(matches!(phase, SnapshotPhase::Inactive)); + } + + #[test] + fn snapshot_catchup_only_starts_for_target_table_with_data() { + let table_id = TableId::new(42); + let mut phase = SnapshotPhase::Catchup { + started: false, + table_id, + }; + + assert!(!phase.start_catchup(Some(TableId::new(43)), true)); + assert!(!phase.start_catchup(Some(table_id), false)); + assert!(!phase.start_catchup(None, true)); + assert!(phase.start_catchup(Some(table_id), true)); + } + + #[test] + fn snapshot_empty_copy_moves_to_catchup_without_commit() { + let table_id = TableId::new(42); + let mut phase = SnapshotPhase::Copy { + started: false, + table_id: Some(table_id), + }; + + assert!(!phase.finish_copy(table_id)); + assert!(matches!( + &phase, + SnapshotPhase::Catchup { + started, + table_id: id, + } if !*started && *id == table_id + )); + } + + #[test] + fn snapshot_unstarted_catchup_finishes_without_commit() { + let mut phase = SnapshotPhase::Catchup { + started: false, + table_id: TableId::new(42), + }; + + assert!(!phase.finish_catchup()); + assert!(matches!(phase, SnapshotPhase::Inactive)); + } + + #[test] + fn snapshot_copy_without_tracking_forces_terminal_barrier() { + assert_eq!( + immediate_snapshot_copy_status(false, true), + Some(DestinationWriteStatus::Accepted) + ); + assert_eq!( + immediate_snapshot_copy_status(false, false), + Some(DestinationWriteStatus::Durable) + ); + } + + #[test] + fn tracked_snapshot_copy_defers_terminal_barrier() { + assert_eq!( + immediate_snapshot_copy_status(true, true), + Some(DestinationWriteStatus::Accepted) + ); + assert_eq!(immediate_snapshot_copy_status(true, false), None); + } + fn test_ack(kind: TestAckKind, tx: std::sync::mpsc::Sender) -> DeferredAck { DeferredAck::Test { kind, tx } } diff --git a/crates/adapters/src/integrated/postgres/test.rs b/crates/adapters/src/integrated/postgres/test.rs index 7810085a94f..739549ed776 100644 --- a/crates/adapters/src/integrated/postgres/test.rs +++ b/crates/adapters/src/integrated/postgres/test.rs @@ -2396,6 +2396,7 @@ mod cdc_tests { use super::*; use crate::test::wait; use feldera_types::config::PipelineConfig; + use feldera_types::transport::postgres::PostgresCdcTransactionMode; use pg::pg_connect; /// Helper: creates a table, publication, and sets REPLICA IDENTITY FULL. @@ -3055,6 +3056,7 @@ mod cdc_tests { streaming_ack_hold_ms: u64, discard_shutdown_errors: bool, discard_table_errors: bool, + transaction_mode: PostgresCdcTransactionMode, max_batch_size: Option, min_batch_size_records: u64, max_buffering_delay_usecs: u64, @@ -3069,6 +3071,7 @@ mod cdc_tests { discard_shutdown_errors: feldera_types::transport::postgres::default_discard_shutdown_errors(), discard_table_errors: false, + transaction_mode: PostgresCdcTransactionMode::None, max_batch_size: None, min_batch_size_records: 0, max_buffering_delay_usecs: 0, @@ -3129,6 +3132,7 @@ mod cdc_tests { "streaming_ack_hold_ms": options.streaming_ack_hold_ms, "discard_shutdown_errors": options.discard_shutdown_errors, "discard_table_errors": options.discard_table_errors, + "transaction_mode": options.transaction_mode, }, }, }, @@ -3229,6 +3233,11 @@ mod cdc_tests { }) } + fn cdc_transaction_idle(controller: &Controller) -> bool { + let metrics = controller.api_status(false).global_metrics; + metrics.transaction_status == feldera_types::adapter_stats::TransactionStatus::NoTransaction + } + // ------------------------------------------------------------------- // Test 1: Basic CDC insert test // ------------------------------------------------------------------- @@ -4844,6 +4853,31 @@ mod cdc_tests { format!("supabase_etl_apply_{pipeline_id}") } + fn table_sync_slot_exists( + table: &mut CdcTestTable, + source_table: &str, + source_table_oid: i32, + ) -> bool { + let connector_url = cdc_connector_url(&table.url); + let pipeline_id = crate::integrated::postgres::cdc_input::pipeline_id( + &connector_url, + &table.publication_name, + source_table, + ); + let slot_name = format!("supabase_etl_table_sync_{pipeline_id}_{source_table_oid}"); + + table + .client + .query_one( + "SELECT EXISTS ( + SELECT 1 FROM pg_replication_slots WHERE slot_name = $1 + )", + &[&slot_name], + ) + .map(|row| row.get(0)) + .unwrap_or(false) + } + fn confirmed_flush_lsn(table: &mut CdcTestTable, source_table: &str) -> Option { let slot_name = apply_slot_name(table, source_table); table @@ -4942,6 +4976,178 @@ mod cdc_tests { .unwrap_or(false) } + /// `transaction_mode: snapshot` uses one transaction for COPY and another + /// for the table-sync WAL catchup, then leaves steady-state CDC outside a + /// connector-managed transaction. + /// + /// Requires: wal_level=logical, user with REPLICATION privilege. + #[test] + #[serial] + fn test_cdc_snapshot_transactions_cover_copy_and_catchup() { + let url = postgres_url(); + let table_name = unique_pg_name("cdc_test_snapshot_transactions"); + let publication = unique_pg_name("cdc_pub_snapshot_transactions"); + const SNAPSHOT_ROWS: usize = 12; + const CATCHUP_ROWS: usize = 12; + const CATCHUP_ANCHOR_ID: i64 = 500; + + let mut table = CdcTestTable::new_simple(&table_name, &publication, &url); + table.execute(&format!( + "INSERT INTO {table_name} (id, b, i, s) \ + SELECT id, true, id * 10, repeat('s', 1024 * 1024) \ + FROM generate_series(1, {SNAPSHOT_ROWS}) AS id" + )); + + let source_table = format!("public.{table_name}"); + let source_table_oid = table_oid(&mut table, &source_table); + let storage = tempfile::tempdir().unwrap(); + let output = NamedTempFile::new().unwrap(); + let (controller, errors) = cdc_ft_test_circuit_with_options( + &url, + &publication, + &source_table, + storage.path(), + output.path(), + CdcFtTestOptions { + streaming_ack_hold_ms: 100, + transaction_mode: PostgresCdcTransactionMode::Snapshot, + max_batch_size: Some(1), + min_batch_size_records: 1_000_000, + max_buffering_delay_usecs: 3_600_000_000, + ..Default::default() + }, + ); + controller.start(); + + wait( + || { + table_sync_slot_exists(&mut table, &source_table, source_table_oid) + || !errors.is_empty() + }, + 60_000, + ) + .expect("timeout: etl table-copy slot was not created"); + assert!( + errors.is_empty(), + "unexpected error starting table COPY: {:?}", + errors.try_recv() + ); + assert_eq!( + count_inserts(&read_output_json(output.path())), + 0, + "COPY rows became visible before the transaction committed" + ); + + // These rows are newer than the COPY snapshot and must be ingested by + // the table-sync worker's catchup transaction. + table.execute(&format!( + "INSERT INTO {table_name} (id, b, i, s) \ + SELECT 100 + id, false, id * 10, repeat('c', 1024 * 1024) \ + FROM generate_series(1, {CATCHUP_ROWS}) AS id" + )); + + let mut first_visible_count = 0; + wait( + || { + first_visible_count = count_inserts(&read_output_json(output.path())); + first_visible_count > 0 || !errors.is_empty() + }, + 120_000, + ) + .expect("COPY transaction did not publish its snapshot rows"); + assert_eq!( + first_visible_count, SNAPSHOT_ROWS, + "COPY transaction exposed only part of the snapshot" + ); + wait( + || cdc_transaction_idle(&controller) || !errors.is_empty(), + 30_000, + ) + .expect("timeout: COPY transaction did not become idle"); + assert!( + errors.is_empty(), + "unexpected error committing COPY transaction: {:?}", + errors.try_recv() + ); + + controller.checkpoint().expect("COPY checkpoint failed"); + wait( + || copy_state_persisted(&mut table, source_table_oid) || !errors.is_empty(), + 60_000, + ) + .expect("timeout: etl did not persist COPY completion"); + + assert_eq!( + count_inserts(&read_output_json(output.path())), + SNAPSHOT_ROWS, + "catchup rows became visible before the transaction committed" + ); + + // As in the other initial-sync tests, drive the replication stream + // after catchup starts so etl can observe that it reached its target + // LSN even when PostgreSQL was otherwise idle. + table.execute(&format!( + "INSERT INTO {table_name} VALUES \ + ({CATCHUP_ANCHOR_ID}, true, {CATCHUP_ANCHOR_ID}, 'catchup_anchor')" + )); + + let mut first_count_after_copy = SNAPSHOT_ROWS; + wait( + || { + first_count_after_copy = count_inserts(&read_output_json(output.path())); + first_count_after_copy > SNAPSHOT_ROWS || !errors.is_empty() + }, + 30_000, + ) + .expect("catchup transaction did not publish its WAL rows atomically"); + assert!( + (SNAPSHOT_ROWS + CATCHUP_ROWS..=SNAPSHOT_ROWS + CATCHUP_ROWS + 1) + .contains(&first_count_after_copy), + "catchup transaction exposed only part of its WAL rows: first visible count was \ + {first_count_after_copy}" + ); + assert!( + errors.is_empty(), + "unexpected error committing catchup transaction: {:?}", + errors.try_recv() + ); + + controller.checkpoint().expect("catchup checkpoint failed"); + wait( + || table_ready(&mut table, source_table_oid) || !errors.is_empty(), + 60_000, + ) + .expect("timeout: etl table did not become ready after catchup"); + wait( + || { + count_inserts(&read_output_json(output.path())) == SNAPSHOT_ROWS + CATCHUP_ROWS + 1 + || !errors.is_empty() + }, + 60_000, + ) + .expect("timeout: main apply worker did not publish remaining WAL rows"); + + table.execute(&format!( + "INSERT INTO {table_name} VALUES (1000, true, 1000, 'steady')" + )); + wait( + || has_insert_id(&read_output_json(output.path()), 1000) || !errors.is_empty(), + 60_000, + ) + .expect("timeout: steady-state row did not reach output"); + assert!( + errors.is_empty(), + "unexpected error during steady-state CDC: {:?}", + errors.try_recv() + ); + assert!( + cdc_transaction_idle(&controller), + "steady-state CDC unexpectedly started a connector transaction" + ); + + controller.stop().unwrap(); + } + /// The terminal snapshot barrier must stay behind every copied row across /// multiple controller queue flushes. /// diff --git a/crates/feldera-types/src/transport/postgres.rs b/crates/feldera-types/src/transport/postgres.rs index 52b72ccfb0e..c4f44bf116f 100644 --- a/crates/feldera-types/src/transport/postgres.rs +++ b/crates/feldera-types/src/transport/postgres.rs @@ -67,6 +67,23 @@ impl PostgresTlsConfig { } } +/// PostgreSQL CDC input transaction mode. +/// +/// Determines whether the connector groups the initial table synchronization +/// into Feldera transactions. +#[derive(Debug, Clone, Copy, Eq, PartialEq, Deserialize, Serialize, ToSchema, Default)] +pub enum PostgresCdcTransactionMode { + /// Do not create transactions automatically. + #[default] + #[serde(rename = "none")] + None, + + /// Ingest the initial table copy and its WAL catchup in two separate + /// transactions. Steady-state changes are not transactional. + #[serde(rename = "snapshot")] + Snapshot, +} + /// Postgres CDC input connector configuration. /// /// Uses logical replication to capture ongoing changes from a Postgres database. @@ -86,6 +103,13 @@ pub struct PostgresCdcReaderConfig { /// Must be included in the publication. pub source_table: String, + /// Transaction mode. + /// + /// See [`PostgresCdcTransactionMode`]. When omitted, defaults to + /// [`PostgresCdcTransactionMode::None`]. + #[serde(default)] + pub transaction_mode: PostgresCdcTransactionMode, + /// TLS/SSL configuration. #[serde(flatten)] #[schema(inline)] @@ -349,6 +373,7 @@ mod tests { uri: "postgres://user:password@localhost:5432/database".to_string(), publication: "publication".to_string(), source_table: "public.table".to_string(), + transaction_mode: PostgresCdcTransactionMode::None, tls, streaming_ack_hold_ms: default_streaming_ack_hold_ms(), discard_shutdown_errors: default_discard_shutdown_errors(), diff --git a/docs.feldera.com/docs/connectors/sources/postgresql-cdc.md b/docs.feldera.com/docs/connectors/sources/postgresql-cdc.md index ad4ea3b20fa..592a1ff5fcc 100644 --- a/docs.feldera.com/docs/connectors/sources/postgresql-cdc.md +++ b/docs.feldera.com/docs/connectors/sources/postgresql-cdc.md @@ -24,16 +24,17 @@ privileges. Use transport name `postgres_cdc_input`. -| Property | Type | Default | Description | -| ------------------------- | ------- | ------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `uri`\* | string | | PostgreSQL connection URI, e.g. `"postgres://postgres:password@localhost:5432/postgres"`. It must include a username, host, and database name. The user needs `REPLICATION` privilege. | -| `publication`\* | string | | Name of an existing PostgreSQL publication. The publication must include `source_table`. | -| `source_table`\* | string | | PostgreSQL table to replicate, usually schema-qualified, e.g. `"public.orders"`. | -| `ssl_ca_pem` | string | | CA certificates in PEM format. Setting this enables TLS and takes precedence over `ssl_ca_location`. | -| `ssl_ca_location` | string | | Path to a PEM file containing CA certificates. Used when `ssl_ca_pem` is not set. | -| `streaming_ack_hold_ms` | integer | `2000` | Time to wait for a nonterminal CDC batch to become durable before ingestion may continue without advancing its PostgreSQL WAL position. Must be greater than zero. | -| `discard_shutdown_errors` | boolean | `true` | Whether to retry a table when the previous connector run stopped before acknowledging pending data. | -| `discard_table_errors` | boolean | `false` | Whether to retry tables with persisted replication errors at startup. See [Discarding table errors](#discarding-table-errors). | +| Property | Type | Default | Description | +| ------------------------- | ------- | ------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `uri`\* | string | | PostgreSQL connection URI, e.g. `"postgres://postgres:password@localhost:5432/postgres"`. It must include a username, host, and database name. The user needs `REPLICATION` privilege. | +| `publication`\* | string | | Name of an existing PostgreSQL publication. The publication must include `source_table`. | +| `source_table`\* | string | | PostgreSQL table to replicate, usually schema-qualified, e.g. `"public.orders"`. | +| `transaction_mode` | enum | `none` | Controls transactions during initial synchronization. Supported values are `none` and `snapshot`. See [Transactions](#transactions). | +| `ssl_ca_pem` | string | | CA certificates in PEM format. Setting this enables TLS and takes precedence over `ssl_ca_location`. | +| `ssl_ca_location` | string | | Path to a PEM file containing CA certificates. Used when `ssl_ca_pem` is not set. | +| `streaming_ack_hold_ms` | integer | `2000` | Time to wait for a nonterminal CDC batch to become durable before ingestion may continue without advancing its PostgreSQL WAL position. Terminal snapshot batches always wait for durability. Must be greater than zero. | +| `discard_shutdown_errors` | boolean | `true` | Whether to retry a table when the previous connector run stopped before acknowledging pending data. | +| `discard_table_errors` | boolean | `false` | Whether to retry tables with persisted replication errors at startup. See [Discarding table errors](#discarding-table-errors). | [*]: Required fields @@ -41,6 +42,21 @@ The CDC connector does not support client-certificate TLS options (`ssl_client_pem`, `ssl_client_location`, `ssl_client_key`, `ssl_client_key_location`, or `ssl_certificate_chain_location`). +## Transactions + +The `transaction_mode` property controls whether the connector groups initial +synchronization into [Feldera transactions](/pipelines/transactions): + +- `none` - do not create transactions automatically. +- `snapshot` - use one transaction for the initial PostgreSQL table copy and a + second transaction for the WAL changes ETL applies while catching up from the + copy's consistent snapshot. Each transaction is committed only after its + complete phase has been queued. + +The two phases cannot share one transaction: ETL waits for the copied rows to +become durable before it starts WAL catchup. Once catchup completes, subsequent +steady-state CDC changes are ingested without transactions. + ## PostgreSQL setup The PostgreSQL server must have logical replication enabled: diff --git a/openapi.json b/openapi.json index 9995cef9f91..e420f9e70fb 100644 --- a/openapi.json +++ b/openapi.json @@ -11969,6 +11969,9 @@ "default": 2000, "minimum": 0 }, + "transaction_mode": { + "$ref": "#/components/schemas/PostgresCdcTransactionMode" + }, "uri": { "type": "string", "description": "Postgres connection URI. The user must have REPLICATION privilege.\nSee: "