diff --git a/CHANGELOG.md b/CHANGELOG.md index f0a20d2a9d..6bb7e3a8e3 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/cmd/tools/integration/packages.json b/cmd/tools/integration/packages.json index 6bcdf7deb0..bcc6dd0812 100644 --- a/cmd/tools/integration/packages.json +++ b/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"}, diff --git a/docs/modules/components/pages/inputs/postgres_cdc.adoc b/docs/modules/components/pages/inputs/postgres_cdc.adoc index 3ae65ba78f..2ca4d50bac 100644 --- a/docs/modules/components/pages/inputs/postgres_cdc.adoc +++ b/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 diff --git a/internal/impl/postgresql/bench/Taskfile.yaml b/internal/impl/postgresql/bench/Taskfile.yaml index d8990ea492..438704e272 100644 --- a/internal/impl/postgresql/bench/Taskfile.yaml +++ b/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 diff --git a/internal/impl/postgresql/bench/benchmark_config.yaml b/internal/impl/postgresql/bench/benchmark_config.yaml index a673fc8503..964b0dbd57 100644 --- a/internal/impl/postgresql/bench/benchmark_config.yaml +++ b/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 diff --git a/internal/impl/postgresql/bench/create.sql b/internal/impl/postgresql/bench/create.sql index d4b8d14d77..2be3bc4842 100644 --- a/internal/impl/postgresql/bench/create.sql +++ b/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; diff --git a/internal/impl/postgresql/bench/users.sql b/internal/impl/postgresql/bench/users.sql index 70898f7cae..607da4aa2e 100644 --- a/internal/impl/postgresql/bench/users.sql +++ b/internal/impl/postgresql/bench/users.sql @@ -1,16 +1,27 @@ -- PostgreSQL Benchmark - Users Data (150K rows, ~500KB per row) -- Prerequisites: Run create.sql first -INSERT INTO public.users (name, surname, about, email, date_of_birth, join_date, created_at, is_active, login_count, balance) -SELECT - 'user-' || n, - 'surname-' || n, - repeat('This is about user ' || n || '. ', 25000), - 'user' || n || '@example.com', - NOW() - (n % 10000 || ' days')::interval, - NOW(), - NOW(), - (n % 2 = 0), - n % 100, - ((n % 1000) + (n % 100) / 100.0)::decimal(10,2) -FROM generate_series(1, 150000) AS n; +DO $$ +DECLARE + tbl text; + num_rows int := 150000; +BEGIN + FOREACH tbl IN ARRAY ARRAY['public.users', 'tenant_a.users', 'tenant_b.users'] + LOOP + EXECUTE format($fmt$ + INSERT INTO %s (name, surname, about, email, date_of_birth, join_date, created_at, is_active, login_count, balance) + SELECT + 'user-' || n, + 'surname-' || n, + repeat('This is about user ' || n || '. ', 25000), + 'user' || n || '@example.com', + NOW() - (n %% 10000 || ' days')::interval, + NOW(), + NOW(), + (n %% 2 = 0), + n %% 100, + ((n %% 1000) + (n %% 100) / 100.0)::decimal(10,2) + FROM generate_series(1, %s) AS n + $fmt$, tbl, num_rows); + END LOOP; +END $$; diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go index 8ff0e62374..6089fe4be9 100644 --- a/internal/impl/postgresql/input_pg_stream.go +++ b/internal/impl/postgresql/input_pg_stream.go @@ -15,6 +15,7 @@ import ( "errors" "fmt" "strconv" + "strings" "sync" "time" @@ -26,6 +27,7 @@ import ( "github.com/redpanda-data/connect/v4/internal/asyncroutine" "github.com/redpanda-data/connect/v4/internal/impl/postgresql/pglogicalstream" + "github.com/redpanda-data/connect/v4/internal/impl/postgresql/pglogicalstream/multischema" "github.com/redpanda-data/connect/v4/internal/impl/postgresql/pglogicalstream/sanitize" "github.com/redpanda-data/connect/v4/internal/license" ) @@ -37,6 +39,8 @@ const ( fieldSnapshotMemSafetyFactor = "snapshot_memory_safety_factor" fieldSnapshotBatchSize = "snapshot_batch_size" fieldSchema = "schema" + fieldSchemaInclude = "schema_include" + fieldSchemaExclude = "schema_exclude" fieldTables = "tables" fieldCheckpointLimit = "checkpoint_limit" fieldTemporarySlot = "temporary_slot" @@ -52,6 +56,8 @@ const ( // FieldAWSIAMAuthEnabled enabled field. FieldAWSIAMAuthEnabled = "enabled" shutdownTimeout = 5 * time.Second + + defaultSchema = "public" ) func notImportedAWSOptFn(_ context.Context, awsConf *service.ParsedConfig, _ *pgconn.Config, _ *service.Logger) (TokenBuilder, error) { @@ -85,6 +91,7 @@ Additionally, if ` + "`" + fieldStreamSnapshot + "`" + ` is set to true, then th This input adds the following metadata fields to each message: - table: Name of the table that the message originated from +- database_schema: The database schema for the table where the message originates from (e.g. "public", "tenant_foo"). Useful for per-schema routing when using schema patterns. - operation: Type of operation that generated the message: "read", "insert", "update", or "delete". "read" is from messages that are read in the initial snapshot phase. This will also be "begin" and "commit" if ` + "`" + fieldIncludeTxnMarkers + "`" + ` is enabled - lsn: the log sequence number in postgres - schema: The table schema in benthos common schema format, compatible with processors like parquet_encode @@ -100,7 +107,7 @@ This input adds the following metadata fields to each message: ShortDescription("Emit empty BEGIN and COMMIT messages at the start and end of each transaction."). Default(false)). Field(service.NewBoolField(fieldStreamSnapshot). - Description("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 `" + fieldTables + "` is left empty, since the snapshot is only planned for tables listed there."). + Description("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 `" + fieldTables + "` is left empty and `" + fieldSchemaInclude + "` is NOT set, since in that case the snapshot is only planned for tables listed in `" + fieldTables + "`. When `" + fieldSchemaInclude + "` IS set, leaving `" + fieldTables + "` empty auto-discovers tables to snapshot instead - see `" + fieldTables + "` below - and every discovered table must have a primary key."). ShortDescription("Stream a snapshot of all existing data before streaming changes. Snapshot tables must have a primary key."). Example(true). Default(false)). @@ -115,14 +122,54 @@ This input adds the following metadata fields to each message: Example(10000). Default(1000)). Field(service.NewStringField(fieldSchema). - Description("The PostgreSQL schema from which to replicate data."). - Examples("public", `"MyCaseSensitiveSchemaNeedingQuotes"`), + Description("The PostgreSQL schema from which to replicate data. Ignored when `"+fieldSchemaInclude+"` is used instead."). + Examples("public", `"MyCaseSensitiveSchemaNeedingQuotes"`). + Optional(). + Default(defaultSchema), + ). + Field(service.NewStringField(fieldSchemaInclude). + Description(`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 `+"`"+fieldStreamSnapshot+"`"+` 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 `+"`"+fieldTables+"`"+` is set. See `+"`"+fieldTables+"`"+` 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; `+"`"+fieldSchema+"`"+` is ignored (with a warning logged if it was explicitly set) rather than combined with it.`). + Examples("tenant_*", "*", `"MyCaseSensitiveSchemaNeedingQuotes"`). + Optional(). + Default(""), + ). + Field(service.NewStringListField(fieldSchemaExclude). + Description(`A list of schema names or glob patterns to exclude from the schemas matched by ` + "`" + fieldSchemaInclude + "`" + `. Only valid when ` + "`" + fieldSchemaInclude + "`" + ` is set. + +Each entry uses the same syntax as ` + "`" + fieldSchemaInclude + "`" + `: 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 ` + "`" + fieldSchemaInclude + "`" + ` and also matches any entry in this list is excluded from replication. An entry that does not match any schema resolved by ` + "`" + fieldSchemaInclude + "`" + ` is silently ignored, so a typo here simply excludes nothing rather than failing startup. + +This exclusion is applied before ` + "`" + fieldTables + "`" + ` is resolved, so it also takes effect when ` + "`" + fieldTables + "`" + ` is left empty and tables are auto-discovered.`). + Examples([]string{"tenant_internal", "tenant_test_*"}). + Default([]string{}), ). Field(service.NewStringListField(fieldTables). Description(`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 ` + "`" + fieldSchema + "`" + `. This also disables ` + "`" + fieldStreamSnapshot + "`" + `, since the initial snapshot is only planned for tables listed here.`). - Example([]string{"my_table_1", `"MyCaseSensitiveTableNeedingQuotes"`})). +When ` + "`" + fieldSchemaInclude + "`" + ` 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 ` + "`" + fieldSchemaInclude + "`" + ` is set, every base table in each matched (and un-excluded, see ` + "`" + fieldSchemaExclude + "`" + `) 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 ` + "`" + fieldSchemaInclude + "`" + ` is NOT set, the underlying PostgreSQL publication is instead created ` + "`FOR ALL TABLES`" + `, which replicates every table in every schema of the database, ignoring ` + "`" + fieldSchema + "`" + `. This also disables ` + "`" + fieldStreamSnapshot + "`" + `, since the initial snapshot is only planned for tables listed here.`). + Example([]string{"my_table_1", `"MyCaseSensitiveTableNeedingQuotes"`}). + Optional(). + Default([]string{})). Field(service.NewIntField(fieldCheckpointLimit). Description("The maximum number of messages that can be processed at a given time. Increasing this limit enables parallel processing and batching at the output level. Any given LSN will not be acknowledged unless all messages under that offset are delivered in order to preserve at least once delivery guarantees."). ShortDescription("The maximum number of messages that can be processed at a given time."). @@ -211,7 +258,7 @@ This connector uses the naming pattern ` + "`pglog_stream_ 0 { + if schemaInclude == "" { + return nil, errors.New("schema_exclude requires schema_include to be set") + } + for i, pattern := range schemaExclude { + if err = validateSchemaPattern(pattern); err != nil { + return nil, fmt.Errorf("invalid schema_exclude entry %q: %w", pattern, err) + } + // Normalize unquoted patterns to lower-case, mirroring schema_include + // above: PostgreSQL folds unquoted identifiers at creation time, so + // TENANT_TEST and tenant_test resolve to the same schema. + if !strings.HasPrefix(pattern, `"`) { + schemaExclude[i] = strings.ToLower(pattern) + } + } + } + if tables, err = conf.FieldStringList(fieldTables); err != nil { return nil, err } @@ -357,6 +445,9 @@ func newPgStreamInput(conf *service.ParsedConfig, mgr *service.Resources) (s ser } if signalTableName != "" { + if schemaInclude != "" { + return nil, fmt.Errorf("%s is not supported when %s is set", fieldSignalTableName, fieldSchemaInclude) + } normalizedSignalTable, err := sanitize.NormalizePostgresIdentifier(signalTableName) if err != nil { return nil, fmt.Errorf("invalid %s %q: %w", fieldSignalTableName, signalTableName, err) @@ -403,12 +494,18 @@ func newPgStreamInput(conf *service.ParsedConfig, mgr *service.Resources) (s ser snapshotMetrics := mgr.Metrics().NewGauge("postgres_snapshot_progress", "table") replicationLag := mgr.Metrics().NewGauge("postgres_replication_lag_bytes") + var schemaResolver *multischema.Resolver + if schemaInclude != "" { + schemaResolver = multischema.NewResolver(schemaInclude, schemaExclude) + } + i := &pgStreamInput{ streamConfig: &pglogicalstream.Config{ DBConfig: pgConnConfig, TLSConfig: pgConnConfig.TLSConfig, DBRawDSN: dsn, DBSchema: schema, + SchemaResolver: schemaResolver, DBTables: tables, RefreshAuthToken: iamAuthTokenBuilder, @@ -453,6 +550,26 @@ func newPgStreamInput(conf *service.ParsedConfig, mgr *service.Resources) (s ser return conf.WrapBatchInputExtractTracingSpanMapping("postgres_cdc", r) } +// validateSchemaPattern validates a schema name or glob pattern. +func validateSchemaPattern(s string) error { + if s == "" { + return errors.New("schema cannot be empty") + } + if strings.HasPrefix(s, `"`) { + if _, err := sanitize.UnquotePostgresIdentifier(s); err != nil { + return fmt.Errorf("invalid quoted schema identifier: %w", err) + } + if strings.ContainsRune(s, '*') { + return errors.New("wildcard '*' is not allowed inside a quoted schema identifier") + } + return nil + } + if strings.ContainsRune(s, '"') { + return fmt.Errorf("unquoted schema pattern %q must not contain '\"'", s) + } + return nil +} + // validateSimpleString ensures we aren't vuln to SQL injection. func validateSimpleString(s string) error { for _, b := range []byte(s) { @@ -615,6 +732,7 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher } batchMsg := service.NewMessage(mb) batchMsg.MetaSet("table", msg.Table) + batchMsg.MetaSet("database_schema", msg.Schema) batchMsg.MetaSet("operation", string(msg.Operation)) if msg.LSN != nil { batchMsg.MetaSet("lsn", *msg.LSN) diff --git a/internal/impl/postgresql/input_pg_stream_test.go b/internal/impl/postgresql/input_pg_stream_test.go index aaf127e82d..56d75118f9 100644 --- a/internal/impl/postgresql/input_pg_stream_test.go +++ b/internal/impl/postgresql/input_pg_stream_test.go @@ -9,15 +9,259 @@ package pgstream import ( + "fmt" + "log/slog" + "strings" "testing" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/redpanda-data/benthos/v4/public/service" + "github.com/redpanda-data/connect/v4/internal/impl/postgresql/pgtest" "github.com/redpanda-data/connect/v4/internal/license" ) +// TestSchemaDefault verifies that the schema field defaults to "public" when +// left unset, matching pre-multi-schema behaviour. +func TestSchemaDefault(t *testing.T) { + yaml := ` +dsn: postgres://testuser:testpass@localhost:5432/testdb?sslmode=disable +slot_name: test_slot +tables: + - events +` + _, err := parsePgStreamInput(t, yaml) + require.NoError(t, err) +} + +// TestSchemaIncludeValidation verifies that the schema_include field is +// validated during config parsing, before any network I/O is attempted. +// Success is asserted via newPgStreamInput returning no error - the +// constructor doesn't dial the database, so a valid pattern implies +// validation passed. +func TestSchemaIncludeValidation(t *testing.T) { + tests := []struct { + pattern string + errContains string + }{ + {"tenant_*", ""}, + {"*", ""}, + {`"MySchema"`, ""}, + // Regression test: validateSchemaPattern must accept the same unicode + // letters/digits that sanitize.NormalizePostgresIdentifier accepts for + // unquoted identifiers (e.g. "münchen"), not just ASCII. + {"münchen", ""}, + {"tenant_ü*", ""}, + // Regression test: len("") == 2 used to pass the old `len(s) < 2` guard. + // Fixed to `len(s) < 3`. + {`""`, "invalid quoted schema identifier"}, + // Regression test: a leading digit is not an identifier-syntax + // violation here - the pattern is compared via ILIKE, never spliced + // into an identifier position - so "1abc" must be as valid as any + // other unquoted pattern. See the "9c0b4ef8-*" case below for the + // motivating real-world scenario (a UUID-suffixed tenant schema). + {"1abc", ""}, + {`"unclosed`, "invalid quoted schema identifier"}, + // Regression test: an unquoted pattern is matched against stored + // schema names, not parsed as an identifier, so hyphens (invalid in + // unquoted Postgres identifiers) must still be accepted - e.g. to + // match a UUID-suffixed tenant schema that had to be created quoted. + {"schema-name", ""}, + {"a0eebc99-*", ""}, + // Regression test: most UUIDs begin with a hex digit, so a tenant + // schema named e.g. "9c0b4ef8-bb6d-6bb9-bd38-0a11a0eebc99" (created + // quoted, per the a0eebc99-* case above) must be matchable by an + // unquoted glob starting with a digit - the pattern is compared via + // ILIKE, never spliced into an identifier position, so there's no + // syntactic reason to require a letter/underscore/'*' lead-in. + {"9c0b4ef8-*", ""}, + {`"quoted*"`, "wildcard"}, + {`a"b`, `must not contain '"'`}, + } + for _, tt := range tests { + t.Run(tt.pattern, func(t *testing.T) { + // Single-quoted so the pattern (which may itself contain double + // quotes, e.g. `"MySchema"`) reaches validateSchemaPattern verbatim. + yaml := fmt.Sprintf(` +dsn: postgres://testuser:testpass@localhost:5432/testdb?sslmode=disable +schema_include: '%s' +slot_name: test_slot +tables: + - events +`, tt.pattern) + + _, err := parsePgStreamInput(t, yaml) + if tt.errContains != "" { + require.Error(t, err) + assert.Contains(t, err.Error(), tt.errContains) + return + } + require.NoError(t, err) + }) + } +} + +func TestSchemaIgnoredWhenSchemaIncludeSet(t *testing.T) { + yaml := ` +dsn: postgres://testuser:testpass@localhost:5432/testdb?sslmode=disable +schema: tenant_foo +schema_include: 'tenant_*' +slot_name: test_slot +tables: + - events +` + conf, err := newPostgresCDCConfig().ParseYAML(yaml, nil) + require.NoError(t, err) + + logs := pgtest.NewTestLogCapture() + mgr := service.MockResources(service.MockResourcesOptUseLogger(service.NewLoggerFromSlog(slog.New(logs)))) + license.InjectTestService(mgr) + + _, err = newPgStreamInput(conf, mgr) + require.NoError(t, err) + + var sawWarning bool + for _, m := range logs.Messages() { + if strings.Contains(m, fieldSchema) && strings.Contains(m, fieldSchemaInclude) { + sawWarning = true + } + } + assert.True(t, sawWarning, "expected a warning that %s is ignored in favor of %s, got: %v", fieldSchema, fieldSchemaInclude, logs.Messages()) +} + +// TestSchemaIncludeWithDefaultSchemaSucceeds verifies that setting +// schema_include while leaving schema untouched (at its "public" default) is +// allowed and logs no warning about schema being ignored. +func TestSchemaIncludeWithDefaultSchemaSucceeds(t *testing.T) { + yaml := ` +dsn: postgres://testuser:testpass@localhost:5432/testdb?sslmode=disable +schema_include: 'tenant_*' +slot_name: test_slot +tables: + - events +` + conf, err := newPostgresCDCConfig().ParseYAML(yaml, nil) + require.NoError(t, err) + + logs := pgtest.NewTestLogCapture() + mgr := service.MockResources(service.MockResourcesOptUseLogger(service.NewLoggerFromSlog(slog.New(logs)))) + license.InjectTestService(mgr) + + _, err = newPgStreamInput(conf, mgr) + require.NoError(t, err) + + for _, m := range logs.Messages() { + assert.NotContains(t, m, fieldSchema+" is set", "schema was left at its default, so no warning about it should be logged, got: %v", m) + } +} + +// TestSchemaExplicitlySetToDefaultAlongsideSchemaIncludeDoesNotWarn documents +// a known, accepted gap: since schema's value is compared against its own +// default ("public") to decide whether to warn, a user who explicitly writes +// schema: public alongside schema_include is indistinguishable from one who +// left schema unset, so no warning is logged either way. This is considered +// acceptable because "public" is inert here regardless of whether it came +// from the user or the default - unlike any other value, which does warn +// (see TestSchemaIgnoredWhenSchemaIncludeSet). +func TestSchemaExplicitlySetToDefaultAlongsideSchemaIncludeDoesNotWarn(t *testing.T) { + yaml := ` +dsn: postgres://testuser:testpass@localhost:5432/testdb?sslmode=disable +schema: public +schema_include: 'tenant_*' +slot_name: test_slot +tables: + - events +` + conf, err := newPostgresCDCConfig().ParseYAML(yaml, nil) + require.NoError(t, err) + + logs := pgtest.NewTestLogCapture() + mgr := service.MockResources(service.MockResourcesOptUseLogger(service.NewLoggerFromSlog(slog.New(logs)))) + license.InjectTestService(mgr) + + _, err = newPgStreamInput(conf, mgr) + require.NoError(t, err) + + for _, m := range logs.Messages() { + assert.NotContains(t, m, fieldSchema+" is set", "explicit schema: public is indistinguishable from the default, so no warning is logged, got: %v", m) + } +} + +// TestSchemaExcludeValidation verifies that each schema_exclude entry is +// validated with the same rules as schema_include - validateSchemaPattern is +// reused rather than re-derived, so this exercises the same error cases +// TestSchemaIncludeValidation covers, just reached through a different field. +func TestSchemaExcludeValidation(t *testing.T) { + tests := []struct { + pattern string + errContains string + }{ + {"tenant_test", ""}, + {"tenant_test_*", ""}, + {`"MySchema"`, ""}, + {"1abc", ""}, + {`"unclosed`, "invalid quoted schema identifier"}, + {"schema-name", ""}, + {`"quoted*"`, "wildcard"}, + } + for _, tt := range tests { + t.Run(tt.pattern, func(t *testing.T) { + // Single-quoted so the pattern (which may itself contain double + // quotes, e.g. `"MySchema"`) reaches validateSchemaPattern verbatim. + yaml := fmt.Sprintf(` +dsn: postgres://testuser:testpass@localhost:5432/testdb?sslmode=disable +schema_include: 'tenant_*' +schema_exclude: ['%s'] +slot_name: test_slot +tables: + - events +`, tt.pattern) + + _, err := parsePgStreamInput(t, yaml) + if tt.errContains != "" { + require.Error(t, err) + assert.Contains(t, err.Error(), tt.errContains) + return + } + require.NoError(t, err) + }) + } +} + +// TestSchemaExcludeRequiresSchemaInclude verifies that schema_exclude is +// rejected at config-parse time when schema_include is left unset. Both +// single-exact-schema mode and FOR ALL TABLES mode (empty tables) have no +// well-defined candidate set to exclude from, so this is a hard error rather +// than a silent no-op. +func TestSchemaExcludeRequiresSchemaInclude(t *testing.T) { + yaml := ` +dsn: postgres://testuser:testpass@localhost:5432/testdb?sslmode=disable +schema_exclude: [tenant_test] +slot_name: test_slot +tables: + - events +` + _, err := parsePgStreamInput(t, yaml) + require.Error(t, err) + assert.Contains(t, err.Error(), "schema_exclude requires schema_include to be set") +} + +// TestSchemaExcludeEmptyWithoutSchemaIncludeSucceeds verifies that leaving +// schema_exclude at its default empty list does not trip the +// requires-schema_include check, since there's nothing to exclude. +func TestSchemaExcludeEmptyWithoutSchemaIncludeSucceeds(t *testing.T) { + yaml := ` +dsn: postgres://testuser:testpass@localhost:5432/testdb?sslmode=disable +slot_name: test_slot +tables: + - events +` + _, err := parsePgStreamInput(t, yaml) + require.NoError(t, err) +} + func TestNewPgStreamInputSignalTableName(t *testing.T) { env := service.NewEnvironment() spec := newPostgresCDCConfig() @@ -93,3 +337,14 @@ signal_table_name: rpcn_signal_table }) } } + +func parsePgStreamInput(t *testing.T, yaml string) (service.BatchInput, error) { + t.Helper() + conf, err := newPostgresCDCConfig().ParseYAML(yaml, nil) + require.NoError(t, err) + + mgr := service.MockResources() + license.InjectTestService(mgr) + + return newPgStreamInput(conf, mgr) +} diff --git a/internal/impl/postgresql/integration_test.go b/internal/impl/postgresql/integration_test.go index bcb67946a1..66d8694d45 100644 --- a/internal/impl/postgresql/integration_test.go +++ b/internal/impl/postgresql/integration_test.go @@ -1235,38 +1235,44 @@ postgres_cdc: outBatches, []any{ map[string]any{ - "operation": "read", - "table": "FlightsCompositePK", + "operation": "read", + "table": "FlightsCompositePK", + "database_schema": "public", }, map[string]any{ - "operation": "read", - "table": "flights", + "operation": "read", + "table": "flights", + "database_schema": "public", }, map[string]any{ - "operation": "insert", - "table": "FlightsCompositePK", - "lsn": "XXX/XXX", - "commit_ts_ms": "SET", + "operation": "insert", + "table": "FlightsCompositePK", + "lsn": "XXX/XXX", + "commit_ts_ms": "SET", + "database_schema": "public", }, map[string]any{ - "operation": "insert", - "table": "flights", - "lsn": "XXX/XXX", - "commit_ts_ms": "SET", + "operation": "insert", + "table": "flights", + "lsn": "XXX/XXX", + "commit_ts_ms": "SET", + "database_schema": "public", }, map[string]any{ - "operation": "update", - "table": "flights", - "lsn": "XXX/XXX", - "commit_ts_ms": "SET", - "before": "SET", + "operation": "update", + "table": "flights", + "lsn": "XXX/XXX", + "commit_ts_ms": "SET", + "before": "SET", + "database_schema": "public", }, map[string]any{ - "operation": "delete", - "table": "flights", - "lsn": "XXX/XXX", - "commit_ts_ms": "SET", - "before": "SET", + "operation": "delete", + "table": "flights", + "lsn": "XXX/XXX", + "commit_ts_ms": "SET", + "before": "SET", + "database_schema": "public", }, }, ) @@ -1614,3 +1620,996 @@ postgres_cdc: } assert.Equal(t, "STRING", byName["extra"], "new 'extra' column should have type STRING") } + +func TestIntegrationMultiSchemaSnapshotAndCDC(t *testing.T) { + integration.CheckSkip(t) + databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16") + require.NoError(t, err) + + // Two tenant schemas with the same table name, replicated on a single slot. + for _, schema := range []string{"tenant_a", "tenant_b"} { + _, err = db.Exec(fmt.Sprintf("CREATE SCHEMA %s", schema)) + require.NoError(t, err) + _, err = db.Exec(fmt.Sprintf( + "CREATE TABLE %s.events (id SERIAL PRIMARY KEY, name TEXT)", schema)) + require.NoError(t, err) + } + + // Pre-load snapshot data: 2 rows in tenant_a, 1 in tenant_b. + _, err = db.Exec("INSERT INTO tenant_a.events (name) VALUES ('alice'), ('bob')") + require.NoError(t, err) + _, err = db.Exec("INSERT INTO tenant_b.events (name) VALUES ('carol')") + require.NoError(t, err) + + type msgMeta struct { + dbSchema string + table string + operation string + lsn string + } + + var ( + mu sync.Mutex + collected []msgMeta + ) + + tmpl := fmt.Sprintf(` +postgres_cdc: + dsn: %s + slot_name: multi_schema_test_slot + stream_snapshot: true + schema_include: tenant_* + tables: + - events +`, databaseURL) + + sb := service.NewStreamBuilder() + require.NoError(t, sb.SetLoggerYAML(`level: WARN`)) + require.NoError(t, sb.AddInputYAML(tmpl)) + require.NoError(t, sb.AddBatchConsumerFunc(func(_ context.Context, batch service.MessageBatch) error { + mu.Lock() + defer mu.Unlock() + for _, msg := range batch { + m := msgMeta{} + m.dbSchema, _ = msg.MetaGet("database_schema") + m.table, _ = msg.MetaGet("table") + m.operation, _ = msg.MetaGet("operation") + m.lsn, _ = msg.MetaGet("lsn") + collected = append(collected, m) + } + return nil + })) + + stream, err := sb.Build() + require.NoError(t, err) + license.InjectTestService(stream.Resources()) + go func() { + if err := stream.Run(t.Context()); err != nil && !errors.Is(err, context.Canceled) { + t.Error(err) + } + }() + t.Cleanup(func() { require.NoError(t, stream.StopWithin(10*time.Second)) }) + + // Wait for all 3 snapshot rows. + assert.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(collected) >= 3 + }, 30*time.Second, 100*time.Millisecond, "timed out waiting for snapshot rows") + + // Insert CDC rows. + _, err = db.Exec("INSERT INTO tenant_a.events (name) VALUES ('dave')") + require.NoError(t, err) + _, err = db.Exec("INSERT INTO tenant_b.events (name) VALUES ('eve')") + require.NoError(t, err) + + // Wait for 2 CDC rows (total 5). + assert.EventuallyWithT(t, func(c *assert.CollectT) { + mu.Lock() + defer mu.Unlock() + assert.Len(c, collected, 5) + }, 30*time.Second, 100*time.Millisecond, "timed out waiting for CDC rows") + + mu.Lock() + defer mu.Unlock() + + var snapshots, cdcMsgs []msgMeta + for _, m := range collected { + if m.operation == "read" { + snapshots = append(snapshots, m) + } else { + cdcMsgs = append(cdcMsgs, m) + } + } + + // Snapshot assertions. + require.Len(t, snapshots, 3) + snapshotSchemas := make(map[string]int) + for _, m := range snapshots { + assert.Equal(t, "events", m.table, "snapshot: table should be bare name without schema prefix") + assert.Empty(t, m.lsn, "snapshot rows have no LSN") + snapshotSchemas[m.dbSchema]++ + } + assert.Equal(t, 2, snapshotSchemas["tenant_a"], "expected 2 snapshot rows from tenant_a") + assert.Equal(t, 1, snapshotSchemas["tenant_b"], "expected 1 snapshot row from tenant_b") + + // CDC assertions. + require.Len(t, cdcMsgs, 2) + cdcSchemas := make(map[string]int) + for _, m := range cdcMsgs { + assert.Equal(t, "insert", m.operation) + assert.Equal(t, "events", m.table) + assert.NotEmpty(t, m.lsn, "CDC rows must have an LSN") + cdcSchemas[m.dbSchema]++ + } + assert.Equal(t, 1, cdcSchemas["tenant_a"], "expected 1 CDC row from tenant_a") + assert.Equal(t, 1, cdcSchemas["tenant_b"], "expected 1 CDC row from tenant_b") +} + +// TestIntegrationSchemaExcludeCarvesOutTenant verifies that schema_exclude +// carves an exception out of a broad schema_include: a schema that matches +// schema_include but also matches a schema_exclude entry contributes no +// rows at all, neither during the initial snapshot nor from subsequent CDC +// changes. +func TestIntegrationMultiSchemaExcludeCarvesOutTenant(t *testing.T) { + integration.CheckSkip(t) + databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16") + require.NoError(t, err) + + // Three tenant schemas match tenant_*; tenant_c is carved out via schema_exclude. + for _, schema := range []string{"tenant_a", "tenant_b", "tenant_c"} { + _, err = db.Exec(fmt.Sprintf("CREATE SCHEMA %s", schema)) + require.NoError(t, err) + _, err = db.Exec(fmt.Sprintf( + "CREATE TABLE %s.events (id SERIAL PRIMARY KEY, name TEXT)", schema)) + require.NoError(t, err) + } + + // Pre-load snapshot data, including a row in the excluded schema that must + // never surface. + _, err = db.Exec("INSERT INTO tenant_a.events (name) VALUES ('alice'), ('bob')") + require.NoError(t, err) + _, err = db.Exec("INSERT INTO tenant_b.events (name) VALUES ('carol')") + require.NoError(t, err) + _, err = db.Exec("INSERT INTO tenant_c.events (name) VALUES ('mallory')") + require.NoError(t, err) + + type msgMeta struct { + dbSchema string + table string + operation string + lsn string + } + + var ( + mu sync.Mutex + collected []msgMeta + ) + collectedLen := func() int { + mu.Lock() + defer mu.Unlock() + return len(collected) + } + + tmpl := fmt.Sprintf(` +postgres_cdc: + dsn: %s + slot_name: schema_exclude_test_slot + stream_snapshot: true + schema_include: tenant_* + schema_exclude: + - tenant_c + tables: + - events +`, databaseURL) + + sb := service.NewStreamBuilder() + require.NoError(t, sb.SetLoggerYAML(`level: WARN`)) + require.NoError(t, sb.AddInputYAML(tmpl)) + require.NoError(t, sb.AddBatchConsumerFunc(func(_ context.Context, batch service.MessageBatch) error { + mu.Lock() + defer mu.Unlock() + for _, msg := range batch { + m := msgMeta{} + m.dbSchema, _ = msg.MetaGet("database_schema") + m.table, _ = msg.MetaGet("table") + m.operation, _ = msg.MetaGet("operation") + m.lsn, _ = msg.MetaGet("lsn") + collected = append(collected, m) + } + return nil + })) + + stream, err := sb.Build() + require.NoError(t, err) + license.InjectTestService(stream.Resources()) + go func() { + if err := stream.Run(t.Context()); err != nil && !errors.Is(err, context.Canceled) { + t.Error(err) + } + }() + t.Cleanup(func() { require.NoError(t, stream.StopWithin(10*time.Second)) }) + + // Wait for the 3 snapshot rows from the two non-excluded schemas; tenant_c's + // row must never contribute to this count. + assert.Eventually(t, func() bool { + return collectedLen() >= 3 + }, 30*time.Second, 100*time.Millisecond, "timed out waiting for snapshot rows") + + // Insert CDC rows into all three schemas, including the excluded one. + _, err = db.Exec("INSERT INTO tenant_a.events (name) VALUES ('dave')") + require.NoError(t, err) + _, err = db.Exec("INSERT INTO tenant_b.events (name) VALUES ('eve')") + require.NoError(t, err) + _, err = db.Exec("INSERT INTO tenant_c.events (name) VALUES ('trudy')") + require.NoError(t, err) + + // Wait for the 2 CDC rows from the non-excluded schemas (total 5). + assert.EventuallyWithT(t, func(c *assert.CollectT) { + assert.Equal(c, 5, collectedLen()) + }, 30*time.Second, 100*time.Millisecond, "timed out waiting for CDC rows") + + // tenant_c's CDC insert above raced the same replication stream as the + // tenant_a/tenant_b inserts already confirmed above, so if it were going + // to leak through it would have by now; assert the count never climbs + // past 5 to catch a delayed leak instead of just checking once. + assert.Never(t, func() bool { + return collectedLen() > 5 + }, 3*time.Second, 200*time.Millisecond, "received unexpected message(s) from excluded schema tenant_c") + + mu.Lock() + defer mu.Unlock() + + require.Len(t, collected, 5) + for _, m := range collected { + assert.NotEqual(t, "tenant_c", m.dbSchema, "tenant_c is excluded and must never appear, got message: %+v", m) + } + + var snapshots, cdcMsgs []msgMeta + for _, m := range collected { + if m.operation == "read" { + snapshots = append(snapshots, m) + } else { + cdcMsgs = append(cdcMsgs, m) + } + } + + // Snapshot assertions. + require.Len(t, snapshots, 3) + snapshotSchemas := make(map[string]int) + for _, m := range snapshots { + assert.Equal(t, "events", m.table, "snapshot: table should be bare name without schema prefix") + assert.Empty(t, m.lsn, "snapshot rows have no LSN") + snapshotSchemas[m.dbSchema]++ + } + assert.Equal(t, 2, snapshotSchemas["tenant_a"], "expected 2 snapshot rows from tenant_a") + assert.Equal(t, 1, snapshotSchemas["tenant_b"], "expected 1 snapshot row from tenant_b") + + // CDC assertions. + require.Len(t, cdcMsgs, 2) + cdcSchemas := make(map[string]int) + for _, m := range cdcMsgs { + assert.Equal(t, "insert", m.operation) + assert.Equal(t, "events", m.table) + assert.NotEmpty(t, m.lsn, "CDC rows must have an LSN") + cdcSchemas[m.dbSchema]++ + } + assert.Equal(t, 1, cdcSchemas["tenant_a"], "expected 1 CDC row from tenant_a") + assert.Equal(t, 1, cdcSchemas["tenant_b"], "expected 1 CDC row from tenant_b") +} + +func TestIntegrationMultiSchemaIncludeMatchesHyphenatedUUIDSchema(t *testing.T) { + integration.CheckSkip(t) + databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16") + require.NoError(t, err) + + const uuidSchema = "a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11" + _, err = db.Exec(fmt.Sprintf(`CREATE SCHEMA "%s"`, uuidSchema)) + require.NoError(t, err) + _, err = db.Exec(fmt.Sprintf(`CREATE TABLE "%s".events (id SERIAL PRIMARY KEY, name TEXT)`, uuidSchema)) + require.NoError(t, err) + _, err = db.Exec(fmt.Sprintf(`INSERT INTO "%s".events (name) VALUES ('alice')`, uuidSchema)) + require.NoError(t, err) + + type msgMeta struct { + dbSchema string + table string + } + + var ( + mu sync.Mutex + collected []msgMeta + ) + + tmpl := fmt.Sprintf(` +postgres_cdc: + dsn: %s + slot_name: hyphenated_schema_include_slot + stream_snapshot: true + schema_include: a0eebc99-* + tables: + - events +`, databaseURL) + + sb := service.NewStreamBuilder() + require.NoError(t, sb.SetLoggerYAML(`level: WARN`)) + require.NoError(t, sb.AddInputYAML(tmpl)) + require.NoError(t, sb.AddBatchConsumerFunc(func(_ context.Context, batch service.MessageBatch) error { + mu.Lock() + defer mu.Unlock() + for _, msg := range batch { + m := msgMeta{} + m.dbSchema, _ = msg.MetaGet("database_schema") + m.table, _ = msg.MetaGet("table") + collected = append(collected, m) + } + return nil + })) + + stream, err := sb.Build() + require.NoError(t, err) + license.InjectTestService(stream.Resources()) + go func() { + if err := stream.Run(t.Context()); err != nil && !errors.Is(err, context.Canceled) { + t.Error(err) + } + }() + t.Cleanup(func() { require.NoError(t, stream.StopWithin(10*time.Second)) }) + + assert.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(collected) >= 1 + }, 30*time.Second, 100*time.Millisecond, "timed out waiting for snapshot row from hyphenated UUID schema") + + mu.Lock() + defer mu.Unlock() + require.Len(t, collected, 1) + assert.Equal(t, uuidSchema, collected[0].dbSchema, "database_schema metadata should be the raw, unquoted, case-preserved schema name") + assert.Equal(t, "events", collected[0].table) +} + +func TestIntegrationMultiSchemaMissingTableDegradesGracefully(t *testing.T) { + integration.CheckSkip(t) + databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16") + require.NoError(t, err) + + // tenant_a is fully provisioned with the "events" table; tenant_b matches + // the schema glob but is missing it (e.g. still being migrated). Before + // this fix, CreatePublication's FOR TABLE clause would reference the + // non-existent tenant_b.events relation and fail publication setup for + // every matched schema, not just the drifted one. + _, err = db.Exec("CREATE SCHEMA tenant_a") + require.NoError(t, err) + _, err = db.Exec("CREATE TABLE tenant_a.events (id SERIAL PRIMARY KEY, name TEXT)") + require.NoError(t, err) + _, err = db.Exec("CREATE SCHEMA tenant_b") + require.NoError(t, err) + + _, err = db.Exec("INSERT INTO tenant_a.events (name) VALUES ('alice')") + require.NoError(t, err) + + type msgMeta struct { + dbSchema string + table string + } + + var ( + mu sync.Mutex + collected []msgMeta + ) + + tmpl := fmt.Sprintf(` +postgres_cdc: + dsn: %s + slot_name: missing_table_degrade_slot + stream_snapshot: true + schema_include: tenant_* + tables: + - events +`, databaseURL) + + sb := service.NewStreamBuilder() + require.NoError(t, sb.SetLoggerYAML(`level: WARN`)) + require.NoError(t, sb.AddInputYAML(tmpl)) + require.NoError(t, sb.AddBatchConsumerFunc(func(_ context.Context, batch service.MessageBatch) error { + mu.Lock() + defer mu.Unlock() + for _, msg := range batch { + m := msgMeta{} + m.dbSchema, _ = msg.MetaGet("database_schema") + m.table, _ = msg.MetaGet("table") + collected = append(collected, m) + } + return nil + })) + + stream, err := sb.Build() + require.NoError(t, err) + license.InjectTestService(stream.Resources()) + go func() { + if err := stream.Run(t.Context()); err != nil && !errors.Is(err, context.Canceled) { + t.Error(err) + } + }() + t.Cleanup(func() { require.NoError(t, stream.StopWithin(10*time.Second)) }) + + // tenant_a should keep streaming even though tenant_b is missing the table. + assert.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(collected) >= 1 + }, 30*time.Second, 100*time.Millisecond, "timed out waiting for tenant_a snapshot row; a missing table in tenant_b should not block replication") + + mu.Lock() + defer mu.Unlock() + require.Len(t, collected, 1) + assert.Equal(t, "tenant_a", collected[0].dbSchema) + assert.Equal(t, "events", collected[0].table) +} + +// TestIntegrationMultiSchemaViewNamesakeDegradesGracefully guards against a +// relation that exists but isn't publishable: tenant_c matches the schema +// glob and has an "events" view (e.g. a compatibility shim over a renamed +// table), not the "events" table configured. Before restricting +// resolveExistingTables to table_type = 'BASE TABLE', this view counted as +// present, so CreatePublication's FOR TABLE clause referenced it and failed +// setup for every matched schema with "... is not supported for views". +// tenant_c's view should instead be skipped with a warning, the same way a +// genuinely missing table is, leaving tenant_a free to stream. +func TestIntegrationMultiSchemaViewNamesakeDegradesGracefully(t *testing.T) { + integration.CheckSkip(t) + databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16") + require.NoError(t, err) + + _, err = db.Exec("CREATE SCHEMA tenant_a") + require.NoError(t, err) + _, err = db.Exec("CREATE TABLE tenant_a.events (id SERIAL PRIMARY KEY, name TEXT)") + require.NoError(t, err) + _, err = db.Exec("CREATE SCHEMA tenant_c") + require.NoError(t, err) + _, err = db.Exec("CREATE VIEW tenant_c.events AS SELECT 1 AS id, 'namesake'::text AS name") + require.NoError(t, err) + + _, err = db.Exec("INSERT INTO tenant_a.events (name) VALUES ('alice')") + require.NoError(t, err) + + type msgMeta struct { + dbSchema string + table string + } + + var ( + mu sync.Mutex + collected []msgMeta + ) + + tmpl := fmt.Sprintf(` +postgres_cdc: + dsn: %s + slot_name: view_namesake_degrade_slot + stream_snapshot: true + schema_include: tenant_* + tables: + - events +`, databaseURL) + + sb := service.NewStreamBuilder() + require.NoError(t, sb.SetLoggerYAML(`level: WARN`)) + require.NoError(t, sb.AddInputYAML(tmpl)) + require.NoError(t, sb.AddBatchConsumerFunc(func(_ context.Context, batch service.MessageBatch) error { + mu.Lock() + defer mu.Unlock() + for _, msg := range batch { + m := msgMeta{} + m.dbSchema, _ = msg.MetaGet("database_schema") + m.table, _ = msg.MetaGet("table") + collected = append(collected, m) + } + return nil + })) + + stream, err := sb.Build() + require.NoError(t, err) + license.InjectTestService(stream.Resources()) + go func() { + if err := stream.Run(t.Context()); err != nil && !errors.Is(err, context.Canceled) { + t.Error(err) + } + }() + t.Cleanup(func() { require.NoError(t, stream.StopWithin(10*time.Second)) }) + + // tenant_a should keep streaming even though tenant_c's "events" is a + // view rather than a publishable table. + assert.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(collected) >= 1 + }, 30*time.Second, 100*time.Millisecond, "timed out waiting for tenant_a snapshot row; a view namesake in tenant_c should not block replication or fail publication setup") + + mu.Lock() + defer mu.Unlock() + require.Len(t, collected, 1) + assert.Equal(t, "tenant_a", collected[0].dbSchema) + assert.Equal(t, "events", collected[0].table) +} + +func TestIntegrationMultiSchemaExcludeCarvesOutTenantAutoDiscover(t *testing.T) { + integration.CheckSkip(t) + databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16") + require.NoError(t, err) + + // Three tenant schemas match tenant_*; tenant_c is carved out via schema_exclude. + for _, schema := range []string{"tenant_a", "tenant_b", "tenant_c"} { + _, err = db.Exec(fmt.Sprintf("CREATE SCHEMA %s", schema)) + require.NoError(t, err) + _, err = db.Exec(fmt.Sprintf( + "CREATE TABLE %s.events (id SERIAL PRIMARY KEY, name TEXT)", schema)) + require.NoError(t, err) + } + + // tenant_a additionally has a partitioned "orders" table, to verify + // auto-discovery excludes the parent while still picking up its leaves. + _, err = db.Exec(` +CREATE TABLE tenant_a.orders ( + id INT NOT NULL, + created_at DATE NOT NULL, + PRIMARY KEY (id, created_at) +) PARTITION BY RANGE (created_at)`) + require.NoError(t, err) + _, err = db.Exec(` +CREATE TABLE tenant_a.orders_2025 PARTITION OF tenant_a.orders + FOR VALUES FROM ('2025-01-01') TO ('2026-01-01')`) + require.NoError(t, err) + _, err = db.Exec(` +CREATE TABLE tenant_a.orders_2026 PARTITION OF tenant_a.orders + FOR VALUES FROM ('2026-01-01') TO ('2027-01-01')`) + require.NoError(t, err) + + // Pre-load snapshot data, including a row in the excluded schema that must + // never surface. + _, err = db.Exec("INSERT INTO tenant_a.events (name) VALUES ('alice'), ('bob')") + require.NoError(t, err) + _, err = db.Exec("INSERT INTO tenant_b.events (name) VALUES ('carol')") + require.NoError(t, err) + _, err = db.Exec("INSERT INTO tenant_c.events (name) VALUES ('mallory')") + require.NoError(t, err) + + // Insert through the parent, as real usage would, letting Postgres route + // each row to the correct leaf partition. + _, err = db.Exec("INSERT INTO tenant_a.orders (id, created_at) VALUES (1, '2025-06-01')") + require.NoError(t, err) + _, err = db.Exec("INSERT INTO tenant_a.orders (id, created_at) VALUES (2, '2026-06-01')") + require.NoError(t, err) + + type msgMeta struct { + dbSchema string + table string + operation string + lsn string + } + + var ( + mu sync.Mutex + collected []msgMeta + ) + collectedLen := func() int { + mu.Lock() + defer mu.Unlock() + return len(collected) + } + + // No `tables` field: every base table in tenant_a/tenant_b must be + // auto-discovered without listing "events" by hand. + tmpl := fmt.Sprintf(` +postgres_cdc: + dsn: %s + slot_name: auto_discover_schema_exclude_test_slot + stream_snapshot: true + schema_include: tenant_* + schema_exclude: + - tenant_c +`, databaseURL) + + sb := service.NewStreamBuilder() + require.NoError(t, sb.SetLoggerYAML(`level: WARN`)) + require.NoError(t, sb.AddInputYAML(tmpl)) + require.NoError(t, sb.AddBatchConsumerFunc(func(_ context.Context, batch service.MessageBatch) error { + mu.Lock() + defer mu.Unlock() + for _, msg := range batch { + m := msgMeta{} + m.dbSchema, _ = msg.MetaGet("database_schema") + m.table, _ = msg.MetaGet("table") + m.operation, _ = msg.MetaGet("operation") + m.lsn, _ = msg.MetaGet("lsn") + collected = append(collected, m) + } + return nil + })) + + stream, err := sb.Build() + require.NoError(t, err) + license.InjectTestService(stream.Resources()) + go func() { + if err := stream.Run(t.Context()); err != nil && !errors.Is(err, context.Canceled) { + t.Error(err) + } + }() + t.Cleanup(func() { require.NoError(t, stream.StopWithin(10*time.Second)) }) + + // Wait for the 5 snapshot rows from the two non-excluded schemas (3 + // "events" rows plus 2 "orders" rows split across its leaf partitions); + // tenant_c's row must never contribute to this count. + assert.Eventually(t, func() bool { + return collectedLen() >= 5 + }, 30*time.Second, 100*time.Millisecond, "timed out waiting for snapshot rows") + + // Give any erroneous duplicate emissions from the partitioned parent a + // chance to surface before moving on to the CDC phase. + assert.Never(t, func() bool { + return collectedLen() > 5 + }, 3*time.Second, 200*time.Millisecond, "received more snapshot rows than were inserted; the partitioned parent was likely auto-discovered alongside its leaf partitions") + + // Insert CDC rows into all three schemas, including the excluded one. + _, err = db.Exec("INSERT INTO tenant_a.events (name) VALUES ('dave')") + require.NoError(t, err) + _, err = db.Exec("INSERT INTO tenant_b.events (name) VALUES ('eve')") + require.NoError(t, err) + _, err = db.Exec("INSERT INTO tenant_c.events (name) VALUES ('trudy')") + require.NoError(t, err) + + // Wait for the 2 CDC rows from the non-excluded schemas (total 7). + assert.EventuallyWithT(t, func(c *assert.CollectT) { + assert.Equal(c, 7, collectedLen()) + }, 30*time.Second, 100*time.Millisecond, "timed out waiting for CDC rows") + + // tenant_c's CDC insert above raced the same replication stream as the + // tenant_a/tenant_b inserts already confirmed above, so if it were going + // to leak through it would have by now; assert the count never climbs + // past 7 to catch a delayed leak instead of just checking once. + assert.Never(t, func() bool { + return collectedLen() > 7 + }, 3*time.Second, 200*time.Millisecond, "received unexpected message(s) from excluded schema tenant_c") + + mu.Lock() + defer mu.Unlock() + + require.Len(t, collected, 7) + for _, m := range collected { + assert.NotEqual(t, "tenant_c", m.dbSchema, "tenant_c is excluded and must never appear, got message: %+v", m) + assert.NotEqual(t, "orders", m.table, "the partitioned parent must not be auto-discovered as a table in its own right") + } + + var snapshots, cdcMsgs []msgMeta + for _, m := range collected { + if m.operation == "read" { + snapshots = append(snapshots, m) + } else { + cdcMsgs = append(cdcMsgs, m) + } + } + + // Snapshot assertions. + require.Len(t, snapshots, 5) + snapshotSchemas := make(map[string]int) + ordersLeaves := make(map[string]int) + for _, m := range snapshots { + assert.Empty(t, m.lsn, "snapshot rows have no LSN") + switch m.table { + case "events": + snapshotSchemas[m.dbSchema]++ + case "orders_2025", "orders_2026": + assert.Equal(t, "tenant_a", m.dbSchema, "orders and its partitions only exist in tenant_a") + ordersLeaves[m.table]++ + default: + t.Errorf("unexpected table in snapshot: %+v", m) + } + } + assert.Equal(t, 2, snapshotSchemas["tenant_a"], "expected 2 \"events\" snapshot rows from tenant_a") + assert.Equal(t, 1, snapshotSchemas["tenant_b"], "expected 1 \"events\" snapshot row from tenant_b") + assert.Equal(t, map[string]int{"orders_2025": 1, "orders_2026": 1}, ordersLeaves, "expected exactly one snapshot row from each leaf partition, and none from the partitioned parent") + + // CDC assertions. + require.Len(t, cdcMsgs, 2) + cdcSchemas := make(map[string]int) + for _, m := range cdcMsgs { + assert.Equal(t, "insert", m.operation) + assert.Equal(t, "events", m.table) + assert.NotEmpty(t, m.lsn, "CDC rows must have an LSN") + cdcSchemas[m.dbSchema]++ + } + assert.Equal(t, 1, cdcSchemas["tenant_a"], "expected 1 CDC row from tenant_a") + assert.Equal(t, 1, cdcSchemas["tenant_b"], "expected 1 CDC row from tenant_b") +} + +func TestIntegrationMultiSchemaAndTableMatchingTest(t *testing.T) { + integration.CheckSkip(t) + + t.Run("exact schema match with missing table fails", func(t *testing.T) { + databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16") + require.NoError(t, err) + + _, err = db.Exec("CREATE TABLE IF NOT EXISTS orders (id SERIAL PRIMARY KEY, name TEXT);") + require.NoError(t, err) + + tmpl := fmt.Sprintf(` +dsn: %s +slot_name: exact_schema_missing_table_slot +schema: public +tables: + - orders + - ordres +`, databaseURL) + + conf, err := newPostgresCDCConfig().ParseYAML(tmpl, nil) + require.NoError(t, err) + + mgr := service.MockResources() + license.InjectTestService(mgr) + + input, err := newPgStreamInput(conf, mgr) + require.NoError(t, err) + + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + + // Bypass the benthos AsyncReader's infinite connect-retry loop, same as + // TestIntegrationSchemaIncludeExcludeConfigValidation. + err = input.Connect(ctx) + require.Error(t, err, "typo'd table %q should fail startup loudly instead of silently streaming only %q", "ordres", "orders") + assert.Contains(t, err.Error(), "ordres") + }) + + t.Run("glob schema matching multiple schemas with one or more missing tables fails", func(t *testing.T) { + databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16") + require.NoError(t, err) + + _, err = db.Exec(` + CREATE SCHEMA tenant_a; + CREATE SCHEMA tenant_b; + + CREATE TABLE tenant_a.orders (id SERIAL PRIMARY KEY, name TEXT); + CREATE TABLE tenant_b.orders (id SERIAL PRIMARY KEY, name TEXT); + `) + require.NoError(t, err) + + tmpl := fmt.Sprintf(` +dsn: %s +slot_name: glob_schema_total_miss_slot +schema_include: tenant_* +tables: + - orders + - ordres +`, databaseURL) + + conf, err := newPostgresCDCConfig().ParseYAML(tmpl, nil) + require.NoError(t, err) + + mgr := service.MockResources() + license.InjectTestService(mgr) + + input, err := newPgStreamInput(conf, mgr) + require.NoError(t, err) + + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + + err = input.Connect(ctx) + require.Error(t, err, "ordres exists in neither tenant_a nor tenant_b, so it should fail startup instead of silently streaming only tenant_a/b.orders") + assert.Contains(t, err.Error(), "ordres") + }) + + t.Run("glob schema matching multiple schemas with matching tables passes", func(t *testing.T) { + databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16") + require.NoError(t, err) + + _, err = db.Exec(` + CREATE SCHEMA tenant_a; + CREATE SCHEMA tenant_b; + + CREATE TABLE tenant_a.orders (id SERIAL PRIMARY KEY, name TEXT); + CREATE TABLE tenant_b.ordres (id SERIAL PRIMARY KEY, name TEXT); + + INSERT INTO tenant_a.orders (name) VALUES ('alice'); + INSERT INTO tenant_b.ordres (name) VALUES ('bob'); + `) + require.NoError(t, err) + + type msgMeta struct { + dbSchema string + table string + } + + var ( + mu sync.Mutex + collected []msgMeta + ) + + tmpl := fmt.Sprintf(` +postgres_cdc: + dsn: %s + slot_name: glob_schema_partial_match_slot + stream_snapshot: true + schema_include: tenant_* + tables: + - orders + - ordres +`, databaseURL) + + sb := service.NewStreamBuilder() + require.NoError(t, sb.SetLoggerYAML(`level: WARN`)) + require.NoError(t, sb.AddInputYAML(tmpl)) + require.NoError(t, sb.AddBatchConsumerFunc(func(_ context.Context, batch service.MessageBatch) error { + mu.Lock() + defer mu.Unlock() + for _, msg := range batch { + m := msgMeta{} + m.dbSchema, _ = msg.MetaGet("database_schema") + m.table, _ = msg.MetaGet("table") + collected = append(collected, m) + } + return nil + })) + + stream, err := sb.Build() + require.NoError(t, err) + license.InjectTestService(stream.Resources()) + go func() { + if err := stream.Run(t.Context()); err != nil && !errors.Is(err, context.Canceled) { + t.Error(err) + } + }() + t.Cleanup(func() { require.NoError(t, stream.StopWithin(10*time.Second)) }) + + assert.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(collected) >= 2 + }, 30*time.Second, 100*time.Millisecond, "timed out waiting for both tenant_a.orders and tenant_b.ordres snapshot rows") + + mu.Lock() + defer mu.Unlock() + require.Len(t, collected, 2) + assert.Contains(t, collected, msgMeta{dbSchema: "tenant_a", table: "orders"}) + assert.Contains(t, collected, msgMeta{dbSchema: "tenant_b", table: "ordres"}) + }) + + t.Run("glob schema matching multiple schemas with all tables present in all schemas passes", func(t *testing.T) { + databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16") + require.NoError(t, err) + + _, err = db.Exec(` + CREATE SCHEMA tenant_a; + CREATE SCHEMA tenant_b; + + CREATE TABLE tenant_a.orders (id SERIAL PRIMARY KEY, name TEXT); + CREATE TABLE tenant_a.ordres (id SERIAL PRIMARY KEY, name TEXT); + CREATE TABLE tenant_b.orders (id SERIAL PRIMARY KEY, name TEXT); + CREATE TABLE tenant_b.ordres (id SERIAL PRIMARY KEY, name TEXT); + + INSERT INTO tenant_a.orders (name) VALUES ('alice'); + INSERT INTO tenant_a.ordres (name) VALUES ('bob'); + INSERT INTO tenant_b.orders (name) VALUES ('carol'); + INSERT INTO tenant_b.ordres (name) VALUES ('dave'); + `) + require.NoError(t, err) + + type msgMeta struct { + dbSchema string + table string + } + + var ( + mu sync.Mutex + collected []msgMeta + ) + + tmpl := fmt.Sprintf(` +postgres_cdc: + dsn: %s + slot_name: glob_schema_full_match_slot + stream_snapshot: true + schema_include: tenant_* + tables: + - orders + - ordres +`, databaseURL) + + sb := service.NewStreamBuilder() + require.NoError(t, sb.SetLoggerYAML(`level: WARN`)) + require.NoError(t, sb.AddInputYAML(tmpl)) + require.NoError(t, sb.AddBatchConsumerFunc(func(_ context.Context, batch service.MessageBatch) error { + mu.Lock() + defer mu.Unlock() + for _, msg := range batch { + m := msgMeta{} + m.dbSchema, _ = msg.MetaGet("database_schema") + m.table, _ = msg.MetaGet("table") + collected = append(collected, m) + } + return nil + })) + + stream, err := sb.Build() + require.NoError(t, err) + license.InjectTestService(stream.Resources()) + go func() { + if err := stream.Run(t.Context()); err != nil && !errors.Is(err, context.Canceled) { + t.Error(err) + } + }() + t.Cleanup(func() { require.NoError(t, stream.StopWithin(10*time.Second)) }) + + assert.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(collected) >= 4 + }, 30*time.Second, 100*time.Millisecond, "timed out waiting for all four tenant_{a,b}.{orders,ordres} snapshot rows") + + mu.Lock() + defer mu.Unlock() + require.Len(t, collected, 4) + assert.Contains(t, collected, msgMeta{dbSchema: "tenant_a", table: "orders"}) + assert.Contains(t, collected, msgMeta{dbSchema: "tenant_a", table: "ordres"}) + assert.Contains(t, collected, msgMeta{dbSchema: "tenant_b", table: "orders"}) + assert.Contains(t, collected, msgMeta{dbSchema: "tenant_b", table: "ordres"}) + }) +} + +func TestIntegrationMultiSchemaExplicitTablesAcceptsPartitionedTable(t *testing.T) { + integration.CheckSkip(t) + databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16") + require.NoError(t, err) + + _, err = db.Exec(` + CREATE SCHEMA tenant_a; + CREATE SCHEMA tenant_b; + + CREATE TABLE tenant_a.orders ( + id INT NOT NULL, + created_at DATE NOT NULL, + PRIMARY KEY (id, created_at) + ) PARTITION BY RANGE (created_at); + CREATE TABLE tenant_a.orders_2025 PARTITION OF tenant_a.orders + FOR VALUES FROM ('2025-01-01') TO ('2026-01-01'); + + CREATE TABLE tenant_b.orders ( + id INT NOT NULL, + created_at DATE NOT NULL, + PRIMARY KEY (id, created_at) + ) PARTITION BY RANGE (created_at); + CREATE TABLE tenant_b.orders_2025 PARTITION OF tenant_b.orders + FOR VALUES FROM ('2025-01-01') TO ('2026-01-01'); + `) + require.NoError(t, err) + + tmpl := fmt.Sprintf(` +dsn: %s +slot_name: partitioned_table_explicit_slot +schema_include: tenant_* +tables: + - orders +`, databaseURL) + + conf, err := newPostgresCDCConfig().ParseYAML(tmpl, nil) + require.NoError(t, err) + + mgr := service.MockResources() + license.InjectTestService(mgr) + + input, err := newPgStreamInput(conf, mgr) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, input.Close(context.Background())) }) + + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + + err = input.Connect(ctx) + require.NoError(t, err, "orders is a partitioned table that exists in every matched schema and should be accepted, not reported as missing") + + require.NoError(t, input.Close(context.Background())) + + err = input.Connect(ctx) + require.NoError(t, err, "reconnecting with the same explicitly-listed partitioned table must reconcile cleanly, not attempt to drop its leaf partitions from the publication") +} diff --git a/internal/impl/postgresql/pglogicalstream/config.go b/internal/impl/postgresql/pglogicalstream/config.go index 93bab82ecf..3c87d0bcaa 100644 --- a/internal/impl/postgresql/pglogicalstream/config.go +++ b/internal/impl/postgresql/pglogicalstream/config.go @@ -16,6 +16,8 @@ import ( "github.com/jackc/pgx/v5/pgconn" "github.com/redpanda-data/benthos/v4/public/service" + + "github.com/redpanda-data/connect/v4/internal/impl/postgresql/pglogicalstream/multischema" ) // Config is the configuration for the pglogicalstream plugin @@ -26,6 +28,11 @@ type Config struct { TLSConfig *tls.Config DBSchema string DBTables []string + + // SchemaResolver resolves schema_include/schema_exclude into the schemas + // to replicate. Non-nil only when schema_include is set. + SchemaResolver *multischema.Resolver + // Refreshes short lived IAM auth token that is treated as a password RefreshAuthToken func(ctx context.Context) error // ReplicationSlotName is the name of the replication slot to use diff --git a/internal/impl/postgresql/pglogicalstream/logical_stream.go b/internal/impl/postgresql/pglogicalstream/logical_stream.go index 64972e0359..05ea8de474 100644 --- a/internal/impl/postgresql/pglogicalstream/logical_stream.go +++ b/internal/impl/postgresql/pglogicalstream/logical_stream.go @@ -26,6 +26,7 @@ import ( "github.com/redpanda-data/benthos/v4/public/service" "github.com/redpanda-data/connect/v4/internal/asyncroutine" + "github.com/redpanda-data/connect/v4/internal/impl/postgresql/pglogicalstream/multischema" "github.com/redpanda-data/connect/v4/internal/impl/postgresql/pglogicalstream/sanitize" ) @@ -98,18 +99,98 @@ func NewPgStream(ctx context.Context, config *Config) (*Stream, error) { return nil, err } - schema, err := sanitize.NormalizePostgresIdentifier(config.DBSchema) - if err != nil { - return nil, fmt.Errorf("invalid schema name %q: %w", config.DBSchema, err) - } + var ( + tables []TableFQN + schema string + ) + if config.SchemaResolver != nil { + if config.SignalTableName != "" { + return nil, errors.New("signal_table_name is not supported when schema_include is set") + } + schemas, err := config.SchemaResolver.Resolve(ctx, dbConn, config.Logger) + if err != nil { + return nil, err + } + + normalizedTables := make([]string, 0, len(config.DBTables)) + for _, table := range config.DBTables { + normalized, err := sanitize.NormalizePostgresIdentifier(table) + if err != nil { + return nil, fmt.Errorf("invalid table name %q: %w", table, err) + } + normalizedTables = append(normalizedTables, normalized) + } + + // tables empty here would otherwise fall through to CreatePublication's + // FOR ALL TABLES fallback, replicating the whole database and defeating + // schema_include/schema_exclude - auto-discover per matched schema instead. + autoDiscoverTables := len(normalizedTables) == 0 + + existingTablesBySchema, err := multischema.ResolveExistingTables(ctx, dbConn, schemas) + if err != nil { + return nil, fmt.Errorf("resolving tables in schema(s) %v: %w", schemas, err) + } - tables := []TableFQN{} - for _, table := range config.DBTables { - normalized, err := sanitize.NormalizePostgresIdentifier(table) + tables = make([]TableFQN, 0, len(schemas)*len(normalizedTables)) + foundTables := make(map[string]bool, len(normalizedTables)) + for _, schema := range schemas { + existingTables := existingTablesBySchema[schema] + if autoDiscoverTables { + // Only ordinary tables: a partitioned parent must not be + // auto-discovered alongside its leaf partitions (which are + // themselves ordinary tables) - see ResolveExistingTables. + for table, relkind := range existingTables { + if relkind != multischema.RelKindOrdinaryTable { + continue + } + tables = append(tables, TableFQN{Schema: schema, Table: table}) + } + continue + } + for _, table := range normalizedTables { + if _, ok := existingTables[table]; !ok { + config.Logger.Warnf("Table %s.%s not found, skipping (schema %s matched schema_include pattern %q but does not contain this table)", schema, table, schema, config.SchemaResolver.Include) + continue + } + tables = append(tables, TableFQN{Schema: schema, Table: table}) + foundTables[table] = true + } + } + if autoDiscoverTables { + if len(tables) == 0 { + return nil, fmt.Errorf("no tables found in schema(s) %v matching schema_include pattern %q", schemas, config.SchemaResolver.Include) + } + config.Logger.Debugf("%q has no `tables` list configured: auto-discovered %d table(s) across %d schema(s)", config.SchemaResolver.Include, len(tables), len(schemas)) + } else { + // A table must exist in at least one matched schema. Missing from some + // (but not all) matched schemas is tolerated above as a multi-tenant gap; + // missing from every matched schema is indistinguishable from a typo and + // must fail loudly rather than silently drop the table. + var missingTables []string + for i, table := range normalizedTables { + if !foundTables[table] { + missingTables = append(missingTables, config.DBTables[i]) + } + } + if len(missingTables) > 0 { + return nil, fmt.Errorf("table(s) %v not found in any schema matching schema_include pattern %q", missingTables, config.SchemaResolver.Include) + } + } + } else { + var err error + schema, err = sanitize.NormalizePostgresIdentifier(config.DBSchema) if err != nil { - return nil, fmt.Errorf("invalid table name %q: %w", table, err) + return nil, fmt.Errorf("invalid schema name %q: %w", config.DBSchema, err) + } + + tables = []TableFQN{} + for _, table := range config.DBTables { + normalized, err := sanitize.NormalizePostgresIdentifier(table) + if err != nil { + return nil, fmt.Errorf("invalid table name %q: %w", table, err) + } + tables = append(tables, TableFQN{Schema: schema, Table: normalized}) } - tables = append(tables, TableFQN{Schema: schema, Table: normalized}) } batchSize := 1000 if config.BatchSize > 0 { @@ -672,7 +753,7 @@ func (s *Stream) processSnapshot(ctx context.Context, snapshotter *snapshotter) if len(ranges) > 1 { s.logger.Infof( - "created plan in %v to split %s into %d chunks of %d and process in parallel", + "Created plan in %v to split %s into %d chunks of %d and process in parallel", time.Since(planStartTime), table, len(ranges), @@ -680,7 +761,7 @@ func (s *Stream) processSnapshot(ctx context.Context, snapshotter *snapshotter) ) } else { s.logger.Infof( - "created plan in %v to scan %s sequentially", + "Created plan in %v to scan %s sequentially", time.Since(planStartTime), table, ) diff --git a/internal/impl/postgresql/pglogicalstream/multischema/resolver.go b/internal/impl/postgresql/pglogicalstream/multischema/resolver.go new file mode 100644 index 0000000000..ad5a94f5cb --- /dev/null +++ b/internal/impl/postgresql/pglogicalstream/multischema/resolver.go @@ -0,0 +1,367 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed as a Redpanda Enterprise file under the Redpanda Community +// License (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// https://github.com/redpanda-data/connect/v4/blob/main/licenses/rcl.md + +// Package multischema resolves a schema_include/schema_exclude +// configuration (replicating from multiple PostgreSQL schemas matched by a +// glob pattern) into the concrete set of schemas and tables to replicate. +package multischema + +import ( + "context" + "fmt" + "regexp" + "slices" + "strings" + + "github.com/jackc/pgx/v5/pgconn" + + "github.com/redpanda-data/benthos/v4/public/service" + + "github.com/redpanda-data/connect/v4/internal/impl/postgresql/pglogicalstream/sanitize" +) + +// Resolver is responsible for helping resolve DB schemas for multi-schema support. +type Resolver struct { + // Include is the schema_include glob pattern. + Include string + // Exclude is the schema_exclude list, evaluated against the schemas + // matched by Include. + Exclude []string + + previouslyResolved []string + previouslyInaccessible []string +} + +// NewResolver returns a Resolver for the given schema_include/schema_exclude +// configuration. Callers should only construct one when schema_include is +// set. +func NewResolver(include string, exclude []string) *Resolver { + return &Resolver{Include: include, Exclude: exclude} +} + +// Resolve resolves r.Include against conn, applies r.Exclude filtering, and +// warns about schema-set drift against the previously resolved set from an +// earlier call to Resolve on this same Resolver. +func (r *Resolver) Resolve(ctx context.Context, conn *pgconn.PgConn, logger *service.Logger) ([]string, error) { + schemas, inaccessibleSchemas, err := resolveSchemas(ctx, conn, r.Include) + if err != nil { + return nil, fmt.Errorf("resolving schema_include pattern %q: %w", r.Include, err) + } + matchedSchemas := schemas + + if len(r.Exclude) > 0 { + var excluded []string + schemas, excluded, err = r.filterExcluded(schemas) + if err != nil { + return nil, err + } + if len(excluded) > 0 { + logger.Debugf("schema_exclude %v excluded %d schema(s) %v from schema_include pattern %q; %d schema(s) remain: %v", r.Exclude, len(excluded), excluded, r.Include, len(schemas), schemas) + } + + if inaccessibleSchemas, _, err = r.filterExcluded(inaccessibleSchemas); err != nil { + return nil, err + } + } + + if newlyInaccessible, _ := diffSchemaSets(r.previouslyInaccessible, inaccessibleSchemas); len(newlyInaccessible) > 0 { + logger.Warnf("schema_include pattern %q matches schema(s) %v that the configured role cannot see (missing USAGE privilege); they will be skipped", r.Include, newlyInaccessible) + } + r.previouslyInaccessible = slices.Clone(inaccessibleSchemas) + + if len(schemas) == 0 { + if len(matchedSchemas) > 0 { + return nil, fmt.Errorf("schema_include pattern %q matched schema(s) %v, but schema_exclude %v excluded all of them", r.Include, matchedSchemas, r.Exclude) + } + return nil, fmt.Errorf("no schemas found matching schema_include pattern %q", r.Include) + } + logger.Debugf("schema_include pattern %q resolved to %d schema(s): %v", r.Include, len(schemas), schemas) + + if r.previouslyResolved != nil { + added, removed := diffSchemaSets(r.previouslyResolved, schemas) + if len(added) > 0 { + logger.Warnf("schema_include pattern %q now also matches schema(s) %v that did not match on the previous connect; their tables are being added to the publication, but any rows already in them will NOT be snapshotted even if stream_snapshot is enabled - only changes made from now on will be captured", r.Include, added) + } + if len(removed) > 0 { + logger.Warnf("schema(s) %v no longer match schema_include pattern %q (dropped, renamed, or the role lost USAGE) since the previous connect; their tables are being removed from the publication and will stop replicating", removed, r.Include) + } + } + r.previouslyResolved = slices.Clone(schemas) + + return schemas, nil +} + +func (r *Resolver) filterExcluded(schemas []string) (remaining, excluded []string, err error) { + remaining = make([]string, 0, len(schemas)) + for _, schema := range schemas { + var isExcluded bool + for _, pattern := range r.Exclude { + matched, err := schemaMatchesExcludePattern(schema, pattern) + if err != nil { + return nil, nil, fmt.Errorf("evaluating schema_exclude pattern %q against schema %q: %w", pattern, schema, err) + } + if matched { + isExcluded = true + break + } + } + if isExcluded { + excluded = append(excluded, schema) + continue + } + remaining = append(remaining, schema) + } + return remaining, excluded, nil +} + +func schemaPatternToLike(pattern string) (likePattern string, caseSensitive bool, err error) { + if strings.HasPrefix(pattern, `"`) { + unquoted, err := sanitize.UnquotePostgresIdentifier(pattern) + if err != nil { + return "", false, fmt.Errorf("invalid quoted schema identifier %q: %w", pattern, err) + } + return escapeLike(unquoted), true, nil + } + return globToLike(strings.ToLower(pattern)), false, nil +} + +func resolveSchemas(ctx context.Context, conn *pgconn.PgConn, pattern string) (visibleSchemas, inaccessibleSchemas []string, err error) { + likePattern, caseSensitive, err := schemaPatternToLike(pattern) + if err != nil { + return nil, nil, err + } + // Fixed, code-chosen operator (never derived from user input), so it's + // safe to splice directly into the query text rather than parameterize. + op := "ILIKE" + if caseSensitive { + op = "LIKE" + } + + q, err := sanitize.SQLQuery( + fmt.Sprintf("SELECT schema_name FROM information_schema.schemata WHERE schema_name %s $1 ESCAPE '!' AND schema_name NOT LIKE 'pg!_%%' ESCAPE '!' AND schema_name != 'information_schema'", op), + likePattern, + ) + if err != nil { + return nil, nil, fmt.Errorf("building schema resolution query: %w", err) + } + + results, err := conn.Exec(ctx, q).ReadAll() + if err != nil { + return nil, nil, fmt.Errorf("querying schemas matching %q: %w", pattern, err) + } + + visible := map[string]struct{}{} + var schemas []string + if len(results) > 0 { + for _, row := range results[0].Rows { + name := string(row[0]) + visible[name] = struct{}{} + // QuotePostgresIdentifier preserves the exact stored name (including + // case for case-sensitive schemas), unlike NormalizePostgresIdentifier + // which would incorrectly fold to lower-case. + schemas = append(schemas, sanitize.QuotePostgresIdentifier(name)) + } + } + + // pg_namespace isn't privilege-filtered, so a match here missing from + // information_schema.schemata means the role lacks USAGE on that schema. + nsQ, err := sanitize.SQLQuery( + fmt.Sprintf("SELECT nspname FROM pg_catalog.pg_namespace WHERE nspname %s $1 ESCAPE '!' AND nspname NOT LIKE 'pg!_%%' ESCAPE '!' AND nspname != 'information_schema'", op), + likePattern, + ) + if err != nil { + return nil, nil, fmt.Errorf("building pg_namespace resolution query: %w", err) + } + + nsResults, err := conn.Exec(ctx, nsQ).ReadAll() + if err != nil { + return nil, nil, fmt.Errorf("querying pg_namespace for schemas matching %q: %w", pattern, err) + } + + var hidden []string + if len(nsResults) > 0 { + for _, row := range nsResults[0].Rows { + name := string(row[0]) + if _, ok := visible[name]; !ok { + hidden = append(hidden, sanitize.QuotePostgresIdentifier(name)) + } + } + } + + return schemas, hidden, nil +} + +// RelKindOrdinaryTable and RelKindPartitionedTable are the pg_class.relkind +// values ResolveExistingTables reports: an ordinary table or leaf partition, +// and a partitioned table's parent. Both are valid ALTER PUBLICATION ... ADD +// TABLE targets from PostgreSQL 13 onward. +const ( + RelKindOrdinaryTable = byte('r') + RelKindPartitionedTable = byte('p') +) + +// ResolveExistingTables returns the relkind of every publishable relation in +// each of the given (already quoted) schemas, keyed by quoted schema name +// and then quoted table name. One query covers every schema, keeping this to +// a single round-trip regardless of tenant count. +// +// Existence checks for explicitly-listed tables should accept either +// relkind. Auto-discovery should keep only RelKindOrdinaryTable - including +// a partitioned parent would double-count rows already covered by its leaf +// partitions, and TABLESAMPLE rejects partitioned parents outright. +func ResolveExistingTables(ctx context.Context, conn *pgconn.PgConn, quotedSchemas []string) (map[string]map[string]byte, error) { + rawToQuoted := make(map[string]string, len(quotedSchemas)) + args := make([]any, len(quotedSchemas)) + placeholders := make([]string, len(quotedSchemas)) + for i, quotedSchema := range quotedSchemas { + schema, err := sanitize.UnquotePostgresIdentifier(quotedSchema) + if err != nil { + return nil, fmt.Errorf("unquoting schema identifier %q: %w", quotedSchema, err) + } + rawToQuoted[schema] = quotedSchema + args[i] = schema + placeholders[i] = fmt.Sprintf("$%d", i+1) + } + + q, err := sanitize.SQLQuery( + fmt.Sprintf(`SELECT n.nspname AS table_schema, c.relname AS table_name, c.relkind +FROM pg_catalog.pg_class c +JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace +WHERE n.nspname IN (%s) AND c.relkind IN ('r', 'p')`, strings.Join(placeholders, ", ")), + args..., + ) + if err != nil { + return nil, fmt.Errorf("building table resolution query for schema(s) %v: %w", quotedSchemas, err) + } + + results, err := conn.Exec(ctx, q).ReadAll() + if err != nil { + return nil, fmt.Errorf("querying tables in schema(s) %v: %w", quotedSchemas, err) + } + + existing := make(map[string]map[string]byte, len(quotedSchemas)) + for _, quotedSchema := range quotedSchemas { + existing[quotedSchema] = map[string]byte{} + } + if len(results) > 0 { + for _, row := range results[0].Rows { + quotedSchema := rawToQuoted[string(row[0])] + existing[quotedSchema][sanitize.QuotePostgresIdentifier(string(row[1]))] = row[2][0] + } + } + return existing, nil +} + +// globToLike converts an unquoted glob pattern (using '*' as wildcard) into a +// PostgreSQL LIKE pattern that uses '!' as the escape character. +// +// Mapping: +// - '*' → '%' (zero or more characters) +// - '_' → '!_' (literal underscore, not the LIKE single-char wildcard) +// - '%' → '!%' (literal percent, not the LIKE multi-char wildcard) +// - '!' → '!!' (literal escape character) +func globToLike(pattern string) string { + var b strings.Builder + b.Grow(len(pattern) + 4) + for _, ch := range pattern { + switch ch { + case '*': + b.WriteByte('%') + case '_': + b.WriteString("!_") + case '%': + b.WriteString("!%") + case '!': + b.WriteString("!!") + default: + b.WriteRune(ch) + } + } + return b.String() +} + +// escapeLike escapes LIKE metacharacters in s without expanding any wildcards. +// Used for exact quoted-identifier lookups. +func escapeLike(s string) string { + var b strings.Builder + b.Grow(len(s)) + for _, ch := range s { + switch ch { + case '_': + b.WriteString("!_") + case '%': + b.WriteString("!%") + case '!': + b.WriteString("!!") + default: + b.WriteRune(ch) + } + } + return b.String() +} + +// schemaMatchesExcludePattern reports whether quotedSchemaName matches +// excludePattern, using the same pattern syntax as schema_include (exact +// name, '*' glob, or quoted exact identifier). Matches in memory against an +// already-resolved schema list, so no extra DB round-trip is needed. Returns +// an error only if a quoted operand fails to unquote. +func schemaMatchesExcludePattern(quotedSchemaName, excludePattern string) (bool, error) { + schemaName, err := sanitize.UnquotePostgresIdentifier(quotedSchemaName) + if err != nil { + return false, fmt.Errorf("unquoting schema identifier %q: %w", quotedSchemaName, err) + } + + if strings.HasPrefix(excludePattern, `"`) { + unquoted, err := sanitize.UnquotePostgresIdentifier(excludePattern) + if err != nil { + return false, fmt.Errorf("invalid quoted schema identifier %q: %w", excludePattern, err) + } + return schemaName == unquoted, nil + } + + re, err := globToRegexp(strings.ToLower(excludePattern)) + if err != nil { + return false, fmt.Errorf("invalid exclude pattern %q: %w", excludePattern, err) + } + return re.MatchString(strings.ToLower(schemaName)), nil +} + +// diffSchemaSets reports schemas present in current but not previous +// (added) and vice versa (removed). Callers should ignore the result when +// previous is nil - that's the first resolution, not real drift. +func diffSchemaSets(previous, current []string) (added, removed []string) { + previousSet := make(map[string]struct{}, len(previous)) + for _, schema := range previous { + previousSet[schema] = struct{}{} + } + currentSet := make(map[string]struct{}, len(current)) + for _, schema := range current { + currentSet[schema] = struct{}{} + } + for _, schema := range current { + if _, ok := previousSet[schema]; !ok { + added = append(added, schema) + } + } + for _, schema := range previous { + if _, ok := currentSet[schema]; !ok { + removed = append(removed, schema) + } + } + return added, removed +} + +// globToRegexp compiles an unquoted glob pattern ('*' as wildcard) into an +// anchored regexp - the in-memory equivalent of globToLike. +func globToRegexp(pattern string) (*regexp.Regexp, error) { + parts := strings.Split(pattern, "*") + for i, part := range parts { + parts[i] = regexp.QuoteMeta(part) + } + return regexp.Compile("^" + strings.Join(parts, ".*") + "$") +} diff --git a/internal/impl/postgresql/pglogicalstream/multischema/resolver_integration_test.go b/internal/impl/postgresql/pglogicalstream/multischema/resolver_integration_test.go new file mode 100644 index 0000000000..f9605425d9 --- /dev/null +++ b/internal/impl/postgresql/pglogicalstream/multischema/resolver_integration_test.go @@ -0,0 +1,431 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed as a Redpanda Enterprise file under the Redpanda Community +// License (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// https://github.com/redpanda-data/connect/v4/blob/main/licenses/rcl.md + +package multischema + +import ( + "context" + "database/sql" + "fmt" + "log/slog" + "strings" + "testing" + "time" + + _ "github.com/lib/pq" // registers "postgres" driver for sql.Open in tests + + "github.com/jackc/pgx/v5/pgconn" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/testcontainers/testcontainers-go" + "github.com/testcontainers/testcontainers-go/wait" + + "github.com/redpanda-data/benthos/v4/public/service" + "github.com/redpanda-data/benthos/v4/public/service/integration" + + "github.com/redpanda-data/connect/v4/internal/impl/postgresql/pgtest" +) + +func TestIntegrationResolveSchemasReportsInaccessibleSchemas(t *testing.T) { + integration.CheckSkip(t) + + adminURL := createDockerInstance(t) + + adminDB, err := sql.Open("postgres", adminURL) + require.NoError(t, err) + defer adminDB.Close() + + _, err = adminDB.Exec("CREATE SCHEMA visible_schema") + require.NoError(t, err) + _, err = adminDB.Exec("CREATE SCHEMA hidden_schema") + require.NoError(t, err) + + _, err = adminDB.Exec("CREATE ROLE restricted_role LOGIN PASSWORD 'restricted_pw'") + require.NoError(t, err) + _, err = adminDB.Exec("GRANT CONNECT ON DATABASE dbname TO restricted_role") + require.NoError(t, err) + _, err = adminDB.Exec("GRANT USAGE ON SCHEMA visible_schema TO restricted_role") + require.NoError(t, err) + // Deliberately no GRANT on hidden_schema. + + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + + restrictedConfig, err := pgconn.ParseConfig(adminURL) + require.NoError(t, err) + restrictedConfig.User = "restricted_role" + restrictedConfig.Password = "restricted_pw" + delete(restrictedConfig.RuntimeParams, "replication") + + restrictedConn, err := pgconn.ConnectConfig(ctx, restrictedConfig) + require.NoError(t, err) + defer closeConn(t, restrictedConn) + + visible, inaccessible, err := resolveSchemas(ctx, restrictedConn, "*_schema") + require.NoError(t, err) + + assert.Equal(t, []string{`"visible_schema"`}, visible) + assert.Equal(t, []string{`"hidden_schema"`}, inaccessible) +} + +func TestIntegrationResolverExcludedInaccessibleSchemaDoesNotWarn(t *testing.T) { + integration.CheckSkip(t) + + adminURL := createDockerInstance(t) + + adminDB, err := sql.Open("postgres", adminURL) + require.NoError(t, err) + defer adminDB.Close() + + _, err = adminDB.Exec("CREATE SCHEMA tenant_visible") + require.NoError(t, err) + _, err = adminDB.Exec("CREATE SCHEMA tenant_hidden") + require.NoError(t, err) + _, err = adminDB.Exec("CREATE SCHEMA tenant_internal") + require.NoError(t, err) + + _, err = adminDB.Exec("CREATE ROLE restricted_role2 LOGIN PASSWORD 'restricted_pw'") + require.NoError(t, err) + _, err = adminDB.Exec("GRANT CONNECT ON DATABASE dbname TO restricted_role2") + require.NoError(t, err) + _, err = adminDB.Exec("GRANT USAGE ON SCHEMA tenant_visible TO restricted_role2") + require.NoError(t, err) + // Deliberately no GRANT on tenant_hidden or tenant_internal. + + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + + restrictedConfig, err := pgconn.ParseConfig(adminURL) + require.NoError(t, err) + restrictedConfig.User = "restricted_role2" + restrictedConfig.Password = "restricted_pw" + delete(restrictedConfig.RuntimeParams, "replication") + + restrictedConn, err := pgconn.ConnectConfig(ctx, restrictedConfig) + require.NoError(t, err) + defer closeConn(t, restrictedConn) + + logs := pgtest.NewTestLogCapture() + logger := service.NewLoggerFromSlog(slog.New(logs)) + + resolver := NewResolver("tenant_*", []string{"tenant_internal"}) + schemas, err := resolver.Resolve(ctx, restrictedConn, logger) + require.NoError(t, err) + + assert.Equal(t, []string{`"tenant_visible"`}, schemas) + + var sawHidden, sawInternal bool + for _, m := range logs.Messages() { + if strings.Contains(m, "tenant_hidden") { + sawHidden = true + } + if strings.Contains(m, "tenant_internal") { + sawInternal = true + } + } + assert.True(t, sawHidden, "expected a warning naming the non-excluded inaccessible schema tenant_hidden, got: %v", logs.Messages()) + assert.False(t, sawInternal, "must not warn about excluded schema tenant_internal, got: %v", logs.Messages()) +} + +func TestIntegrationResolverWarnsOnSchemaSetDriftBetweenReconnects(t *testing.T) { + integration.CheckSkip(t) + + adminURL := createDockerInstance(t) + + adminDB, err := sql.Open("postgres", adminURL) + require.NoError(t, err) + defer adminDB.Close() + + _, err = adminDB.Exec("CREATE SCHEMA tenant_a") + require.NoError(t, err) + + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + + conn, err := pgconn.Connect(ctx, adminURL) + require.NoError(t, err) + defer closeConn(t, conn) + + logs := pgtest.NewTestLogCapture() + logger := service.NewLoggerFromSlog(slog.New(logs)) + resolver := NewResolver("tenant_*", nil) + + // First resolve: only tenant_a matches. previouslyResolved is nil going + // in, so this must not be treated as drift - no added/removed warning. + schemas, err := resolver.Resolve(ctx, conn, logger) + require.NoError(t, err) + assert.Equal(t, []string{`"tenant_a"`}, schemas) + assertNoMessageContains(t, logs.Messages(), "now also matches", "no longer match") + + // Second resolve (simulating a reconnect): tenant_b now also matches. + // Must warn that it was added, and must not also claim anything was + // removed. + before := len(logs.Messages()) + _, err = adminDB.Exec("CREATE SCHEMA tenant_b") + require.NoError(t, err) + + schemas, err = resolver.Resolve(ctx, conn, logger) + require.NoError(t, err) + assert.ElementsMatch(t, []string{`"tenant_a"`, `"tenant_b"`}, schemas) + + added := logs.Messages()[before:] + assertAnyMessageContains(t, added, "now also matches", "tenant_b") + assertNoMessageContains(t, added, "no longer match") + + // Third resolve (another reconnect): tenant_a is dropped. Must warn that + // it was removed, and must not re-warn about tenant_b (it's no longer + // new - it was already part of the previously resolved set). + before = len(logs.Messages()) + _, err = adminDB.Exec("DROP SCHEMA tenant_a") + require.NoError(t, err) + + schemas, err = resolver.Resolve(ctx, conn, logger) + require.NoError(t, err) + assert.Equal(t, []string{`"tenant_b"`}, schemas) + + removed := logs.Messages()[before:] + assertAnyMessageContains(t, removed, "no longer match", "tenant_a") + assertNoMessageContains(t, removed, "now also matches") +} + +func assertAnyMessageContains(t *testing.T, messages []string, substrs ...string) { + t.Helper() + for _, m := range messages { + matchesAll := true + for _, s := range substrs { + if !strings.Contains(m, s) { + matchesAll = false + break + } + } + if matchesAll { + return + } + } + assert.Fail(t, "expected a log message containing all of the given substrings", "substrings: %v, got: %v", substrs, messages) +} + +func assertNoMessageContains(t *testing.T, messages []string, substrs ...string) { + t.Helper() + for _, m := range messages { + for _, s := range substrs { + assert.NotContains(t, m, s, "unexpected log message: %v", m) + } + } +} + +func TestIntegrationResolveSchemasUUIDSuffixedSchemas(t *testing.T) { + integration.CheckSkip(t) + + adminURL := createDockerInstance(t) + + adminDB, err := sql.Open("postgres", adminURL) + require.NoError(t, err) + defer adminDB.Close() + + const ( + lowerCaseSchema = `"tenant_a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11"` + mixedCaseSchema = `"Tenant_9c0b4ef8-bb6d-6bb9-bd38-0a11a0eebc99"` + ) + _, err = adminDB.Exec(`CREATE SCHEMA ` + lowerCaseSchema) + require.NoError(t, err) + _, err = adminDB.Exec(`CREATE SCHEMA ` + mixedCaseSchema) + require.NoError(t, err) + + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + + conn, err := pgconn.Connect(ctx, adminURL) + require.NoError(t, err) + defer closeConn(t, conn) + + visible, inaccessible, err := resolveSchemas(ctx, conn, "tenant_*") + require.NoError(t, err) + + assert.ElementsMatch(t, []string{lowerCaseSchema, mixedCaseSchema}, visible) + assert.Empty(t, inaccessible) + + exactVisible, _, err := resolveSchemas(ctx, conn, mixedCaseSchema) + require.NoError(t, err) + assert.Equal(t, []string{mixedCaseSchema}, exactVisible) +} + +func TestIntegrationResolveSchemasBareUUIDSchema(t *testing.T) { + integration.CheckSkip(t) + + adminURL := createDockerInstance(t) + + adminDB, err := sql.Open("postgres", adminURL) + require.NoError(t, err) + defer adminDB.Close() + + // Upper-case hex digits, no prefix - quoting is mandatory purely because + // of the hyphens, not because of anything alphabetic. + const bareUUIDSchema = `"A0EEBC99-9C0B-4EF8-BB6D-6BB9BD380A11"` + _, err = adminDB.Exec(`CREATE SCHEMA ` + bareUUIDSchema) + require.NoError(t, err) + + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + + conn, err := pgconn.Connect(ctx, adminURL) + require.NoError(t, err) + defer closeConn(t, conn) + + visible, _, err := resolveSchemas(ctx, conn, "*") + require.NoError(t, err) + assert.Contains(t, visible, bareUUIDSchema) + + visible, _, err = resolveSchemas(ctx, conn, "a0eebc99-*") + require.NoError(t, err) + assert.Equal(t, []string{bareUUIDSchema}, visible) + + visible, _, err = resolveSchemas(ctx, conn, "*-bb6d-*") + require.NoError(t, err) + assert.Equal(t, []string{bareUUIDSchema}, visible) + + visible, _, err = resolveSchemas(ctx, conn, `"a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11"`) + require.NoError(t, err) + assert.Empty(t, visible) +} + +// TestIntegrationResolveExistingTablesReportsRelKind verifies that +// ResolveExistingTables reports a partitioned table's parent and its leaf +// partitions with distinct relkinds, rather than excluding the parent +// outright: callers need both, since a partitioned parent is a valid +// publication member if explicitly listed by the user, but must not be +// auto-discovered alongside its own leaf partitions (that filtering is the +// auto-discovery caller's responsibility, not this function's - see +// logical_stream.go's use of RelKindOrdinaryTable). +func TestIntegrationResolveExistingTablesReportsRelKind(t *testing.T) { + integration.CheckSkip(t) + + adminURL := createDockerInstance(t) + + adminDB, err := sql.Open("postgres", adminURL) + require.NoError(t, err) + defer adminDB.Close() + + _, err = adminDB.Exec("CREATE SCHEMA tenant_a") + require.NoError(t, err) + _, err = adminDB.Exec(` +CREATE TABLE tenant_a.orders ( + id INT NOT NULL, + created_at DATE NOT NULL, + PRIMARY KEY (id, created_at) +) PARTITION BY RANGE (created_at)`) + require.NoError(t, err) + _, err = adminDB.Exec(` +CREATE TABLE tenant_a.orders_2025 PARTITION OF tenant_a.orders + FOR VALUES FROM ('2025-01-01') TO ('2026-01-01')`) + require.NoError(t, err) + _, err = adminDB.Exec(` +CREATE TABLE tenant_a.orders_2026 PARTITION OF tenant_a.orders + FOR VALUES FROM ('2026-01-01') TO ('2027-01-01')`) + require.NoError(t, err) + + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + + conn, err := pgconn.Connect(ctx, adminURL) + require.NoError(t, err) + defer closeConn(t, conn) + + existing, err := ResolveExistingTables(ctx, conn, []string{`"tenant_a"`}) + require.NoError(t, err) + + tables := existing[`"tenant_a"`] + require.Len(t, tables, 3, "the parent and both leaf partitions should all be reported") + assert.Equal(t, RelKindPartitionedTable, tables[`"orders"`], "the partitioned parent should be reported, not excluded") + assert.Equal(t, RelKindOrdinaryTable, tables[`"orders_2025"`]) + assert.Equal(t, RelKindOrdinaryTable, tables[`"orders_2026"`]) +} + +func TestIntegrationResolveSchemasConfigValidation(t *testing.T) { + integration.CheckSkip(t) + + adminURL := createDockerInstance(t) + + adminDB, err := sql.Open("postgres", adminURL) + require.NoError(t, err) + defer adminDB.Close() + + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + + conn, err := pgconn.Connect(ctx, adminURL) + require.NoError(t, err) + defer closeConn(t, conn) + + logger := service.MockResources().Logger() + + t.Run("schema_include matches nothing", func(t *testing.T) { + resolver := NewResolver("nonexistent_schema_zzz_*", nil) + _, err := resolver.Resolve(ctx, conn, logger) + require.Error(t, err) + assert.Contains(t, err.Error(), "no schemas found matching schema_include pattern") + }) + + t.Run("schema_exclude excludes every matched schema", func(t *testing.T) { + _, err = adminDB.Exec("CREATE SCHEMA tenant_a") + require.NoError(t, err) + _, err = adminDB.Exec("CREATE SCHEMA tenant_b") + require.NoError(t, err) + + resolver := NewResolver("tenant_*", []string{"tenant_*"}) + _, err := resolver.Resolve(ctx, conn, logger) + require.Error(t, err) + assert.Contains(t, err.Error(), "matched schema(s)") + assert.Contains(t, err.Error(), "excluded all of them") + assert.NotContains(t, err.Error(), "no schemas found matching schema_include pattern") + }) +} + +func closeConn(t testing.TB, conn *pgconn.PgConn) { + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + require.NoError(t, conn.Close(ctx)) +} + +func createDockerInstance(t *testing.T) (dbURL string) { + ctr, err := testcontainers.Run(t.Context(), "postgres:16", + testcontainers.WithExposedPorts("5432/tcp"), + testcontainers.WithEnv(map[string]string{ + "POSTGRES_PASSWORD": "secret", + "POSTGRES_USER": "user_name", + "POSTGRES_DB": "dbname", + }), + testcontainers.WithCmd("postgres", "-c", "wal_level=logical"), + testcontainers.WithWaitStrategy( + wait.ForListeningPort("5432/tcp").WithStartupTimeout(2*time.Minute), + ), + ) + testcontainers.CleanupContainer(t, ctr) + require.NoError(t, err) + + host, err := ctr.Host(t.Context()) + require.NoError(t, err) + mp, err := ctr.MappedPort(t.Context(), "5432/tcp") + require.NoError(t, err) + + databaseURL := fmt.Sprintf("user=user_name password=secret dbname=dbname sslmode=disable host=%s port=%s replication=database", host, mp.Port()) + + var db *sql.DB + require.Eventually(t, func() bool { + if db != nil { + db.Close() + } + if db, err = sql.Open("postgres", databaseURL); err != nil { + return false + } + return db.Ping() == nil + }, 2*time.Minute, time.Second) + t.Cleanup(func() { require.NoError(t, db.Close()) }) + + return databaseURL +} diff --git a/internal/impl/postgresql/pglogicalstream/multischema/resolver_test.go b/internal/impl/postgresql/pglogicalstream/multischema/resolver_test.go new file mode 100644 index 0000000000..14c5532d89 --- /dev/null +++ b/internal/impl/postgresql/pglogicalstream/multischema/resolver_test.go @@ -0,0 +1,199 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed as a Redpanda Enterprise file under the Redpanda Community +// License (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// https://github.com/redpanda-data/connect/v4/blob/main/licenses/rcl.md + +package multischema + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestGlobToLike(t *testing.T) { + tests := []struct { + input string + expected string + }{ + {"public", "public"}, + {"tenant_*", "tenant!_%"}, + {"*", "%"}, + {"tenant_a", "tenant!_a"}, + {"100%", "100!%"}, + {"a!b", "a!!b"}, + {"multi_*_end", "multi!_%!_end"}, + {"", ""}, + } + for _, tt := range tests { + t.Run(tt.input, func(t *testing.T) { + assert.Equal(t, tt.expected, globToLike(tt.input)) + }) + } +} + +func TestSchemaPatternToLike(t *testing.T) { + tests := []struct { + pattern string + expected string + caseSensitive bool + errContains string + }{ + // Unquoted glob patterns — folded to lower-case, '*' → '%', '_' escaped, + // matched case-insensitively regardless of how the matched schema was created. + {pattern: "public", expected: "public"}, + {pattern: "tenant_*", expected: "tenant!_%"}, + {pattern: "*", expected: "%"}, + {pattern: "schema_1", expected: "schema!_1"}, + // Upper-case is folded: TENANT_* matches the same rows as tenant_*. + {pattern: "TENANT_*", expected: "tenant!_%"}, + // Quoted exact identifier — case preserved, no wildcard expansion, + // matched case-sensitively. + {pattern: `"MySchema"`, expected: "MySchema", caseSensitive: true}, + {pattern: `"schema_1"`, expected: "schema!_1", caseSensitive: true}, + {pattern: `"has%bang!"`, expected: "has!%bang!!", caseSensitive: true}, + // Unterminated quoted identifier → error. + {pattern: `"bad`, errContains: "invalid quoted schema identifier"}, + } + for _, tt := range tests { + t.Run(tt.pattern, func(t *testing.T) { + got, caseSensitive, err := schemaPatternToLike(tt.pattern) + if tt.errContains != "" { + require.Error(t, err) + assert.Contains(t, err.Error(), tt.errContains) + return + } + require.NoError(t, err) + assert.Equal(t, tt.expected, got) + assert.Equal(t, tt.caseSensitive, caseSensitive) + }) + } +} + +func TestEscapeLike(t *testing.T) { + tests := []struct { + input string + expected string + }{ + {"MySchema", "MySchema"}, + {"schema_1", "schema!_1"}, + {"100%", "100!%"}, + {"bang!bang", "bang!!bang"}, + {"has_a%b!c", "has!_a!%b!!c"}, + {"", ""}, + } + for _, tt := range tests { + t.Run(tt.input, func(t *testing.T) { + assert.Equal(t, tt.expected, escapeLike(tt.input)) + }) + } +} + +func TestSchemaMatchesExcludePattern(t *testing.T) { + tests := []struct { + name string + schema string + pattern string + expected bool + errContains string + }{ + // Unquoted patterns - case-insensitive, '*' as wildcard. + {name: "unquoted exact match", schema: `"tenant_a"`, pattern: "tenant_a", expected: true}, + {name: "unquoted exact no match", schema: `"tenant_a"`, pattern: "tenant_b", expected: false}, + {name: "unquoted glob match", schema: `"tenant_test_x"`, pattern: "tenant_test_*", expected: true}, + {name: "unquoted glob no match", schema: `"tenant_prod_x"`, pattern: "tenant_test_*", expected: false}, + {name: "bare wildcard matches everything", schema: `"anything"`, pattern: "*", expected: true}, + // Case-folding: unquoted patterns and unquoted-origin schema names both + // fold to lower-case, mirroring PostgreSQL's identifier folding. + {name: "case-insensitive exact match", schema: `"tenant_a"`, pattern: "TENANT_A", expected: true}, + {name: "case-insensitive glob match", schema: `"Tenant_Test_X"`, pattern: "tenant_test_*", expected: true}, + // Quoted patterns - exact, case-sensitive, no wildcard expansion. + {name: "quoted exact case-sensitive match", schema: `"MySchema"`, pattern: `"MySchema"`, expected: true}, + {name: "quoted exact case mismatch does not match", schema: `"MySchema"`, pattern: `"myschema"`, expected: false}, + {name: "quoted pattern does not expand wildcard", schema: `"tenant_a"`, pattern: `"tenant_*"`, expected: false}, + // Errors - only from a malformed pattern, never from the candidate + // schema name, since that's always freshly quoted by resolveSchemas. + {name: "unterminated quoted pattern errors", schema: `"tenant_a"`, pattern: `"unterminated`, errContains: "invalid quoted schema identifier"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := schemaMatchesExcludePattern(tt.schema, tt.pattern) + if tt.errContains != "" { + require.Error(t, err) + assert.Contains(t, err.Error(), tt.errContains) + return + } + require.NoError(t, err) + assert.Equal(t, tt.expected, got) + }) + } +} + +func TestDiffSchemaSets(t *testing.T) { + tests := []struct { + name string + previous []string + current []string + expectedAdded []string + expectedRemoved []string + }{ + // nil previous is the first resolution, not drift - Resolve's caller + // is responsible for ignoring this result in that case (see the + // doc comment on diffSchemaSets), but the function itself still + // reports every current schema as "added" since it has nothing to + // compare against. + { + name: "nil previous, empty current", + previous: nil, + current: nil, + expectedAdded: nil, expectedRemoved: nil, + }, + { + name: "nil previous, non-empty current", + previous: nil, + current: []string{`"a"`, `"b"`}, + expectedAdded: []string{`"a"`, `"b"`}, expectedRemoved: nil, + }, + { + name: "no change", + previous: []string{`"a"`, `"b"`}, + current: []string{`"a"`, `"b"`}, + expectedAdded: nil, expectedRemoved: nil, + }, + { + name: "added only", + previous: []string{`"a"`}, + current: []string{`"a"`, `"b"`}, + expectedAdded: []string{`"b"`}, expectedRemoved: nil, + }, + { + name: "removed only", + previous: []string{`"a"`, `"b"`}, + current: []string{`"a"`}, + expectedAdded: nil, expectedRemoved: []string{`"b"`}, + }, + { + name: "added and removed", + previous: []string{`"a"`, `"b"`}, + current: []string{`"b"`, `"c"`}, + expectedAdded: []string{`"c"`}, expectedRemoved: []string{`"a"`}, + }, + { + name: "everything replaced", + previous: []string{`"a"`, `"b"`}, + current: []string{`"c"`, `"d"`}, + expectedAdded: []string{`"c"`, `"d"`}, expectedRemoved: []string{`"a"`, `"b"`}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + added, removed := diffSchemaSets(tt.previous, tt.current) + assert.Equal(t, tt.expectedAdded, added, "added") + assert.Equal(t, tt.expectedRemoved, removed, "removed") + }) + } +} diff --git a/internal/impl/postgresql/pglogicalstream/pglogrepl.go b/internal/impl/postgresql/pglogicalstream/pglogrepl.go index f92e222d0d..67f0a2419a 100644 --- a/internal/impl/postgresql/pglogicalstream/pglogrepl.go +++ b/internal/impl/postgresql/pglogicalstream/pglogrepl.go @@ -23,7 +23,6 @@ import ( "encoding/binary" "errors" "fmt" - "slices" "strconv" "strings" "time" @@ -329,47 +328,74 @@ func CreatePublication(ctx context.Context, conn *pgconn.PgConn, publicationName return fmt.Errorf("getting publication tables: %w", err) } - // list of tables to publish is empty and publication is for all tables - // no update is needed - if forAllTables && len(pubTables) == 0 { + // No-op only if the caller still wants FOR ALL TABLES too. Checking + // pubTables instead of tables would always be true here (a FOR ALL + // TABLES publication has no pg_publication_rel rows), silently ignoring + // a caller that has since narrowed to an explicit table list. + if forAllTables && len(tables) == 0 { return nil } - tablesToRemoveFromPublication := []TableFQN{} - tablesToAddToPublication := []TableFQN{} - for _, table := range tables { - if !slices.Contains(pubTables, table) { - tablesToAddToPublication = append(tablesToAddToPublication, table) - } + // Build sets for O(1) lookup — avoids O(n²) slices.Contains when reconciling + // large publication table lists (e.g. 100 schemas × 5 tables = 500 entries). + wantSet := make(map[TableFQN]struct{}, len(tables)) + for _, t := range tables { + wantSet[t] = struct{}{} + } + haveSet := make(map[TableFQN]struct{}, len(pubTables)) + for _, t := range pubTables { + haveSet[t] = struct{}{} } - for _, table := range pubTables { - if !slices.Contains(tables, table) { - tablesToRemoveFromPublication = append(tablesToRemoveFromPublication, table) + var tablesToAdd, tablesToRemove []TableFQN + for _, t := range tables { + if _, ok := haveSet[t]; !ok { + tablesToAdd = append(tablesToAdd, t) + } + } + for _, t := range pubTables { + if _, ok := wantSet[t]; !ok { + tablesToRemove = append(tablesToRemove, t) } } - // remove tables from publication - for _, dropTable := range tablesToRemoveFromPublication { - sq, err := sanitize.SQLQuery(fmt.Sprintf(`ALTER PUBLICATION %s DROP TABLE %s;`, publicationName, dropTable.String())) + // Batch DROP: single ALTER statement for all removed tables. + if len(tablesToRemove) > 0 { + var sb strings.Builder + fmt.Fprintf(&sb, "ALTER PUBLICATION %s DROP TABLE ", publicationName) + for i, t := range tablesToRemove { + if i > 0 { + sb.WriteString(", ") + } + sb.WriteString(t.String()) + } + sb.WriteByte(';') + sq, err := sanitize.SQLQuery(sb.String()) if err != nil { - return fmt.Errorf("sanitizing drop table query: %w", err) + return fmt.Errorf("sanitizing drop tables query: %w", err) } - result = conn.Exec(ctx, sq) - if _, err := result.ReadAll(); err != nil { - return fmt.Errorf("removing table from publication: %w", err) + if _, err := conn.Exec(ctx, sq).ReadAll(); err != nil { + return fmt.Errorf("removing tables from publication: %w", err) } } - // add tables to publication - for _, addTable := range tablesToAddToPublication { - sq, err := sanitize.SQLQuery(fmt.Sprintf("ALTER PUBLICATION %s ADD TABLE %s;", publicationName, addTable.String())) + // Batch ADD: single ALTER statement for all new tables. + if len(tablesToAdd) > 0 { + var sb strings.Builder + fmt.Fprintf(&sb, "ALTER PUBLICATION %s ADD TABLE ", publicationName) + for i, t := range tablesToAdd { + if i > 0 { + sb.WriteString(", ") + } + sb.WriteString(t.String()) + } + sb.WriteByte(';') + sq, err := sanitize.SQLQuery(sb.String()) if err != nil { - return fmt.Errorf("sanitizing add table query: %w", err) + return fmt.Errorf("sanitizing add tables query: %w", err) } - result = conn.Exec(ctx, sq) - if _, err := result.ReadAll(); err != nil { - return fmt.Errorf("adding table to publication: %w", err) + if _, err := conn.Exec(ctx, sq).ReadAll(); err != nil { + return fmt.Errorf("adding tables to publication: %w", err) } } @@ -379,12 +405,33 @@ func CreatePublication(ctx context.Context, conn *pgconn.PgConn, publicationName // GetPublicationTables returns a list of tables currently in the publication // Arguments, in order: list of the tables, exist for all tables, error. func GetPublicationTables(ctx context.Context, conn *pgconn.PgConn, publicationName string) ([]TableFQN, bool, error) { + pubQuery, err := sanitize.SQLQuery(` + SELECT puballtables + FROM pg_publication + WHERE pubname = $1; + `, publicationName) + if err != nil { + return nil, false, fmt.Errorf("getting publication tables: %w", err) + } + + pubRows, err := conn.Exec(ctx, pubQuery).ReadAll() + if err != nil { + return nil, false, fmt.Errorf("getting publication tables: %w", err) + } + if len(pubRows) == 0 || len(pubRows[0].Rows) == 0 { + return nil, false, fmt.Errorf("publication %q does not exist", publicationName) + } + forAllTables := string(pubRows[0].Rows[0][0]) == "t" + query, err := sanitize.SQLQuery(` SELECT DISTINCT - tablename as table_name, - schemaname as schema_name - FROM pg_publication_tables - WHERE pubname = $1 + c.relname AS table_name, + n.nspname AS schema_name + FROM pg_publication_rel pr + JOIN pg_publication p ON p.oid = pr.prpubid + JOIN pg_class c ON c.oid = pr.prrelid + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE p.pubname = $1 ORDER BY schema_name, table_name; `, publicationName) if err != nil { @@ -400,7 +447,7 @@ func GetPublicationTables(ctx context.Context, conn *pgconn.PgConn, publicationN } if len(rows) == 0 || len(rows[0].Rows) == 0 { - return nil, true, nil // Publication exists and is for all tables + return nil, forAllTables, nil } tables := make([]TableFQN, 0, len(rows)) @@ -412,7 +459,7 @@ func GetPublicationTables(ctx context.Context, conn *pgconn.PgConn, publicationN tables = append(tables, TableFQN{Table: table, Schema: schema}) } - return tables, false, nil + return tables, forAllTables, nil } // StartReplicationOptions are the options for the START_REPLICATION command. diff --git a/internal/impl/postgresql/pglogicalstream/pglogrepl_test.go b/internal/impl/postgresql/pglogicalstream/pglogrepl_test.go index 4cc6b81547..b001aa6178 100644 --- a/internal/impl/postgresql/pglogicalstream/pglogrepl_test.go +++ b/internal/impl/postgresql/pglogicalstream/pglogrepl_test.go @@ -255,7 +255,7 @@ func TestIntegrationCreatePublication(t *testing.T) { err = CreatePublication(t.Context(), conn, publicationWithTables, []TableFQN{{schema, `"test_table"`}}) require.NoError(t, err) - tables, forAllTables, err = GetPublicationTables(t.Context(), conn, publicationName) + tables, forAllTables, err = GetPublicationTables(t.Context(), conn, publicationWithTables) require.NoError(t, err) assert.NotEmpty(t, tables) assert.Len(t, tables, 1) @@ -344,6 +344,56 @@ func TestIntegrationCreatePublication(t *testing.T) { assert.False(t, forAllTables) } +// TestIntegrationCreatePublicationNarrowingFromForAllTablesFailsLoudly +// guards against a silent no-op when a pipeline that previously ran with an +// empty table list (FOR ALL TABLES) reconnects with a narrowed, explicit +// table list on the same slot/publication: CreatePublication must attempt +// the reconcile - which Postgres rejects, since tables can't be added to or +// dropped from a FOR ALL TABLES publication - rather than silently leaving +// the publication as FOR ALL TABLES with no error and no warning. +func TestIntegrationCreatePublicationNarrowingFromForAllTablesFailsLoudly(t *testing.T) { + integration.CheckSkip(t) + + cleanup, dbURL := createDockerInstance(t) + defer cleanup() + + ctx, cancel := context.WithTimeout(t.Context(), time.Second*5) + defer cancel() + + conn, err := pgconn.Connect(ctx, dbURL) + require.NoError(t, err) + defer closeConn(t, conn) + + multiReader := conn.Exec(t.Context(), "CREATE TABLE orders (id serial PRIMARY KEY, name text);") + _, err = multiReader.ReadAll() + require.NoError(t, err) + + publicationName := "narrowing_test_publication" + schema := `"public"` + + // First connect: tables left empty, so the publication is created FOR + // ALL TABLES. + err = CreatePublication(t.Context(), conn, publicationName, []TableFQN{}) + require.NoError(t, err) + + tables, forAllTables, err := GetPublicationTables(t.Context(), conn, publicationName) + require.NoError(t, err) + assert.Empty(t, tables) + assert.True(t, forAllTables) + + // Reconnect: the config has since been narrowed to an explicit table + // list. This must fail loudly, not silently leave the publication + // unchanged. + err = CreatePublication(t.Context(), conn, publicationName, []TableFQN{{schema, `"orders"`}}) + require.Error(t, err, "narrowing an existing FOR ALL TABLES publication to an explicit table list should fail, not silently no-op") + + // The publication must still be untouched - still FOR ALL TABLES. + tables, forAllTables, err = GetPublicationTables(t.Context(), conn, publicationName) + require.NoError(t, err) + assert.Empty(t, tables) + assert.True(t, forAllTables) +} + func TestIntegrationStartReplication(t *testing.T) { integration.CheckSkip(t) diff --git a/internal/impl/postgresql/pglogicalstream/sanitize/sanitize.go b/internal/impl/postgresql/pglogicalstream/sanitize/sanitize.go index febdb3311f..c3826b0037 100644 --- a/internal/impl/postgresql/pglogicalstream/sanitize/sanitize.go +++ b/internal/impl/postgresql/pglogicalstream/sanitize/sanitize.go @@ -384,7 +384,7 @@ func QuotePostgresIdentifier(name string) string { // UnquotePostgresIdentifier returns the valid unescaped identifier. func UnquotePostgresIdentifier(quoted string) (string, error) { var output strings.Builder - if !strings.HasPrefix(quoted, `"`) || !strings.HasSuffix(quoted, `"`) || len(quoted) < 2 { + if !strings.HasPrefix(quoted, `"`) || !strings.HasSuffix(quoted, `"`) || len(quoted) < 3 { return "", errors.New("missing quotes for identifier") } unquoted := quoted[1 : len(quoted)-1] diff --git a/internal/impl/postgresql/pglogicalstream/sanitize/sanitize_test.go b/internal/impl/postgresql/pglogicalstream/sanitize/sanitize_test.go index e79b233fb7..4e013ee634 100644 --- a/internal/impl/postgresql/pglogicalstream/sanitize/sanitize_test.go +++ b/internal/impl/postgresql/pglogicalstream/sanitize/sanitize_test.go @@ -280,6 +280,7 @@ func TestIdentifierValidation(t *testing.T) { `_Foobar`, strings.Repeat("a", 63), strings.Repeat("A", 63), + `münchen`, } for _, i := range unquoted { diff --git a/internal/impl/postgresql/signaller_integration_test.go b/internal/impl/postgresql/signaller_integration_test.go index fb70c7ff2d..d3ef8a5a2f 100644 --- a/internal/impl/postgresql/signaller_integration_test.go +++ b/internal/impl/postgresql/signaller_integration_test.go @@ -449,6 +449,7 @@ func startSignallingStream(t *testing.T, inputYAML string) (*pgtest.ReceivedMess } delete(m, "schema") delete(m, "commit_ts_ms") + delete(m, "database_schema") received.Add(m) } return nil