fix(connectors): fix postgres source CDC producing no messages by mattp5657 · Pull Request #3640 · apache/iggy

mattp5657 · 2026-07-09T23:32:30Z

Which issue does this PR address?

Relates to #3582

Rationale

Builtin CDC mode currently produces zero messages against a real
database - wrong plugin/read-path pairing, and a parser matching a
format no real Postgres output plugin emits. This PR is scoped to
getting CDC actually working and provably correct, nothing more.

I have reduce the scope of this specific PR for reviewability, in the future
would look to create issues and work on these issues:

pgoutput as the long-term plugin - real binary decoder + streaming
connection is a much bigger lift; test_decoding unblocks correctness
today. Target once cdc_backend = "pg_replicate" gets picked up.
At-least-once / crash safety (slot advances before send is confirmed,
no LSN tracked) - needs its own review since it touches the
state-persistence contract. Follow-up: #TODO.
Cosmetic items (UUID vs PK as message id, processing time vs WAL
commit time, old_data unpopulated) - mechanical, separate PR.

What changed?

Before: wrong slot plugin, a parser matching a format no real Postgres
output emits, and invalid setup SQL - CDC produced zero messages, and
several related bugs failed silently too. After: parser rewritten
against real captured output, setup/poll SQL fixed and bounded, and
those failure modes now work correctly or fail loudly. Covered by
corpus-driven unit tests plus integration tests against a live
wal_level=logical container.

Parser rewritten against real test_decoding output
Slot/setup SQL fixed, including the upgrade path and table filtering
Batch handling hardened (bad rows, unbounded backlog)
Dead config fields removed

Local Execution

Passed
Pre-commit hooks ran

AI Usage

Claude was used to help generate and review this PR.

…ng output

codecov · 2026-07-09T23:43:35Z

Codecov Report

❌ Patch coverage is 94.57364% with 35 lines in your changes missing coverage. Please review.
✅ Project coverage is 40.85%. Comparing base (6ae792a) to head (b5a8604).
⚠️ Report is 32 commits behind head on master.

Files with missing lines	Patch %	Lines
core/connectors/sources/postgres_source/src/lib.rs	94.57%	27 Missing and 8 partials ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             master    #3640       +/-   ##
=============================================
- Coverage     73.53%   40.85%   -32.69%     
  Complexity      937      937               
=============================================
  Files          1283     1294       +11     
  Lines        138809   128504    -10305     
  Branches     114699   104394    -10305     
=============================================
- Hits         102077    52498    -49579     
- Misses        33563    73221    +39658     
+ Partials       3169     2785      -384

Components	Coverage Δ
Rust Core	`33.18% <94.57%> (-40.60%)`	⬇️
Java SDK	`62.44% <ø> (ø)`
C# SDK	`72.10% <ø> (ø)`
Python SDK	`91.65% <ø> (ø)`
PHP SDK	`84.29% <ø> (ø)`
Node SDK	`91.35% <ø> (ø)`
Go SDK	`42.28% <ø> (ø)`

Files with missing lines	Coverage Δ
core/connectors/sources/postgres_source/src/lib.rs	`73.20% <94.57%> (+16.52%)`	⬆️

... and 443 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

hubcio · 2026-07-10T08:38:44Z

i think this is somewhat related to #3313

hubcio · 2026-07-15T09:00:38Z

@mmodzelewski could you please check this one?

mmodzelewski

Thanks, @mattp5657. The change looks solid, I've posted a few comments where the code could be improved.

mmodzelewski · 2026-07-15T10:16:32Z

+            break;
+        };
+        let name_end = pos + bracket_offset;
+        let column_name = &data[pos..name_end];


Quoted column names keep their literal quotes as JSON keys. test_decoding applies quote_identifier to column names, so a mixed-case or keyword column (createdAt, user) arrives as "createdAt"[timestamp with time zone]:'...' and this parser emits the JSON key \"createdAt\" with the double-quote characters included. Table names already go through unquote_pg_identifier; column names should get the same treatment, otherwise downstream consumers looking up the plain column name miss. Worth a unit test with a quoted column name fixture.

mmodzelewski · 2026-07-15T10:16:32Z

+    result
+}
+
+fn parse_column_value(data: &str, start: usize) -> (serde_json::Value, usize) {


The unchanged-toast-datum sentinel leaks into records as a real value. When an UPDATE does not touch a TOASTed column (large text/jsonb), test_decoding emits the bare token unchanged-toast-datum, and the parser stores data["col"] = "unchanged-toast-datum", which is indistinguishable from a legitimate string. Consumers will silently ingest the sentinel as the column value. Suggest handling it explicitly in parse_bare_scalar (map to null, or omit the key) and adding a fixture for it.

mmodzelewski · 2026-07-15T10:16:32Z

+    let http = Client::new();
+    wait_for_source_status(&http, &api_url, ConnectorStatus::Running).await;
+
+    sqlx::query("SELECT pg_drop_replication_slot($1)")


Flake risk: this first pg_drop_replication_slot races the still-running connector, which calls pg_logical_slot_get_changes every 50ms and briefly holds the slot on each call. If the drop lands inside that window, Postgres raises ERROR 55006: replication slot "iggy_slot" is active for PID ... and the expect panics. Rough odds are a few percent per run, which is enough to flake CI. Wrapping the drop in a short retry loop fixes it. The second drop below is fine since the failed restart leaves no poller running.

mmodzelewski · 2026-07-15T10:16:32Z

-enable_wal_cdc = false
-publication_name = "iggy_publication"
+# CDC options (only used when mode = "cdc")
 replication_slot = "iggy_slot"


Two operational hazards of the new slot handling deserve README coverage:

Slot sharing: setup_cdc accepts any pre-existing test_decoding slot, so two CDC connectors against the same database with the default replication_slot = "iggy_slot" silently share one slot. pg_logical_slot_get_changes consumes on read, so each connector steals a subset of the other's changes. The README should state that every CDC connector needs a unique slot name.

WAL retention: a slot orphaned after a connector is decommissioned retains WAL indefinitely (default max_slot_wal_keep_size = -1) until the disk fills. The README should document dropping the slot (SELECT pg_drop_replication_slot('iggy_slot')) as part of decommissioning.

mmodzelewski · 2026-07-15T10:16:32Z

@@ -303,10 +301,6 @@ impl PostgresSource {
    }

    async fn setup_cdc(&self) -> Result<(), Error> {


Not on this line, but related to CDC startup: open() validates mode but not cdc_backend. A typo like cdc_backend = "built-in" passes startup, then every poll_cdc call returns an error that the SDK loop logs and swallows, leaving a Running connector that produces nothing. That is the same silent-death failure mode this PR fixes for the slot mismatch. Validating the backend in the "cdc" arm of open(), where the slot mismatch already fails loudly, would close it.

I resolved this but also handled two other edge cases.

capture_operations - an entry outside INSERT/UPDATE/DELETE (e.g. a typo like INSRT) previously passed startup silently and would just never match, quietly dropping that operation type forever.

payload_format - an unsupported value (e.g. btea instead of bytea) would similarly pass startup and only surface as a failure deep in row processing on the first poll, not at startup.

Both now validate in open() alongside cdc_backend, before setup_cdc() runs, so any of the three fails loudly at startup instead of leaving a Running connector that silently drops or mishandles data.

Then I rewrote the integration test to make sure it covered this as well.

mmodzelewski · 2026-07-15T10:33:38Z

-            });
-        }
-        None
+        let rest = rest


The old-key columns are parsed past and discarded here, so DatabaseRecord.old_data is always None even though the struct has the field and the data is sitting in the row. For a PK-changing UPDATE (and for DELETE under REPLICA IDENTITY FULL) consumers only see the new tuple and cannot learn which row it replaced, which breaks e.g. keyed upsert/delete downstream. Issue #3582 lists unpopulated old_data as well. Suggest running parse_record_columns over the old-key section and populating old_data instead of dropping it.

mattp5657 · 2026-07-19T23:04:53Z

These should be fixed, let me know if you have any other feedback :)

mattp5657 · 2026-07-19T23:04:57Z

/ready

Matthew Patton added 3 commits July 9, 2026 15:44

fix(connectors): rewrite postgres CDC parser against real test_decodi…

c9b0b18

…ng output

test(connectors): add postgres CDC integration test fixtures

ac8ae1a

test(connectors): add end-to-end postgres CDC integration tests

af5bbc4

mattp5657 changed the title ~~Fix/connectors postgres builtin cdc~~ fix(connectors): fix postgres source CDC producing no messages Jul 9, 2026

mattp5657 marked this pull request as ready for review July 9, 2026 23:35

github-actions Bot added the S-waiting-on-review PR is waiting on a reviewer label Jul 9, 2026

mmodzelewski requested changes Jul 15, 2026

View reviewed changes

github-actions Bot added S-waiting-on-author PR is waiting on author response and removed S-waiting-on-review PR is waiting on a reviewer labels Jul 15, 2026

Matthew Patton added 6 commits July 18, 2026 20:25

fix(connectors): unquote CDC column names from test_decoding output

2969da4

fix(connectors): map unchanged-toast-datum sentinel to null

b743b05

test(connectors): retry postgres slot drop past transient in-use errors

d327d5a

docs(connectors): document CDC slot sharing and decommissioning hazards

714faea

fix(connectors): validate postgres CDC plugin_config at startup

76429db

fix(connectors): populate old_data from CDC old-key section

b5a8604

github-actions Bot added S-waiting-on-review PR is waiting on a reviewer and removed S-waiting-on-author PR is waiting on author response labels Jul 19, 2026

		@@ -303,10 +301,6 @@ impl PostgresSource {
		}

		async fn setup_cdc(&self) -> Result<(), Error> {

Uh oh!

Conversation

mattp5657 commented Jul 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR address?

Rationale

What changed?

Local Execution

AI Usage

Uh oh!

codecov Bot commented Jul 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

hubcio commented Jul 10, 2026

Uh oh!

hubcio commented Jul 15, 2026

Uh oh!

mmodzelewski left a comment

Choose a reason for hiding this comment

Uh oh!

mmodzelewski Jul 15, 2026

Choose a reason for hiding this comment

Uh oh!

mmodzelewski Jul 15, 2026

Choose a reason for hiding this comment

Uh oh!

mmodzelewski Jul 15, 2026

Choose a reason for hiding this comment

Uh oh!

mmodzelewski Jul 15, 2026

Choose a reason for hiding this comment

Uh oh!

mmodzelewski Jul 15, 2026

Choose a reason for hiding this comment

Uh oh!

mattp5657 Jul 19, 2026

Choose a reason for hiding this comment

Uh oh!

mmodzelewski Jul 15, 2026

Choose a reason for hiding this comment

Uh oh!

mattp5657 commented Jul 19, 2026

Uh oh!

mattp5657 commented Jul 19, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

mattp5657 commented Jul 9, 2026 •

edited

Loading

codecov Bot commented Jul 9, 2026 •

edited

Loading