This is an automated email from the ASF dual-hosted git repository.
mmodzelewski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 6513b2f55 fix(connectors): fix postgres source CDC producing no
messages (#3640)
6513b2f55 is described below
commit 6513b2f5542b87461449c6d67e198eeffeb0536b
Author: Matthew Patton <[email protected]>
AuthorDate: Thu Jul 23 08:52:57 2026 -0400
fix(connectors): fix postgres source CDC producing no messages (#3640)
---
.gitignore | 2 +
core/connectors/sources/postgres_source/README.md | 31 +-
.../sources/postgres_source/src/cdc_fixtures.rs | 89 ++
core/connectors/sources/postgres_source/src/lib.rs | 1100 ++++++++++++++++----
core/integration/tests/connectors/fixtures/mod.rs | 5 +-
.../tests/connectors/fixtures/postgres/cdc.rs | 206 ++++
.../connectors/fixtures/postgres/container.rs | 21 +-
.../tests/connectors/fixtures/postgres/mod.rs | 2 +
.../postgres/cdc_restart_connectors/config.toml | 42 +
core/integration/tests/connectors/postgres/mod.rs | 1 +
.../connectors/postgres/postgres_source_cdc.rs | 584 +++++++++++
.../connectors/postgres/source_cdc_restart.toml | 26 +
12 files changed, 1886 insertions(+), 223 deletions(-)
diff --git a/.gitignore b/.gitignore
index f0d8b6250..45260878f 100644
--- a/.gitignore
+++ b/.gitignore
@@ -40,5 +40,7 @@ foreign/node/.npm
*.out
go.work
core/bench/dashboard/frontend/dist
+core/integration/tests/connectors/postgres/cdc_restart_connectors/*
+!core/integration/tests/connectors/postgres/cdc_restart_connectors/config.toml
LICENSE-binary
**/LICENSE-binary
diff --git a/core/connectors/sources/postgres_source/README.md
b/core/connectors/sources/postgres_source/README.md
index ba03b51bf..d0619bdea 100644
--- a/core/connectors/sources/postgres_source/README.md
+++ b/core/connectors/sources/postgres_source/README.md
@@ -41,9 +41,7 @@ primary_key_column = "id"
# Custom query (optional)
custom_query = "SELECT * FROM $table WHERE id > $offset ORDER BY id LIMIT
$limit"
-# CDC options (optional)
-enable_wal_cdc = false
-publication_name = "iggy_publication"
+# CDC options (only used when mode = "cdc")
replication_slot = "iggy_slot"
capture_operations = ["INSERT", "UPDATE", "DELETE"]
cdc_backend = "builtin"
@@ -69,9 +67,7 @@ cdc_backend = "builtin"
| `processed_column` | string | none | Boolean column to mark as processed |
| `primary_key_column` | string | tracking_column | PK for delete/mark
operations |
| `custom_query` | string | none | Custom SQL with parameter substitution |
-| `enable_wal_cdc` | bool | `false` | Enable WAL-based CDC |
-| `publication_name` | string | `iggy_publication` | Logical replication
publication |
-| `replication_slot` | string | `iggy_slot` | Replication slot name |
+| `replication_slot` | string | `iggy_slot` | Replication slot name (only used
when `mode = "cdc"`) |
| `capture_operations` | array | `["INSERT","UPDATE","DELETE"]` | CDC
operations to capture |
| `cdc_backend` | string | `builtin` | `builtin` or `pg_replicate` |
| `verbose_logging` | bool | `false` | Log at info level instead of debug |
@@ -267,13 +263,33 @@ CDC requires PostgreSQL logical replication setup:
```toml
[plugin_config]
mode = "cdc"
-enable_wal_cdc = true
tables = ["users", "orders"]
capture_operations = ["INSERT", "UPDATE", "DELETE"]
```
The `pg_replicate` backend requires the `cdc_pg_replicate` feature flag at
build time.
+### Slot Naming
+
+Each CDC connector must use a unique `replication_slot`. Setup accepts any
+pre-existing `test_decoding` slot, so two connectors pointed at the same
+database with the default `replication_slot = "iggy_slot"` will silently
+share one slot. `pg_logical_slot_get_changes` consumes changes on read, so
+each connector only sees a subset of the other's changes instead of erroring.
+Set an explicit, distinct `replication_slot` per connector instance.
+
+### Decommissioning
+
+A replication slot retains WAL for as long as it exists, regardless of
+whether a connector is still consuming it (`max_slot_wal_keep_size` defaults
+to `-1`, i.e. unbounded). Dropping a connector without dropping its slot
+leaves an orphaned slot that accumulates WAL indefinitely and can fill the
+disk. When decommissioning a CDC connector, drop its slot:
+
+```sql
+SELECT pg_drop_replication_slot('iggy_slot');
+```
+
## Example Configs
### Basic Polling (JSON Mode)
@@ -344,7 +360,6 @@ batch_length = 100
[plugin_config]
connection_string = "postgresql://user:pass@localhost:5432/mydb"
mode = "cdc"
-enable_wal_cdc = true
tables = ["users", "orders"]
capture_operations = ["INSERT", "UPDATE"]
```
diff --git a/core/connectors/sources/postgres_source/src/cdc_fixtures.rs
b/core/connectors/sources/postgres_source/src/cdc_fixtures.rs
new file mode 100644
index 000000000..0c7c3b611
--- /dev/null
+++ b/core/connectors/sources/postgres_source/src/cdc_fixtures.rs
@@ -0,0 +1,89 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Real `test_decoding` output rows, captured from
+// pg_logical_slot_get_changes() on postgres:16 with the test_decoding
+// plugin (BEGIN/COMMIT rows omitted, parser never sees them).
+
+pub(crate) const INSERT_SINGLE_ROW_ALL_TYPES: &str = r#"table
public.probe_events: INSERT: id[integer]:2 name[text]:'alice' note[text]:'first
note' amount[numeric]:12.50 active[boolean]:true tags[text[]]:'{a,b}'
payload[jsonb]:'{"k": 1}' created_at[timestamp with time zone]:'2026-07-05
17:58:23.202192+00' small_int[smallint]:null big_int[bigint]:null
real_val[real]:null double_val[double precision]:null numeric_val[numeric]:null
uuid_val[uuid]:null bytea_val[bytea]:null date_val[date]:nu [...]
+
+pub(crate) const INSERT_WITH_NULLS: &str = r#"table public.probe_events:
INSERT: id[integer]:3 name[text]:'bob' note[text]:null amount[numeric]:null
active[boolean]:null tags[text[]]:null payload[jsonb]:null created_at[timestamp
with time zone]:'2026-07-05 17:58:23.924391+00' small_int[smallint]:null
big_int[bigint]:null real_val[real]:null double_val[double precision]:null
numeric_val[numeric]:null uuid_val[uuid]:null bytea_val[bytea]:null
date_val[date]:null time_val[time without time [...]
+
+pub(crate) const INSERT_MULTI_ROW_SINGLE_STATEMENT: [&str; 2] = [
+ r#"table public.probe_events: INSERT: id[integer]:4 name[text]:'carol'
note[text]:null amount[numeric]:null active[boolean]:true tags[text[]]:null
payload[jsonb]:null created_at[timestamp with time zone]:'2026-07-05
17:58:24.628431+00' small_int[smallint]:null big_int[bigint]:null
real_val[real]:null double_val[double precision]:null numeric_val[numeric]:null
uuid_val[uuid]:null bytea_val[bytea]:null date_val[date]:null time_val[time
without time zone]:null interval_val[interval]:nul [...]
+ r#"table public.probe_events: INSERT: id[integer]:5 name[text]:'dave'
note[text]:null amount[numeric]:null active[boolean]:true tags[text[]]:null
payload[jsonb]:null created_at[timestamp with time zone]:'2026-07-05
17:58:24.628431+00' small_int[smallint]:null big_int[bigint]:null
real_val[real]:null double_val[double precision]:null numeric_val[numeric]:null
uuid_val[uuid]:null bytea_val[bytea]:null date_val[date]:null time_val[time
without time zone]:null interval_val[interval]:null [...]
+];
+
+pub(crate) const INSERT_MULTI_STATEMENT_ONE_TRANSACTION: [&str; 2] = [
+ r#"table public.probe_events: INSERT: id[integer]:6 name[text]:'eve'
note[text]:null amount[numeric]:null active[boolean]:true tags[text[]]:null
payload[jsonb]:null created_at[timestamp with time zone]:'2026-07-05
17:58:25.344503+00' small_int[smallint]:null big_int[bigint]:null
real_val[real]:null double_val[double precision]:null numeric_val[numeric]:null
uuid_val[uuid]:null bytea_val[bytea]:null date_val[date]:null time_val[time
without time zone]:null interval_val[interval]:null [...]
+ r#"table public.probe_events: INSERT: id[integer]:7 name[text]:'frank'
note[text]:null amount[numeric]:null active[boolean]:true tags[text[]]:null
payload[jsonb]:null created_at[timestamp with time zone]:'2026-07-05
17:58:25.344503+00' small_int[smallint]:null big_int[bigint]:null
real_val[real]:null double_val[double precision]:null numeric_val[numeric]:null
uuid_val[uuid]:null bytea_val[bytea]:null date_val[date]:null time_val[time
without time zone]:null interval_val[interval]:nul [...]
+];
+
+pub(crate) const UPDATE_FULL_ROW: &str = r#"table public.probe_events: UPDATE:
id[integer]:2 name[text]:'alice2' note[text]:'updated' amount[numeric]:99.99
active[boolean]:false tags[text[]]:'{a,b}' payload[jsonb]:'{"k": 1}'
created_at[timestamp with time zone]:'2026-07-05 17:58:23.202192+00'
small_int[smallint]:null big_int[bigint]:null real_val[real]:null
double_val[double precision]:null numeric_val[numeric]:null uuid_val[uuid]:null
bytea_val[bytea]:null date_val[date]:null time_val[t [...]
+
+pub(crate) const UPDATE_SINGLE_COLUMN: &str = r#"table public.probe_events:
UPDATE: id[integer]:3 name[text]:'bob' note[text]:'only note changed'
amount[numeric]:null active[boolean]:null tags[text[]]:null payload[jsonb]:null
created_at[timestamp with time zone]:'2026-07-05 17:58:23.924391+00'
small_int[smallint]:null big_int[bigint]:null real_val[real]:null
double_val[double precision]:null numeric_val[numeric]:null uuid_val[uuid]:null
bytea_val[bytea]:null date_val[date]:null time_val[ [...]
+
+pub(crate) const UPDATE_TO_NULL: &str = r#"table public.probe_events: UPDATE:
id[integer]:2 name[text]:'alice2' note[text]:null amount[numeric]:99.99
active[boolean]:false tags[text[]]:'{a,b}' payload[jsonb]:'{"k": 1}'
created_at[timestamp with time zone]:'2026-07-05 17:58:23.202192+00'
small_int[smallint]:null big_int[bigint]:null real_val[real]:null
double_val[double precision]:null numeric_val[numeric]:null uuid_val[uuid]:null
bytea_val[bytea]:null date_val[date]:null time_val[time wi [...]
+
+// PK change: test_decoding emits the pre-update key separately from the
+// post-update tuple, `old-key: ... new-tuple: ...`, instead of the plain
+// `UPDATE: <cols>` shape every other row here uses.
+pub(crate) const UPDATE_PRIMARY_KEY: &str = r#"table public.probe_events:
UPDATE: old-key: id[integer]:4 new-tuple: id[integer]:1004 name[text]:'carol'
note[text]:null amount[numeric]:null active[boolean]:true tags[text[]]:null
payload[jsonb]:null created_at[timestamp with time zone]:'2026-07-05
17:58:24.628431+00' small_int[smallint]:null big_int[bigint]:null
real_val[real]:null double_val[double precision]:null numeric_val[numeric]:null
uuid_val[uuid]:null bytea_val[bytea]:null date_va [...]
+
+// DELETE only carries replica-identity columns (here, just the PK), not
+// the full row.
+pub(crate) const DELETE_ROW: &str = "table public.probe_events: DELETE:
id[integer]:5";
+
+pub(crate) const DELETE_MULTIPLE_ROWS: [&str; 2] = [
+ "table public.probe_events: DELETE: id[integer]:6",
+ "table public.probe_events: DELETE: id[integer]:7",
+];
+
+pub(crate) const TRUNCATE_TABLE: &str = "table public.probe_events: TRUNCATE:
(no-flags)";
+
+// Value contains a literal embedded newline and an escaped quote (`''`) -
+// the row spans two lines in the corpus capture.
+pub(crate) const UNICODE_AND_SPECIAL_CHARS: &str = "table public.probe_events:
INSERT: id[integer]:9 name[text]:'unicode' note[text]:'emoji \u{1F680} quote''
backslash\\ newline\nend' amount[numeric]:null active[boolean]:true
tags[text[]]:null payload[jsonb]:null created_at[timestamp with time
zone]:'2026-07-05 17:58:31.874185+00' small_int[smallint]:null
big_int[bigint]:null real_val[real]:null double_val[double precision]:null
numeric_val[numeric]:null uuid_val[uuid]:null bytea_val[byt [...]
+
+pub(crate) const INSERT_EXTENDED_TYPES_NORMAL_VALUES: &str = r#"table
public.probe_events: INSERT: id[integer]:10 name[text]:'extended'
note[text]:null amount[numeric]:null active[boolean]:true tags[text[]]:null
payload[jsonb]:null created_at[timestamp with time zone]:'2026-07-05
17:58:32.595761+00' small_int[smallint]:42 big_int[bigint]:9000000000
real_val[real]:3.14 double_val[double precision]:2.718281828
numeric_val[numeric]:123.456789
uuid_val[uuid]:'11111111-1111-1111-1111-11111111 [...]
+
+pub(crate) const INSERT_NUMERIC_NAN: &str = r#"table public.probe_events:
INSERT: id[integer]:11 name[text]:'nan_row' note[text]:null
amount[numeric]:null active[boolean]:true tags[text[]]:null payload[jsonb]:null
created_at[timestamp with time zone]:'2026-07-05 17:58:33.30327+00'
small_int[smallint]:null big_int[bigint]:null real_val[real]:NaN
double_val[double precision]:NaN numeric_val[numeric]:NaN uuid_val[uuid]:null
bytea_val[bytea]:null date_val[date]:null time_val[time without tim [...]
+
+pub(crate) const INSERT_NUMERIC_INFINITY: &str = r#"table public.probe_events:
INSERT: id[integer]:12 name[text]:'inf_row' note[text]:null
amount[numeric]:null active[boolean]:true tags[text[]]:null payload[jsonb]:null
created_at[timestamp with time zone]:'2026-07-05 17:58:34.023511+00'
small_int[smallint]:null big_int[bigint]:null real_val[real]:Infinity
double_val[double precision]:-Infinity numeric_val[numeric]:Infinity
uuid_val[uuid]:null bytea_val[bytea]:null date_val[date]:null tim [...]
+
+pub(crate) const INSERT_NEGATIVE_AND_BOUNDARY_NUMBERS: &str = r#"table
public.probe_events: INSERT: id[integer]:13 name[text]:'neg_row'
note[text]:null amount[numeric]:null active[boolean]:true tags[text[]]:null
payload[jsonb]:null created_at[timestamp with time zone]:'2026-07-05
17:58:34.743797+00' small_int[smallint]:-32768
big_int[bigint]:-9223372036854775808 real_val[real]:-3.14 double_val[double
precision]:-2.71828 numeric_val[numeric]:-999999.9999 uuid_val[uuid]:null
bytea_val[byte [...]
+
+pub(crate) const INSERT_MAX_BOUNDARY_NUMBERS: &str = r#"table
public.probe_events: INSERT: id[integer]:14 name[text]:'max_row'
note[text]:null amount[numeric]:null active[boolean]:true tags[text[]]:null
payload[jsonb]:null created_at[timestamp with time zone]:'2026-07-05
17:58:35.473792+00' small_int[smallint]:32767
big_int[bigint]:9223372036854775807 real_val[real]:null double_val[double
precision]:null numeric_val[numeric]:null uuid_val[uuid]:null
bytea_val[bytea]:null date_val[date]:n [...]
+
+pub(crate) const INSERT_NEGATIVE_ZERO_FLOAT: &str = r#"table
public.probe_events: INSERT: id[integer]:15 name[text]:'negzero_row'
note[text]:null amount[numeric]:null active[boolean]:true tags[text[]]:null
payload[jsonb]:null created_at[timestamp with time zone]:'2026-07-05
17:58:36.198193+00' small_int[smallint]:null big_int[bigint]:null
real_val[real]:-0 double_val[double precision]:-0 numeric_val[numeric]:null
uuid_val[uuid]:null bytea_val[bytea]:null date_val[date]:null time_val[time
[...]
+
+pub(crate) const INSERT_EMPTY_STRING_VS_NULL: [&str; 2] = [
+ r#"table public.probe_events: INSERT: id[integer]:16
name[text]:'empty_string_row' note[text]:'' amount[numeric]:null
active[boolean]:true tags[text[]]:null payload[jsonb]:null created_at[timestamp
with time zone]:'2026-07-05 17:58:36.931897+00' small_int[smallint]:null
big_int[bigint]:null real_val[real]:null double_val[double precision]:null
numeric_val[numeric]:null uuid_val[uuid]:null bytea_val[bytea]:null
date_val[date]:null time_val[time without time zone]:null interval_val[int [...]
+ r#"table public.probe_events: INSERT: id[integer]:17 name[text]:'null_row'
note[text]:null amount[numeric]:null active[boolean]:true tags[text[]]:null
payload[jsonb]:null created_at[timestamp with time zone]:'2026-07-05
17:58:36.931897+00' small_int[smallint]:null big_int[bigint]:null
real_val[real]:null double_val[double precision]:null numeric_val[numeric]:null
uuid_val[uuid]:null bytea_val[bytea]:null date_val[date]:null time_val[time
without time zone]:null interval_val[interval] [...]
+];
+
+pub(crate) const INSERT_ARRAY_WITH_NULL_ELEMENT: &str = r#"table
public.probe_events: INSERT: id[integer]:18 name[text]:'array_null_elem_row'
note[text]:null amount[numeric]:null active[boolean]:true tags[text[]]:null
payload[jsonb]:null created_at[timestamp with time zone]:'2026-07-05
17:58:37.658172+00' small_int[smallint]:null big_int[bigint]:null
real_val[real]:null double_val[double precision]:null numeric_val[numeric]:null
uuid_val[uuid]:null bytea_val[bytea]:null date_val[date]:nu [...]
+
+pub(crate) const INSERT_EMPTY_ARRAY: &str = r#"table public.probe_events:
INSERT: id[integer]:19 name[text]:'empty_array_row' note[text]:null
amount[numeric]:null active[boolean]:true tags[text[]]:null payload[jsonb]:null
created_at[timestamp with time zone]:'2026-07-05 17:58:38.384294+00'
small_int[smallint]:null big_int[bigint]:null real_val[real]:null
double_val[double precision]:null numeric_val[numeric]:null uuid_val[uuid]:null
bytea_val[bytea]:null date_val[date]:null time_val[time [...]
+
+pub(crate) const UPDATE_ARRAY_COLUMN: &str = r#"table public.probe_events:
UPDATE: id[integer]:10 name[text]:'extended' note[text]:null
amount[numeric]:null active[boolean]:true tags[text[]]:null payload[jsonb]:null
created_at[timestamp with time zone]:'2026-07-05 17:58:32.595761+00'
small_int[smallint]:42 big_int[bigint]:9000000000 real_val[real]:3.14
double_val[double precision]:2.718281828 numeric_val[numeric]:123.456789
uuid_val[uuid]:'11111111-1111-1111-1111-111111111111' bytea_val[ [...]
+
+pub(crate) const INSERT_CHAR_PADDING_VS_VARCHAR: &str = r#"table
public.probe_events: INSERT: id[integer]:20 name[text]:'padding_row'
note[text]:null amount[numeric]:null active[boolean]:true tags[text[]]:null
payload[jsonb]:null created_at[timestamp with time zone]:'2026-07-05
17:58:39.824702+00' small_int[smallint]:null big_int[bigint]:null
real_val[real]:null double_val[double precision]:null numeric_val[numeric]:null
uuid_val[uuid]:null bytea_val[bytea]:null date_val[date]:null time_ [...]
+
+pub(crate) const INSERT_QUOTED_MIXED_CASE_COLUMN: &str = r#"table
public.probe_events: INSERT: id[integer]:21 "createdAt"[timestamp with time
zone]:'2026-07-05 17:58:40.000000+00' "user"[text]:'quoted_row'"#;
+
+pub(crate) const UPDATE_UNCHANGED_TOAST_COLUMN: &str = r#"table
public.probe_events: UPDATE: id[integer]:22 name[text]:'toast_row'
note[text]:unchanged-toast-datum amount[numeric]:null active[boolean]:true
tags[text[]]:null payload[jsonb]:unchanged-toast-datum created_at[timestamp
with time zone]:'2026-07-05 17:58:41.000000+00' small_int[smallint]:null
big_int[bigint]:null real_val[real]:null double_val[double precision]:null
numeric_val[numeric]:null uuid_val[uuid]:null bytea_val[bytea] [...]
diff --git a/core/connectors/sources/postgres_source/src/lib.rs
b/core/connectors/sources/postgres_source/src/lib.rs
index 7593d1786..8deb35516 100644
--- a/core/connectors/sources/postgres_source/src/lib.rs
+++ b/core/connectors/sources/postgres_source/src/lib.rs
@@ -61,12 +61,10 @@ pub struct PostgresSourceConfig {
pub tracking_column: Option<String>,
pub initial_offset: Option<String>,
pub max_connections: Option<u32>,
- pub enable_wal_cdc: Option<bool>,
pub custom_query: Option<String>,
pub snake_case_columns: Option<bool>,
pub include_metadata: Option<bool>,
pub replication_slot: Option<String>,
- pub publication_name: Option<String>,
pub capture_operations: Option<Vec<String>>,
pub cdc_backend: Option<String>,
pub delete_after_read: Option<bool>,
@@ -185,10 +183,13 @@ impl Source for PostgresSource {
self.connect().await?;
+ validate_payload_format(self.config.payload_format.as_deref())?;
+
match self.config.mode.as_str() {
"cdc" => {
+ let backend =
validate_cdc_backend(self.config.cdc_backend.as_deref())?;
+
validate_capture_operations(self.config.capture_operations.as_deref())?;
self.setup_cdc().await?;
- let backend =
self.config.cdc_backend.as_deref().unwrap_or("builtin");
info!(
"PostgreSQL CDC mode enabled (backend: {backend}) for
connector ID: {}",
self.id
@@ -303,10 +304,6 @@ impl PostgresSource {
}
async fn setup_cdc(&self) -> Result<(), Error> {
- if !self.config.enable_wal_cdc.unwrap_or(false) {
- return Ok(());
- }
-
let pool = self.get_pool()?;
let wal_level: String = sqlx::query_scalar("SHOW wal_level")
@@ -320,31 +317,35 @@ impl PostgresSource {
));
}
- let publication_name = self
- .config
- .publication_name
- .as_deref()
- .unwrap_or("iggy_publication");
- let quoted_publication = quote_identifier(publication_name)?;
- let tables_clause = if self.config.tables.is_empty() {
- "FOR ALL TABLES".to_string()
- } else {
- let quoted_tables = self
- .config
- .tables
- .iter()
- .map(|t| quote_qualified_identifier(t))
- .collect::<Result<Vec<_>, _>>()?;
- format!("FOR TABLE {}", quoted_tables.join(", "))
- };
-
- let create_publication_sql =
- format!("CREATE PUBLICATION IF NOT EXISTS {quoted_publication}
{tables_clause}");
+ for table in &self.config.tables {
+ let exists: bool = if let Some((schema, name)) =
table.split_once('.') {
+ sqlx::query_scalar(
+ "SELECT EXISTS (SELECT 1 FROM information_schema.tables \
+ WHERE table_schema = $1 AND table_name = $2)",
+ )
+ .bind(schema)
+ .bind(name)
+ .fetch_one(pool)
+ .await
+ } else {
+ sqlx::query_scalar(
+ "SELECT EXISTS (SELECT 1 FROM information_schema.tables
WHERE table_name = $1)",
+ )
+ .bind(table)
+ .fetch_one(pool)
+ .await
+ }
+ .map_err(|e| Error::InitError(format!("Failed to validate table
'{table}': {e}")))?;
- sqlx::query(sqlx::AssertSqlSafe(create_publication_sql))
- .execute(pool)
- .await
- .map_err(|e| Error::InitError(format!("Failed to create
publication: {e}")))?;
+ // Not fatal - the table may just not exist yet (e.g. a migration
+ // runs shortly after startup).
+ if !exists {
+ warn!(
+ "Configured table '{table}' does not exist yet. If this is
unexpected, \
+ check config.tables for typos."
+ );
+ }
+ }
let slot_name = self
.config
@@ -352,40 +353,49 @@ impl PostgresSource {
.as_deref()
.unwrap_or("iggy_slot");
- sqlx::query(
- "SELECT pg_create_logical_replication_slot($1, 'pgoutput') \
- WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots WHERE
slot_name = $1)",
- )
- .bind(slot_name)
- .fetch_optional(pool)
- .await
- .map_err(|e| Error::InitError(format!("Failed to create replication
slot: {e}")))?;
+ let existing_plugin: Option<String> =
+ sqlx::query_scalar("SELECT plugin FROM pg_replication_slots WHERE
slot_name = $1")
+ .bind(slot_name)
+ .fetch_optional(pool)
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to check
replication slot: {e}")))?;
- info!("PostgreSQL CDC setup completed. Publication:
{publication_name}, Slot: {slot_name}");
+ match existing_plugin.as_deref() {
+ Some("test_decoding") => {}
+ Some(other) => {
+ return Err(Error::InitError(format!(
+ "Replication slot '{slot_name}' already exists with plugin
'{other}', \
+ expected 'test_decoding' (likely created by an older
version). Drop it \
+ with SELECT pg_drop_replication_slot('{slot_name}') or
set \
+ replication_slot to a new name."
+ )));
+ }
+ None => {
+ // test_decoding ignores publications entirely, so none is
created
+ // here. A future pgoutput-based backend would need one, since
+ // that's the only way to get server-side table filtering under
+ // that plugin.
+ sqlx::query("SELECT pg_create_logical_replication_slot($1,
'test_decoding')")
+ .bind(slot_name)
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ Error::InitError(format!("Failed to create replication
slot: {e}"))
+ })?;
+ }
+ }
+
+ info!("PostgreSQL CDC setup completed. Slot: {slot_name}");
Ok(())
}
async fn poll_cdc(&self) -> Result<Vec<ProducedMessage>, Error> {
- let backend = self.config.cdc_backend.as_deref().unwrap_or("builtin");
- match backend {
+ match self.config.cdc_backend.as_deref().unwrap_or("builtin") {
"builtin" => self.poll_cdc_builtin().await,
- "pg_replicate" => {
- #[cfg(feature = "cdc_pg_replicate")]
- {
- Err(Error::InitError(
- "pg_replicate backend not yet implemented".to_string(),
- ))
- }
- #[cfg(not(feature = "cdc_pg_replicate"))]
- {
- Err(Error::InitError(
- "cdc_backend 'pg_replicate' requested but feature
'cdc_pg_replicate' is not enabled at build time".to_string(),
- ))
- }
- }
- other => Err(Error::InitError(format!(
- "Unsupported cdc_backend '{other}'. Use 'builtin' or
'pg_replicate'"
- ))),
+ "pg_replicate" => Err(Error::InitError(
+ "pg_replicate backend not yet implemented".to_string(),
+ )),
+ other => unreachable!("validate_cdc_backend already rejected
'{other}'"),
}
}
@@ -397,40 +407,53 @@ impl PostgresSource {
.replication_slot
.as_deref()
.unwrap_or("iggy_slot");
- let publication_name = self
- .config
- .publication_name
- .as_deref()
- .unwrap_or("iggy_publication");
let capture_ops = self
.config
.capture_operations
.as_ref()
.map(|ops| ops.iter().map(|s| s.as_str()).collect::<Vec<_>>())
.unwrap_or_else(|| vec!["INSERT", "UPDATE", "DELETE"]);
-
- let logical_repl_sql = format!(
- "SELECT lsn, xid, data FROM
pg_logical_slot_get_changes('{slot_name}', NULL, NULL, 'proto_version', '1',
'publication_names', '{publication_name}')"
- );
-
- // Database I/O without holding the lock
- let rows = sqlx::query(sqlx::AssertSqlSafe(logical_repl_sql))
- .fetch_all(pool)
- .await
- .map_err(|e| {
- error!("Failed to fetch CDC changes: {e}");
- Error::InvalidRecord
- })?;
+ let captured_tables =
+
(!self.config.tables.is_empty()).then_some(self.config.tables.as_slice());
+ let batch_size = self.config.batch_size.unwrap_or(1000) as i32;
+
+ // Database I/O without holding the lock. upto_nchanges is only
+ // checked at transaction-commit boundaries (a single huge transaction
+ // can still exceed it), so this isn't a hard per-call cap - but it
+ // stops the backlog from growing unbounded across many transactions
+ // the way NULL (no limit at all) did.
+ let rows =
+ sqlx::query("SELECT lsn, xid, data FROM
pg_logical_slot_get_changes($1, NULL, $2)")
+ .bind(slot_name)
+ .bind(batch_size)
+ .fetch_all(pool)
+ .await
+ .map_err(|e| {
+ error!("Failed to fetch CDC changes: {e}");
+ Error::InvalidRecord
+ })?;
let mut messages = Vec::new();
for row in rows {
- let data: String = row.try_get("data").map_err(|_|
Error::InvalidRecord)?;
+ let data: String = match row.try_get("data") {
+ Ok(data) => data,
+ Err(e) => {
+ error!("Skipping CDC row with unreadable data column:
{e}");
+ continue;
+ }
+ };
- if let Some(change_record) =
self.parse_logical_replication_message(&data, &capture_ops)
+ if let Some(change_record) =
+ self.parse_logical_replication_message(&data, &capture_ops,
captured_tables)
{
- let payload =
- simd_json::to_vec(&change_record).map_err(|_|
Error::InvalidRecord)?;
+ let payload = match simd_json::to_vec(&change_record) {
+ Ok(payload) => payload,
+ Err(e) => {
+ error!("Skipping CDC row that failed to serialize:
{e}");
+ continue;
+ }
+ };
let message = ProducedMessage {
id: Some(Uuid::new_v4().as_u128()),
@@ -728,99 +751,63 @@ impl PostgresSource {
&self,
data: &str,
capture_ops: &[&str],
+ captured_tables: Option<&[String]>,
) -> Option<DatabaseRecord> {
if data.starts_with("BEGIN") || data.starts_with("COMMIT") {
return None;
}
- if data.starts_with("INSERT:") && capture_ops.contains(&"INSERT") {
- return self.parse_insert_message(data);
- }
-
- if data.starts_with("UPDATE:") && capture_ops.contains(&"UPDATE") {
- return self.parse_update_message(data);
- }
-
- if data.starts_with("DELETE:") && capture_ops.contains(&"DELETE") {
- return self.parse_delete_message(data);
- }
-
- None
- }
+ let rest = data.strip_prefix("table ")?;
+ let (qualified_table, rest) = rest.split_once(": ")?;
+ let (operation, rest) = rest.split_once(": ")?;
- fn parse_insert_message(&self, data: &str) -> Option<DatabaseRecord> {
- if let Some(table_start) = data.find("table ")
- && let Some(colon_pos) = data[table_start..].find(':')
+ if !matches!(operation, "INSERT" | "UPDATE" | "DELETE") ||
!capture_ops.contains(&operation)
{
- let table_part = &data[table_start + 6..table_start + colon_pos];
- let table_name = table_part
- .split('.')
- .next_back()
- .unwrap_or(table_part)
- .to_string();
-
- let data_part = &data[table_start + colon_pos + 1..];
- let parsed_data = parse_record_data(data_part);
-
- return Some(DatabaseRecord {
- table_name,
- operation_type: "INSERT".to_string(),
- timestamp: Utc::now(),
- data: serde_json::Value::Object(parsed_data),
- old_data: None,
- });
+ return None;
}
- None
- }
- fn parse_update_message(&self, data: &str) -> Option<DatabaseRecord> {
- if let Some(table_start) = data.find("table ")
- && let Some(colon_pos) = data[table_start..].find(':')
+ let table_name = unquote_pg_identifier(
+ qualified_table
+ .rsplit('.')
+ .next()
+ .unwrap_or(qualified_table),
+ );
+ let unquoted_qualified_table = qualified_table
+ .split('.')
+ .map(unquote_pg_identifier)
+ .collect::<Vec<_>>()
+ .join(".");
+
+ // test_decoding has no server-side table filter, so Postgres already
+ // sent us every table's changes - config.tables scoping happens here.
+ if let Some(tables) = captured_tables
+ && !tables.iter().any(|t| {
+ if t.contains('.') {
+ t == &unquoted_qualified_table
+ } else {
+ t == &table_name
+ }
+ })
{
- let table_part = &data[table_start + 6..table_start + colon_pos];
- let table_name = table_part
- .split('.')
- .next_back()
- .unwrap_or(table_part)
- .to_string();
-
- let data_part = &data[table_start + colon_pos + 1..];
- let parsed_data = parse_record_data(data_part);
-
- return Some(DatabaseRecord {
- table_name,
- operation_type: "UPDATE".to_string(),
- timestamp: Utc::now(),
- data: serde_json::Value::Object(parsed_data),
- old_data: None,
- });
+ return None;
}
- None
- }
- fn parse_delete_message(&self, data: &str) -> Option<DatabaseRecord> {
- if let Some(table_start) = data.find("table ")
- && let Some(colon_pos) = data[table_start..].find(':')
+ let (old_key, rest) = match rest
+ .strip_prefix("old-key: ")
+ .and_then(|old_key| old_key.split_once("new-tuple: "))
{
- let table_part = &data[table_start + 6..table_start + colon_pos];
- let table_name = table_part
- .split('.')
- .next_back()
- .unwrap_or(table_part)
- .to_string();
-
- let data_part = &data[table_start + colon_pos + 1..];
- let parsed_data = parse_record_data(data_part);
-
- return Some(DatabaseRecord {
- table_name,
- operation_type: "DELETE".to_string(),
- timestamp: Utc::now(),
- data: serde_json::Value::Object(parsed_data),
- old_data: None,
- });
- }
- None
+ Some((old_key, new_tuple)) => (Some(old_key), new_tuple),
+ None => (None, rest),
+ };
+
+ Some(DatabaseRecord {
+ table_name,
+ operation_type: operation.to_string(),
+ timestamp: Utc::now(),
+ data: serde_json::Value::Object(parse_record_columns(rest)),
+ old_data: old_key
+ .map(|old_key|
serde_json::Value::Object(parse_record_columns(old_key))),
+ })
}
fn process_row(
@@ -1497,44 +1484,169 @@ fn to_snake_case(input: &str) -> String {
result
}
-fn parse_record_data(data: &str) -> serde_json::Map<String, serde_json::Value>
{
- let mut result = serde_json::Map::new();
+fn validate_cdc_backend(cdc_backend: Option<&str>) -> Result<&str, Error> {
+ let backend = cdc_backend.unwrap_or("builtin");
+ match backend {
+ "builtin" => Ok(backend),
+ "pg_replicate" => {
+ #[cfg(feature = "cdc_pg_replicate")]
+ {
+ Ok(backend)
+ }
+ #[cfg(not(feature = "cdc_pg_replicate"))]
+ {
+ Err(Error::InitError(
+ "cdc_backend 'pg_replicate' requested but feature
'cdc_pg_replicate' is not enabled at build time".to_string(),
+ ))
+ }
+ }
+ other => Err(Error::InitError(format!(
+ "Unsupported cdc_backend '{other}'. Use 'builtin' or
'pg_replicate'"
+ ))),
+ }
+}
- for part in data.split_whitespace() {
- if let Some(bracket_pos) = part.find('[')
- && let Some(_close_bracket) = part.find(']')
- && let Some(colon_pos) = part.find(':')
- {
- let column_name = &part[..bracket_pos];
- let value_str = &part[colon_pos + 1..];
+fn validate_capture_operations(capture_operations: Option<&[String]>) ->
Result<(), Error> {
+ const VALID: [&str; 3] = ["INSERT", "UPDATE", "DELETE"];
+ let Some(ops) = capture_operations else {
+ return Ok(());
+ };
+ for op in ops {
+ if !VALID.contains(&op.as_str()) {
+ return Err(Error::InitError(format!(
+ "Unsupported capture_operations value '{op}'. Use any of
'INSERT', 'UPDATE', 'DELETE'"
+ )));
+ }
+ }
+ Ok(())
+}
- let cleaned_value = if value_str.starts_with('\'') &&
value_str.ends_with('\'') {
- &value_str[1..value_str.len() - 1]
- } else {
- value_str
- };
+fn validate_payload_format(payload_format: Option<&str>) -> Result<(), Error> {
+ const VALID: [&str; 7] = [
+ "json",
+ "bytea",
+ "raw",
+ "text",
+ "json_direct",
+ "jsonb",
+ "jsonb_direct",
+ ];
+ let Some(fmt) = payload_format else {
+ return Ok(());
+ };
+ if !VALID.contains(&fmt.to_lowercase().as_str()) {
+ return Err(Error::InitError(format!(
+ "Unsupported payload_format '{fmt}'. Use 'json', 'bytea'/'raw',
'text', or 'json_direct'/'jsonb'/'jsonb_direct'"
+ )));
+ }
+ Ok(())
+}
- let value = if let Ok(num) = cleaned_value.parse::<i64>() {
- serde_json::Value::Number(serde_json::Number::from(num))
- } else if let Ok(float) = cleaned_value.parse::<f64>() {
- serde_json::Value::Number(
-
serde_json::Number::from_f64(float).unwrap_or(serde_json::Number::from(0)),
- )
- } else if cleaned_value.eq_ignore_ascii_case("true") {
- serde_json::Value::Bool(true)
- } else if cleaned_value.eq_ignore_ascii_case("false") {
- serde_json::Value::Bool(false)
- } else {
- serde_json::Value::String(cleaned_value.to_string())
- };
+fn unquote_pg_identifier(segment: &str) -> String {
+ match segment.strip_prefix('"').and_then(|s| s.strip_suffix('"')) {
+ Some(inner) => inner.replace("\"\"", "\""),
+ None => segment.to_string(),
+ }
+}
- result.insert(column_name.to_string(), value);
+fn parse_record_columns(data: &str) -> serde_json::Map<String,
serde_json::Value> {
+ let mut result = serde_json::Map::new();
+ let bytes = data.as_bytes();
+ let len = bytes.len();
+ let mut pos = 0;
+
+ while pos < len {
+ while pos < len && bytes[pos] == b' ' {
+ pos += 1;
+ }
+ if pos >= len {
+ break;
+ }
+
+ let Some(bracket_offset) = data[pos..].find('[') else {
+ break;
+ };
+ let name_end = pos + bracket_offset;
+ let column_name = unquote_pg_identifier(&data[pos..name_end]);
+
+ let mut depth = 0;
+ let mut type_end = name_end;
+ loop {
+ match bytes.get(type_end) {
+ Some(b'[') => depth += 1,
+ Some(b']') => {
+ depth -= 1;
+ if depth == 0 {
+ break;
+ }
+ }
+ Some(_) => {}
+ None => return result,
+ }
+ type_end += 1;
+ }
+ if bytes.get(type_end + 1) != Some(&b':') {
+ break;
}
+
+ let (value, next_pos) = parse_column_value(data, type_end + 2);
+ result.insert(column_name, value);
+ pos = next_pos;
}
result
}
+fn parse_column_value(data: &str, start: usize) -> (serde_json::Value, usize) {
+ let bytes = data.as_bytes();
+
+ if bytes.get(start) != Some(&b'\'') {
+ let end = data[start..]
+ .find(' ')
+ .map_or(data.len(), |offset| start + offset);
+ return (parse_bare_scalar(&data[start..end]), end);
+ }
+
+ let mut value = String::new();
+ let mut pos = start + 1;
+ while pos < bytes.len() {
+ if bytes[pos] == b'\'' {
+ if bytes.get(pos + 1) == Some(&b'\'') {
+ value.push('\'');
+ pos += 2;
+ continue;
+ }
+ pos += 1;
+ break;
+ }
+ let ch = data[pos..].chars().next().unwrap_or('\u{FFFD}');
+ value.push(ch);
+ pos += ch.len_utf8();
+ }
+ (serde_json::Value::String(value), pos)
+}
+
+fn parse_bare_scalar(token: &str) -> serde_json::Value {
+ // test_decoding emits this sentinel for TOASTed columns the UPDATE didn't
touch;
+ // treating it as null avoids leaking the literal token as a fake column
value.
+ match token {
+ "null" | "unchanged-toast-datum" => serde_json::Value::Null,
+ "true" => serde_json::Value::Bool(true),
+ "false" => serde_json::Value::Bool(false),
+ _ => {
+ if let Ok(i) = token.parse::<i64>() {
+ serde_json::Value::Number(serde_json::Number::from(i))
+ } else if let Ok(f) = token.parse::<f64>()
+ && let Some(num) = serde_json::Number::from_f64(f)
+ {
+ serde_json::Value::Number(num)
+ } else {
+ serde_json::Value::String(token.to_string())
+ }
+ }
+ }
+}
+
async fn with_retry<T, F, Fut>(operation: F, max_retries: u32, delay_ms: u64)
-> Result<T, Error>
where
F: Fn() -> Fut,
@@ -1586,6 +1698,10 @@ fn redact_connection_string(conn_str: &str) -> String {
format!("{preview}***")
}
+#[cfg(test)]
+#[path = "cdc_fixtures.rs"]
+mod cdc_fixtures;
+
#[cfg(test)]
mod tests {
use super::*;
@@ -1600,12 +1716,10 @@ mod tests {
tracking_column: Some("updated_at".to_string()),
initial_offset: None,
max_connections: None,
- enable_wal_cdc: None,
custom_query: None,
snake_case_columns: None,
include_metadata: None,
replication_slot: None,
- publication_name: None,
capture_operations: None,
cdc_backend: None,
delete_after_read: None,
@@ -1703,46 +1817,610 @@ mod tests {
assert!(quote_qualified_identifier(".users").is_err());
}
- #[test]
- fn given_insert_message_should_parse_correctly() {
+ fn cdc_source() -> PostgresSource {
let mut config = test_config();
config.mode = "cdc".to_string();
- let src = PostgresSource::new(1, config, None);
+ PostgresSource::new(1, config, None)
+ }
- let data = "INSERT: table public.users: id[1] name['Alice']
active[true]";
+ #[test]
+ fn given_insert_single_row_all_types_should_parse_correctly() {
+ let src = cdc_source();
let rec = src
- .parse_logical_replication_message(data, &["INSERT"])
+ .parse_logical_replication_message(
+ cdc_fixtures::INSERT_SINGLE_ROW_ALL_TYPES,
+ &["INSERT"],
+ None,
+ )
.unwrap();
- assert_eq!(rec.table_name, "users");
+
+ assert_eq!(rec.table_name, "probe_events");
assert_eq!(rec.operation_type, "INSERT");
+ assert_eq!(rec.data["id"], serde_json::json!(2));
+ assert_eq!(rec.data["name"], serde_json::json!("alice"));
+ assert_eq!(rec.data["note"], serde_json::json!("first note"));
+ assert_eq!(rec.data["amount"], serde_json::json!(12.50));
+ assert_eq!(rec.data["active"], serde_json::json!(true));
+ assert_eq!(rec.data["tags"], serde_json::json!("{a,b}"));
+ assert_eq!(rec.data["payload"], serde_json::json!(r#"{"k": 1}"#));
+ assert_eq!(rec.data["small_int"], serde_json::Value::Null);
}
#[test]
- fn given_update_message_should_parse_correctly() {
- let mut config = test_config();
- config.mode = "cdc".to_string();
- let src = PostgresSource::new(1, config, None);
+ fn given_insert_with_nulls_should_parse_correctly() {
+ let src = cdc_source();
+ let rec = src
+
.parse_logical_replication_message(cdc_fixtures::INSERT_WITH_NULLS,
&["INSERT"], None)
+ .unwrap();
+
+ assert_eq!(rec.data["name"], serde_json::json!("bob"));
+ assert_eq!(rec.data["note"], serde_json::Value::Null);
+ assert_eq!(rec.data["amount"], serde_json::Value::Null);
+ assert_eq!(rec.data["active"], serde_json::Value::Null);
+ }
- let data = "UPDATE: table public.orders: id[42] total[99.5]";
+ #[test]
+ fn given_multi_row_insert_statement_should_parse_each_row() {
+ let src = cdc_source();
+ for (fixture, name) in cdc_fixtures::INSERT_MULTI_ROW_SINGLE_STATEMENT
+ .iter()
+ .zip(["carol", "dave"])
+ {
+ let rec = src
+ .parse_logical_replication_message(fixture, &["INSERT"], None)
+ .unwrap();
+ assert_eq!(rec.data["name"], serde_json::json!(name));
+ }
+ }
+
+ #[test]
+ fn given_multi_statement_transaction_should_parse_each_insert() {
+ let src = cdc_source();
+ for (fixture, name) in
cdc_fixtures::INSERT_MULTI_STATEMENT_ONE_TRANSACTION
+ .iter()
+ .zip(["eve", "frank"])
+ {
+ let rec = src
+ .parse_logical_replication_message(fixture, &["INSERT"], None)
+ .unwrap();
+ assert_eq!(rec.data["name"], serde_json::json!(name));
+ }
+ }
+
+ #[test]
+ fn given_update_single_column_should_keep_other_columns() {
+ let src = cdc_source();
+ let rec = src
+ .parse_logical_replication_message(
+ cdc_fixtures::UPDATE_SINGLE_COLUMN,
+ &["UPDATE"],
+ None,
+ )
+ .unwrap();
+
+ assert_eq!(rec.data["note"], serde_json::json!("only note changed"));
+ assert_eq!(rec.data["name"], serde_json::json!("bob"));
+ }
+
+ #[test]
+ fn given_delete_multiple_rows_should_parse_each_row() {
+ let src = cdc_source();
+ for (fixture, id) in cdc_fixtures::DELETE_MULTIPLE_ROWS.iter().zip([6,
7]) {
+ let rec = src
+ .parse_logical_replication_message(fixture, &["DELETE"], None)
+ .unwrap();
+ assert_eq!(rec.data["id"], serde_json::json!(id));
+ }
+ }
+
+ #[test]
+ fn given_update_array_column_should_parse_new_array() {
+ let src = cdc_source();
+ let rec = src
+
.parse_logical_replication_message(cdc_fixtures::UPDATE_ARRAY_COLUMN,
&["UPDATE"], None)
+ .unwrap();
+
+ assert_eq!(rec.data["int_array"], serde_json::json!("{9,8,7}"));
+ }
+
+ #[test]
+ fn given_negative_zero_float_should_parse_as_zero() {
+ let src = cdc_source();
+ let rec = src
+ .parse_logical_replication_message(
+ cdc_fixtures::INSERT_NEGATIVE_ZERO_FLOAT,
+ &["INSERT"],
+ None,
+ )
+ .unwrap();
+
+ assert_eq!(rec.data["real_val"], serde_json::json!(0));
+ assert_eq!(rec.data["double_val"], serde_json::json!(0));
+ }
+
+ #[test]
+ fn given_quoted_mixed_case_column_should_strip_quotes_from_key() {
+ let src = cdc_source();
let rec = src
- .parse_logical_replication_message(data, &["UPDATE"])
+ .parse_logical_replication_message(
+ cdc_fixtures::INSERT_QUOTED_MIXED_CASE_COLUMN,
+ &["INSERT"],
+ None,
+ )
.unwrap();
- assert_eq!(rec.table_name, "orders");
+
+ assert_eq!(rec.data["user"], serde_json::json!("quoted_row"));
+ assert!(rec.data.get("createdAt").is_some());
+ assert!(rec.data.get("\"user\"").is_none());
+ }
+
+ #[test]
+ fn given_unknown_cdc_backend_should_fail_validation() {
+ let err = validate_cdc_backend(Some("built-in")).unwrap_err();
+ assert!(matches!(err, Error::InitError(_)));
+ }
+
+ #[test]
+ fn given_no_cdc_backend_should_default_to_builtin() {
+ assert_eq!(validate_cdc_backend(None).unwrap(), "builtin");
+ }
+
+ #[test]
+ fn given_unsupported_capture_operation_should_fail_validation() {
+ let ops = vec!["INSRT".to_string()];
+ let err = validate_capture_operations(Some(&ops)).unwrap_err();
+ assert!(matches!(err, Error::InitError(_)));
+ }
+
+ #[test]
+ fn given_valid_capture_operations_should_pass_validation() {
+ let ops = vec!["INSERT".to_string(), "DELETE".to_string()];
+ assert!(validate_capture_operations(Some(&ops)).is_ok());
+ assert!(validate_capture_operations(None).is_ok());
+ }
+
+ #[test]
+ fn given_unsupported_payload_format_should_fail_validation() {
+ let err = validate_payload_format(Some("btea")).unwrap_err();
+ assert!(matches!(err, Error::InitError(_)));
+ }
+
+ #[test]
+ fn given_valid_payload_format_should_pass_validation() {
+ assert!(validate_payload_format(Some("bytea")).is_ok());
+ assert!(validate_payload_format(Some("JSON")).is_ok());
+ assert!(validate_payload_format(None).is_ok());
+ }
+
+ #[test]
+ fn given_unchanged_toast_datum_should_parse_as_null() {
+ let src = cdc_source();
+ let rec = src
+ .parse_logical_replication_message(
+ cdc_fixtures::UPDATE_UNCHANGED_TOAST_COLUMN,
+ &["UPDATE"],
+ None,
+ )
+ .unwrap();
+
+ assert_eq!(rec.data["note"], serde_json::Value::Null);
+ assert_eq!(rec.data["payload"], serde_json::Value::Null);
+ assert_eq!(rec.data["name"], serde_json::json!("toast_row"));
+ }
+
+ #[test]
+ fn given_update_full_row_should_parse_correctly() {
+ let src = cdc_source();
+ let rec = src
+ .parse_logical_replication_message(cdc_fixtures::UPDATE_FULL_ROW,
&["UPDATE"], None)
+ .unwrap();
+
+ assert_eq!(rec.table_name, "probe_events");
assert_eq!(rec.operation_type, "UPDATE");
+ assert_eq!(rec.data["name"], serde_json::json!("alice2"));
+ assert_eq!(rec.data["amount"], serde_json::json!(99.99));
+ assert_eq!(rec.data["active"], serde_json::json!(false));
}
#[test]
- fn given_delete_message_should_parse_correctly() {
- let mut config = test_config();
- config.mode = "cdc".to_string();
- let src = PostgresSource::new(1, config, None);
+ fn given_update_to_null_should_parse_correctly() {
+ let src = cdc_source();
+ let rec = src
+ .parse_logical_replication_message(cdc_fixtures::UPDATE_TO_NULL,
&["UPDATE"], None)
+ .unwrap();
- let data = "DELETE: table public.products: id[7]";
+ assert_eq!(rec.data["note"], serde_json::Value::Null);
+ assert_eq!(rec.data["amount"], serde_json::json!(99.99));
+ }
+
+ #[test]
+ fn given_update_primary_key_should_parse_new_tuple_only() {
+ let src = cdc_source();
+ let rec = src
+
.parse_logical_replication_message(cdc_fixtures::UPDATE_PRIMARY_KEY,
&["UPDATE"], None)
+ .unwrap();
+
+ assert_eq!(rec.data["id"], serde_json::json!(1004));
+ assert_eq!(rec.data["name"], serde_json::json!("carol"));
+ }
+
+ #[test]
+ fn given_delete_row_should_parse_replica_identity_columns() {
+ let src = cdc_source();
let rec = src
- .parse_logical_replication_message(data, &["DELETE"])
+ .parse_logical_replication_message(cdc_fixtures::DELETE_ROW,
&["DELETE"], None)
.unwrap();
- assert_eq!(rec.table_name, "products");
+
+ assert_eq!(rec.table_name, "probe_events");
assert_eq!(rec.operation_type, "DELETE");
+ assert_eq!(rec.data["id"], serde_json::json!(5));
+ assert_eq!(rec.data.as_object().unwrap().len(), 1);
+ }
+
+ #[test]
+ fn given_structurally_broken_rows_should_return_none() {
+ let src = cdc_source();
+ let capture_ops = ["INSERT", "UPDATE", "DELETE"];
+ let bad_rows = [
+ "",
+ "garbage",
+ "table",
+ "table public.probe_events",
+ "table public.probe_events:",
+ "table public.probe_events: INSERT",
+ "table public.probe_events: INSERT:",
+ ];
+
+ for bad_row in bad_rows {
+ assert!(
+ src.parse_logical_replication_message(bad_row, &capture_ops,
None)
+ .is_none(),
+ "Expected None for structurally broken row: {bad_row:?}"
+ );
+ }
+ }
+
+ #[test]
+ fn given_broken_column_syntax_should_not_panic() {
+ let src = cdc_source();
+ let capture_ops = ["INSERT", "UPDATE", "DELETE"];
+ let bad_rows = [
+ "table : INSERT: id[integer]:1",
+ "table public.probe_events: INSERT: id[integer",
+ "table public.probe_events: INSERT: id[integer]",
+ "table public.probe_events: INSERT: id[integer]:",
+ "table public.probe_events: INSERT: id[integer]:'unterminated",
+ ];
+
+ for bad_row in bad_rows {
+ let result =
std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+ src.parse_logical_replication_message(bad_row, &capture_ops,
None)
+ }));
+ assert!(
+ result.is_ok(),
+ "Parsing must not panic on malformed column data: {bad_row:?}"
+ );
+ }
+ }
+
+ #[test]
+ fn given_truncate_message_should_be_ignored() {
+ let src = cdc_source();
+ assert!(
+ src.parse_logical_replication_message(
+ cdc_fixtures::TRUNCATE_TABLE,
+ &["INSERT", "UPDATE", "DELETE"],
+ None,
+ )
+ .is_none()
+ );
+ }
+
+ #[test]
+ fn given_operation_not_in_capture_ops_should_be_ignored() {
+ let src = cdc_source();
+ assert!(
+ src.parse_logical_replication_message(
+ cdc_fixtures::DELETE_ROW,
+ &["INSERT", "UPDATE"],
+ None
+ )
+ .is_none()
+ );
+ }
+
+ #[test]
+ fn given_schema_qualified_captured_table_should_match_exact_schema_only() {
+ let src = cdc_source();
+ let matching = vec!["public.probe_events".to_string()];
+ let other_schema = vec!["other_schema.probe_events".to_string()];
+
+ assert!(
+ src.parse_logical_replication_message(
+ cdc_fixtures::INSERT_SINGLE_ROW_ALL_TYPES,
+ &["INSERT"],
+ Some(&matching),
+ )
+ .is_some()
+ );
+ assert!(
+ src.parse_logical_replication_message(
+ cdc_fixtures::INSERT_SINGLE_ROW_ALL_TYPES,
+ &["INSERT"],
+ Some(&other_schema),
+ )
+ .is_none()
+ );
+ }
+
+ #[test]
+ fn given_bare_captured_table_should_match_regardless_of_schema() {
+ let src = cdc_source();
+ let bare = vec!["probe_events".to_string()];
+
+ assert!(
+ src.parse_logical_replication_message(
+ cdc_fixtures::INSERT_SINGLE_ROW_ALL_TYPES,
+ &["INSERT"],
+ Some(&bare),
+ )
+ .is_some()
+ );
+ }
+
+ #[test]
+ fn given_mixed_case_table_name_should_unquote_for_table_name_and_filter() {
+ let src = cdc_source();
+ let data = r#"table public."MyTable": INSERT: id[integer]:1
name[text]:'alice'"#;
+
+ let rec = src
+ .parse_logical_replication_message(data, &["INSERT"], None)
+ .unwrap();
+ assert_eq!(rec.table_name, "MyTable");
+
+ let bare = vec!["MyTable".to_string()];
+ assert!(
+ src.parse_logical_replication_message(data, &["INSERT"],
Some(&bare))
+ .is_some(),
+ "Unquoted config entry must match the unquoted table name"
+ );
+
+ let qualified = vec!["public.MyTable".to_string()];
+ assert!(
+ src.parse_logical_replication_message(data, &["INSERT"],
Some(&qualified))
+ .is_some(),
+ "Schema-qualified config entry must match against the unquoted
qualified name"
+ );
+
+ let wrong_case = vec!["mytable".to_string()];
+ assert!(
+ src.parse_logical_replication_message(data, &["INSERT"],
Some(&wrong_case))
+ .is_none(),
+ "Postgres identifiers are case-sensitive once quoted - must not
fuzzy-match"
+ );
+ }
+
+ #[test]
+ fn given_quoted_identifier_with_embedded_quote_should_unescape() {
+ let src = cdc_source();
+ let data = r#"table public."Weird""Table": INSERT: id[integer]:1"#;
+
+ let rec = src
+ .parse_logical_replication_message(data, &["INSERT"], None)
+ .unwrap();
+ assert_eq!(rec.table_name, "Weird\"Table");
+ }
+
+ #[test]
+ fn
given_new_tuple_substring_in_value_should_not_truncate_ordinary_update() {
+ let src = cdc_source();
+ let data = "table public.probe_events: UPDATE: id[integer]:2 \
+ note[text]:'see new-tuple: format docs'
amount[numeric]:99.99";
+
+ let rec = src
+ .parse_logical_replication_message(data, &["UPDATE"], None)
+ .unwrap();
+
+ assert_eq!(rec.data["id"], serde_json::json!(2));
+ assert_eq!(
+ rec.data["note"],
+ serde_json::json!("see new-tuple: format docs")
+ );
+ assert_eq!(rec.data["amount"], serde_json::json!(99.99));
+ }
+
+ #[test]
+ fn given_old_key_section_should_populate_old_data() {
+ let src = cdc_source();
+ let data = "table public.probe_events: UPDATE: old-key: id[integer]:1 \
+ new-tuple: id[integer]:2 amount[numeric]:99.99";
+
+ let rec = src
+ .parse_logical_replication_message(data, &["UPDATE"], None)
+ .unwrap();
+
+ assert_eq!(rec.data["id"], serde_json::json!(2));
+ assert_eq!(rec.data["amount"], serde_json::json!(99.99));
+ assert_eq!(rec.old_data.unwrap()["id"], serde_json::json!(1));
+ }
+
+ #[test]
+ fn given_no_old_key_section_should_leave_old_data_none() {
+ let src = cdc_source();
+ let data = "table public.probe_events: INSERT: id[integer]:1";
+
+ let rec = src
+ .parse_logical_replication_message(data, &["INSERT"], None)
+ .unwrap();
+
+ assert!(rec.old_data.is_none());
+ }
+
+ #[test]
+ fn given_unicode_and_escaped_quote_should_parse_correctly() {
+ let src = cdc_source();
+ let rec = src
+ .parse_logical_replication_message(
+ cdc_fixtures::UNICODE_AND_SPECIAL_CHARS,
+ &["INSERT"],
+ None,
+ )
+ .unwrap();
+
+ assert_eq!(
+ rec.data["note"],
+ serde_json::json!("emoji \u{1F680} quote' backslash\\
newline\nend")
+ );
+ }
+
+ #[test]
+ fn given_extended_types_should_parse_correctly() {
+ let src = cdc_source();
+ let rec = src
+ .parse_logical_replication_message(
+ cdc_fixtures::INSERT_EXTENDED_TYPES_NORMAL_VALUES,
+ &["INSERT"],
+ None,
+ )
+ .unwrap();
+
+ assert_eq!(rec.data["small_int"], serde_json::json!(42));
+ assert_eq!(rec.data["big_int"], serde_json::json!(9000000000i64));
+ assert_eq!(
+ rec.data["real_val"],
+ serde_json::json!("3.14".parse::<f64>().unwrap())
+ );
+ assert_eq!(
+ rec.data["uuid_val"],
+ serde_json::json!("11111111-1111-1111-1111-111111111111")
+ );
+ assert_eq!(rec.data["bytea_val"], serde_json::json!("\\xdeadbeef"));
+ assert_eq!(rec.data["date_val"], serde_json::json!("2024-01-15"));
+ assert_eq!(rec.data["int_array"], serde_json::json!("{1,2,3}"));
+ assert_eq!(rec.data["char_val"], serde_json::json!("ab "));
+ }
+
+ #[test]
+ fn given_nan_should_parse_as_string() {
+ let src = cdc_source();
+ let rec = src
+
.parse_logical_replication_message(cdc_fixtures::INSERT_NUMERIC_NAN,
&["INSERT"], None)
+ .unwrap();
+
+ assert_eq!(rec.data["real_val"], serde_json::json!("NaN"));
+ assert_eq!(rec.data["numeric_val"], serde_json::json!("NaN"));
+ }
+
+ #[test]
+ fn given_infinity_should_parse_as_string() {
+ let src = cdc_source();
+ let rec = src
+ .parse_logical_replication_message(
+ cdc_fixtures::INSERT_NUMERIC_INFINITY,
+ &["INSERT"],
+ None,
+ )
+ .unwrap();
+
+ assert_eq!(rec.data["real_val"], serde_json::json!("Infinity"));
+ assert_eq!(rec.data["double_val"], serde_json::json!("-Infinity"));
+ }
+
+ #[test]
+ fn given_negative_and_boundary_numbers_should_parse_correctly() {
+ let src = cdc_source();
+ let rec = src
+ .parse_logical_replication_message(
+ cdc_fixtures::INSERT_NEGATIVE_AND_BOUNDARY_NUMBERS,
+ &["INSERT"],
+ None,
+ )
+ .unwrap();
+
+ assert_eq!(rec.data["small_int"], serde_json::json!(-32768));
+ assert_eq!(
+ rec.data["big_int"],
+ serde_json::json!(-9223372036854775808i64)
+ );
+ }
+
+ #[test]
+ fn given_max_boundary_numbers_should_parse_correctly() {
+ let src = cdc_source();
+ let rec = src
+ .parse_logical_replication_message(
+ cdc_fixtures::INSERT_MAX_BOUNDARY_NUMBERS,
+ &["INSERT"],
+ None,
+ )
+ .unwrap();
+
+ assert_eq!(rec.data["small_int"], serde_json::json!(32767));
+ assert_eq!(
+ rec.data["big_int"],
+ serde_json::json!(9223372036854775807i64)
+ );
+ }
+
+ #[test]
+ fn given_empty_string_vs_null_should_be_distinct() {
+ let src = cdc_source();
+ let empty_row = src
+ .parse_logical_replication_message(
+ cdc_fixtures::INSERT_EMPTY_STRING_VS_NULL[0],
+ &["INSERT"],
+ None,
+ )
+ .unwrap();
+ let null_row = src
+ .parse_logical_replication_message(
+ cdc_fixtures::INSERT_EMPTY_STRING_VS_NULL[1],
+ &["INSERT"],
+ None,
+ )
+ .unwrap();
+
+ assert_eq!(empty_row.data["note"], serde_json::json!(""));
+ assert_eq!(null_row.data["note"], serde_json::Value::Null);
+ }
+
+ #[test]
+ fn given_array_with_null_element_should_parse_as_raw_string() {
+ let src = cdc_source();
+ let rec = src
+ .parse_logical_replication_message(
+ cdc_fixtures::INSERT_ARRAY_WITH_NULL_ELEMENT,
+ &["INSERT"],
+ None,
+ )
+ .unwrap();
+
+ assert_eq!(rec.data["int_array"], serde_json::json!("{1,2,NULL,4}"));
+ }
+
+ #[test]
+ fn given_empty_array_should_parse_correctly() {
+ let src = cdc_source();
+ let rec = src
+
.parse_logical_replication_message(cdc_fixtures::INSERT_EMPTY_ARRAY,
&["INSERT"], None)
+ .unwrap();
+
+ assert_eq!(rec.data["int_array"], serde_json::json!("{}"));
+ }
+
+ #[test]
+ fn given_char_padding_vs_varchar_should_preserve_padding() {
+ let src = cdc_source();
+ let rec = src
+ .parse_logical_replication_message(
+ cdc_fixtures::INSERT_CHAR_PADDING_VS_VARCHAR,
+ &["INSERT"],
+ None,
+ )
+ .unwrap();
+
+ assert_eq!(rec.data["char_val"], serde_json::json!("ab "));
+ assert_eq!(rec.data["varchar_val"], serde_json::json!("ab"));
}
#[test]
diff --git a/core/integration/tests/connectors/fixtures/mod.rs
b/core/integration/tests/connectors/fixtures/mod.rs
index 885867de9..6d75fbf97 100644
--- a/core/integration/tests/connectors/fixtures/mod.rs
+++ b/core/integration/tests/connectors/fixtures/mod.rs
@@ -74,8 +74,9 @@ pub use mongodb::{
};
pub use postgres::{
PostgresOps, PostgresSinkByteaFixture, PostgresSinkFixture,
PostgresSinkJsonFixture,
- PostgresSourceByteaFixture, PostgresSourceDeleteFixture,
PostgresSourceJsonFixture,
- PostgresSourceJsonbFixture, PostgresSourceMarkFixture, PostgresSourceOps,
+ PostgresSourceByteaFixture, PostgresSourceCdcFixture,
PostgresSourceDeleteFixture,
+ PostgresSourceJsonFixture, PostgresSourceJsonbFixture,
PostgresSourceMarkFixture,
+ PostgresSourceOps,
};
pub use quickwit::{QuickwitFixture, QuickwitOps, QuickwitPreCreatedFixture};
pub use s3::{S3SinkFixture, S3SinkOps, S3SinkRotationFixture};
diff --git a/core/integration/tests/connectors/fixtures/postgres/cdc.rs
b/core/integration/tests/connectors/fixtures/postgres/cdc.rs
new file mode 100644
index 000000000..88d69a054
--- /dev/null
+++ b/core/integration/tests/connectors/fixtures/postgres/cdc.rs
@@ -0,0 +1,206 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::container::{
+ DEFAULT_TEST_STREAM, DEFAULT_TEST_TOPIC, ENV_SOURCE_CONNECTION_STRING,
+ ENV_SOURCE_INCLUDE_METADATA, ENV_SOURCE_MODE, ENV_SOURCE_PATH,
ENV_SOURCE_POLL_INTERVAL,
+ ENV_SOURCE_STREAMS_0_SCHEMA, ENV_SOURCE_STREAMS_0_STREAM,
ENV_SOURCE_STREAMS_0_TOPIC,
+ ENV_SOURCE_TABLES, ENV_SOURCE_TRACKING_COLUMN, PostgresContainer,
PostgresOps,
+ PostgresSourceOps,
+};
+use async_trait::async_trait;
+use integration::harness::{TestBinaryError, TestFixture};
+use sqlx::{Pool, Postgres};
+use std::collections::HashMap;
+
+/// PostgreSQL source fixture for CDC (WAL logical decoding) mode.
+///
+/// Starts the container with `wal_level = logical` so the connector's
+/// `setup_cdc` can create a real `test_decoding` replication slot.
+pub struct PostgresSourceCdcFixture {
+ container: PostgresContainer,
+}
+
+impl PostgresOps for PostgresSourceCdcFixture {
+ fn container(&self) -> &PostgresContainer {
+ &self.container
+ }
+}
+
+impl PostgresSourceOps for PostgresSourceCdcFixture {
+ fn table_name(&self) -> &str {
+ Self::TABLE
+ }
+}
+
+impl PostgresSourceCdcFixture {
+ const TABLE: &'static str = "cdc_events";
+ const UNTRACKED_TABLE: &'static str = "cdc_events_untracked";
+
+ pub async fn create_table(&self, pool: &Pool<Postgres>) {
+ let query = format!(
+ "CREATE TABLE IF NOT EXISTS {} (
+ id SERIAL PRIMARY KEY,
+ name VARCHAR(255) NOT NULL,
+ count INTEGER NOT NULL,
+ amount DOUBLE PRECISION NOT NULL,
+ active BOOLEAN NOT NULL,
+ timestamp BIGINT NOT NULL,
+ tag CHAR(10) NOT NULL
+ )",
+ Self::TABLE
+ );
+ sqlx::query(sqlx::AssertSqlSafe(query))
+ .execute(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to create table: {e}"));
+ }
+
+ #[allow(clippy::too_many_arguments)]
+ pub async fn insert_row(
+ &self,
+ pool: &Pool<Postgres>,
+ id: i32,
+ name: &str,
+ count: i32,
+ amount: f64,
+ active: bool,
+ timestamp: i64,
+ ) {
+ let tag = format!("{:<10}", format!("tag_{id}"));
+ let query = format!(
+ "INSERT INTO {} (id, name, count, amount, active, timestamp, tag)
VALUES ($1, $2, $3, $4, $5, $6, $7)",
+ Self::TABLE
+ );
+ sqlx::query(sqlx::AssertSqlSafe(query))
+ .bind(id)
+ .bind(name)
+ .bind(count)
+ .bind(amount)
+ .bind(active)
+ .bind(timestamp)
+ .bind(&tag)
+ .execute(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to insert row: {e}"));
+ }
+
+ pub async fn update_row(&self, pool: &Pool<Postgres>, id: i32, new_count:
i32) {
+ let query = format!("UPDATE {} SET count = $1 WHERE id = $2",
Self::TABLE);
+ sqlx::query(sqlx::AssertSqlSafe(query))
+ .bind(new_count)
+ .bind(id)
+ .execute(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to update row: {e}"));
+ }
+
+ pub async fn update_primary_key(&self, pool: &Pool<Postgres>, old_id: i32,
new_id: i32) {
+ let query = format!("UPDATE {} SET id = $1 WHERE id = $2",
Self::TABLE);
+ sqlx::query(sqlx::AssertSqlSafe(query))
+ .bind(new_id)
+ .bind(old_id)
+ .execute(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to update primary key: {e}"));
+ }
+
+ pub async fn delete_row(&self, pool: &Pool<Postgres>, id: i32) {
+ let query = format!("DELETE FROM {} WHERE id = $1", Self::TABLE);
+ sqlx::query(sqlx::AssertSqlSafe(query))
+ .bind(id)
+ .execute(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to delete row: {e}"));
+ }
+
+ pub async fn insert_row_rolled_back(&self, pool: &Pool<Postgres>, id: i32,
name: &str) {
+ let mut tx = pool
+ .begin()
+ .await
+ .unwrap_or_else(|e| panic!("Failed to begin transaction: {e}"));
+ let query = format!(
+ "INSERT INTO {} (id, name, count, amount, active, timestamp, tag) \
+ VALUES ($1, $2, 0, 0.0, true, 0, 'rollback ')",
+ Self::TABLE
+ );
+ sqlx::query(sqlx::AssertSqlSafe(query))
+ .bind(id)
+ .bind(name)
+ .execute(&mut *tx)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to insert row inside
transaction: {e}"));
+ tx.rollback()
+ .await
+ .unwrap_or_else(|e| panic!("Failed to roll back transaction:
{e}"));
+ }
+
+ pub async fn create_untracked_table(&self, pool: &Pool<Postgres>) {
+ let query = format!(
+ "CREATE TABLE IF NOT EXISTS {} (id SERIAL PRIMARY KEY, name
VARCHAR(255) NOT NULL)",
+ Self::UNTRACKED_TABLE
+ );
+ sqlx::query(sqlx::AssertSqlSafe(query))
+ .execute(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to create untracked table:
{e}"));
+ }
+
+ pub async fn insert_untracked_row(&self, pool: &Pool<Postgres>, name:
&str) {
+ let query = format!("INSERT INTO {} (name) VALUES ($1)",
Self::UNTRACKED_TABLE);
+ sqlx::query(sqlx::AssertSqlSafe(query))
+ .bind(name)
+ .execute(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to insert untracked row: {e}"));
+ }
+}
+
+#[async_trait]
+impl TestFixture for PostgresSourceCdcFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let container =
PostgresContainer::start_with_logical_replication().await?;
+ Ok(Self { container })
+ }
+
+ fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+ let mut envs = HashMap::new();
+ envs.insert(
+ ENV_SOURCE_CONNECTION_STRING.to_string(),
+ self.container.connection_string.clone(),
+ );
+ envs.insert(ENV_SOURCE_MODE.to_string(), "cdc".to_string());
+ envs.insert(ENV_SOURCE_TABLES.to_string(), format!("[{}]",
Self::TABLE));
+ envs.insert(ENV_SOURCE_TRACKING_COLUMN.to_string(), "id".to_string());
+ envs.insert(ENV_SOURCE_INCLUDE_METADATA.to_string(),
"true".to_string());
+ envs.insert(
+ ENV_SOURCE_STREAMS_0_STREAM.to_string(),
+ DEFAULT_TEST_STREAM.to_string(),
+ );
+ envs.insert(
+ ENV_SOURCE_STREAMS_0_TOPIC.to_string(),
+ DEFAULT_TEST_TOPIC.to_string(),
+ );
+ envs.insert(ENV_SOURCE_STREAMS_0_SCHEMA.to_string(),
"json".to_string());
+ envs.insert(ENV_SOURCE_POLL_INTERVAL.to_string(), "50ms".to_string());
+ envs.insert(
+ ENV_SOURCE_PATH.to_string(),
+ "../../target/debug/libiggy_connector_postgres_source".to_string(),
+ );
+ envs
+ }
+}
diff --git a/core/integration/tests/connectors/fixtures/postgres/container.rs
b/core/integration/tests/connectors/fixtures/postgres/container.rs
index 642341a12..508607f0c 100644
--- a/core/integration/tests/connectors/fixtures/postgres/container.rs
+++ b/core/integration/tests/connectors/fixtures/postgres/container.rs
@@ -22,7 +22,7 @@ use sqlx::{Pool, Postgres};
use crate::connectors::fixtures;
use testcontainers_modules::{
postgres,
- testcontainers::{ContainerAsync, ImageExt, runners::AsyncRunner},
+ testcontainers::{ContainerAsync, ContainerRequest, ImageExt,
runners::AsyncRunner},
};
pub(super) const POSTGRES_PORT: u16 = 5432;
@@ -68,6 +68,7 @@ pub(super) const ENV_SOURCE_PROCESSED_COLUMN: &str =
"IGGY_CONNECTORS_SOURCE_POSTGRES_PLUGIN_CONFIG_PROCESSED_COLUMN";
pub(super) const ENV_SOURCE_INCLUDE_METADATA: &str =
"IGGY_CONNECTORS_SOURCE_POSTGRES_PLUGIN_CONFIG_INCLUDE_METADATA";
+pub(super) const ENV_SOURCE_MODE: &str =
"IGGY_CONNECTORS_SOURCE_POSTGRES_PLUGIN_CONFIG_MODE";
pub(super) const DEFAULT_TEST_STREAM: &str = "test_stream";
pub(super) const DEFAULT_TEST_TOPIC: &str = "test_topic";
@@ -128,7 +129,23 @@ pub struct PostgresContainer {
impl PostgresContainer {
pub(super) async fn start() -> Result<Self, TestBinaryError> {
- let container = postgres::Postgres::default()
+ Self::start_with_image(postgres::Postgres::default().into()).await
+ }
+
+ pub(super) async fn start_with_logical_replication() -> Result<Self,
TestBinaryError> {
+ Self::start_with_image(postgres::Postgres::default().with_cmd([
+ "-c",
+ "wal_level=logical",
+ "-c",
+ "fsync=off",
+ ]))
+ .await
+ }
+
+ async fn start_with_image(
+ image: ContainerRequest<postgres::Postgres>,
+ ) -> Result<Self, TestBinaryError> {
+ let container = image
.with_container_name(fixtures::unique_container_name("postgres"))
.start()
.await
diff --git a/core/integration/tests/connectors/fixtures/postgres/mod.rs
b/core/integration/tests/connectors/fixtures/postgres/mod.rs
index 327012672..cca22e6e3 100644
--- a/core/integration/tests/connectors/fixtures/postgres/mod.rs
+++ b/core/integration/tests/connectors/fixtures/postgres/mod.rs
@@ -15,10 +15,12 @@
// specific language governing permissions and limitations
// under the License.
+mod cdc;
mod container;
mod sink;
mod source;
+pub use cdc::PostgresSourceCdcFixture;
pub use container::{PostgresOps, PostgresSourceOps};
pub use sink::{PostgresSinkByteaFixture, PostgresSinkFixture,
PostgresSinkJsonFixture};
pub use source::{
diff --git
a/core/integration/tests/connectors/postgres/cdc_restart_connectors/config.toml
b/core/integration/tests/connectors/postgres/cdc_restart_connectors/config.toml
new file mode 100644
index 000000000..1ea3757f2
--- /dev/null
+++
b/core/integration/tests/connectors/postgres/cdc_restart_connectors/config.toml
@@ -0,0 +1,42 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+type = "source"
+key = "postgres"
+enabled = true
+version = 0
+name = "Postgres source"
+path = "../../target/release/libiggy_connector_postgres_source"
+verbose = false
+
+[[streams]]
+stream = "user_events"
+topic = "users"
+schema = "json"
+batch_length = 100
+
+[plugin_config]
+connection_string = "postgresql://user:pass@localhost:5432/database"
+mode = "polling"
+tables = ["users", "orders"]
+poll_interval = "1s"
+batch_size = 1000
+tracking_column = "id"
+initial_offset = "0"
+max_connections = 10
+snake_case_columns = false
+include_metadata = true
diff --git a/core/integration/tests/connectors/postgres/mod.rs
b/core/integration/tests/connectors/postgres/mod.rs
index 6cb4e679e..ee992b36b 100644
--- a/core/integration/tests/connectors/postgres/mod.rs
+++ b/core/integration/tests/connectors/postgres/mod.rs
@@ -17,6 +17,7 @@
mod postgres_sink;
mod postgres_source;
+mod postgres_source_cdc;
mod restart;
use crate::connectors::TestMessage;
diff --git a/core/integration/tests/connectors/postgres/postgres_source_cdc.rs
b/core/integration/tests/connectors/postgres/postgres_source_cdc.rs
new file mode 100644
index 000000000..b9a485e83
--- /dev/null
+++ b/core/integration/tests/connectors/postgres/postgres_source_cdc.rs
@@ -0,0 +1,584 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::{POLL_ATTEMPTS, POLL_INTERVAL_MS};
+use crate::connectors::create_test_messages;
+use crate::connectors::fixtures::{PostgresOps, PostgresSourceCdcFixture,
PostgresSourceOps};
+use iggy::prelude::IggyClient;
+use iggy_common::MessageClient;
+use iggy_common::{Consumer, Identifier, PollingStrategy};
+use iggy_connector_sdk::api::{ConnectorStatus, SourceInfoResponse};
+use integration::harness::seeds;
+use integration::iggy_harness;
+use reqwest::Client;
+use serde::Deserialize;
+use std::time::Duration;
+use tokio::time::sleep;
+
+const API_KEY: &str = "test-api-key";
+const SOURCE_KEY: &str = "postgres";
+const DEFAULT_SLOT: &str = "iggy_slot";
+
+#[derive(Debug, Deserialize)]
+struct CdcRecord {
+ table_name: String,
+ operation_type: String,
+ data: serde_json::Value,
+}
+
+async fn poll_cdc_records(
+ client: &IggyClient,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ consumer_id: &Identifier,
+ want: usize,
+) -> Vec<CdcRecord> {
+ let mut received = Vec::new();
+ for _ in 0..POLL_ATTEMPTS {
+ if let Ok(polled) = client
+ .poll_messages(
+ stream_id,
+ topic_id,
+ None,
+ &Consumer::new(consumer_id.clone()),
+ &PollingStrategy::next(),
+ 10,
+ true,
+ )
+ .await
+ {
+ for msg in polled.messages {
+ if let Ok(record) =
serde_json::from_slice::<CdcRecord>(&msg.payload) {
+ received.push(record);
+ }
+ }
+ if received.len() >= want {
+ break;
+ }
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
+ received
+}
+
+// End-to-end CDC coverage against a real wal_level=logical container:
+// INSERT, UPDATE, PK-changing UPDATE, DELETE, a rolled-back transaction
+// (must produce nothing), an untracked table (must be filtered out), a
+// 25-row sustained batch, and a final INSERT proving the slot/connector
+// are still healthy after all of the above.
+#[iggy_harness(
+ server(connectors_runtime(config_path =
"tests/connectors/postgres/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn cdc_source_captures_insert_update_delete(
+ harness: &TestHarness,
+ fixture: PostgresSourceCdcFixture,
+) {
+ const BATCH_SIZE: usize = 25;
+
+ let client = harness.root_client().await.unwrap();
+ let pool = fixture.create_pool().await.expect("Failed to create pool");
+ fixture.create_table(&pool).await;
+ fixture.create_untracked_table(&pool).await;
+
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let consumer_id: Identifier = "cdc_test_consumer".try_into().unwrap();
+
+ // INSERT: basic capture, full column fidelity
(id/name/count/amount/active).
+ let [msg] = create_test_messages(1).try_into().unwrap();
+ fixture
+ .insert_row(
+ &pool,
+ msg.id as i32,
+ &msg.name,
+ msg.count as i32,
+ msg.amount,
+ msg.active,
+ msg.timestamp,
+ )
+ .await;
+ let received = poll_cdc_records(&client, &stream_id, &topic_id,
&consumer_id, 1).await;
+ assert_eq!(received.len(), 1, "Expected 1 INSERT message");
+ assert_eq!(received[0].table_name, fixture.table_name());
+ assert_eq!(received[0].operation_type, "INSERT");
+ assert_eq!(received[0].data["id"], serde_json::json!(msg.id));
+ assert_eq!(received[0].data["name"], serde_json::json!(msg.name));
+ assert_eq!(received[0].data["count"], serde_json::json!(msg.count));
+ assert_eq!(received[0].data["amount"].as_f64(), Some(msg.amount));
+ assert_eq!(received[0].data["active"], serde_json::json!(msg.active));
+
+ // UPDATE: non-key column change.
+ let updated_count = msg.count + 1;
+ fixture
+ .update_row(&pool, msg.id as i32, updated_count as i32)
+ .await;
+ let received = poll_cdc_records(&client, &stream_id, &topic_id,
&consumer_id, 1).await;
+ assert_eq!(received.len(), 1, "Expected 1 UPDATE message");
+ assert_eq!(received[0].operation_type, "UPDATE");
+ assert_eq!(received[0].data["id"], serde_json::json!(msg.id));
+ assert_eq!(received[0].data["count"], serde_json::json!(updated_count));
+
+ // UPDATE: primary key change - test_decoding emits "old-key: ...
new-tuple: ...",
+ // the parser must report the new-tuple's id, not the old one.
+ let new_id = msg.id as i32 + 1000;
+ fixture
+ .update_primary_key(&pool, msg.id as i32, new_id)
+ .await;
+ let received = poll_cdc_records(&client, &stream_id, &topic_id,
&consumer_id, 1).await;
+ assert_eq!(received.len(), 1, "Expected 1 UPDATE message for PK change");
+ assert_eq!(received[0].operation_type, "UPDATE");
+ assert_eq!(
+ received[0].data["id"],
+ serde_json::json!(new_id),
+ "PK-changing UPDATE must report the new-tuple id, not the old-key one"
+ );
+ assert_eq!(received[0].data["name"], serde_json::json!(msg.name));
+
+ // DELETE: only replica-identity columns (here, just id) are present.
+ fixture.delete_row(&pool, new_id).await;
+ let received = poll_cdc_records(&client, &stream_id, &topic_id,
&consumer_id, 1).await;
+ assert_eq!(received.len(), 1, "Expected 1 DELETE message");
+ assert_eq!(received[0].operation_type, "DELETE");
+ assert_eq!(received[0].data["id"], serde_json::json!(new_id));
+
+ // Negative cases, verified together with the batch below: a rolled-back
+ // transaction must never reach test_decoding, and a table outside
+ // config.tables must be dropped by the client-side filter.
+ fixture
+ .insert_row_rolled_back(&pool, 9000, "should_not_appear")
+ .await;
+ fixture.insert_untracked_row(&pool, "not_configured").await;
+
+ // Sustained batch: BATCH_SIZE separate INSERT rows, asserting order and
+ // full data fidelity hold under a longer run, and that neither the
+ // rolled-back row nor the untracked-table row leaked into the output.
+ let batch = create_test_messages(BATCH_SIZE);
+ for row in &batch {
+ fixture
+ .insert_row(
+ &pool,
+ row.id as i32,
+ &row.name,
+ row.count as i32,
+ row.amount,
+ row.active,
+ row.timestamp,
+ )
+ .await;
+ }
+
+ let received = poll_cdc_records(&client, &stream_id, &topic_id,
&consumer_id, BATCH_SIZE).await;
+ assert_eq!(
+ received.len(),
+ BATCH_SIZE,
+ "Expected exactly the {BATCH_SIZE} batch of INSERT rows - the
rolled-back row and the \
+ untracked-table row must not appear"
+ );
+ for (i, record) in received.iter().enumerate() {
+ assert_eq!(record.operation_type, "INSERT");
+ assert_eq!(record.table_name, fixture.table_name());
+ assert_eq!(
+ record.data["id"],
+ serde_json::json!(batch[i].id),
+ "Out-of-order or missing row at index {i}"
+ );
+ assert_eq!(record.data["name"], serde_json::json!(batch[i].name));
+ }
+
+ // Final sanity check: the slot/connector must still be healthy after the
+ // earlier PK-changing UPDATE and the sustained batch, not left in some
+ // degraded state that silently swallows further changes.
+ let final_id = new_id + BATCH_SIZE as i32 + 1;
+ fixture
+ .insert_row(
+ &pool,
+ final_id,
+ "after_pk_change",
+ 1,
+ 1.0,
+ true,
+ msg.timestamp,
+ )
+ .await;
+ let received = poll_cdc_records(&client, &stream_id, &topic_id,
&consumer_id, 1).await;
+ assert_eq!(
+ received.len(),
+ 1,
+ "Expected the final INSERT after the PK change to still be captured"
+ );
+ assert_eq!(received[0].operation_type, "INSERT");
+ assert_eq!(received[0].data["id"], serde_json::json!(final_id));
+ assert_eq!(
+ received[0].data["name"],
+ serde_json::json!("after_pk_change")
+ );
+
+ pool.close().await;
+}
+
+async fn wait_for_source_status(
+ http: &Client,
+ api_url: &str,
+ expected: ConnectorStatus,
+) -> SourceInfoResponse {
+ for _ in 0..POLL_ATTEMPTS {
+ if let Ok(resp) = http
+ .get(format!("{api_url}/sources/{SOURCE_KEY}"))
+ .header("api-key", API_KEY)
+ .send()
+ .await
+ && let Ok(info) = resp.json::<SourceInfoResponse>().await
+ && info.status == expected
+ {
+ return info;
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
+ panic!("Source connector did not reach {expected:?} status in time");
+}
+
+async fn get_active_source_config(http: &Client, api_url: &str) ->
serde_json::Value {
+ http.get(format!("{api_url}/sources/{SOURCE_KEY}/configs/active"))
+ .header("api-key", API_KEY)
+ .send()
+ .await
+ .expect("Failed to fetch active source config")
+ .json()
+ .await
+ .expect("Failed to parse active source config")
+}
+
+// CreateSourceConfig has no key/version/active fields - strip what GET
+// .../configs/active added before POSTing the payload back as a new version.
+async fn push_source_config(http: &Client, api_url: &str, mut config:
serde_json::Value) -> u64 {
+ if let Some(obj) = config.as_object_mut() {
+ obj.remove("key");
+ obj.remove("version");
+ obj.remove("active");
+ }
+ let resp = http
+ .post(format!("{api_url}/sources/{SOURCE_KEY}/configs"))
+ .header("api-key", API_KEY)
+ .json(&config)
+ .send()
+ .await
+ .expect("Failed to call create-config endpoint");
+ assert!(
+ resp.status().is_success(),
+ "Failed to create source config version, got {}",
+ resp.status()
+ );
+ let created: serde_json::Value = resp.json().await.expect("Failed to parse
created config");
+ created["version"]
+ .as_u64()
+ .expect("Created config response missing version")
+}
+
+async fn activate_source_config(http: &Client, api_url: &str, version: u64) {
+ let resp = http
+ .put(format!("{api_url}/sources/{SOURCE_KEY}/configs/active"))
+ .header("api-key", API_KEY)
+ .json(&serde_json::json!({ "version": version }))
+ .send()
+ .await
+ .expect("Failed to call activate-config endpoint");
+ assert!(
+ resp.status().is_success(),
+ "Failed to activate source config version {version}, got {}",
+ resp.status()
+ );
+}
+
+async fn restart_source_expect(http: &Client, api_url: &str, should_succeed:
bool, context: &str) {
+ let resp = http
+ .post(format!("{api_url}/sources/{SOURCE_KEY}/restart"))
+ .header("api-key", API_KEY)
+ .send()
+ .await
+ .expect("Failed to call restart endpoint");
+ assert_eq!(
+ resp.status().is_success(),
+ should_succeed,
+ "{context}, got {}",
+ resp.status()
+ );
+}
+
+// The local config provider globs every *.toml under config_dir on startup
+// (see LocalConnectorsConfigProvider::init), and config_dir here is the real,
+// shared postgres_source crate directory - so any version this test pushes
+// must be deleted again, or it leaks into every other process that later
+// points at that same directory.
+async fn delete_source_config_version(http: &Client, api_url: &str, version:
u64) {
+ let resp = http
+ .delete(format!(
+ "{api_url}/sources/{SOURCE_KEY}/configs?version={version}"
+ ))
+ .header("api-key", API_KEY)
+ .send()
+ .await
+ .expect("Failed to call delete-config endpoint");
+ assert!(
+ resp.status().is_success(),
+ "Failed to delete source config version {version}, got {}",
+ resp.status()
+ );
+}
+
+// The connector calls pg_logical_slot_get_changes on a fixed poll interval and
+// briefly holds the slot active during each call. A drop landing in that
window
+// gets ERROR 55006 (slot is active for PID ...), so retry past transient hits
+// instead of dropping while the poller is guaranteed stopped.
+const PG_OBJECT_IN_USE: &str = "55006";
+
+async fn drop_replication_slot_retrying(pool: &sqlx::PgPool, slot: &str) {
+ for attempt in 0..POLL_ATTEMPTS {
+ match sqlx::query("SELECT pg_drop_replication_slot($1)")
+ .bind(slot)
+ .execute(pool)
+ .await
+ {
+ Ok(_) => return,
+ Err(sqlx::Error::Database(ref db_err))
+ if attempt + 1 < POLL_ATTEMPTS
+ && db_err.code().as_deref() == Some(PG_OBJECT_IN_USE) =>
+ {
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
+ Err(e) => panic!("Failed to drop replication slot {slot}: {e}"),
+ }
+ }
+}
+
+// Three restart scenarios against one container, run in sequence:
+// 1. Simulates upgrading from the old broken version, which left a slot
+// behind created with the pgoutput plugin. setup_cdc must refuse to
+// silently reuse a mismatched slot (which used to fail forever on
+// every poll instead) - it should fail loudly at startup, and recover
+// cleanly once the operator drops the bad slot and restarts.
+// 2. Invalid plugin_config fields (capture_operations, payload_format).
+// open() must reject each at restart instead of leaving a Running
+// connector that silently drops every change or emits wrong data - the
+// same silent-death shape as the slot mismatch above. Config is fixed
+// one field at a time until restart succeeds and CDC resumes.
+// 3. Changes written while the connector is down (the slot retains WAL
+// regardless of consumer state) - not the at-least-once crash window
+// where the slot has already been consumed but send/state-persist
+// hasn't happened yet. That gap remains open until the slot-peek/LSN
+// work lands.
+#[iggy_harness(
+ server(connectors_runtime(config_path =
"tests/connectors/postgres/source_cdc_restart.toml")),
+ seed = seeds::connector_stream
+)]
+async fn cdc_source_recovers_from_slot_mismatch_and_restart(
+ harness: &mut TestHarness,
+ fixture: PostgresSourceCdcFixture,
+) {
+ let client = harness.root_client().await.unwrap();
+ let pool = fixture.create_pool().await.expect("Failed to create pool");
+ fixture.create_table(&pool).await;
+
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let consumer_id: Identifier = "cdc_restart_consumer".try_into().unwrap();
+
+ let api_url = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
+ let http = Client::new();
+ wait_for_source_status(&http, &api_url, ConnectorStatus::Running).await;
+
+ drop_replication_slot_retrying(&pool, DEFAULT_SLOT).await;
+ sqlx::query("SELECT pg_create_logical_replication_slot($1, 'pgoutput')")
+ .bind(DEFAULT_SLOT)
+ .execute(&pool)
+ .await
+ .expect("Failed to create a wrong-plugin slot");
+
+ // Assert on the restart call's own HTTP status, not last_error - the
+ // still-running old connector keeps polling in the background and
+ // could race a transient poll failure into last_error first.
+ let restart_resp = http
+ .post(format!("{api_url}/sources/{SOURCE_KEY}/restart"))
+ .header("api-key", API_KEY)
+ .send()
+ .await
+ .expect("Failed to call restart endpoint");
+ assert!(
+ !restart_resp.status().is_success(),
+ "Restart with a mismatched slot should not report success, got {}",
+ restart_resp.status()
+ );
+
+ // Operator fix: drop the bad slot, restart so the connector recreates it.
+ sqlx::query("SELECT pg_drop_replication_slot($1)")
+ .bind(DEFAULT_SLOT)
+ .execute(&pool)
+ .await
+ .expect("Failed to drop the bad slot");
+
+ let restart_resp = http
+ .post(format!("{api_url}/sources/{SOURCE_KEY}/restart"))
+ .header("api-key", API_KEY)
+ .send()
+ .await
+ .expect("Failed to call restart endpoint");
+ assert!(
+ restart_resp.status().is_success(),
+ "Restart after fixing the slot should succeed, got {}",
+ restart_resp.status()
+ );
+ wait_for_source_status(&http, &api_url, ConnectorStatus::Running).await;
+
+ let [before] = create_test_messages(1).try_into().unwrap();
+ fixture
+ .insert_row(
+ &pool,
+ before.id as i32,
+ &before.name,
+ before.count as i32,
+ before.amount,
+ before.active,
+ before.timestamp,
+ )
+ .await;
+ let received = poll_cdc_records(&client, &stream_id, &topic_id,
&consumer_id, 1).await;
+ assert_eq!(
+ received.len(),
+ 1,
+ "Expected CDC to resume capturing changes after the slot was fixed"
+ );
+
+ // Scenario 2: invalid plugin_config fields must be rejected at restart,
+ // one field at a time, until the config is fully valid again. Every
+ // pushed version is deleted again at the end (see
+ // delete_source_config_version) so nothing outlives this test on disk.
+ let baseline_config = get_active_source_config(&http, &api_url).await;
+ let mut pushed_versions = Vec::new();
+
+ let mut bad_ops = baseline_config.clone();
+ bad_ops["plugin_config"]["capture_operations"] =
serde_json::json!(["INSRT"]);
+ let version = push_source_config(&http, &api_url, bad_ops).await;
+ pushed_versions.push(version);
+ activate_source_config(&http, &api_url, version).await;
+ restart_source_expect(
+ &http,
+ &api_url,
+ false,
+ "Restart with an invalid capture_operations entry should not report
success",
+ )
+ .await;
+
+ let mut bad_format = baseline_config.clone();
+ bad_format["plugin_config"]["payload_column"] = serde_json::json!("name");
+ bad_format["plugin_config"]["payload_format"] = serde_json::json!("btea");
+ let version = push_source_config(&http, &api_url, bad_format).await;
+ pushed_versions.push(version);
+ activate_source_config(&http, &api_url, version).await;
+ restart_source_expect(
+ &http,
+ &api_url,
+ false,
+ "Restart with an invalid payload_format should not report success",
+ )
+ .await;
+
+ let version = push_source_config(&http, &api_url, baseline_config).await;
+ pushed_versions.push(version);
+ activate_source_config(&http, &api_url, version).await;
+ restart_source_expect(
+ &http,
+ &api_url,
+ true,
+ "Restart with a valid config should succeed",
+ )
+ .await;
+ wait_for_source_status(&http, &api_url, ConnectorStatus::Running).await;
+
+ let reconfigured_id = before.id as i32 + 1;
+ fixture
+ .insert_row(
+ &pool,
+ reconfigured_id,
+ "reconfigured_row",
+ 1,
+ 1.0,
+ true,
+ before.timestamp,
+ )
+ .await;
+ let received = poll_cdc_records(&client, &stream_id, &topic_id,
&consumer_id, 1).await;
+ assert_eq!(
+ received.len(),
+ 1,
+ "Expected CDC to resume capturing changes once the config was fixed"
+ );
+
+ // Deleting the still-active version last falls back to version 0 (the
+ // original config.toml), restoring the pre-test state exactly.
+ for version in pushed_versions {
+ delete_source_config_version(&http, &api_url, version).await;
+ }
+
+ harness
+ .server_mut()
+ .stop_dependents()
+ .expect("Failed to stop connectors");
+
+ // The replication slot retains WAL for changes made while nothing is
+ // consuming - this row must still arrive once the connector restarts.
+ let after_id = reconfigured_id + 1;
+ fixture
+ .insert_row(
+ &pool,
+ after_id,
+ "written_while_down",
+ 1,
+ 1.0,
+ true,
+ before.timestamp,
+ )
+ .await;
+
+ harness
+ .server_mut()
+ .start_dependents()
+ .await
+ .expect("Failed to restart connectors");
+ let api_url = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
+ wait_for_source_status(&http, &api_url, ConnectorStatus::Running).await;
+
+ let received = poll_cdc_records(&client, &stream_id, &topic_id,
&consumer_id, 1).await;
+ assert_eq!(
+ received.len(),
+ 1,
+ "Expected the change written while the connector was down to arrive
after restart"
+ );
+ assert_eq!(received[0].operation_type, "INSERT");
+ assert_eq!(received[0].data["id"], serde_json::json!(after_id));
+ assert_eq!(
+ received[0].data["name"],
+ serde_json::json!("written_while_down")
+ );
+
+ pool.close().await;
+}
diff --git a/core/integration/tests/connectors/postgres/source_cdc_restart.toml
b/core/integration/tests/connectors/postgres/source_cdc_restart.toml
new file mode 100644
index 000000000..0cf9403ba
--- /dev/null
+++ b/core/integration/tests/connectors/postgres/source_cdc_restart.toml
@@ -0,0 +1,26 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Dedicated scratch config_dir (see cdc_restart_connectors/): this test posts
+# new plugin_config versions at runtime via the local config provider, which
+# globs every *.toml under config_dir on startup. Pointing that at the shared
+# ../connectors/sources/postgres_source directory (used by other postgres
+# source tests) would let a stray version leak into whichever of those tests
+# happens to boot its own runtime while this one is mid-run.
+[connectors]
+config_type = "local"
+config_dir = "tests/connectors/postgres/cdc_restart_connectors"
