bug(connectors): PostgreSQL source builtin CDC mode is non-functional

[mmodzelewski]

Summary

The PostgreSQL source connector's builtin CDC mode (mode = "cdc",
cdc_backend = "builtin") does not work. It cannot produce a single
change message against a real database. Polling mode is unaffected.

File: core/connectors/sources/postgres_source/src/lib.rs

Details

1. Invalid publication SQL (setup fails). setup_cdc runs
   CREATE PUBLICATION IF NOT EXISTS .... PostgreSQL has no
   IF NOT EXISTS clause for CREATE PUBLICATION, so the statement is
   a syntax error and CDC setup fails outright.

2. Plugin/reader mismatch (hard error on every poll). setup_cdc
   creates the replication slot with the pgoutput plugin (binary
   output), but poll_cdc_builtin reads it via the textual
   pg_logical_slot_get_changes. PostgreSQL rejects this:
   output plugin "pgoutput" produces binary output, but function ... expects textual data. Every poll fails.

3. Parser targets a format no plugin emits.
   parse_logical_replication_message matches lines starting with
   INSERT:/UPDATE:/DELETE:. Real test_decoding output is
   table public.users: INSERT: id[integer]:1 ..., so the prefix
   never matches and all changes are dropped. The synthetic format in
   the unit tests corresponds to no real Postgres output.

4. Column data lost even on the synthetic format. For input like
   id[1], parse_record_data requires a : that isn't present, so
   the data object comes out empty. Tests only assert table_name
   and operation_type, so they pass while extracting nothing.

5. Data-loss risk if it did run. pg_logical_slot_get_changes
   consumes (advances) the slot at poll time, before the runtime
   confirms the Iggy send and persists state. A crash in between loses
   those changes - violates the at-least-once contract. The selected
   lsn/xid columns are never used; CDC tracks no LSN in state.

Smaller issues

• mode = "cdc" with enable_wal_cdc unset silently skips slot
  creation, then errors on every poll. Two flags for one feature.
• slot_name / publication_name interpolated raw into SQL string
  literals (rest of the file uses quote_identifier).
• CDC messages use a random UUID id, not the row PK - no dedup.
• timestamp / origin_timestamp use Utc::now(), not WAL commit time.
• old_data is never populated for UPDATE/DELETE.

Acceptance criteria

The fix must be well tested - current unit tests pass against a
fictional format and gave false confidence. Required:

• Integration test under
  core/integration/tests/connectors/postgres/ using
  testcontainers-modules, against a real PostgreSQL with
  wal_level = logical, asserting INSERT/UPDATE/DELETE are captured
  with correct column data.
• A restart test proving no change loss across a connector restart
  (state/LSN survives).
• Unit tests fed real decoding output, not a synthetic format.

[mattp5657]

Ok if I give this a try?

[mmodzelewski]

@mattp5657 sure, go ahead

[mattp5657]

@mmodzelewski

While digging into postgres_source CDC (it's currently fully broken due to invalid CREATE PUBLICATION syntax, plus the slot's created with pgoutput but read via a function that only accepts text output, so every poll errors), I hit a fork in the road on how to actually fix it, and wanted to check before opening a PR.

The config already stubs out cdc_backend = "pg_replicate" pointing at Supabase's ETL framework, but that depends on an unreleased, git-pinned fork of rust-postgres with no upstream path. This was added in 2025, so I don't want to be waiting on it to get CDC working.

Given that, three options:

1. Fix the existing builtin backend against test_decoding smallest change, but Postgres explicitly documents test_decoding as an example plugin with no output-format stability guarantee.
2. Implement real pgoutput over streaming replication ourselves the production-grade approach, but no mature Rust crate exists for it; this would be its own multi-PR effort.
3. Wait for a standardized Rust CDC solution to land

Leaning toward (1) now + (2) as tracked follow-up work, but wanted input before committing to a direction.

[mattp5657]

Also sorry this took a week - I was sick the last few days :)

[mmodzelewski]

@mattp5657 no worries, I see you've already opened the PR covering the 1st option. I think that's a good direction :) I'll try to review the PR today.

[mattp5657]

Yes, wanted to get something out there so we have a concrete baseline :)

Main goal for this PR was just getting CDC actually working. test_decoding gets us there without a binary decoder, but I think theres still a lot to do if this is the approach chosen. No server-side table filtering, and its format isn't a stable API the way pgoutput's is. I'd like to follow up with a real pgoutput backend once this lands.