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 95.91078% with 22 lines in your changes missing coverage. Please review.
✅ Project coverage is 42.58%. Comparing base (6ae792a) to head (af5bbc4).
⚠️ Report is 23 commits behind head on master.

Files with missing lines	Patch %	Lines
core/connectors/sources/postgres_source/src/lib.rs	95.91%	18 Missing and 4 partials ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             master    #3640       +/-   ##
=============================================
- Coverage     73.53%   42.58%   -30.95%     
  Complexity      937      937               
=============================================
  Files          1283     1281        -2     
  Lines        138809   121901    -16908     
  Branches     114699    97791    -16908     
=============================================
- Hits         102077    51917    -50160     
- Misses        33563    67257    +33694     
+ Partials       3169     2727      -442

Components	Coverage Δ
Rust Core	`34.85% <95.91%> (-38.94%)`	⬇️
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	`72.02% <95.91%> (+15.33%)`	⬆️

... and 404 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.

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.

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

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

fix(connectors): fix postgres source CDC producing no messages#3640

fix(connectors): fix postgres source CDC producing no messages#3640
mattp5657 wants to merge 3 commits into
apache:masterfrom
mattp5657:fix/connectors-postgres-builtin-cdc

mattp5657 commented Jul 9, 2026 •

edited

Loading

Uh oh!

codecov Bot commented Jul 9, 2026 •

edited

Loading

Uh oh!

hubcio commented Jul 10, 2026

Uh oh!

hubcio commented Jul 15, 2026

Uh oh!

mmodzelewski left a comment

Uh oh!

mmodzelewski Jul 15, 2026

Uh oh!

mmodzelewski Jul 15, 2026

Uh oh!

mmodzelewski Jul 15, 2026

Uh oh!

mmodzelewski Jul 15, 2026

Uh oh!

mmodzelewski Jul 15, 2026

Uh oh!

mmodzelewski Jul 15, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

		@@ -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!

mmodzelewski Jul 15, 2026

Choose a reason for hiding this comment

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