diff --git a/CHANGELOG.md b/CHANGELOG.md index f0a20d2a9d..3196a0dc19 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -92,6 +92,8 @@ All notable changes to this project will be documented in this file. ### Added +- postgres_cdc: Added a new `schema_include` field accepting a glob pattern (e.g. `tenant_*`), replicating all matching schemas through a single replication slot. Useful for multi-tenant databases where each tenant has its own schema. Leaving `tables` unset auto-discovers every table in each matched schema instead of listing them by hand. The existing `schema` field is unaffected and continues to take a single exact schema name (defaulting to `public`); `schema` and `schema_include` are mutually exclusive. ([@ness-david-dedu](https://github.com/ness-david-dedu), [#4589](https://github.com/redpanda-data/connect/pull/4589)) +- postgres_cdc: Added a new `schema_exclude` field to carve exceptions out of a broad `schema_include` (e.g. `schema_include: tenant_*` while skipping `tenant_test`). Accepts the same exact-name/glob/quoted syntax as `schema_include`, matches entries against the already-resolved schema list in memory with no extra database round-trips, and requires `schema_include` to be set. ([@ness-david-dedu](https://github.com/ness-david-dedu), [#4589](https://github.com/redpanda-data/connect/pull/4589)) - aws_dynamodb_cdc: DynamoDB CDC now supports an optional checkpoint_namespace field, allowing multiple independent pipelines to share a single checkpoint table without overwriting each other's checkpoints. ([@squiidz](https://github.com/squiidz), [#4602](https://github.com/redpanda-data/connect/pull/4602)) ### Fixed diff --git a/docs/modules/components/pages/inputs/postgres_cdc.adoc b/docs/modules/components/pages/inputs/postgres_cdc.adoc index 3ae65ba78f..1f419c285d 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` @@ -176,6 +180,7 @@ The PostgreSQL schema from which to replicate data. *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. 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 and are silently missing from the output - only changes made after the schema is picked up are streamed. +- A schema that is dropped, renamed, or loses its `USAGE` grant between reconnects stops matching and its tables are silently removed from the publication on the next reconnect. A warning is logged when a schema becomes inaccessible due to a lost `USAGE` grant, but that warning is about the schema being inaccessible - it is not logged for a dropped or renamed schema, and nothing is ever logged about the resulting publication drop itself. + +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-*`. + +This field is mutually exclusive with `schema`; when set, it takes over schema resolution entirely and `schema` must be left at its default. + + +*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. + +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/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go index 8ff0e62374..84b8d2044c 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" @@ -85,6 +89,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 +105,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)). @@ -116,13 +121,54 @@ This input adds the following metadata fields to each message: Default(1000)). Field(service.NewStringField(fieldSchema). Description("The PostgreSQL schema from which to replicate data."). - Examples("public", `"MyCaseSensitiveSchemaNeedingQuotes"`), + Examples("public", `"MyCaseSensitiveSchemaNeedingQuotes"`). + Optional(). + Default("public"), + ). + 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. 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 and are silently missing from the output - only changes made after the schema is picked up are streamed. +- A schema that is dropped, renamed, or loses its `+"`USAGE`"+` grant between reconnects stops matching and its tables are silently removed from the publication on the next reconnect. A warning is logged when a schema becomes inaccessible due to a lost `+"`USAGE`"+` grant, but that warning is about the schema being inaccessible - it is not logged for a dropped or renamed schema, and nothing is ever logged about the resulting publication drop itself. + +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-*`"+`. + +This field is mutually exclusive with `+"`"+fieldSchema+"`"+`; when set, it takes over schema resolution entirely and `+"`"+fieldSchema+"`"+` must be left at its default.`). + 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_*"}). + Optional(). + 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. + +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 +257,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 +443,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 +492,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 +548,34 @@ func newPgStreamInput(conf *service.ParsedConfig, mgr *service.Resources) (s ser return conf.WrapBatchInputExtractTracingSpanMapping("postgres_cdc", r) } +// validateSchemaPattern validates a schema name or glob pattern. +// +// Unquoted patterns are matched via ILIKE against stored schema names (see +// resolveSchemas), not parsed as an identifier, so any character or leading +// character is accepted - including hyphens and leading digits - except a +// literal '"'. This lets a glob like "a0eebc99-*" match a UUID-suffixed +// schema that itself had to be created quoted. +// Double-quoted identifiers (e.g. "MySchema") are accepted as exact names; +// wildcards are not allowed inside quotes. +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 +738,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..322de40f24 100644 --- a/internal/impl/postgresql/input_pg_stream_test.go +++ b/internal/impl/postgresql/input_pg_stream_test.go @@ -9,8 +9,10 @@ package pgstream import ( + "fmt" "testing" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/redpanda-data/benthos/v4/public/service" @@ -18,6 +20,220 @@ import ( "github.com/redpanda-data/connect/v4/internal/license" ) +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) +} + +// 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) + }) + } +} + +// TestSchemaAndSchemaIncludeMutuallyExclusive verifies that setting both +// schema (to a non-default value) and schema_include is rejected at config +// construction time. +func TestSchemaAndSchemaIncludeMutuallyExclusive(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 +` + _, err := parsePgStreamInput(t, yaml) + require.Error(t, err) + assert.Contains(t, err.Error(), "schema and schema_include are mutually exclusive") +} + +// TestSchemaIncludeWithDefaultSchemaSucceeds verifies that setting +// schema_include while leaving schema untouched (at its "public" default) is +// allowed. +func TestSchemaIncludeWithDefaultSchemaSucceeds(t *testing.T) { + yaml := ` +dsn: postgres://testuser:testpass@localhost:5432/testdb?sslmode=disable +schema_include: 'tenant_*' +slot_name: test_slot +tables: + - events +` + _, err := parsePgStreamInput(t, yaml) + require.NoError(t, err) +} + +// 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) +} + +// TestSchemaAcceptsUnicodeIdentifier verifies that the schema field (single +// exact-name path) still accepts unquoted unicode identifiers like +// "münchen", matching sanitize.NormalizePostgresIdentifier which is the sole +// validator on this path (see NewPgStream). schema no longer runs through +// validateSchemaPattern, so this guards against that ASCII-only validator +// regressing this path again in the future. +func TestSchemaAcceptsUnicodeIdentifier(t *testing.T) { + yaml := ` +dsn: postgres://testuser:testpass@localhost:5432/testdb?sslmode=disable +schema: münchen +slot_name: test_slot +tables: + - events +` + _, err := parsePgStreamInput(t, yaml) + require.NoError(t, err) +} + func TestNewPgStreamInputSignalTableName(t *testing.T) { env := service.NewEnvironment() spec := newPostgresCDCConfig() diff --git a/internal/impl/postgresql/integration_test.go b/internal/impl/postgresql/integration_test.go index bcb67946a1..afd4d811fa 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,1023 @@ 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 + operation 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: 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") + m.operation, _ = msg.MetaGet("operation") + 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 { + return collectedLen() >= 1 + }, 30*time.Second, 100*time.Millisecond, "timed out waiting for snapshot row from hyphenated UUID schema") + + _, err = db.Exec(fmt.Sprintf(`INSERT INTO "%s".events (name) VALUES ('bob')`, uuidSchema)) + require.NoError(t, err) + + assert.EventuallyWithT(t, func(c *assert.CollectT) { + assert.Equal(c, 2, collectedLen()) + }, 30*time.Second, 100*time.Millisecond, "timed out waiting for CDC row from hyphenated UUID schema") + + mu.Lock() + defer mu.Unlock() + require.Len(t, collected, 2) + for _, m := range collected { + assert.Equal(t, uuidSchema, m.dbSchema, "database_schema metadata should be the raw, unquoted, case-preserved schema name") + assert.Equal(t, "events", m.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 TestIntegrationMultiSchemaIncludeExcludeConfigValidation(t *testing.T) { + integration.CheckSkip(t) + + t.Run("schema_include matches nothing", func(t *testing.T) { + databaseURL, _, err := ResourceWithPostgreSQLVersion(t, "16") + require.NoError(t, err) + + tmpl := fmt.Sprintf(` +dsn: %s +slot_name: no_schema_match_slot +schema_include: nonexistent_schema_zzz_* +tables: + - events +`, 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) + assert.Contains(t, err.Error(), "no schemas found matching schema_include pattern") + }) + + t.Run("schema_include matches nothing with empty tables", func(t *testing.T) { + databaseURL, _, err := ResourceWithPostgreSQLVersion(t, "16") + require.NoError(t, err) + + tmpl := fmt.Sprintf(` +dsn: %s +slot_name: no_schema_match_empty_tables_slot +schema_include: nonexistent_schema_zzz_* +`, 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 + // the "schema_include matches nothing" case above. + err = input.Connect(ctx) + 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) { + databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16") + require.NoError(t, err) + + // Both schemas match tenant_*, but schema_exclude below excludes them all. + 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) + } + + tmpl := fmt.Sprintf(` +dsn: %s +slot_name: schema_exclude_all_matched_slot +schema_include: tenant_* +schema_exclude: + - tenant_* +tables: + - events +`, 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 + // the "schema_include matches nothing" case above. + err = input.Connect(ctx) + 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") + }) +} + +// TestIntegrationSchemaExcludeCarvesOutTenantAutoDiscover is +// TestIntegrationSchemaExcludeCarvesOutTenant with `tables` left unset: it +// verifies that leaving `tables` empty under schema_include auto-discovers +// the "events" table in each matched schema instead of falling back to a +// database-wide FOR ALL TABLES publication, so schema_exclude still carves +// tenant_c out of both the snapshot and CDC. +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) + } + + // 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) + } + + // 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 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 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"}) + }) +} 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..9b29cfdb95 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,92 @@ 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 { + for table := range existingTables { + 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 { diff --git a/internal/impl/postgresql/pglogicalstream/multischema/resolver.go b/internal/impl/postgresql/pglogicalstream/multischema/resolver.go new file mode 100644 index 0000000000..4579d90eb2 --- /dev/null +++ b/internal/impl/postgresql/pglogicalstream/multischema/resolver.go @@ -0,0 +1,344 @@ +// 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" +) + +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 { + // Filtering happens entirely against the schemas slice we already + // fetched above - no extra DB round-trips per exclude pattern. + var excluded []string + 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, 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) + } + 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(remaining), remaining) + } + schemas = remaining + } + + if len(inaccessibleSchemas) > 0 && !slices.Equal(inaccessibleSchemas, r.previouslyInaccessible) { + 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, inaccessibleSchemas) + } + 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 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 +} + +// ResolveExistingTables returns the quoted names of the publishable base +// tables in each of the given (already quoted) schemas, keyed by quoted +// schema name. A single query covering every schema, rather than one per +// schema, keeps this to one round-trip regardless of tenant count - the +// difference between one query and, say, one hundred on a multi-tenant, +// schema-per-tenant database on every connect and reconnect. Restricted to +// table_type = 'BASE TABLE' so a same-named view or foreign table is treated +// as missing rather than breaking CreatePublication. +func ResolveExistingTables(ctx context.Context, conn *pgconn.PgConn, quotedSchemas []string) (map[string]map[string]struct{}, 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 table_schema, table_name FROM information_schema.tables WHERE table_schema IN (%s) AND table_type = 'BASE TABLE'", 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]struct{}, len(quotedSchemas)) + for _, quotedSchema := range quotedSchemas { + existing[quotedSchema] = map[string]struct{}{} + } + if len(results) > 0 { + for _, row := range results[0].Rows { + quotedSchema := rawToQuoted[string(row[0])] + existing[quotedSchema][sanitize.QuotePostgresIdentifier(string(row[1]))] = struct{}{} + } + } + 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..a2497e08b4 --- /dev/null +++ b/internal/impl/postgresql/pglogicalstream/multischema/resolver_integration_test.go @@ -0,0 +1,212 @@ +// 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" + "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/integration" +) + +// closeConn and createDockerInstance mirror the identically named helpers in +// pglogicalstream/pglogrepl_test.go - duplicated here rather than shared, +// since they're small, self-contained, and this package must not import the +// pglogicalstream test package. +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) (cleanup func(), 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, err = sql.Open("postgres", databaseURL); err != nil { + return false + } + return db.Ping() == nil + }, 2*time.Minute, time.Second) + + cleanup = func() { + // Container cleanup is handled by testcontainers.CleanupContainer + } + + return cleanup, databaseURL +} + +// TestIntegrationResolveSchemasReportsInaccessibleSchemas verifies that a +// schema pattern matching a schema the connecting role lacks USAGE on is +// reported via inaccessibleSchemas rather than silently dropped, since +// information_schema.schemata alone would make it indistinguishable from a +// schema that simply doesn't exist. +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 TestIntegrationResolveSchemasUUIDSuffixedSchemas(t *testing.T) { + integration.CheckSkip(t) + + _, adminURL := createDockerInstance(t) + + adminDB, err := sql.Open("postgres", adminURL) + require.NoError(t, err) + defer adminDB.Close() + + // Quoting is mandatory here because of the UUID's hyphens, regardless of + // case - so both of these preserve their literal casing exactly as written. + 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) + + // Both schemas match the unquoted glob despite the case difference in + // their literal prefix - matching is case-insensitive independent of how + // the schema itself was created. + assert.ElementsMatch(t, []string{lowerCaseSchema, mixedCaseSchema}, visible) + assert.Empty(t, inaccessible) + + // A quoted pattern is still exact and case-sensitive: it picks out only + // the schema whose case matches the pattern, with no wildcard expansion. + 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) + + // A bare wildcard has no literal characters to case-compare, so this is + // unaffected by hex-digit casing either way - included as a baseline. + visible, _, err := resolveSchemas(ctx, conn, "*") + require.NoError(t, err) + assert.Contains(t, visible, bareUUIDSchema) + + // The interesting case: an unquoted pattern whose only literal portion is + // a lower-case chunk of the UUID itself (no prefix) must still match the + // upper-case schema case-insensitively. + visible, _, err = resolveSchemas(ctx, conn, "a0eebc99-*") + require.NoError(t, err) + assert.Equal(t, []string{bareUUIDSchema}, visible) + + // Same, but the literal chunk sits in the middle rather than at the start. + visible, _, err = resolveSchemas(ctx, conn, "*-bb6d-*") + require.NoError(t, err) + assert.Equal(t, []string{bareUUIDSchema}, visible) + + // A quoted pattern remains an exact, case-sensitive lookup: the + // differently-cased quoted form matches nothing. + visible, _, err = resolveSchemas(ctx, conn, `"a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11"`) + require.NoError(t, err) + assert.Empty(t, visible) +} 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..d06b5362dd --- /dev/null +++ b/internal/impl/postgresql/pglogicalstream/multischema/resolver_test.go @@ -0,0 +1,134 @@ +// 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) + }) + } +} diff --git a/internal/impl/postgresql/pglogicalstream/pglogrepl.go b/internal/impl/postgresql/pglogicalstream/pglogrepl.go index f92e222d0d..69385f753f 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" @@ -335,41 +334,66 @@ func CreatePublication(ctx context.Context, conn *pgconn.PgConn, publicationName 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) } } 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/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 diff --git a/internal/impl/postgresql/tests/current/Taskfile.yaml b/internal/impl/postgresql/tests/current/Taskfile.yaml new file mode 100644 index 0000000000..da2e348d06 --- /dev/null +++ b/internal/impl/postgresql/tests/current/Taskfile.yaml @@ -0,0 +1,120 @@ +version: "3" + +vars: + PG_DSN: '{{.PG_DSN | default "postgres://postgres:postgres@localhost:5433/testdb?sslmode=disable"}}' + +tasks: + # ── Infrastructure ──────────────────────────────────────────────────────────── + + up: + desc: Start PostgreSQL and run schema/data setup + cmds: + - docker compose up -d postgres + - docker compose run --rm setup + + down: + desc: Stop and remove all containers and volumes + cmds: + - docker compose down -v + + reset: + desc: Full teardown + bring back up (also drops and recreates the replication slot) + cmds: + - task: down + - task: up + + # ── Slot management ─────────────────────────────────────────────────────────── + + slot:drop: + desc: Drop the replication slot so the test can be re-run from scratch + cmds: + - | + psql "{{.PG_DSN}}" -c \ + "SELECT pg_drop_replication_slot('multi_schema_test_slot') + FROM pg_replication_slots + WHERE slot_name = 'multi_schema_test_slot';" + + # ── Run the pipeline ────────────────────────────────────────────────────────── + + run: + desc: Run the multi-schema CDC pipeline (streams snapshot then live CDC) + env: + PG_DSN: '{{.PG_DSN}}' + cmds: + - task: slot:drop + - go run ../../../../../cmd/redpanda-connect/main.go run ./test_config.yaml + + # ── Test data ───────────────────────────────────────────────────────────────── + + data:insert: + desc: Insert CDC rows into all three tenant schemas (tenant_c must not appear in the output — it's excluded) + cmds: + - psql "{{.PG_DSN}}" -c "INSERT INTO tenant_a.events (name) VALUES ('dave'), ('eve');" + - psql "{{.PG_DSN}}" -c "INSERT INTO tenant_b.events (name) VALUES ('frank');" + - psql "{{.PG_DSN}}" -c "INSERT INTO tenant_c.events (name) VALUES ('trudy');" + + data:update: + desc: Update a row in tenant_a (triggers update event with 'before' field) + cmds: + - psql "{{.PG_DSN}}" -c "UPDATE tenant_a.events SET status = 'updated' WHERE name = 'dave';" + + data:delete: + desc: Delete a row from tenant_b (triggers delete event with 'before' field) + cmds: + - psql "{{.PG_DSN}}" -c "DELETE FROM tenant_b.events WHERE name = 'frank';" + + data:all: + desc: Run all test data mutations in sequence (insert → update → delete) + cmds: + - task: data:insert + - task: data:update + - task: data:delete + + # ── Schema validation smoke test ────────────────────────────────────────────── + + test:invalid-schema: + desc: Confirm that a malformed schema_include is rejected at startup, before any DB connection is attempted + env: + PG_DSN: '{{.PG_DSN}}' + cmds: + - | + set +e + go run ../../../../../cmd/redpanda-connect/main.go run \ + --set 'input.postgres_cdc.schema_include=1abc' \ + ./test_config.yaml 2>&1 | head -5 + echo "exit $?" + # Expects: "invalid schema_include" error printed, process exits non-zero. + # schema_include is validated eagerly in newPgStreamInput + # (validateSchemaPattern), unlike schema which is only checked later, + # against the live DB connection, inside NewPgStream. + # Note: an empty schema_include is NOT an error - it means "unset", and + # the connector falls back to the schema field (default "public"). + + test:schema-exclude-requires-include: + desc: Confirm schema_exclude is rejected at startup when schema_include is left unset + env: + PG_DSN: '{{.PG_DSN}}' + cmds: + - | + set +e + go run ../../../../../cmd/redpanda-connect/main.go run \ + ./test_config_schema_exclude_no_include.yaml 2>&1 | head -5 + echo "exit $?" + # Expects: "schema_exclude requires schema_include to be set" error + # printed, process exits non-zero. Uses a dedicated fixture file rather + # than --set on test_config.yaml, since --set rejects an empty RHS + # ("foo=" -> "expected foo=bar syntax"), so schema_include can't be + # cleared that way. + + # ── Quick sanity ────────────────────────────────────────────────────────────── + + psql: + desc: Open a psql shell to the test database + cmds: + - psql "{{.PG_DSN}}" + + status: + desc: Show active replication slots and publications + cmds: + - psql "{{.PG_DSN}}" -c "SELECT slot_name, active FROM pg_replication_slots;" + - psql "{{.PG_DSN}}" -c "SELECT pubname FROM pg_publication;" diff --git a/internal/impl/postgresql/tests/current/docker-compose.yaml b/internal/impl/postgresql/tests/current/docker-compose.yaml new file mode 100644 index 0000000000..896a807027 --- /dev/null +++ b/internal/impl/postgresql/tests/current/docker-compose.yaml @@ -0,0 +1,36 @@ +services: + postgres: + image: postgres:16 + container_name: pgtest-postgres + ports: + - "5433:5432" + environment: + POSTGRES_USER: postgres + POSTGRES_PASSWORD: postgres + POSTGRES_DB: testdb + # Enable logical replication — required for postgres_cdc + command: postgres -c wal_level=logical -c max_replication_slots=10 -c max_wal_senders=10 + healthcheck: + test: ["CMD-SHELL", "pg_isready -U postgres"] + interval: 3s + timeout: 5s + retries: 10 + + # One-shot setup container: creates schemas, tables, and seed data, then exits. + setup: + image: postgres:16 + container_name: pgtest-setup + depends_on: + postgres: + condition: service_healthy + environment: + PGPASSWORD: postgres + volumes: + - ./setup.sql:/setup.sql:ro + entrypoint: /bin/sh + command: + - -c + - | + psql -h postgres -U postgres -d testdb -f /setup.sql + echo "setup complete" + restart: "no" diff --git a/internal/impl/postgresql/tests/current/setup.sql b/internal/impl/postgresql/tests/current/setup.sql new file mode 100644 index 0000000000..d231c381f7 --- /dev/null +++ b/internal/impl/postgresql/tests/current/setup.sql @@ -0,0 +1,49 @@ +-- Multi-schema CDC test setup +-- Tests: schema glob (tenant_*), schema_exclude (tenant_c), database_schema +-- metadata, commit_ts_ms, before (update/delete) + +-- ── Tenant schemas ──────────────────────────────────────────────────────────── +-- tenant_a and tenant_b are replicated; tenant_c matches the tenant_* glob in +-- test_config.yaml but is carved out via schema_exclude — its rows must +-- never appear in the pipeline output. + +CREATE SCHEMA IF NOT EXISTS tenant_a; +CREATE SCHEMA IF NOT EXISTS tenant_b; +CREATE SCHEMA IF NOT EXISTS tenant_c; + +-- ── Events table (same shape in each schema) ────────────────────────────────── + +CREATE TABLE IF NOT EXISTS tenant_a.events ( + id SERIAL PRIMARY KEY, + name TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'active', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE TABLE IF NOT EXISTS tenant_b.events ( + id SERIAL PRIMARY KEY, + name TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'active', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE TABLE IF NOT EXISTS tenant_c.events ( + id SERIAL PRIMARY KEY, + name TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'active', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +-- REPLICA IDENTITY FULL so update/delete messages carry the full before-row. +ALTER TABLE tenant_a.events REPLICA IDENTITY FULL; +ALTER TABLE tenant_b.events REPLICA IDENTITY FULL; +ALTER TABLE tenant_c.events REPLICA IDENTITY FULL; + +-- ── Seed snapshot rows ──────────────────────────────────────────────────────── +-- These are visible during the initial snapshot (stream_snapshot: true). +-- tenant_c's row (mallory) must NOT appear in the output — see schema_exclude +-- in test_config.yaml. + +INSERT INTO tenant_a.events (name) VALUES ('alice'), ('bob'); +INSERT INTO tenant_b.events (name) VALUES ('carol'); +INSERT INTO tenant_c.events (name) VALUES ('mallory'); diff --git a/internal/impl/postgresql/tests/current/test_config.yaml b/internal/impl/postgresql/tests/current/test_config.yaml new file mode 100644 index 0000000000..b440817dd1 --- /dev/null +++ b/internal/impl/postgresql/tests/current/test_config.yaml @@ -0,0 +1,45 @@ +input: + postgres_cdc: + dsn: ${PG_DSN:postgres://postgres:postgres@localhost:5433/testdb?sslmode=disable} + slot_name: multi_schema_test_slot + stream_snapshot: true + # Glob pattern: replicates both tenant_a and tenant_b via one slot. + schema_include: tenant_* + # tenant_c matches the glob above but is carved out here — its rows must + # never appear in the output below, neither during snapshot nor CDC. + schema_exclude: + - tenant_c + tables: + - events + +pipeline: + processors: + # Annotate each message with all new metadata fields so the output clearly + # shows what the feature delivers. + - mapping: | + let op = @operation + let tbl = @table + let schema = @database_schema + let lsn = @lsn + let ts_ms = @commit_ts_ms + let before = @before + + root = { + "operation": $op, + "database_schema": $schema, + "table": $tbl, + "payload": this, + "lsn": if $lsn != null { $lsn } else { null }, + "commit_ts_ms": if $ts_ms != null { $ts_ms } else { null }, + "before": if $before != null { $before.string().parse_json() } else { null }, + } + +output: + stdout: + codec: lines + +logger: + level: INFO + +metrics: + none: {} diff --git a/internal/impl/postgresql/tests/current/test_config_schema_exclude_no_include.yaml b/internal/impl/postgresql/tests/current/test_config_schema_exclude_no_include.yaml new file mode 100644 index 0000000000..014332ed82 --- /dev/null +++ b/internal/impl/postgresql/tests/current/test_config_schema_exclude_no_include.yaml @@ -0,0 +1,22 @@ +# Negative-validation fixture for schema_exclude: schema_include is +# deliberately left unset. --set can't assign an empty string (the CLI +# rejects "foo=" as "expected foo=bar syntax"), so this exists as its own +# file instead of overriding test_config.yaml at run time. +input: + postgres_cdc: + dsn: ${PG_DSN:postgres://postgres:postgres@localhost:5433/testdb?sslmode=disable} + slot_name: schema_exclude_no_include_slot + schema_exclude: + - tenant_c + tables: + - events + +output: + stdout: + codec: lines + +logger: + level: INFO + +metrics: + none: {}