[adapters] Add Postgres CDC input connector with crash-safe replication - #5988

@flak153 this is fantastic! Thank you for your contribution. We'll get this some reviews.

Thanks everyone for the thorough reviews. Pushed fixes for all comments:

@abhizer — Good catch on the Datagen format. Found two mismatches and fixed both:

• Cell::Timestamp now uses RFC 3339 format (T separator) instead of space-separated
• Cell::Bytes now encodes as byte arrays instead of hex strings to match BinaryFormat::Array
• Same fixes for the ArrayCell variants
• Added 47 unit tests covering all cell_to_json/array_cell_to_json variants plus URI parsing and table matching

@blp — All comments addressed:

• completion_watcher default impl removed, added explicit None to all 3 implementations
• with-postgres-cdc added to default features
• mpsc → oneshot for init status channel
• mem::take instead of drain().collect()
• Kept thread::Builder because open() is sync and needs blocking_recv for init status — happy to switch if there's a better pattern
• On FT: the path is switching MemoryStore → PostgresStore so etl persists table phases across restarts, then implementing Resume::Seek. Planning that as a follow-up since it's a significant addition.

@mythical-fred — Both blockers addressed:

• etl deps pinned to rev = "05cb11ae"
• Added 3 integration tests (test_cdc_basic_insert, test_cdc_all_data_types, test_cdc_update_delete) marked #[ignore] since they require wal_level=logical

Squashed into a single commit.

Question — beyond the current tests, there are several edge cases we could cover. Would you like any of these before merge, or are they better as follow-ups?

High priority (most likely to surface real bugs):

• Crash-resume / failover: start pipeline, insert data, kill without confirming, restart, verify data replays
• Schema changes during streaming: column added/dropped while CDC is running
• Large transactions: single transaction with thousands of rows (tests 2MB buffer splitting)
• Multiple tables in publication: verify source_table filtering ignores other tables

Medium priority:

• TRUNCATE events (currently ignored — verify it doesn't crash)
• Empty table snapshot (zero rows, then streaming starts)
• REPLICA IDENTITY behavior without FULL (no old row on updates/deletes)
• Rapid insert/update/delete cycles on same row

Lower priority (edge cases):

• Connection loss / Postgres restart mid-stream
• Slot invalidation (slot dropped externally)
• Long-running transactions
• Special float values end-to-end (NaN, Infinity)

Happy to add whichever subset you think is important for this PR vs follow-ups.

Pushed updates:

• Fault tolerance implemented: Switched MemoryStore → PostgresStore so etl persists table replication phases across restarts. The connector now returns Some(FtModel::AtLeastOnce). On restart, etl reads the stored phases from Postgres and resumes from the replication slot position instead of re-snapshotting.
• Column names pre-constructed outside the per-row loop in write_table_rows (blp's comment)
• Numeric: Keeping .to_string() — PgNumeric is a custom enum (NaN/Infinity/Value) without Serialize, so json!(n) wouldn't work even with arbitrary_precision
• Pipeline ID: Uses a deterministic hash of (URI, publication, source_table) so the same replication slot and stored state are reused across restarts

Question on fault tolerance implementation — looking for guidance on the right pattern.

The connector now uses PostgresStore so etl persists table replication phases in Postgres. On restart, etl checks the stored phases and resumes from the replication slot position (no re-snapshot). This works regardless of what fault_tolerance() returns.

To properly return Some(FtModel::AtLeastOnce), we need to provide Resume::Seek metadata via consumer.extended(). The issue is that InputQueue::queue() hardcodes resume: None in its internal extended() call, and we use queue() to avoid reimplementing its batching/transaction logic.

Would it be valid to call extended() a second time after queue() with empty data + resume metadata?

self.inner.queue.queue();  // flushes data, calls extended(total, None, watermarks)
self.inner.consumer.extended(
    BufferSize::empty(),
    Some(Resume::Seek { seek: json!({"pipeline_id": pipeline_id}) }),
    vec![],
);

Or is there a better pattern for integrated endpoints that need resume metadata but want to use queue() for the data path?

For now I've set fault_tolerance() to AtLeastOnce with the PostgresStore backing — the crash recovery works because etl manages its own state. But want to make sure the Feldera checkpoint integration is correct too.

Question on fault tolerance implementation — looking for guidance on the right pattern.

The connector now uses PostgresStore so etl persists table replication phases in Postgres. On restart, etl checks the stored phases and resumes from the replication slot position (no re-snapshot). This works regardless of what fault_tolerance() returns.

To properly return Some(FtModel::AtLeastOnce), we need to provide Resume::Seek metadata via consumer.extended(). The issue is that InputQueue::queue() hardcodes resume: None in its internal extended() call, and we use queue() to avoid reimplementing its batching/transaction logic.

Right, InputQueue::queue can't be used for fault tolerance.

One way to achieve fault tolerance along with InputQueue is to add auxiliary data (the A generic parameter to InputQueue) for the resume information to each batch in the queue, and then use flush_with_aux instead of queue and subsequently call extended directly. The nats connector does this.

Would it be valid to call extended() a second time after queue() with empty data + resume metadata?

No, it needs to be called exactly once.

Switched to flush_with_aux + manual extended() with Resume::Seek per blp's guidance (nats pattern). The Queue handler now calls extended exactly once with resume metadata containing the pipeline_id. Also moved pipeline_id computation to new() so it's available in the Queue handler.

Switched to flush_with_aux + manual extended() with Resume::Seek per blp's guidance (nats pattern). The Queue handler now calls extended exactly once with resume metadata containing the pipeline_id. Also moved pipeline_id computation to new() so it's available in the Queue handler.

Awesome! Thanks for the update. I will take a look at the revised version as soon as I can. Unfortunately that will be a few days (probably Tuesday) because I am leaving soon for a long weekend.

Small follow-up fix: the pipeline_id hash was using config.uri directly, which includes the password. Rotating the password would change the pipeline_id and orphan the replication slot. Fixed with a new stable_connection_identity() helper that parses the URI and hashes only host:port/db + publication + source_table. Added 5 unit tests covering password rotation, username changes, host changes, publication changes, and URI parse failure fallback.

@blp gentle ping — ready for another look whenever you have time. mythical-fred's comments are all addressed (approved) and the password-rotation fix is in. No rush, just want to make sure it doesn't fall off your radar.

Hi, here are the issues I'd raise that potentially haven't been flagged yet, in rough priority order:

HIGH: Unqualified table name matching is too broad

crates/adapters/src/integrated/postgres/cdc_input.rs

  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   // ← danger
          || self.source_table == format!("\"{schema_name}\".\"{table_name}\"")
  }

If a user configures source_table = "orders" (no schema) and the publication covers public.orders AND audit.orders, both will match. The connector will silently ingest both tables and interleave their rows. Either reject unqualified names at config validation time or enforce schema-qualified matching always.

HIGH: Missing Relation event causes silent data loss

WAL events arrive in order: Relation → Insert/Update/Delete. If a Relation event is ever missed (e.g. slot created after the schema was set up, slot invalidated and recreated), subsequent DML events for that table fall through to the else branch that generates fallback column names (col_0, col_1...) rather than aborting or warning. Schema mismatch goes undetected.

Suggest: emit a structured warning/error when DML arrives for an unknown table_id rather than silently falling back to generic column names.

MEDIUM: Timestamp format %.f may silently truncate

format!("%Y-%m-%dT%H:%M:%S%.f")

In chrono, %.f formats fractional seconds with trailing zeros suppressed. So 12:00:00.000000000 formats as "12:00:00" — indistinguishable from a timestamp with no sub-second component. Whether the downstream Datagen parser handles this correctly depends on whether it accepts both "12:00:00" and "12:00:00.123".
Worth an explicit test with a timestamp that has nanosecond precision to verify round-trip fidelity.

MEDIUM: Resume metadata doesn't encode actual LSN position

The current Resume::Seek payload is just { "pipeline_id": } — a static identifier, not a position. Feldera's fault-tolerance model for Resume::Seek semantics implies the reader can skip to that position; but here the actual LSN tracking is entirely in PostgresStore (etl-side), with no coordination with Feldera's checkpoint system.

This means: if Feldera's checkpoint store is wiped independently of PostgresStore (or vice versa), the two systems can diverge. The pipeline_id in the seek metadata has no operational effect on restart — it's identity, not position. This is worth calling out in documentation and probably in a // TODO comment
explaining the limitation, since it affects the semantic correctness claim of AtLeastOnce.

LOW: stable_connection_identity strips password but not all volatile fields

sslmode, connect_timeout, application_name can appear in connection URIs and are currently included in the identity hash. Changing application_name (common in monitoring setups) would change pipeline_id, orphaning the replication slot. Recommend extracting only host, port, dbname as the stable identity, not the full path component after stripping password.

LOW: PgNumeric string representation for special values

Cell::Numeric(n) => json!(n.to_string()) 

etl's PgNumeric can represent NaN and Infinity. If to_string() produces "NaN" or "Infinity", these are valid JSON strings but the downstream Feldera decimal type parser may not accept them, silently dropping or erroring on those rows. Same pattern as the float handling: should check and special-case. Looking at etl source, PgNumeric::to_string() does in fact produce "NaN" for NaN variants.

@nmarasoiu on your point about the resume metadata being identity rather than position — you're right that it's a bit of a half-measure. Want to flag the tradeoff and get input.

Actual resume state lives in etl's PostgresStore + the replication slot's confirmed_flush_lsn. On restart, etl checks its stored table phases and resumes from the slot position. Feldera's Resume::Seek is really just "yes, I support resumption" — the pipeline_id is there so Feldera's controller doesn't think we're stateless.

For stateless pipelines (forward CDC events to a sink), this is functionally fine. Outputs are the source of truth, reprocessing the same inputs produces the same outputs.

For stateful circuits it's murkier. Step completion fires our async_result and advances the slot, but Feldera's circuit state is only durable after a checkpoint. If Feldera crashes between those two events, the slot could be ahead of the last checkpoint on restart — data between those points wouldn't be redelivered but also wasn't persisted in Feldera's state.

The cleaner fix is to wait on checkpoint_watcher instead of completion_watcher — only advance the slot after Feldera has actually checkpointed. The tradeoff is real and user-facing: since etl is single-inflight, batches would be gated on checkpoint cadence instead of step completion. If Feldera checkpoints every ~30s, downstream latency goes from few-second to ~30s. Data from Postgres would sit in etl's buffers until the next checkpoint.

So the choice is between faster but racy (current) vs stricter but slower (checkpoint-based). Probably a config flag is the right answer — default to completion-based for the common stateless use case, opt into checkpoint_watcher for stateful pipelines that need the guarantee. Reasonable for this PR, or do you think strict should be the default?

Thanks.

On the completion_watcher vs checkpoint_watcher flag. I agree with the flag direction and would argue for flipping the default: checkpoint_watcher (strict) as default, completion_watcher (fast) as an explicit opt-in.

The precondition for the fast path isn't "stateless circuit" alone: it's stateless circuit and idempotent output commit:

Only the top-left cell is safe. For the other three, the slot can advance past Feldera's last durable checkpoint: data in that window is either already ack'd but lost from circuit state (stateful case), or delivered twice to non-idempotent outputs.

The connector has no way to detect which cell a given pipeline lands in: circuit statefulness isn't exposed to input adapters, and sink idempotency is a property of the downstream system and the specific operations emitted, not something knowable from the CDC side. So "pick the fast path when it's safe" isn't an option; the user has to declare it.

Correctness-opt-in is thus a wrong default shape imo; performance-opt-in is the healthy one. Worth documenting the preconditions explicitly wherever fast mode is enabled.

One operational item we hadn't raised and think is worth flagging prominently — PostgresStore::new(pipeline_id, source_config) uses the source connection, and etl/migrations/20250827000000_base.sql opens with "Base schema for etl-replicator
store (applied on the source DB)". So Feldera is creating an etl.* schema (four tables) inside the user's source Postgres. Implications:

• Requires CREATE SCHEMA / CREATE TABLE privileges on the source, not just REPLICATION + SELECT.
• Read-replicas can't serve as sources.
• An etl schema appears in the customer's DB with nothing in Feldera's config making that explicit.

Not necessarily blocking, but worth documenting very prominently — and a follow-up to support a separate state DB via config would be a meaningful hardening.

One concrete one-line fix: cell_to_json at cdc_input.rs:697-710 converts f32/f64 NaN/Infinity to Value::Null, but Cell::Numeric at line 711-714 passes through n.to_string() unconditionally. PgNumeric::Display produces literal
"NaN" / "Infinity" / "-Infinity" (etl/etl/src/conversions/numeric.rs:524-530), which downstream decimal parsers won't accept. Suggest:

Cell::Numeric(n) => match n {
    PgNumeric::NaN
    | PgNumeric::PositiveInfinity
    | PgNumeric::NegativeInfinity => Value::Null,
    _ => json!(n.to_string()),
}

Future follow-up (not blocking this PR): Event::Relation overwrites the connector's relation_cache without diffing against the deserializer schema configured at startup. An ALTER TABLE ADD/DROP COLUMN mid-stream silently produces misaligned JSON.

Worth a dedicated design pass in a follow-up: schema evolution deserves more than a patch here.

PS. On keepalive I want to retract a concern I raised earlier about wal_sender_timeout risks during long confirmation waits. Reading etl more carefully, etl/etl/src/replication/apply.rs:792-809 (PRIORITY 6 in run_active_iteration) explicitly fires PeriodicKeepAlive on a deadline including while blocked on an in-flight flush, with a comment calling out exactly that scenario. Connection drop on slow waits isn't a live concern. Credit where due: that path is solidly handled.

@blp rebased onto current main. Conflicts resolved in Cargo.lock and mock_input_consumer.rs (both new trait methods kept). All 52 unit tests pass. Ready to merge.

Separately — I've been tracking a list of follow-up items from the review (both from reviewers here and things we flagged ourselves): 16 items total, ranging from quick-win fixes (TLS config wiring is a real bug, PgNumeric NaN handling, unqualified table matching, etc.) to larger work (separate state DB config, strict checkpoint sync flag, schema evolution, broader test coverage). Where would you prefer I put this — a GitHub issue, a section in the connector docs as known limitations, split into multiple tracked issues, or somewhere else?

@blp rebased onto current main. Conflicts resolved in Cargo.lock and mock_input_consumer.rs (both new trait methods kept). All 52 unit tests pass. Ready to merge.

Thank you! I started the process. (I wouldn't be surprised if it took a few tries. Thanks for being patient with me!)

Separately — I've been tracking a list of follow-up items from the review (both from reviewers here and things we flagged ourselves): 16 items total, ranging from quick-win fixes (TLS config wiring is a real bug, PgNumeric NaN handling, unqualified table matching, etc.) to larger work (separate state DB config, strict checkpoint sync flag, schema evolution, broader test coverage). Where would you prefer I put this — a GitHub issue, a section in the connector docs as known limitations, split into multiple tracked issues, or somewhere else?

For things that are mostly internal, I think that a GitHub issue is a good idea. I leave it to your judgment and preferences whether that's better as one issue with a checklist, or separate issues, or some mix.

For things that are user-visible, I think it's good to put them into the connector docs. If there's an issue to reference then it's kind to reference it from the docs.

I see that this failed in the pre-merge check. That's not a huge surprise, it happens to me regularly too.

From the output from the job (which I think you should be able to read, too; let me know if you can't), it looks like the biggest issue is formatting. cargo fmt should be able to fix that. In addition, openapi.json needs to be regenerated. You can either do these things or apply the diff that appears in the job output.

@blp ran cargo fmt and regenerated openapi.json. Pushed.

Thanks for the guidance on the follow-up list — I'll open a GitHub issue with a checklist for the internal items, and add a known-limitations section to the connector docs for the user-visible ones (cross-referencing the issue). Will do that after this merges.

@blp ran cargo fmt and regenerated openapi.json. Pushed.

Almost there! I see that there is one remaining issue in the rerun: it wants to convert if a { if b { c } } into if a && b { c }.

Collapsed the nested if. Clippy clean now.

CI found PostgresCdcReaderConfig was referenced in openapi.json but not registered in the pipeline-manager ApiDoc schemas. Added it there and regenerated openapi.json. Pushed.

CI failure wasn't related to the changes in this PR, requeuing.