postgres_cdc: multi-schema support

CHANGELOG.md

@@ -3,6 +3,48 @@ Changelog
 
 All notable changes to this project will be documented in this file.
 
+## 4.106.0 - 2026-08-20
+
+### Added
+
+- salesforce_cdc: Added decode-failure bounds and classification of schema-fetch errors (deterministic vs transient) to prevent infinite retry loops and livelocks, with terminal failures surfaced clearly to the health check. ([@squiidz](https://github.com/squiidz), [#4689](https://github.com/redpanda-data/connect/pull/4689))
+- mongodb, mongodb_cdc: Added AWS IAM authentication (`MONGODB-AWS`) for MongoDB Atlas to the `mongodb` input, output, processor and cache, and to the `mongodb_cdc` input, via a new `aws` configuration block supporting the ambient credential chain, static keys, and assume-role chaining. ([@squiidz](https://github.com/squiidz), [#4690](https://github.com/redpanda-data/connect/pull/4690))
+- mongodb_cdc: The input now checkpoints as soon as the initial snapshot completes and is fully acknowledged, so restarts resume the stream instead of re-running the snapshot; a stream position that can no longer be resumed from (for example one that has aged out of the oplog) is recovered by re-running the snapshot, bounded by a breaker that fails loudly instead of churning, with a new `on_unresumable_position` field controlling the lossy no-snapshot case (default `fail`) and a new `checkpoint_write_timeout` field bounding the detached checkpoint writes (the post-snapshot store and the recovery clear). ([@squiidz](https://github.com/squiidz), [#4690](https://github.com/redpanda-data/connect/pull/4690))
+
+### Fixed
+
+- aws_dynamodb_cdc: Fixed silent data loss in snapshot handling by gating checkpoint persistence on downstream acknowledgments, ensuring rejected batches are redelivered instead of skipped. ([@squiidz](https://github.com/squiidz), [#4687](https://github.com/redpanda-data/connect/pull/4687))
+- aws_dynamodb_cdc: Fixed stream rotation and restart scenarios where start_from: latest was incorrectly applied to child shards and checkpoint-less shards discovered after initial setup, causing silent loss of backlog. ([@squiidz](https://github.com/squiidz), [#4687](https://github.com/redpanda-data/connect/pull/4687))
+- cockroachdb_changefeed: Fixed unbounded silent data loss where transaction rows and backfill batches sharing timestamps could skip data on restart; now checkpoints only persist resolved timestamps to guarantee no loss. ([@squiidz](https://github.com/squiidz), [#4688](https://github.com/redpanda-data/connect/pull/4688))
+- salesforce_cdc: Fixed multiple silent-loss paths in Pub/Sub gRPC handling and ack functions: full buffer now applies backpressure instead of dropping events, schema/decode failures reconnect without losing batches, and nacks now pin checkpoints. ([@squiidz](https://github.com/squiidz), [#4689](https://github.com/redpanda-data/connect/pull/4689))
+- salesforce_cdc: Fixed off-by-one error in schema-retry budgeting and credential refresh in unanchored schema retries to prevent indefinite stalls under the default unlimited reconnect policy. ([@squiidz](https://github.com/squiidz), [#4689](https://github.com/redpanda-data/connect/pull/4689))
+
+### Changed
+
+- aws_dynamodb_cdc: Added auto_replay_nacks support to automatically retry transient downstream failures in-process, with nacks now advancing checkpoints when auto_replay_nacks is disabled. ([@squiidz](https://github.com/squiidz), [#4687](https://github.com/redpanda-data/connect/pull/4687))
+- cockroachdb_changefeed: Changed nack handling to advance cursors when auto_replay_nacks is disabled, treating it as an opt-in to drop rejected messages per the framework contract. ([@squiidz](https://github.com/squiidz), [#4688](https://github.com/redpanda-data/connect/pull/4688))
+- general: Updated CDC connector documentation across Microsoft SQL Server, MongoDB, and OracleDB with measured performance characteristics, scaling limitations, and configuration guidance based on real-world benchmarking. ([@prakhargarg105](https://github.com/prakhargarg105), [#4691](https://github.com/redpanda-data/connect/pull/4691))
+
+## 4.105.0 - 2026-08-13
+
+### Added
+
+- postgres_cdc: Added support for control signals in PostgreSQL CDC by detecting and forwarding rows inserted into a configurable signal table downstream like regular messages. ([@josephwoodward](https://github.com/josephwoodward), [#4637](https://github.com/redpanda-data/connect/pull/4637))
+- iceberg: Added an opt-in `merge_strategy: copy-on-write` for row-level `upsert`/`delete`, which materialises mutations by rewriting whole data files so the table only ever contains plain data files. This makes mutations readable by engine-backed catalogs that cannot handle merge-on-read equality deletes, such as the Databricks Unity Catalog and Snowflake. The default remains `merge-on-read`. ([@Jeffail](https://github.com/Jeffail), [#4666](https://github.com/redpanda-data/connect/pull/4666))
+- iceberg: Added a `commit.cleanup_on_failure` field (default `true`) to disable connector-side cleanup of files written by failed commits, as an escape hatch for incident recovery. Disabling it can only leak orphan files, which regular orphan-file maintenance reclaims. ([@Jeffail](https://github.com/Jeffail), [#4666](https://github.com/redpanda-data/connect/pull/4666))
+
+### Fixed
+
+- iceberg: Fixed a regression introduced in 4.99.0 where a commit that landed server-side but was reported as failed (ambiguous 5xx, timeout, lost acknowledgement, or an unclassified error) had its just-written parquet files deleted by the failure-path cleanup, leaving the table unreadable. Failure cleanup is now gated on a provable catalog rejection, and commits detected as landed are reported as success, which also prevents the duplicate rows that redelivery produced. ([@Jeffail](https://github.com/Jeffail), [#4666](https://github.com/redpanda-data/connect/pull/4666))
+- iceberg: Fixed no-timezone `timestamp` columns being written to parquet with `isAdjustedToUTC=true`, which is spec-incorrect and made them read back as `timestamptz`. New tables are written correctly; the encoding is pinned per table via a `redpanda-connect.timestamp-encoding` property so an existing table never changes or mixes encodings. ([@Jeffail](https://github.com/Jeffail), [#4666](https://github.com/redpanda-data/connect/pull/4666))
+- iceberg: Fixed commits failing against catalogs that prohibit clients setting particular table properties (for example the Databricks Unity Catalog and `schema.name-mapping.default`) by learning the prohibited keys from the catalog's rejection and stripping them from subsequent commits. ([@Jeffail](https://github.com/Jeffail), [#4666](https://github.com/redpanda-data/connect/pull/4666))
+- iceberg: Fixed several `identifier_fields` value shapes that silently matched no rows on `upsert`/`delete` — non-UTC `time` values, decimal floating-point ties, and `[]byte` values for string key columns — and fixed base64 mangling of binary and fixed column values during copy-on-write rewrites. All write paths now share a single value canonicaliser with the insert path. ([@Jeffail](https://github.com/Jeffail), [#4666](https://github.com/redpanda-data/connect/pull/4666))
+
+### Change
+
+- oracledb_cdc: Snapshot performance improvements by reusing seeded schema metadata [@josephwoodward](https://github.com/josephwoodward), [#4695](https://github.com/redpanda-data/connect/pull/4695))
+- iceberg: Merge-key input strictness now matches the insert path: string-typed values for integer and boolean key columns (for example `{"id": "42"}` against a `BIGINT` key) previously matched by accident and are now rejected with an actionable error, and nanosecond-precision timestamp `identifier_fields` are now rejected under `merge-on-read` as they already were under copy-on-write. A table whose `write.delete.mode` property is explicitly `merge-on-read` also now rejects `copy-on-write` mutations rather than silently overriding the property. ([@Jeffail](https://github.com/Jeffail), [#4666](https://github.com/redpanda-data/connect/pull/4666))
+
 ## 4.104.0 - 2026-08-06
 
 ### Fixed

cmd/tools/integration/packages.json

@@ -47,6 +47,7 @@
   {"path":"./internal/impl/otlp"},
   {"path":"./internal/impl/postgresql"},
   {"path":"./internal/impl/postgresql/pglogicalstream"},
+  {"path":"./internal/impl/postgresql/pglogicalstream/multischema"},
   {"path":"./internal/impl/pulsar","timeout":"10m"},
   {"path":"./internal/impl/qdrant"},
   {"path":"./internal/impl/questdb"},

docs/modules/components/pages/inputs/postgres_cdc.adoc

@@ -43,8 +43,10 @@ input:
     include_transaction_markers: false
     stream_snapshot: false
     snapshot_batch_size: 1000
-    schema: public # No default (required)
-    tables: [] # No default (required)
+    schema: public
+    schema_include: ""
+    schema_exclude: []
+    tables: []
     checkpoint_limit: 1024
     temporary_slot: false
     slot_name: my_test_slot # No default (required)
@@ -73,8 +75,10 @@ input:
     include_transaction_markers: false
     stream_snapshot: false
     snapshot_batch_size: 1000
-    schema: public # No default (required)
-    tables: [] # No default (required)
+    schema: public
+    schema_include: ""
+    schema_exclude: []
+    tables: []
     checkpoint_limit: 1024
     temporary_slot: false
     slot_name: my_test_slot # No default (required)
@@ -141,7 +145,7 @@ When set to true, empty messages with operation types BEGIN and COMMIT are gener
 
 === `stream_snapshot`
 
-When set to true, the plugin will first stream a snapshot of all existing data in the database before streaming changes. In order to use this the tables that are being snapshot MUST have a primary key set so that reading from the table can be parallelized. Note that this has no effect if `tables` is left empty, since the snapshot is only planned for tables listed there.
+When set to true, the plugin will first stream a snapshot of all existing data in the database before streaming changes. In order to use this the tables that are being snapshot MUST have a primary key set so that reading from the table can be parallelized. Note that this has no effect if `tables` is left empty and `schema_include` is NOT set, since in that case the snapshot is only planned for tables listed in `tables`. When `schema_include` IS set, leaving `tables` empty auto-discovers tables to snapshot instead - see `tables` below - and every discovered table must have a primary key.
 
 
 *Type*: `bool`
@@ -171,11 +175,12 @@ snapshot_batch_size: 10000
 
 === `schema`
 
-The PostgreSQL schema from which to replicate data.
+The PostgreSQL schema from which to replicate data. Ignored when `schema_include` is used instead.
 
 
 *Type*: `string`
 
+*Default*: `"public"`
 
 ```yml
 # Examples
@@ -185,15 +190,77 @@ schema: public
 schema: '"MyCaseSensitiveSchemaNeedingQuotes"'
 ```
 
+=== `schema_include`
+
+The PostgreSQL schema pattern to replicate data from. Accepts an exact schema name or a glob pattern using `*` as a wildcard to match multiple schemas.
+
+When a pattern is used, all schemas whose names match the pattern are replicated using a single replication slot and publication. This is useful for multi-tenant databases where each tenant has its own schema (e.g. `tenant_*` matches `tenant_foo`, `tenant_bar`, etc.).
+
+Double-quoted identifiers are treated as exact names and do not support wildcards.
+
+Schema pattern matching is re-evaluated every time the connector connects or reconnects - including the automatic reconnects that follow a transient replication failure - not just once at pipeline startup. This has two consequences that are easy to miss:
+
+- A schema created after the pipeline started that matches the pattern is picked up on the next reconnect and its tables are added to the publication - a warning naming the schema is logged when this happens. However, because the replication slot already exists by then, those tables are treated as already caught up, so with `stream_snapshot` enabled the rows that existed in that schema before it was picked up are never snapshotted - only changes made after the schema is picked up are streamed, and the missed rows cannot be recovered short of resetting the replication slot.
+- A schema that is dropped, renamed, or loses its `USAGE` grant between reconnects stops matching and its tables are removed from the publication on the next reconnect - a warning naming the schema is logged for all three cases, whether it dropped out via `USAGE` loss, a rename, or being dropped outright.
+
+If this pattern matches no schema in the database, startup fails with an error - this holds whether or not `tables` is set. See `tables` below for what happens when it's left empty.
+
+This pattern can contain characters that wouldn't be allowed in an unquoted schema name, because it's only ever compared against the real name of each schema in the database - it doesn't have to be a valid name itself. For example, a schema literally named `a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11` (which must have been created using double quotes, since hyphens aren't allowed in an unquoted `CREATE SCHEMA` statement) can still be matched using the unquoted pattern `a0eebc99-*`.
+
+When set, this field takes over schema resolution entirely; `schema` is ignored (with a warning logged if it was explicitly set) rather than combined with it.
+
+
+*Type*: `string`
+
+*Default*: `""`
+
+```yml
+# Examples
+
+schema_include: tenant_*
+
+schema_include: '*'
+
+schema_include: '"MyCaseSensitiveSchemaNeedingQuotes"'
+```
+
+=== `schema_exclude`
+
+A list of schema names or glob patterns to exclude from the schemas matched by `schema_include`. Only valid when `schema_include` is set.
+
+Each entry uses the same syntax as `schema_include`: an exact schema name, a glob pattern using `*` as a wildcard, or a double-quoted exact identifier for an exact, case-sensitive match.
+
+A schema that matches `schema_include` and also matches any entry in this list is excluded from replication. An entry that does not match any schema resolved by `schema_include` is silently ignored, so a typo here simply excludes nothing rather than failing startup.
+
+This exclusion is applied before `tables` is resolved, so it also takes effect when `tables` is left empty and tables are auto-discovered.
+
+
+*Type*: `array`
+
+*Default*: `[]`
+
+```yml
+# Examples
+
+schema_exclude:
+  - tenant_internal
+  - tenant_test_*
+```
+
 === `tables`
 
 A list of table names to include in the logical replication. Each table should be specified as a separate item.
 
-If left empty, the underlying PostgreSQL publication is created `FOR ALL TABLES`, which replicates every table in every schema of the database, ignoring `schema`. This also disables `stream_snapshot`, since the initial snapshot is only planned for tables listed here.
+When `schema_include` is set, this list is resolved against each matched schema independently: a table missing from some (but not all) of the matched schemas is skipped for those schemas only (with a warning logged), tolerating multi-tenant setups where a table hasn't been provisioned in every schema yet. A table that's missing from every matched schema, however, is treated as a configuration error (most likely a typo) and startup fails, naming the missing table.
+
+If left empty while `schema_include` is set, every base table in each matched (and un-excluded, see `schema_exclude`) schema is auto-discovered and published explicitly, instead of listing tables by hand - this is the expected way to replicate "every table" in a multi-tenant, schema-per-tenant setup without also picking up unrelated schemas. Startup fails if no matched schema contains any table. A partitioned table is auto-discovered as its individual leaf partitions rather than the table itself, so the `table` metadata on emitted messages will be the partition's name (e.g. `orders_2025`), not the parent's - list the parent explicitly in this field instead if you want it published as a single relation.
+
+If left empty while `schema_include` is NOT set, the underlying PostgreSQL publication is instead created `FOR ALL TABLES`, which replicates every table in every schema of the database, ignoring `schema`. This also disables `stream_snapshot`, since the initial snapshot is only planned for tables listed here.
 
 
 *Type*: `array`
 
+*Default*: `[]`
 
 ```yml
 # Examples
@@ -566,7 +633,7 @@ Optional external ID for the role assumption.
 
 === `signal_table_name`
 
-The name of the table used to send control signals to the connector, excluding the schema. The table must
+The name of the table used to send control signals to the connector, excluding the schema. Not supported when `schema_include` is set, since there is no single schema to anchor the signal table to. The table must
 exist in the schema configured via the `schema` field, and must not also appear in `tables`
 — the signal table is implicitly added to the publication and excluded from snapshot scans, so listing
 it in both places is rejected at startup. It must have at least these columns — startup validation checks

internal/impl/postgresql/bench/Taskfile.yaml

@@ -57,7 +57,7 @@ tasks:
   psql:truncate:
     desc: Truncate all benchmark tables
     cmds:
-      - psql {{.PG_DSN}} -c "TRUNCATE public.users, public.products, public.cart;"
+      - psql {{.PG_DSN}} -c "TRUNCATE public.users, public.products, public.cart, tenant_a.users, tenant_b.users, tenant_c.users;"
 
   psql:drop-slot:
     desc: Drop the bench_slot replication slot so benchmarks can be re-run

internal/impl/postgresql/bench/benchmark_config.yaml

@@ -6,6 +6,9 @@ input:
     dsn: ${PG_DSN:postgres://postgres:postgres@localhost:5432/testdb?sslmode=disable}
     stream_snapshot: true
     schema: public
+    # schema_include: "*"
+    # schema_exclude:
+    #   - tenant_c
     tables:
       - users
       - products

internal/impl/postgresql/bench/create.sql

@@ -1,9 +1,4 @@
 -- PostgreSQL Benchmark Setup Script
-CREATE TABLE IF NOT EXISTS public.rpcn_signal_table (
-    id SERIAL PRIMARY KEY,
-    type VARCHAR(32),
-    data TEXT
-);
 
 CREATE TABLE IF NOT EXISTS public.users (
     id            SERIAL PRIMARY KEY,
@@ -42,3 +37,54 @@ CREATE TABLE IF NOT EXISTS public.cart (
     info       TEXT          NOT NULL
 );
 ALTER TABLE public.cart REPLICA IDENTITY FULL;
+
+-- schema 1
+CREATE SCHEMA IF NOT EXISTS tenant_a;
+CREATE TABLE IF NOT EXISTS tenant_a.users (
+    id            SERIAL PRIMARY KEY,
+    name          VARCHAR(100)    NOT NULL,
+    surname       VARCHAR(100)    NOT NULL,
+    about         TEXT            NOT NULL,
+    email         VARCHAR(255)    NOT NULL,
+    date_of_birth DATE,
+    join_date     DATE,
+    created_at    TIMESTAMPTZ     NOT NULL DEFAULT NOW(),
+    is_active     BOOLEAN         NOT NULL DEFAULT TRUE,
+    login_count   INT             NOT NULL DEFAULT 0,
+    balance       DECIMAL(10,2)   NOT NULL DEFAULT 0.00
+);
+ALTER TABLE tenant_a.users REPLICA IDENTITY FULL;
+
+-- schema 2
+CREATE SCHEMA IF NOT EXISTS tenant_b;
+CREATE TABLE IF NOT EXISTS tenant_b.users (
+    id            SERIAL PRIMARY KEY,
+    name          VARCHAR(100)    NOT NULL,
+    surname       VARCHAR(100)    NOT NULL,
+    about         TEXT            NOT NULL,
+    email         VARCHAR(255)    NOT NULL,
+    date_of_birth DATE,
+    join_date     DATE,
+    created_at    TIMESTAMPTZ     NOT NULL DEFAULT NOW(),
+    is_active     BOOLEAN         NOT NULL DEFAULT TRUE,
+    login_count   INT             NOT NULL DEFAULT 0,
+    balance       DECIMAL(10,2)   NOT NULL DEFAULT 0.00
+);
+ALTER TABLE tenant_b.users REPLICA IDENTITY FULL;
+
+-- schema 3
+CREATE SCHEMA IF NOT EXISTS tenant_c;
+CREATE TABLE IF NOT EXISTS tenant_c.users (
+    id            SERIAL PRIMARY KEY,
+    name          VARCHAR(100)    NOT NULL,
+    surname       VARCHAR(100)    NOT NULL,
+    about         TEXT            NOT NULL,
+    email         VARCHAR(255)    NOT NULL,
+    date_of_birth DATE,
+    join_date     DATE,
+    created_at    TIMESTAMPTZ     NOT NULL DEFAULT NOW(),
+    is_active     BOOLEAN         NOT NULL DEFAULT TRUE,
+    login_count   INT             NOT NULL DEFAULT 0,
+    balance       DECIMAL(10,2)   NOT NULL DEFAULT 0.00
+);
+ALTER TABLE tenant_c.users REPLICA IDENTITY FULL;