Here's my proposed design:

Approach: New IntegratedInputEndpoint using supabase/etl. It handles snapshot (parallel COPY), WAL streaming (pgoutput), and the snapshot→follow transition.

Config: New TransportConfig::PostgresCdcInput variant (separate from existing PostgresInput):

{
  "transport": {
    "name": "postgres_cdc_input",
    "config": {
      "uri": "postgresql://user:pass@host:5432/mydb",
      "publication": "my_publication",
      "state_store_uri": "postgresql://user:pass@host:5432/mydb"
    }
  }
}

How it works:

Snapshot existing data via COPY
Catch up WAL between snapshot LSN and current LSN
Stream ongoing changes indefinitely (never calls eoi())

CDC events map to DeCollectionStream: Insert → insert(), Delete → delete(), Update → delete(before) + insert(after).

State persisted via etl's PostgresStore so restarts resume from LSN without re-snapshotting.

Requirements: Postgres 14+, wal_level=logical, tables must have PKs, pre-created publication.

Initial PR scope: No Feldera-level fault tolerance (fault_tolerance() -> None); resumption handled internally by etl's replication slots. FT integration as a follow-up.

Files touched:

crates/feldera-types/src/config.rs — new TransportConfig variant
crates/adapters/src/integrated/postgres/ — new cdc_input.rs
crates/adapters/src/integrated.rs — register connector
crates/adapters/Cargo.toml — etl dependency

Questions:

Preferred transport name — postgres_cdc_input, postgres_replication, or something else?
Should the connector auto-create the publication if it doesn't exist, or require it to be pre-created?
supabase/etl is async (Tokio). The existing postgres connector uses std::thread. OK to spawn a Tokio runtime inside the connector, or is there an existing runtime to hook into?
Is there CI infrastructure with wal_level=logical for integration tests?

Connector for postgres snapshot-and-follow #5208

Description

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions