From f903238464baf836923e8c5141c82de5cd712abc Mon Sep 17 00:00:00 2001 From: ness-david-dedu Date: Mon, 6 Jul 2026 21:51:30 +0300 Subject: [PATCH 01/20] postgres_cdc: add multi-schema support --- internal/impl/postgresql/input_pg_stream.go | 56 +++++- internal/impl/postgresql/integration_test.go | 166 +++++++++++++++++- .../impl/postgresql/pglogicalstream/config.go | 6 +- .../pglogicalstream/logical_stream.go | 19 +- .../postgresql/pglogicalstream/pglogrepl.go | 72 +++++--- .../pglogicalstream/schema_resolver.go | 129 ++++++++++++++ .../pglogicalstream/schema_resolver_test.go | 90 ++++++++++ 7 files changed, 496 insertions(+), 42 deletions(-) create mode 100644 internal/impl/postgresql/pglogicalstream/schema_resolver.go create mode 100644 internal/impl/postgresql/pglogicalstream/schema_resolver_test.go diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go index b61792a444..f08856dee5 100644 --- a/internal/impl/postgresql/input_pg_stream.go +++ b/internal/impl/postgresql/input_pg_stream.go @@ -14,6 +14,8 @@ import ( "encoding/json" "errors" "fmt" + "strconv" + "strings" "time" "github.com/Jeffail/checkpoint" @@ -81,6 +83,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 +- pg_schema: The PostgreSQL schema name that the table belongs to (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 @@ -105,8 +108,12 @@ This input adds the following metadata fields to each message: Example(10000). Default(1000)). Field(service.NewStringField(fieldSchema). - Description("The PostgreSQL schema from which to replicate data."). - Examples("public", `"MyCaseSensitiveSchemaNeedingQuotes"`), + Description(`The PostgreSQL schema 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.`). + Examples("public", `"MyCaseSensitiveSchemaNeedingQuotes"`, "tenant_*", "*"), ). 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."). @@ -242,6 +249,15 @@ func newPgStreamInput(conf *service.ParsedConfig, mgr *service.Resources) (s ser if schema, err = conf.FieldString(fieldSchema); err != nil { return nil, err } + if err = validateSchemaPattern(schema); err != nil { + return nil, fmt.Errorf("invalid schema: %w", err) + } + // Normalize unquoted patterns to lower-case: PostgreSQL folds unquoted + // identifiers at creation time, so TENANT_* and tenant_* resolve identically. + // Normalizing early avoids silent case-folding surprises in resolveSchemas. + if !strings.HasPrefix(schema, `"`) { + schema = strings.ToLower(schema) + } if tables, err = conf.FieldStringList(fieldTables); err != nil { return nil, err @@ -321,8 +337,8 @@ func newPgStreamInput(conf *service.ParsedConfig, mgr *service.Resources) (s ser DBConfig: pgConnConfig, TLSConfig: pgConnConfig.TLSConfig, DBRawDSN: dsn, - DBSchema: schema, - DBTables: tables, + DBSchemaPattern: schema, + DBTables: tables, RefreshAuthToken: iamAuthTokenBuilder, IncludeTxnMarkers: includeTxnMarkers, @@ -361,6 +377,37 @@ func newPgStreamInput(conf *service.ParsedConfig, mgr *service.Resources) (s ser return conf.WrapBatchInputExtractTracingSpanMapping("postgres_cdc", r) } +// validateSchemaPattern validates a schema name or glob pattern. +// Accepts exact postgres identifiers (letters/digits/underscores) and glob +// patterns that additionally allow '*' as a wildcard character. +// 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 !strings.HasSuffix(s, `"`) || len(s) < 2 { + return errors.New("unterminated quoted identifier in schema") + } + if strings.ContainsRune(s, '*') { + return errors.New("wildcard '*' is not allowed inside a quoted schema identifier") + } + return nil + } + for i, ch := range s { + if (ch >= 'a' && ch <= 'z') || (ch >= 'A' && ch <= 'Z') || (ch >= '0' && ch <= '9') || ch == '_' || ch == '*' { + continue + } + return fmt.Errorf("invalid character %q at position %d in schema pattern %q", ch, i, s) + } + first := rune(s[0]) + if !(first == '_' || first == '*' || (first >= 'a' && first <= 'z') || (first >= 'A' && first <= 'Z')) { + return fmt.Errorf("schema pattern %q must start with a letter, underscore, or '*'", s) + } + return nil +} + // validateSimpleString ensures we aren't vuln to SQL injection. func validateSimpleString(s string) error { for _, b := range []byte(s) { @@ -475,6 +522,7 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher } batchMsg := service.NewMessage(mb) batchMsg.MetaSet("table", msg.Table) + batchMsg.MetaSet("pg_schema", msg.Schema) batchMsg.MetaSet("operation", string(msg.Operation)) if msg.LSN != nil { batchMsg.MetaSet("lsn", *msg.LSN) diff --git a/internal/impl/postgresql/integration_test.go b/internal/impl/postgresql/integration_test.go index c3115986e3..31eb6b8a05 100644 --- a/internal/impl/postgresql/integration_test.go +++ b/internal/impl/postgresql/integration_test.go @@ -1052,13 +1052,20 @@ postgres_cdc: _, err = db.Exec(`INSERT INTO "FlightsCompositePK" ("Seq", "Name", "CreatedAt") VALUES ($1, $2, $3);`, 2, "bravo", "2006-01-02T15:04:05Z07:00") require.NoError(t, err) - _, err = db.Exec(`INSERT INTO flights (name, created_at) VALUES ($1, $2);`, "bravo", "2006-01-02T15:04:05Z07:00") + var flightsID int + err = db.QueryRow(`INSERT INTO flights (name, created_at) VALUES ($1, $2) RETURNING id;`, "bravo", "2006-01-02T15:04:05Z07:00").Scan(&flightsID) + require.NoError(t, err) + + _, err = db.Exec(`UPDATE flights SET name = $1 WHERE id = $2;`, "charlie", flightsID) + require.NoError(t, err) + + _, err = db.Exec(`DELETE FROM flights WHERE id = $1;`, flightsID) require.NoError(t, err) assert.EventuallyWithT(t, func(c *assert.CollectT) { outBatchMut.Lock() defer outBatchMut.Unlock() - assert.Len(c, outBatches, 4, "got: %#v", outBatches) + assert.Len(c, outBatches, 6, "got: %#v", outBatches) }, time.Second*25, time.Millisecond*100) require.ElementsMatch( @@ -1068,20 +1075,42 @@ postgres_cdc: map[string]any{ "operation": "read", "table": "FlightsCompositePK", + "pg_schema": "public", }, map[string]any{ "operation": "read", "table": "flights", + "pg_schema": "public", }, map[string]any{ - "operation": "insert", - "table": "flights", - "lsn": "XXX/XXX", + "operation": "insert", + "table": "FlightsCompositePK", + "lsn": "XXX/XXX", + "commit_ts_ms": "SET", + "pg_schema": "public", }, map[string]any{ - "operation": "insert", - "table": "FlightsCompositePK", - "lsn": "XXX/XXX", + "operation": "insert", + "table": "flights", + "lsn": "XXX/XXX", + "commit_ts_ms": "SET", + "pg_schema": "public", + }, + map[string]any{ + "operation": "update", + "table": "flights", + "lsn": "XXX/XXX", + "commit_ts_ms": "SET", + "before": "SET", + "pg_schema": "public", + }, + map[string]any{ + "operation": "delete", + "table": "flights", + "lsn": "XXX/XXX", + "commit_ts_ms": "SET", + "before": "SET", + "pg_schema": "public", }, }, ) @@ -1429,3 +1458,124 @@ 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 { + pgSchema 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: 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.pgSchema, _ = msg.MetaGet("pg_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() { _ = stream.Run(t.Context()) }() + 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.pgSchema]++ + } + 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.pgSchema]++ + } + 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") +} diff --git a/internal/impl/postgresql/pglogicalstream/config.go b/internal/impl/postgresql/pglogicalstream/config.go index 73ff4c1c4d..8fed67862e 100644 --- a/internal/impl/postgresql/pglogicalstream/config.go +++ b/internal/impl/postgresql/pglogicalstream/config.go @@ -24,8 +24,10 @@ type Config struct { DBConfig *pgconn.Config DBRawDSN string TLSConfig *tls.Config - DBSchema string - DBTables []string + // DBSchemaPattern is the schema to replicate from. Accepts an exact schema + // name or a glob pattern using '*' as a wildcard (e.g. "tenant_*", "*"). + DBSchemaPattern string + DBTables []string // 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 5ebce38064..092669cba8 100644 --- a/internal/impl/postgresql/pglogicalstream/logical_stream.go +++ b/internal/impl/postgresql/pglogicalstream/logical_stream.go @@ -91,18 +91,29 @@ func NewPgStream(ctx context.Context, config *Config) (*Stream, error) { return nil, err } - schema, err := sanitize.NormalizePostgresIdentifier(config.DBSchema) + schemas, err := resolveSchemas(ctx, dbConn, config.DBSchemaPattern) if err != nil { - return nil, fmt.Errorf("invalid schema name %q: %w", config.DBSchema, err) + return nil, fmt.Errorf("resolving schema pattern %q: %w", config.DBSchemaPattern, err) } + if len(schemas) == 0 { + return nil, fmt.Errorf("no schemas found matching pattern %q", config.DBSchemaPattern) + } + config.Logger.Infof("Schema pattern %q resolved to %d schema(s): %v", config.DBSchemaPattern, len(schemas), schemas) - tables := []TableFQN{} + 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) } - tables = append(tables, TableFQN{Schema: schema, Table: normalized}) + normalizedTables = append(normalizedTables, normalized) + } + + tables := make([]TableFQN, 0, len(schemas)*len(normalizedTables)) + for _, schema := range schemas { + for _, table := range normalizedTables { + tables = append(tables, TableFQN{Schema: schema, Table: table}) + } } batchSize := 1000 if config.BatchSize > 0 { diff --git a/internal/impl/postgresql/pglogicalstream/pglogrepl.go b/internal/impl/postgresql/pglogicalstream/pglogrepl.go index f92e222d0d..df2f80c933 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 + sb.WriteString(fmt.Sprintf("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 + sb.WriteString(fmt.Sprintf("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/schema_resolver.go b/internal/impl/postgresql/pglogicalstream/schema_resolver.go new file mode 100644 index 0000000000..8af4f7712d --- /dev/null +++ b/internal/impl/postgresql/pglogicalstream/schema_resolver.go @@ -0,0 +1,129 @@ +// Copyright 2024 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 pglogicalstream + +import ( + "context" + "fmt" + "strings" + + "github.com/jackc/pgx/v5/pgconn" + + "github.com/redpanda-data/connect/v4/internal/impl/postgresql/pglogicalstream/sanitize" +) + +// resolveSchemas expands a schema name or glob pattern into the set of +// quoted PostgreSQL identifiers that exist in the database. +// +// For unquoted patterns (e.g. "tenant_*") the pattern is matched +// case-insensitively against information_schema.schemata using LIKE, because +// PostgreSQL folds unquoted identifiers to lower-case at creation time. +// +// For quoted identifiers (e.g. `"MySchema"`) an exact case-sensitive lookup +// is performed. +// +// System schemas (pg_* and information_schema) are always excluded so that +// wildcard patterns like "*" do not attempt to replicate catalog tables. +// +// Returns an error if the query fails or if no schemas match the pattern. +// schemaPatternToLike converts a schema name or glob pattern into the LIKE +// pattern used by resolveSchemas. Extracted for unit testing. +// +// For quoted identifiers the inner name is exact-escaped (no wildcard expansion). +// For unquoted patterns the '*' wildcard is converted to '%' and the input is +// folded to lower-case to match PostgreSQL's identifier folding. +func schemaPatternToLike(pattern string) (string, error) { + if strings.HasPrefix(pattern, `"`) { + unquoted, err := sanitize.UnquotePostgresIdentifier(pattern) + if err != nil { + return "", fmt.Errorf("invalid quoted schema identifier %q: %w", pattern, err) + } + return escapeLike(unquoted), nil + } + return globToLike(strings.ToLower(pattern)), nil +} + +func resolveSchemas(ctx context.Context, conn *pgconn.PgConn, pattern string) ([]string, error) { + likePattern, err := schemaPatternToLike(pattern) + if err != nil { + return nil, err + } + + q, err := sanitize.SQLQuery( + "SELECT schema_name FROM information_schema.schemata WHERE schema_name LIKE $1 ESCAPE '!' AND schema_name NOT LIKE 'pg!_%' ESCAPE '!' AND schema_name != 'information_schema'", + likePattern, + ) + if err != nil { + return nil, fmt.Errorf("building schema resolution query: %w", err) + } + + results, err := conn.Exec(ctx, q).ReadAll() + if err != nil { + return nil, fmt.Errorf("querying schemas matching %q: %w", pattern, err) + } + + var schemas []string + if len(results) > 0 { + for _, row := range results[0].Rows { + // 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(string(row[0]))) + } + } + return schemas, 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() +} diff --git a/internal/impl/postgresql/pglogicalstream/schema_resolver_test.go b/internal/impl/postgresql/pglogicalstream/schema_resolver_test.go new file mode 100644 index 0000000000..7ee213f6e4 --- /dev/null +++ b/internal/impl/postgresql/pglogicalstream/schema_resolver_test.go @@ -0,0 +1,90 @@ +// Copyright 2024 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 pglogicalstream + +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 + errContains string + }{ + // Unquoted glob patterns — folded to lower-case, '*' → '%', '_' escaped. + {"public", "public", ""}, + {"tenant_*", "tenant!_%", ""}, + {"*", "%", ""}, + {"schema_1", "schema!_1", ""}, + // Upper-case is folded: TENANT_* matches the same rows as tenant_*. + {"TENANT_*", "tenant!_%", ""}, + // Quoted exact identifier — case preserved, no wildcard expansion. + {`"MySchema"`, "MySchema", ""}, + {`"schema_1"`, "schema!_1", ""}, + {`"has%bang!"`, "has!%bang!!", ""}, + // Unterminated quoted identifier → error. + {`"bad`, "", "invalid quoted schema identifier"}, + } + for _, tt := range tests { + t.Run(tt.pattern, func(t *testing.T) { + got, 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) + }) + } +} + +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)) + }) + } +} From c38fd7a709654da2c871c1acd994ab8084a71be8 Mon Sep 17 00:00:00 2001 From: ness-david-dedu Date: Mon, 6 Jul 2026 22:27:03 +0300 Subject: [PATCH 02/20] postgres_cdc: reject empty quoted schema identifier and fix misleading godoc --- internal/impl/postgresql/input_pg_stream.go | 2 +- internal/impl/postgresql/pglogicalstream/schema_resolver.go | 4 +++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go index f08856dee5..733155a281 100644 --- a/internal/impl/postgresql/input_pg_stream.go +++ b/internal/impl/postgresql/input_pg_stream.go @@ -387,7 +387,7 @@ func validateSchemaPattern(s string) error { return errors.New("schema cannot be empty") } if strings.HasPrefix(s, `"`) { - if !strings.HasSuffix(s, `"`) || len(s) < 2 { + if !strings.HasSuffix(s, `"`) || len(s) < 3 { return errors.New("unterminated quoted identifier in schema") } if strings.ContainsRune(s, '*') { diff --git a/internal/impl/postgresql/pglogicalstream/schema_resolver.go b/internal/impl/postgresql/pglogicalstream/schema_resolver.go index 8af4f7712d..8920121e99 100644 --- a/internal/impl/postgresql/pglogicalstream/schema_resolver.go +++ b/internal/impl/postgresql/pglogicalstream/schema_resolver.go @@ -31,7 +31,9 @@ import ( // System schemas (pg_* and information_schema) are always excluded so that // wildcard patterns like "*" do not attempt to replicate catalog tables. // -// Returns an error if the query fails or if no schemas match the pattern. +// Returns an error if the query fails. Returns (nil, nil) if no schemas match. +// The caller is responsible for treating an empty result as an error. + // schemaPatternToLike converts a schema name or glob pattern into the LIKE // pattern used by resolveSchemas. Extracted for unit testing. // From bda218d4b793dd1a8e91841f3f9172e4521dfd5b Mon Sep 17 00:00:00 2001 From: ness-david-dedu Date: Tue, 7 Jul 2026 17:54:33 +0300 Subject: [PATCH 03/20] postgres_cdc: add tests/current/ Docker Compose + Taskfile manual test harness Two-schema (tenant_a, tenant_b) Postgres 16 setup that exercises the multi-schema CDC pipeline end-to-end. Also adds schema_validation unit tests that verify invalid patterns are rejected at startup without a DB. --- .../postgresql/tests/current/Taskfile.yaml | 98 +++++++++++++++++++ .../tests/current/docker-compose.yaml | 36 +++++++ .../impl/postgresql/tests/current/setup.sql | 33 +++++++ .../postgresql/tests/current/test_config.yaml | 41 ++++++++ 4 files changed, 208 insertions(+) create mode 100644 internal/impl/postgresql/tests/current/Taskfile.yaml create mode 100644 internal/impl/postgresql/tests/current/docker-compose.yaml create mode 100644 internal/impl/postgresql/tests/current/setup.sql create mode 100644 internal/impl/postgresql/tests/current/test_config.yaml diff --git a/internal/impl/postgresql/tests/current/Taskfile.yaml b/internal/impl/postgresql/tests/current/Taskfile.yaml new file mode 100644 index 0000000000..75551c4d34 --- /dev/null +++ b/internal/impl/postgresql/tests/current/Taskfile.yaml @@ -0,0 +1,98 @@ +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 both tenant schemas (triggers insert events) + 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');" + + 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 an empty quoted schema identifier is rejected at startup + env: + PG_DSN: '{{.PG_DSN}}' + cmds: + - | + set +e + go run ../../../../../cmd/redpanda-connect/main.go run \ + --set 'input.postgres_cdc.schema=""' \ + ./test_config.yaml 2>&1 | head -5 + echo "exit $?" + # Expects: "invalid schema" error printed, process exits non-zero. + + # ── 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..ee40a376cc --- /dev/null +++ b/internal/impl/postgresql/tests/current/setup.sql @@ -0,0 +1,33 @@ +-- Multi-schema CDC test setup +-- Tests: schema glob (tenant_*), pg_schema metadata, commit_ts_ms, before (update/delete) + +-- ── Tenant schemas ──────────────────────────────────────────────────────────── + +CREATE SCHEMA IF NOT EXISTS tenant_a; +CREATE SCHEMA IF NOT EXISTS tenant_b; + +-- ── 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() +); + +-- 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; + +-- ── Seed snapshot rows ──────────────────────────────────────────────────────── +-- These are visible during the initial snapshot (stream_snapshot: true). + +INSERT INTO tenant_a.events (name) VALUES ('alice'), ('bob'); +INSERT INTO tenant_b.events (name) VALUES ('carol'); 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..4a994e30c2 --- /dev/null +++ b/internal/impl/postgresql/tests/current/test_config.yaml @@ -0,0 +1,41 @@ +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: tenant_* + 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 = @pg_schema + let lsn = @lsn + let ts_ms = @commit_ts_ms + let before = @before + + root = { + "operation": $op, + "pg_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: {} From df2dca6dff49c4f9c50ca795d89815a7d107859a Mon Sep 17 00:00:00 2001 From: ness-david-dedu Date: Tue, 7 Jul 2026 17:55:08 +0300 Subject: [PATCH 04/20] postgres_cdc: add commit_ts_ms and before metadata fields --- internal/impl/postgresql/input_pg_stream.go | 8 + internal/impl/postgresql/integration_test.go | 6 + .../pglogicalstream/logical_stream.go | 14 ++ .../replication_message_decoders.go | 23 +++ .../pglogicalstream/stream_message.go | 6 + .../tests/schema_validation/validate_test.go | 147 ++++++++++++++++++ 6 files changed, 204 insertions(+) create mode 100644 internal/impl/postgresql/tests/schema_validation/validate_test.go diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go index 733155a281..7acbe0dd62 100644 --- a/internal/impl/postgresql/input_pg_stream.go +++ b/internal/impl/postgresql/input_pg_stream.go @@ -530,6 +530,14 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher if msg.ColumnSchema != nil { batchMsg.MetaSetImmut("schema", service.ImmutableAny{V: msg.ColumnSchema}) } + if msg.CommitTs != nil { + batchMsg.MetaSet("commit_ts_ms", strconv.FormatInt(msg.CommitTs.UnixMilli(), 10)) + } + if msg.Before != nil { + if beforeBytes, err := json.Marshal(msg.Before); err == nil { + batchMsg.MetaSet("before", string(beforeBytes)) + } + } if batcher.Add(batchMsg) { flush = true } diff --git a/internal/impl/postgresql/integration_test.go b/internal/impl/postgresql/integration_test.go index 31eb6b8a05..6087b62a96 100644 --- a/internal/impl/postgresql/integration_test.go +++ b/internal/impl/postgresql/integration_test.go @@ -1029,6 +1029,12 @@ postgres_cdc: if _, ok := d["lsn"]; ok { d["lsn"] = "XXX/XXX" // Consistent LSN for assertions below } + if _, ok := d["commit_ts_ms"]; ok { + d["commit_ts_ms"] = "SET" + } + if _, ok := d["before"]; ok { + d["before"] = "SET" + } delete(d, "schema") // Schema metadata tested separately in TestIntegrationPostgresCDCSchemaMetadata outBatches = append(outBatches, data) } diff --git a/internal/impl/postgresql/pglogicalstream/logical_stream.go b/internal/impl/postgresql/pglogicalstream/logical_stream.go index 092669cba8..bd231a30f2 100644 --- a/internal/impl/postgresql/pglogicalstream/logical_stream.go +++ b/internal/impl/postgresql/pglogicalstream/logical_stream.go @@ -55,6 +55,7 @@ type Stream struct { heartbeat *heartbeat maxSnapshotWorkers int unchangedToastValue any + currentTxCommitTime *time.Time } // NewPgStream creates a new instance of the Stream struct. @@ -517,6 +518,14 @@ func (s *Stream) processChange(ctx context.Context, msgLSN LSN, xld XLogData, re delete(schemaCache, rel.RelationID) } + // Track commit timestamp: set on BEGIN (available for DML), clear on COMMIT. + if begin, ok := logicalMsg.(*BeginMessage); ok { + t := begin.CommitTime + s.currentTxCommitTime = &t + } else if _, ok := logicalMsg.(*CommitMessage); ok { + s.currentTxCommitTime = nil + } + // parse changes inside the transaction message, err := toStreamMessage(logicalMsg, relations, typeMap, s.unchangedToastValue) if err != nil { @@ -562,6 +571,11 @@ func (s *Stream) processChange(ctx context.Context, msgLSN LSN, xld XLogData, re } } + switch message.Operation { + case InsertOpType, UpdateOpType, DeleteOpType: + message.CommitTs = s.currentTxCommitTime + } + lsn := msgLSN.String() message.LSN = &lsn select { diff --git a/internal/impl/postgresql/pglogicalstream/replication_message_decoders.go b/internal/impl/postgresql/pglogicalstream/replication_message_decoders.go index 02466e85ed..1f93398c15 100644 --- a/internal/impl/postgresql/pglogicalstream/replication_message_decoders.go +++ b/internal/impl/postgresql/pglogicalstream/replication_message_decoders.go @@ -133,6 +133,28 @@ func toStreamMessage(logicalMsg Message, relations map[uint32]*RelationMessage, } } message.Data = values + if logicalMsg.OldTuple != nil { + before := map[string]any{} + for idx, col := range logicalMsg.OldTuple.Columns { + if idx >= len(rel.Columns) { + break + } + colName := rel.Columns[idx].Name + switch col.DataType { + case 'n': + before[colName] = nil + case 'u': + before[colName] = unchangedToastValue + case 't': + val, err := decodeTextColumnData(typeMap, col.Data, rel.Columns[idx].DataType, rel.Columns[idx].TypeModifier) + if err != nil { + return nil, fmt.Errorf("unable to decode before column data: %w", err) + } + before[colName] = val + } + } + message.Before = before + } case *DeleteMessage: rel, ok := relations[logicalMsg.RelationID] if !ok { @@ -159,6 +181,7 @@ func toStreamMessage(logicalMsg Message, relations map[uint32]*RelationMessage, } } message.Data = values + message.Before = values case *TruncateMessage: case *TypeMessage: case *OriginMessage: diff --git a/internal/impl/postgresql/pglogicalstream/stream_message.go b/internal/impl/postgresql/pglogicalstream/stream_message.go index 73e6795fee..0fe927f81f 100644 --- a/internal/impl/postgresql/pglogicalstream/stream_message.go +++ b/internal/impl/postgresql/pglogicalstream/stream_message.go @@ -8,6 +8,8 @@ package pglogicalstream +import "time" + // StreamMode represents the mode of the stream at the time of the message type StreamMode string @@ -47,4 +49,8 @@ type StreamMessage struct { // ColumnSchema contains the table's column schema in benthos common schema format. // It is set as message metadata and excluded from JSON serialization. ColumnSchema any `json:"-"` + // CommitTs is the commit timestamp of the enclosing transaction. Nil for snapshot reads. + CommitTs *time.Time `json:"-"` + // Before holds the pre-change row state for update and delete operations. Nil otherwise. + Before any `json:"-"` } diff --git a/internal/impl/postgresql/tests/schema_validation/validate_test.go b/internal/impl/postgresql/tests/schema_validation/validate_test.go new file mode 100644 index 0000000000..a2acdc2e42 --- /dev/null +++ b/internal/impl/postgresql/tests/schema_validation/validate_test.go @@ -0,0 +1,147 @@ +// Copyright 2024 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 schema_validation tests postgres_cdc schema pattern validation +// through the public service config API. No database connection is required: +// invalid schemas are rejected during stream construction (before any network +// I/O), so stream.Run returns synchronously for the invalid-schema cases. +package schema_validation_test + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/redpanda-data/benthos/v4/public/service" + _ "github.com/redpanda-data/benthos/v4/public/components/pure" // registers none tracer and other built-ins + + "github.com/redpanda-data/connect/v4/internal/license" + _ "github.com/redpanda-data/connect/v4/internal/impl/postgresql" // registers postgres_cdc +) + +// postgresStream builds a postgres_cdc stream with the given YAML schema value +// (the raw fragment that appears after "schema: " in YAML) and injects a test +// enterprise license. Returns the stream ready to run. +func postgresStream(t *testing.T, schemaYAML string) *service.Stream { + t.Helper() + yaml := fmt.Sprintf(` +postgres_cdc: + dsn: postgres://testuser:testpass@localhost:5432/testdb?sslmode=disable + schema: %s + slot_name: test_slot + tables: + - events +`, schemaYAML) + + sb := service.NewStreamBuilder() + require.NoError(t, sb.SetLoggerYAML(`level: ERROR`)) + require.NoError(t, sb.AddInputYAML(yaml)) + require.NoError(t, sb.AddBatchConsumerFunc(func(_ context.Context, _ service.MessageBatch) error { + return nil + })) + + stream, err := sb.Build() + require.NoError(t, err) + license.InjectTestService(stream.Resources()) + return stream +} + +// TestInvalidSchemaPatterns verifies that invalid schema values are rejected +// during stream construction — before any database connection is attempted. +// stream.Run returns synchronously (no goroutine needed) when the constructor +// fails. +func TestInvalidSchemaPatterns(t *testing.T) { + tests := []struct { + name string + schemaYAML string + wantErrMsg string + }{ + { + // Regression test: len("") == 2 used to pass the old `len(s) < 2` + // guard. Fixed to `len(s) < 3`. + name: "empty quoted identifier", + schemaYAML: `'""'`, + wantErrMsg: "invalid schema", + }, + { + name: "digit-first unquoted pattern", + schemaYAML: `"1abc"`, + wantErrMsg: "invalid schema", + }, + { + name: "unterminated quoted identifier", + schemaYAML: `'"unclosed'`, + wantErrMsg: "invalid schema", + }, + { + name: "hyphen in unquoted pattern", + schemaYAML: `schema-name`, + wantErrMsg: "invalid schema", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + stream := postgresStream(t, tt.schemaYAML) + + // Run returns synchronously when the input constructor fails — + // no timeout context needed, but use one as a safety net. + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + err := stream.Run(ctx) + require.Error(t, err, "expected stream construction to fail") + assert.Contains(t, err.Error(), tt.wantErrMsg, + "error should indicate schema validation failure") + }) + } +} + +// TestValidSchemaPatterns verifies that valid schema values pass construction +// and are only rejected later (at DB-connect time). We run the stream briefly +// and confirm no "invalid schema" error surfaces. +func TestValidSchemaPatterns(t *testing.T) { + tests := []struct { + name string + schemaYAML string + }{ + {name: "exact unquoted schema", schemaYAML: `public`}, + {name: "glob pattern", schemaYAML: `tenant_*`}, + {name: "wildcard", schemaYAML: `"*"`}, + {name: "quoted exact identifier", schemaYAML: `'"MySchema"'`}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + stream := postgresStream(t, tt.schemaYAML) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + errCh := make(chan error, 1) + go func() { errCh <- stream.Run(ctx) }() + + select { + case err := <-errCh: + // Stream stopped before timeout — must not be a schema error. + if err != nil { + assert.NotContains(t, err.Error(), "invalid schema", + "valid schema %q should not trigger schema validation error", tt.schemaYAML) + } + case <-ctx.Done(): + // Stream is still running after timeout — constructor succeeded, + // stream is attempting DB connection. This is the expected path. + _ = stream.StopWithin(2 * time.Second) + } + }) + } +} From fcb0427d6a1a030a36adf5b963fddf3380565917 Mon Sep 17 00:00:00 2001 From: ness-david-dedu Date: Fri, 10 Jul 2026 12:55:07 +0300 Subject: [PATCH 05/20] postgres_cdc: fix lint and docs --- docs/modules/components/pages/inputs/postgres_cdc.adoc | 10 +++++++++- internal/impl/postgresql/input_pg_stream.go | 8 ++++---- internal/impl/postgresql/pglogicalstream/pglogrepl.go | 4 ++-- .../tests/schema_validation/validate_test.go | 4 ++-- 4 files changed, 17 insertions(+), 9 deletions(-) diff --git a/docs/modules/components/pages/inputs/postgres_cdc.adoc b/docs/modules/components/pages/inputs/postgres_cdc.adoc index a7d64a843b..e1c2257518 100644 --- a/docs/modules/components/pages/inputs/postgres_cdc.adoc +++ b/docs/modules/components/pages/inputs/postgres_cdc.adoc @@ -170,7 +170,11 @@ snapshot_batch_size: 10000 === `schema` -The PostgreSQL schema from which to replicate data. +The PostgreSQL schema 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. *Type*: `string` @@ -182,6 +186,10 @@ The PostgreSQL schema from which to replicate data. schema: public schema: '"MyCaseSensitiveSchemaNeedingQuotes"' + +schema: tenant_* + +schema: '*' ``` === `tables` diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go index e6e1b509f8..dfec0b4881 100644 --- a/internal/impl/postgresql/input_pg_stream.go +++ b/internal/impl/postgresql/input_pg_stream.go @@ -110,9 +110,9 @@ This input adds the following metadata fields to each message: Example(10000). Default(1000)). Field(service.NewStringField(fieldSchema). - Description(`The PostgreSQL schema to replicate data from. Accepts an exact schema name or a glob pattern using ` + "`*`" + ` as a wildcard to match multiple schemas. + Description(`The PostgreSQL schema 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.). +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.`). Examples("public", `"MyCaseSensitiveSchemaNeedingQuotes"`, "tenant_*", "*"), @@ -339,8 +339,8 @@ func newPgStreamInput(conf *service.ParsedConfig, mgr *service.Resources) (s ser DBConfig: pgConnConfig, TLSConfig: pgConnConfig.TLSConfig, DBRawDSN: dsn, - DBSchemaPattern: schema, - DBTables: tables, + DBSchemaPattern: schema, + DBTables: tables, RefreshAuthToken: iamAuthTokenBuilder, IncludeTxnMarkers: includeTxnMarkers, diff --git a/internal/impl/postgresql/pglogicalstream/pglogrepl.go b/internal/impl/postgresql/pglogicalstream/pglogrepl.go index df2f80c933..69385f753f 100644 --- a/internal/impl/postgresql/pglogicalstream/pglogrepl.go +++ b/internal/impl/postgresql/pglogicalstream/pglogrepl.go @@ -360,7 +360,7 @@ func CreatePublication(ctx context.Context, conn *pgconn.PgConn, publicationName // Batch DROP: single ALTER statement for all removed tables. if len(tablesToRemove) > 0 { var sb strings.Builder - sb.WriteString(fmt.Sprintf("ALTER PUBLICATION %s DROP TABLE ", publicationName)) + fmt.Fprintf(&sb, "ALTER PUBLICATION %s DROP TABLE ", publicationName) for i, t := range tablesToRemove { if i > 0 { sb.WriteString(", ") @@ -380,7 +380,7 @@ func CreatePublication(ctx context.Context, conn *pgconn.PgConn, publicationName // Batch ADD: single ALTER statement for all new tables. if len(tablesToAdd) > 0 { var sb strings.Builder - sb.WriteString(fmt.Sprintf("ALTER PUBLICATION %s ADD TABLE ", publicationName)) + fmt.Fprintf(&sb, "ALTER PUBLICATION %s ADD TABLE ", publicationName) for i, t := range tablesToAdd { if i > 0 { sb.WriteString(", ") diff --git a/internal/impl/postgresql/tests/schema_validation/validate_test.go b/internal/impl/postgresql/tests/schema_validation/validate_test.go index a2acdc2e42..e45e4bf2f2 100644 --- a/internal/impl/postgresql/tests/schema_validation/validate_test.go +++ b/internal/impl/postgresql/tests/schema_validation/validate_test.go @@ -21,11 +21,11 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "github.com/redpanda-data/benthos/v4/public/service" _ "github.com/redpanda-data/benthos/v4/public/components/pure" // registers none tracer and other built-ins + "github.com/redpanda-data/benthos/v4/public/service" - "github.com/redpanda-data/connect/v4/internal/license" _ "github.com/redpanda-data/connect/v4/internal/impl/postgresql" // registers postgres_cdc + "github.com/redpanda-data/connect/v4/internal/license" ) // postgresStream builds a postgres_cdc stream with the given YAML schema value From 7bcc3ac5af7cb21e4bbabd97e379443fd477810a Mon Sep 17 00:00:00 2001 From: ness-david-dedu Date: Fri, 10 Jul 2026 13:21:46 +0300 Subject: [PATCH 06/20] postgres_cdc: review fixes and test coverage --- CHANGELOG.md | 4 +++ .../components/pages/inputs/postgres_cdc.adoc | 2 ++ internal/impl/postgresql/input_pg_stream.go | 9 +++-- internal/impl/postgresql/integration_test.go | 33 +++++++++++++++++++ 4 files changed, 45 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 77f8a2e534..93fa77403b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,10 @@ All notable changes to this project will be documented in this file. ## Unreleased +### Added + +- postgres_cdc: Postgres CDC now accepts a glob pattern for the `schema` field (e.g. `tenant_*`), replicating all matching schemas through a single replication slot. Useful for multi-tenant databases where each tenant has its own schema. ([@ness-david-dedu](https://github.com/ness-david-dedu), [#4589](https://github.com/redpanda-data/connect/pull/4589)) + ### Fixed - general: The CGO-enabled distribution binary now embeds the IANA time zone database via the `timetzdata` build tag, matching the other distributions, so `time.LoadLocation` works in minimal runtimes without system tzdata instead of silently falling back to UTC (which shifts JQL date predicates in the `jira` input). ([@squiidz](https://github.com/squiidz), [#4583](https://github.com/redpanda-data/connect/pull/4583)) diff --git a/docs/modules/components/pages/inputs/postgres_cdc.adoc b/docs/modules/components/pages/inputs/postgres_cdc.adoc index e1c2257518..069c0f1bfe 100644 --- a/docs/modules/components/pages/inputs/postgres_cdc.adoc +++ b/docs/modules/components/pages/inputs/postgres_cdc.adoc @@ -176,6 +176,8 @@ When a pattern is used, all schemas whose names match the pattern are replicated Double-quoted identifiers are treated as exact names and do not support wildcards. +Schema pattern matching runs once at pipeline startup. Schemas created after the pipeline starts will not be picked up until the pipeline is restarted. + *Type*: `string` diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go index dfec0b4881..f8d0b0fde3 100644 --- a/internal/impl/postgresql/input_pg_stream.go +++ b/internal/impl/postgresql/input_pg_stream.go @@ -26,6 +26,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/sanitize" "github.com/redpanda-data/connect/v4/internal/license" ) @@ -114,7 +115,9 @@ This input adds the following metadata fields to each message: 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.`). +Double-quoted identifiers are treated as exact names and do not support wildcards. + +Schema pattern matching runs once at pipeline startup. Schemas created after the pipeline starts will not be picked up until the pipeline is restarted.`). Examples("public", `"MyCaseSensitiveSchemaNeedingQuotes"`, "tenant_*", "*"), ). Field(service.NewStringListField(fieldTables). @@ -389,8 +392,8 @@ func validateSchemaPattern(s string) error { return errors.New("schema cannot be empty") } if strings.HasPrefix(s, `"`) { - if !strings.HasSuffix(s, `"`) || len(s) < 3 { - return errors.New("unterminated quoted identifier in schema") + 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") diff --git a/internal/impl/postgresql/integration_test.go b/internal/impl/postgresql/integration_test.go index aeae22b0c7..30068b6d2c 100644 --- a/internal/impl/postgresql/integration_test.go +++ b/internal/impl/postgresql/integration_test.go @@ -1600,3 +1600,36 @@ postgres_cdc: 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 TestIntegrationNoSchemasMatchedError(t *testing.T) { + integration.CheckSkip(t) + databaseURL, _, err := ResourceWithPostgreSQLVersion(t, "16") + require.NoError(t, err) + + tmpl := fmt.Sprintf(` +postgres_cdc: + dsn: %s + slot_name: no_schema_match_slot + schema: nonexistent_schema_zzz_* + tables: + - events +`, databaseURL) + + sb := service.NewStreamBuilder() + require.NoError(t, sb.SetLoggerYAML(`level: ERROR`)) + require.NoError(t, sb.AddInputYAML(tmpl)) + require.NoError(t, sb.AddBatchConsumerFunc(func(_ context.Context, _ service.MessageBatch) error { + return nil + })) + + stream, err := sb.Build() + require.NoError(t, err) + license.InjectTestService(stream.Resources()) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + err = stream.Run(ctx) + require.Error(t, err) + assert.Contains(t, err.Error(), "no schemas found matching pattern") +} From d39de9a3e5976a9868ee97674ab54273da689001 Mon Sep 17 00:00:00 2001 From: ness-david-dedu Date: Mon, 13 Jul 2026 10:26:28 +0300 Subject: [PATCH 07/20] postgres_cdc: fix tests --- .../impl/postgresql/pglogicalstream/sanitize/sanitize.go | 2 +- internal/plugins/cdctest/cdc_conformance_test.go | 7 ------- 2 files changed, 1 insertion(+), 8 deletions(-) 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/plugins/cdctest/cdc_conformance_test.go b/internal/plugins/cdctest/cdc_conformance_test.go index cd5eac4de5..31141ddc2a 100644 --- a/internal/plugins/cdctest/cdc_conformance_test.go +++ b/internal/plugins/cdctest/cdc_conformance_test.go @@ -92,13 +92,6 @@ var knownNonConformant = map[string]map[string]string{ "salesforce_cdc": { "max_parallel_snapshot_tables": "uses max_parallel_snapshot_objects; migrate", }, - "tigerbeetle_cdc": { - "checkpoint_cache": "non-relational; §5 applicability under triage", - "checkpoint_limit": "non-relational; §5 applicability under triage", - "snapshot_max_batch_size": "non-relational; §5 applicability under triage", - "max_parallel_snapshot_tables": "non-relational; §5 applicability under triage", - "stream_snapshot": "non-relational; §5 applicability under triage", - }, } // componentSchema is the subset of the docs.ComponentSpec JSON emitted by From 0aa387adfd3d9a05ecf4ed8464e6584bd1267d71 Mon Sep 17 00:00:00 2001 From: ness-david-dedu Date: Wed, 15 Jul 2026 15:35:56 +0300 Subject: [PATCH 08/20] test(cdctest): waive tigerbeetle_cdc conformance fields --- .../plugins/cdctest/cdc_conformance_test.go | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/internal/plugins/cdctest/cdc_conformance_test.go b/internal/plugins/cdctest/cdc_conformance_test.go index 31141ddc2a..6a3980caba 100644 --- a/internal/plugins/cdctest/cdc_conformance_test.go +++ b/internal/plugins/cdctest/cdc_conformance_test.go @@ -54,6 +54,13 @@ var canonicalFields = []string{ "stream_snapshot", } +// conditionalConnectors lists CDC inputs that are only registered under specific +// build tags (e.g. cgo). They are exempt from the stale-entry guard when not +// registered, but are still subject to conformance checks when they are. +var conditionalConnectors = map[string]bool{ + "tigerbeetle_cdc": true, // requires cgo +} + // knownNonConformant waives specific (connector → field → reason) checks that // have not yet been migrated. New connectors default to strict. Populated from // the actual registry state; shrink it as connectors converge. @@ -92,6 +99,13 @@ var knownNonConformant = map[string]map[string]string{ "salesforce_cdc": { "max_parallel_snapshot_tables": "uses max_parallel_snapshot_objects; migrate", }, + "tigerbeetle_cdc": { + "checkpoint_cache": "uses progress_cache; migrate to checkpoint_cache", + "checkpoint_limit": "no discrete checkpoint limit; TigerBeetle CDC is pure streaming", + "snapshot_max_batch_size": "no snapshot phase; TigerBeetle CDC is pure streaming with no initial snapshot", + "max_parallel_snapshot_tables": "no snapshot phase; TigerBeetle CDC is pure streaming with no initial snapshot", + "stream_snapshot": "no snapshot phase; TigerBeetle CDC is pure streaming with no initial snapshot", + }, } // componentSchema is the subset of the docs.ComponentSpec JSON emitted by @@ -172,6 +186,10 @@ func TestCDCConformance(t *testing.T) { } for name := range knownNonConformant { if _, ok := registered[name]; !ok { + if conditionalConnectors[name] { + t.Logf("SKIPPED stale check for %q: not registered in this build (conditional build tag)", name) + continue + } t.Errorf("stale knownNonConformant entry %q is not a registered CDC input; remove it", name) } } From ba7c3849c22472df2dbd46d92abce94610903f54 Mon Sep 17 00:00:00 2001 From: ness-david-dedu Date: Wed, 15 Jul 2026 17:17:41 +0300 Subject: [PATCH 09/20] postgres_cdc: fix lint --- internal/impl/postgresql/input_pg_stream.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go index deb3edc5bb..11ad70d6c4 100644 --- a/internal/impl/postgresql/input_pg_stream.go +++ b/internal/impl/postgresql/input_pg_stream.go @@ -408,7 +408,7 @@ func validateSchemaPattern(s string) error { return fmt.Errorf("invalid character %q at position %d in schema pattern %q", ch, i, s) } first := rune(s[0]) - if !(first == '_' || first == '*' || (first >= 'a' && first <= 'z') || (first >= 'A' && first <= 'Z')) { + if first != '_' && first != '*' && (first < 'a' || first > 'z') && (first < 'A' || first > 'Z') { return fmt.Errorf("schema pattern %q must start with a letter, underscore, or '*'", s) } return nil From caa0b1ace5d20b919def4ef14191a438f7b67abe Mon Sep 17 00:00:00 2001 From: ness-david-dedu Date: Wed, 22 Jul 2026 23:42:19 +0300 Subject: [PATCH 10/20] postgres_cdc: skip missing tables per-schema instead of failing whole publication --- .../components/pages/inputs/postgres_cdc.adoc | 2 + internal/impl/postgresql/input_pg_stream.go | 4 +- internal/impl/postgresql/integration_test.go | 75 +++++++++++++++++++ .../pglogicalstream/logical_stream.go | 11 +++ .../pglogicalstream/schema_resolver.go | 37 +++++++++ 5 files changed, 128 insertions(+), 1 deletion(-) diff --git a/docs/modules/components/pages/inputs/postgres_cdc.adoc b/docs/modules/components/pages/inputs/postgres_cdc.adoc index 069c0f1bfe..70818a8a1d 100644 --- a/docs/modules/components/pages/inputs/postgres_cdc.adoc +++ b/docs/modules/components/pages/inputs/postgres_cdc.adoc @@ -198,6 +198,8 @@ schema: '*' A list of table names to include in the logical replication. Each table should be specified as a separate item. +When `schema` is a glob pattern, this list is resolved against each matched schema independently: a table missing from a given schema is skipped (with a warning logged) rather than failing replication for every matched schema. + *Type*: `array` diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go index 11ad70d6c4..0ad7e1c887 100644 --- a/internal/impl/postgresql/input_pg_stream.go +++ b/internal/impl/postgresql/input_pg_stream.go @@ -122,7 +122,9 @@ Schema pattern matching runs once at pipeline startup. Schemas created after the Examples("public", `"MyCaseSensitiveSchemaNeedingQuotes"`, "tenant_*", "*"), ). 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."). + Description(`A list of table names to include in the logical replication. Each table should be specified as a separate item. + +When `+"`schema`"+` is a glob pattern, this list is resolved against each matched schema independently: a table missing from a given schema is skipped (with a warning logged) rather than failing replication for every matched schema.`). Example([]string{"my_table_1", `"MyCaseSensitiveTableNeedingQuotes"`})). 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."). diff --git a/internal/impl/postgresql/integration_test.go b/internal/impl/postgresql/integration_test.go index 55ebafe4f1..344c20f006 100644 --- a/internal/impl/postgresql/integration_test.go +++ b/internal/impl/postgresql/integration_test.go @@ -1738,6 +1738,81 @@ postgres_cdc: assert.Equal(t, 1, cdcSchemas["tenant_b"], "expected 1 CDC row from tenant_b") } +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 { + pgSchema 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: 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.pgSchema, _ = msg.MetaGet("pg_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() { _ = stream.Run(t.Context()) }() + 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].pgSchema) + assert.Equal(t, "events", collected[0].table) +} + func TestIntegrationNoSchemasMatchedError(t *testing.T) { integration.CheckSkip(t) databaseURL, _, err := ResourceWithPostgreSQLVersion(t, "16") diff --git a/internal/impl/postgresql/pglogicalstream/logical_stream.go b/internal/impl/postgresql/pglogicalstream/logical_stream.go index 1c0ecf322d..96f4953d46 100644 --- a/internal/impl/postgresql/pglogicalstream/logical_stream.go +++ b/internal/impl/postgresql/pglogicalstream/logical_stream.go @@ -118,10 +118,21 @@ func NewPgStream(ctx context.Context, config *Config) (*Stream, error) { tables := make([]TableFQN, 0, len(schemas)*len(normalizedTables)) for _, schema := range schemas { + existingTables, err := resolveExistingTables(ctx, dbConn, schema) + if err != nil { + return nil, fmt.Errorf("resolving tables in schema %q: %w", schema, err) + } for _, table := range normalizedTables { + if _, ok := existingTables[table]; !ok { + config.Logger.Warnf("table %s.%s not found, skipping (schema %s matched pattern %q but does not contain this table)", schema, table, schema, config.DBSchemaPattern) + continue + } tables = append(tables, TableFQN{Schema: schema, Table: table}) } } + if len(tables) == 0 && len(normalizedTables) > 0 { + return nil, fmt.Errorf("none of the configured tables %v were found in any schema matching pattern %q", config.DBTables, config.DBSchemaPattern) + } batchSize := 1000 if config.BatchSize > 0 { batchSize = config.BatchSize diff --git a/internal/impl/postgresql/pglogicalstream/schema_resolver.go b/internal/impl/postgresql/pglogicalstream/schema_resolver.go index 8920121e99..117b822da9 100644 --- a/internal/impl/postgresql/pglogicalstream/schema_resolver.go +++ b/internal/impl/postgresql/pglogicalstream/schema_resolver.go @@ -82,6 +82,43 @@ func resolveSchemas(ctx context.Context, conn *pgconn.PgConn, pattern string) ([ return schemas, nil } +// resolveExistingTables returns the set of quoted table identifiers that +// actually exist in the given (already quoted) schema. +// +// This is used to resolve a schema glob × table list combination per-schema +// rather than assuming every matched schema contains every listed table. A +// schema matching the glob but missing one of the configured tables (e.g. a +// tenant schema that's still being provisioned) would otherwise cause +// CreatePublication's FOR TABLE clause to reference a non-existent relation, +// failing publication setup for every schema, not just the drifted one. +func resolveExistingTables(ctx context.Context, conn *pgconn.PgConn, quotedSchema string) (map[string]struct{}, error) { + schema, err := sanitize.UnquotePostgresIdentifier(quotedSchema) + if err != nil { + return nil, fmt.Errorf("unquoting schema identifier %q: %w", quotedSchema, err) + } + + q, err := sanitize.SQLQuery( + "SELECT table_name FROM information_schema.tables WHERE table_schema = $1", + schema, + ) + if err != nil { + return nil, fmt.Errorf("building table resolution query for schema %q: %w", quotedSchema, err) + } + + results, err := conn.Exec(ctx, q).ReadAll() + if err != nil { + return nil, fmt.Errorf("querying tables in schema %q: %w", quotedSchema, err) + } + + existing := map[string]struct{}{} + if len(results) > 0 { + for _, row := range results[0].Rows { + existing[sanitize.QuotePostgresIdentifier(string(row[0]))] = struct{}{} + } + } + return existing, nil +} + // globToLike converts an unquoted glob pattern (using '*' as wildcard) into a // PostgreSQL LIKE pattern that uses '!' as the escape character. // From c161947c20bf8051a5fe14e9020beafb446346d9 Mon Sep 17 00:00:00 2001 From: ness-david-dedu Date: Wed, 22 Jul 2026 23:48:02 +0300 Subject: [PATCH 11/20] postgres_cdc: fix lint --- internal/gateway/authz_endpoint_test.go | 6 +++--- internal/impl/otlp/mock_policy_server_test.go | 6 +++--- internal/impl/postgresql/input_pg_stream.go | 2 +- internal/impl/protobuf/processor_protobuf_test.go | 4 ++-- 4 files changed, 9 insertions(+), 9 deletions(-) diff --git a/internal/gateway/authz_endpoint_test.go b/internal/gateway/authz_endpoint_test.go index b7f240b4f6..899e04b048 100644 --- a/internal/gateway/authz_endpoint_test.go +++ b/internal/gateway/authz_endpoint_test.go @@ -20,7 +20,7 @@ import ( policymaterializerv1 "buf.build/gen/go/redpandadata/common/protocolbuffers/go/redpanda/policymaterializer/v1" "connectrpc.com/connect" "golang.org/x/net/http2" - "golang.org/x/net/http2/h2c" //nolint:staticcheck + "golang.org/x/net/http2/h2c" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -65,8 +65,8 @@ func startPolicyMaterializerServer(t *testing.T, svc policymaterializerv1connect lis, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0") require.NoError(t, err) - srv := &http.Server{Handler: h2c.NewHandler(mux, &http2.Server{})} //nolint:staticcheck - go srv.Serve(lis) //nolint:errcheck // test server + srv := &http.Server{Handler: h2c.NewHandler(mux, &http2.Server{})} + go srv.Serve(lis) //nolint:errcheck // test server t.Cleanup(func() { srv.Close() }) return "http://" + lis.Addr().String() diff --git a/internal/impl/otlp/mock_policy_server_test.go b/internal/impl/otlp/mock_policy_server_test.go index 43ae225c7c..64d5781551 100644 --- a/internal/impl/otlp/mock_policy_server_test.go +++ b/internal/impl/otlp/mock_policy_server_test.go @@ -18,7 +18,7 @@ import ( policymaterializerv1 "buf.build/gen/go/redpandadata/common/protocolbuffers/go/redpanda/policymaterializer/v1" "connectrpc.com/connect" "golang.org/x/net/http2" - "golang.org/x/net/http2/h2c" //nolint:staticcheck + "golang.org/x/net/http2/h2c" "github.com/stretchr/testify/require" ) @@ -59,8 +59,8 @@ func startMockPolicyEndpoint(t *testing.T, svc policymaterializerv1connect.Polic lis, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0") require.NoError(t, err) - srv := &http.Server{Handler: h2c.NewHandler(mux, &http2.Server{})} //nolint:staticcheck - go srv.Serve(lis) //nolint:errcheck + srv := &http.Server{Handler: h2c.NewHandler(mux, &http2.Server{})} + go srv.Serve(lis) //nolint:errcheck t.Cleanup(func() { srv.Close() }) return "http://" + lis.Addr().String() diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go index 0ad7e1c887..1bbac06ed1 100644 --- a/internal/impl/postgresql/input_pg_stream.go +++ b/internal/impl/postgresql/input_pg_stream.go @@ -124,7 +124,7 @@ Schema pattern matching runs once at pipeline startup. Schemas created after the 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. -When `+"`schema`"+` is a glob pattern, this list is resolved against each matched schema independently: a table missing from a given schema is skipped (with a warning logged) rather than failing replication for every matched schema.`). +When ` + "`schema`" + ` is a glob pattern, this list is resolved against each matched schema independently: a table missing from a given schema is skipped (with a warning logged) rather than failing replication for every matched schema.`). Example([]string{"my_table_1", `"MyCaseSensitiveTableNeedingQuotes"`})). 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."). diff --git a/internal/impl/protobuf/processor_protobuf_test.go b/internal/impl/protobuf/processor_protobuf_test.go index d7f6fd01d9..92d8f6bd90 100644 --- a/internal/impl/protobuf/processor_protobuf_test.go +++ b/internal/impl/protobuf/processor_protobuf_test.go @@ -51,7 +51,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "golang.org/x/net/http2" - "golang.org/x/net/http2/h2c" //nolint:staticcheck + "golang.org/x/net/http2/h2c" "google.golang.org/protobuf/types/descriptorpb" "github.com/redpanda-data/benthos/v4/public/service" @@ -463,7 +463,7 @@ func runMockBSRServer(t *testing.T, importPath string) string { fileDescriptorSetServer := &fileDescriptorSetServer{fileDescriptorSet: files} mux.Handle(reflectv1beta1connect.NewFileDescriptorSetServiceHandler(fileDescriptorSetServer)) go func() { - if err := http.Serve(listener, h2c.NewHandler(mux, &http2.Server{})); err != nil && !errors.Is(err, http.ErrServerClosed) { //nolint:staticcheck + if err := http.Serve(listener, h2c.NewHandler(mux, &http2.Server{})); err != nil && !errors.Is(err, http.ErrServerClosed) { require.NoError(t, err) } }() From 0ed96fdf0cd8364c58b62e55963d8cc7cb208fbb Mon Sep 17 00:00:00 2001 From: ness-david-dedu Date: Thu, 23 Jul 2026 13:03:26 +0300 Subject: [PATCH 12/20] Update internal/impl/postgresql/pglogicalstream/schema_resolver.go Co-authored-by: Joseph Woodward --- internal/impl/postgresql/pglogicalstream/schema_resolver.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/impl/postgresql/pglogicalstream/schema_resolver.go b/internal/impl/postgresql/pglogicalstream/schema_resolver.go index 117b822da9..8f547bcff7 100644 --- a/internal/impl/postgresql/pglogicalstream/schema_resolver.go +++ b/internal/impl/postgresql/pglogicalstream/schema_resolver.go @@ -1,4 +1,4 @@ -// Copyright 2024 Redpanda Data, Inc. +// 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 From d893e1ac9bc9efc93464b6e776ffea9d240e3884 Mon Sep 17 00:00:00 2001 From: ness-david-dedu Date: Mon, 27 Jul 2026 15:38:08 +0300 Subject: [PATCH 13/20] postgres_cdc: warn when schema pattern matches privilege-hidden schemas information_schema.schemata only lists schemas the connecting role can see, so a schema hidden by missing USAGE was silently dropped from a matched pattern with no signal to the user. resolveSchemas now cross-checks pg_catalog.pg_namespace, which isn't privilege-filtered, and reports those as inaccessible so logical_stream.go can warn instead of skipping silently. --- .../pglogicalstream/logical_stream.go | 5 +- .../pglogicalstream/schema_resolver.go | 49 +++++++++++-- .../schema_resolver_integration_test.go | 71 +++++++++++++++++++ 3 files changed, 118 insertions(+), 7 deletions(-) create mode 100644 internal/impl/postgresql/pglogicalstream/schema_resolver_integration_test.go diff --git a/internal/impl/postgresql/pglogicalstream/logical_stream.go b/internal/impl/postgresql/pglogicalstream/logical_stream.go index 96f4953d46..b7cf77d571 100644 --- a/internal/impl/postgresql/pglogicalstream/logical_stream.go +++ b/internal/impl/postgresql/pglogicalstream/logical_stream.go @@ -98,10 +98,13 @@ func NewPgStream(ctx context.Context, config *Config) (*Stream, error) { return nil, err } - schemas, err := resolveSchemas(ctx, dbConn, config.DBSchemaPattern) + schemas, inaccessibleSchemas, err := resolveSchemas(ctx, dbConn, config.DBSchemaPattern) if err != nil { return nil, fmt.Errorf("resolving schema pattern %q: %w", config.DBSchemaPattern, err) } + if len(inaccessibleSchemas) > 0 { + config.Logger.Warnf("schema pattern %q matches schema(s) %v that the configured role cannot see (missing USAGE privilege); they will be skipped", config.DBSchemaPattern, inaccessibleSchemas) + } if len(schemas) == 0 { return nil, fmt.Errorf("no schemas found matching pattern %q", config.DBSchemaPattern) } diff --git a/internal/impl/postgresql/pglogicalstream/schema_resolver.go b/internal/impl/postgresql/pglogicalstream/schema_resolver.go index 8f547bcff7..3859ccface 100644 --- a/internal/impl/postgresql/pglogicalstream/schema_resolver.go +++ b/internal/impl/postgresql/pglogicalstream/schema_resolver.go @@ -51,10 +51,17 @@ func schemaPatternToLike(pattern string) (string, error) { return globToLike(strings.ToLower(pattern)), nil } -func resolveSchemas(ctx context.Context, conn *pgconn.PgConn, pattern string) ([]string, error) { +// resolveSchemas returns the schemas matching pattern that the connection's +// role has access to (visibleSchemas), plus any schemas that also match +// pattern but are hidden from information_schema.schemata by privileges +// (inaccessibleSchemas). The latter is surfaced separately so callers can +// warn the user instead of silently dropping schemas they expected to be +// included, e.g. `"tenant_*"` matching a schema the configured role can't +// see yet. +func resolveSchemas(ctx context.Context, conn *pgconn.PgConn, pattern string) (visibleSchemas, inaccessibleSchemas []string, err error) { likePattern, err := schemaPatternToLike(pattern) if err != nil { - return nil, err + return nil, nil, err } q, err := sanitize.SQLQuery( @@ -62,24 +69,54 @@ func resolveSchemas(ctx context.Context, conn *pgconn.PgConn, pattern string) ([ likePattern, ) if err != nil { - return nil, fmt.Errorf("building schema resolution query: %w", err) + return nil, nil, fmt.Errorf("building schema resolution query: %w", err) } results, err := conn.Exec(ctx, q).ReadAll() if err != nil { - return nil, fmt.Errorf("querying schemas matching %q: %w", pattern, err) + 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(string(row[0]))) + schemas = append(schemas, sanitize.QuotePostgresIdentifier(name)) } } - return schemas, nil + + // pg_namespace is not privilege-filtered, so any pattern match here that's + // missing from information_schema.schemata means the role lacks USAGE (or + // similar) on that schema rather than the schema simply not existing. + nsQ, err := sanitize.SQLQuery( + "SELECT nspname FROM pg_catalog.pg_namespace WHERE nspname LIKE $1 ESCAPE '!' AND nspname NOT LIKE 'pg!_%' ESCAPE '!' AND nspname != 'information_schema'", + 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 set of quoted table identifiers that diff --git a/internal/impl/postgresql/pglogicalstream/schema_resolver_integration_test.go b/internal/impl/postgresql/pglogicalstream/schema_resolver_integration_test.go new file mode 100644 index 0000000000..6af1d4ad7e --- /dev/null +++ b/internal/impl/postgresql/pglogicalstream/schema_resolver_integration_test.go @@ -0,0 +1,71 @@ +// Copyright 2024 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 pglogicalstream + +import ( + "context" + "database/sql" + "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/redpanda-data/benthos/v4/public/service/integration" +) + +// 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) +} From 0e662de9d78a8903fcb2a77d0a709127442ae83f Mon Sep 17 00:00:00 2001 From: Joseph Woodward Date: Tue, 4 Aug 2026 12:15:05 +0100 Subject: [PATCH 14/20] postgres_cdc: Address minor issues --- docs/modules/components/pages/inputs/postgres_cdc.adoc | 1 + internal/gateway/authz_endpoint_test.go | 6 +++--- internal/impl/otlp/mock_policy_server_test.go | 6 +++--- internal/impl/protobuf/processor_protobuf_test.go | 4 ++-- 4 files changed, 9 insertions(+), 8 deletions(-) diff --git a/docs/modules/components/pages/inputs/postgres_cdc.adoc b/docs/modules/components/pages/inputs/postgres_cdc.adoc index 37adc7fe1b..849adbd8ae 100644 --- a/docs/modules/components/pages/inputs/postgres_cdc.adoc +++ b/docs/modules/components/pages/inputs/postgres_cdc.adoc @@ -199,6 +199,7 @@ schema: '*' A list of table names to include in the logical replication. Each table should be specified as a separate item. When `schema` is a glob pattern, this list is resolved against each matched schema independently: a table missing from a given schema is skipped (with a warning logged) rather than failing replication for every matched schema. + 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. diff --git a/internal/gateway/authz_endpoint_test.go b/internal/gateway/authz_endpoint_test.go index 899e04b048..b7f240b4f6 100644 --- a/internal/gateway/authz_endpoint_test.go +++ b/internal/gateway/authz_endpoint_test.go @@ -20,7 +20,7 @@ import ( policymaterializerv1 "buf.build/gen/go/redpandadata/common/protocolbuffers/go/redpanda/policymaterializer/v1" "connectrpc.com/connect" "golang.org/x/net/http2" - "golang.org/x/net/http2/h2c" + "golang.org/x/net/http2/h2c" //nolint:staticcheck "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -65,8 +65,8 @@ func startPolicyMaterializerServer(t *testing.T, svc policymaterializerv1connect lis, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0") require.NoError(t, err) - srv := &http.Server{Handler: h2c.NewHandler(mux, &http2.Server{})} - go srv.Serve(lis) //nolint:errcheck // test server + srv := &http.Server{Handler: h2c.NewHandler(mux, &http2.Server{})} //nolint:staticcheck + go srv.Serve(lis) //nolint:errcheck // test server t.Cleanup(func() { srv.Close() }) return "http://" + lis.Addr().String() diff --git a/internal/impl/otlp/mock_policy_server_test.go b/internal/impl/otlp/mock_policy_server_test.go index 64d5781551..43ae225c7c 100644 --- a/internal/impl/otlp/mock_policy_server_test.go +++ b/internal/impl/otlp/mock_policy_server_test.go @@ -18,7 +18,7 @@ import ( policymaterializerv1 "buf.build/gen/go/redpandadata/common/protocolbuffers/go/redpanda/policymaterializer/v1" "connectrpc.com/connect" "golang.org/x/net/http2" - "golang.org/x/net/http2/h2c" + "golang.org/x/net/http2/h2c" //nolint:staticcheck "github.com/stretchr/testify/require" ) @@ -59,8 +59,8 @@ func startMockPolicyEndpoint(t *testing.T, svc policymaterializerv1connect.Polic lis, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0") require.NoError(t, err) - srv := &http.Server{Handler: h2c.NewHandler(mux, &http2.Server{})} - go srv.Serve(lis) //nolint:errcheck + srv := &http.Server{Handler: h2c.NewHandler(mux, &http2.Server{})} //nolint:staticcheck + go srv.Serve(lis) //nolint:errcheck t.Cleanup(func() { srv.Close() }) return "http://" + lis.Addr().String() diff --git a/internal/impl/protobuf/processor_protobuf_test.go b/internal/impl/protobuf/processor_protobuf_test.go index 92d8f6bd90..d7f6fd01d9 100644 --- a/internal/impl/protobuf/processor_protobuf_test.go +++ b/internal/impl/protobuf/processor_protobuf_test.go @@ -51,7 +51,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "golang.org/x/net/http2" - "golang.org/x/net/http2/h2c" + "golang.org/x/net/http2/h2c" //nolint:staticcheck "google.golang.org/protobuf/types/descriptorpb" "github.com/redpanda-data/benthos/v4/public/service" @@ -463,7 +463,7 @@ func runMockBSRServer(t *testing.T, importPath string) string { fileDescriptorSetServer := &fileDescriptorSetServer{fileDescriptorSet: files} mux.Handle(reflectv1beta1connect.NewFileDescriptorSetServiceHandler(fileDescriptorSetServer)) go func() { - if err := http.Serve(listener, h2c.NewHandler(mux, &http2.Server{})); err != nil && !errors.Is(err, http.ErrServerClosed) { + if err := http.Serve(listener, h2c.NewHandler(mux, &http2.Server{})); err != nil && !errors.Is(err, http.ErrServerClosed) { //nolint:staticcheck require.NoError(t, err) } }() From b0c39d294c3021c09f531f9c56c559e12b75bcaa Mon Sep 17 00:00:00 2001 From: Joseph Woodward Date: Tue, 4 Aug 2026 12:53:19 +0100 Subject: [PATCH 15/20] postgres_cdc: fix broken test --- internal/impl/postgresql/integration_test.go | 35 ++++++++++---------- 1 file changed, 18 insertions(+), 17 deletions(-) diff --git a/internal/impl/postgresql/integration_test.go b/internal/impl/postgresql/integration_test.go index f675014071..f61148c4a4 100644 --- a/internal/impl/postgresql/integration_test.go +++ b/internal/impl/postgresql/integration_test.go @@ -1831,35 +1831,36 @@ postgres_cdc: assert.Equal(t, "events", collected[0].table) } -func TestIntegrationNoSchemasMatchedError(t *testing.T) { +func TestIntegrationNoSchemasMatchedReturnsError(t *testing.T) { integration.CheckSkip(t) databaseURL, _, err := ResourceWithPostgreSQLVersion(t, "16") require.NoError(t, err) tmpl := fmt.Sprintf(` -postgres_cdc: - dsn: %s - slot_name: no_schema_match_slot - schema: nonexistent_schema_zzz_* - tables: - - events +dsn: %s +slot_name: no_schema_match_slot +schema: nonexistent_schema_zzz_* +tables: + - events `, databaseURL) - sb := service.NewStreamBuilder() - require.NoError(t, sb.SetLoggerYAML(`level: ERROR`)) - require.NoError(t, sb.AddInputYAML(tmpl)) - require.NoError(t, sb.AddBatchConsumerFunc(func(_ context.Context, _ service.MessageBatch) error { - return nil - })) + conf, err := newPostgresCDCConfig().ParseYAML(tmpl, nil) + require.NoError(t, err) - stream, err := sb.Build() + mgr := service.MockResources() + license.InjectTestService(mgr) + + input, err := newPgStreamInput(conf, mgr) require.NoError(t, err) - license.InjectTestService(stream.Resources()) - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - err = stream.Run(ctx) + // Bypass the benthos AsyncReader's infinite connect-retry loop by calling + // Connect directly: a schema-pattern-not-found error is permanent, but + // stream.Run has no path to surface it (it only returns once ctx is done), + // so going through StreamBuilder/Run here would just time out instead. + err = input.Connect(ctx) require.Error(t, err) assert.Contains(t, err.Error(), "no schemas found matching pattern") } From 3d253f881e1315f09329f18ca35ee97b53baa75c Mon Sep 17 00:00:00 2001 From: Joseph Woodward Date: Tue, 4 Aug 2026 13:09:51 +0100 Subject: [PATCH 16/20] postgres_cdc: replace pg_schema with database_schema --- internal/impl/postgresql/input_pg_stream.go | 4 +- internal/impl/postgresql/integration_test.go | 70 +++++++++---------- .../impl/postgresql/tests/current/setup.sql | 2 +- .../postgresql/tests/current/test_config.yaml | 4 +- 4 files changed, 40 insertions(+), 40 deletions(-) diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go index 13c0a41fd6..5cecc17457 100644 --- a/internal/impl/postgresql/input_pg_stream.go +++ b/internal/impl/postgresql/input_pg_stream.go @@ -85,7 +85,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 -- pg_schema: The PostgreSQL schema name that the table belongs to (e.g. "public", "tenant_foo"). Useful for per-schema routing when using schema patterns. +- 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 @@ -586,7 +586,7 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher } batchMsg := service.NewMessage(mb) batchMsg.MetaSet("table", msg.Table) - batchMsg.MetaSet("pg_schema", msg.Schema) + 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/integration_test.go b/internal/impl/postgresql/integration_test.go index f61148c4a4..e183f0f517 100644 --- a/internal/impl/postgresql/integration_test.go +++ b/internal/impl/postgresql/integration_test.go @@ -1249,44 +1249,44 @@ postgres_cdc: outBatches, []any{ map[string]any{ - "operation": "read", - "table": "FlightsCompositePK", - "pg_schema": "public", + "operation": "read", + "table": "FlightsCompositePK", + "database_schema": "public", }, map[string]any{ - "operation": "read", - "table": "flights", - "pg_schema": "public", + "operation": "read", + "table": "flights", + "database_schema": "public", }, map[string]any{ - "operation": "insert", - "table": "FlightsCompositePK", - "lsn": "XXX/XXX", - "commit_ts_ms": "SET", - "pg_schema": "public", + "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", - "pg_schema": "public", + "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", - "pg_schema": "public", + "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", - "pg_schema": "public", + "operation": "delete", + "table": "flights", + "lsn": "XXX/XXX", + "commit_ts_ms": "SET", + "before": "SET", + "database_schema": "public", }, }, ) @@ -1656,7 +1656,7 @@ func TestIntegrationMultiSchemaSnapshotAndCDC(t *testing.T) { require.NoError(t, err) type msgMeta struct { - pgSchema string + dbSchema string table string operation string lsn string @@ -1685,7 +1685,7 @@ postgres_cdc: defer mu.Unlock() for _, msg := range batch { m := msgMeta{} - m.pgSchema, _ = msg.MetaGet("pg_schema") + m.dbSchema, _ = msg.MetaGet("database_schema") m.table, _ = msg.MetaGet("table") m.operation, _ = msg.MetaGet("operation") m.lsn, _ = msg.MetaGet("lsn") @@ -1738,7 +1738,7 @@ postgres_cdc: 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.pgSchema]++ + 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") @@ -1750,7 +1750,7 @@ postgres_cdc: 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.pgSchema]++ + 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") @@ -1777,7 +1777,7 @@ func TestIntegrationMultiSchemaMissingTableDegradesGracefully(t *testing.T) { require.NoError(t, err) type msgMeta struct { - pgSchema string + dbSchema string table string } @@ -1804,7 +1804,7 @@ postgres_cdc: defer mu.Unlock() for _, msg := range batch { m := msgMeta{} - m.pgSchema, _ = msg.MetaGet("pg_schema") + m.dbSchema, _ = msg.MetaGet("database_schema") m.table, _ = msg.MetaGet("table") collected = append(collected, m) } @@ -1827,7 +1827,7 @@ postgres_cdc: mu.Lock() defer mu.Unlock() require.Len(t, collected, 1) - assert.Equal(t, "tenant_a", collected[0].pgSchema) + assert.Equal(t, "tenant_a", collected[0].dbSchema) assert.Equal(t, "events", collected[0].table) } diff --git a/internal/impl/postgresql/tests/current/setup.sql b/internal/impl/postgresql/tests/current/setup.sql index ee40a376cc..512f8373fa 100644 --- a/internal/impl/postgresql/tests/current/setup.sql +++ b/internal/impl/postgresql/tests/current/setup.sql @@ -1,5 +1,5 @@ -- Multi-schema CDC test setup --- Tests: schema glob (tenant_*), pg_schema metadata, commit_ts_ms, before (update/delete) +-- Tests: schema glob (tenant_*), database_schema metadata, commit_ts_ms, before (update/delete) -- ── Tenant schemas ──────────────────────────────────────────────────────────── diff --git a/internal/impl/postgresql/tests/current/test_config.yaml b/internal/impl/postgresql/tests/current/test_config.yaml index 4a994e30c2..9e40b5a1c1 100644 --- a/internal/impl/postgresql/tests/current/test_config.yaml +++ b/internal/impl/postgresql/tests/current/test_config.yaml @@ -15,14 +15,14 @@ pipeline: - mapping: | let op = @operation let tbl = @table - let schema = @pg_schema + let schema = @database_schema let lsn = @lsn let ts_ms = @commit_ts_ms let before = @before root = { "operation": $op, - "pg_schema": $schema, + "database_schema": $schema, "table": $tbl, "payload": this, "lsn": if $lsn != null { $lsn } else { null }, From 1f1b68f77a31004c3b7ab3f977f36d8be764dbb3 Mon Sep 17 00:00:00 2001 From: Joseph Woodward Date: Tue, 4 Aug 2026 13:20:58 +0100 Subject: [PATCH 17/20] postgres_cdc: move schema validation to unit test closer to use --- .../impl/postgresql/input_pg_stream_test.go | 72 +++++++++ .../tests/schema_validation/validate_test.go | 147 ------------------ 2 files changed, 72 insertions(+), 147 deletions(-) create mode 100644 internal/impl/postgresql/input_pg_stream_test.go delete mode 100644 internal/impl/postgresql/tests/schema_validation/validate_test.go diff --git a/internal/impl/postgresql/input_pg_stream_test.go b/internal/impl/postgresql/input_pg_stream_test.go new file mode 100644 index 0000000000..19d9723bc2 --- /dev/null +++ b/internal/impl/postgresql/input_pg_stream_test.go @@ -0,0 +1,72 @@ +// Copyright 2024 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 pgstream + +import ( + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/redpanda-data/benthos/v4/public/service" + + "github.com/redpanda-data/connect/v4/internal/license" +) + +// TestSchemaPatternValidation verifies that the schema 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 TestSchemaPatternValidation(t *testing.T) { + tests := []struct { + pattern string + errContains string + }{ + {"public", ""}, + {"tenant_*", ""}, + {"*", ""}, + {`"MySchema"`, ""}, + // Regression test: len("") == 2 used to pass the old `len(s) < 2` guard. + // Fixed to `len(s) < 3`. + {`""`, "invalid quoted schema identifier"}, + {"1abc", "must start with a letter"}, + {`"unclosed`, "invalid quoted schema identifier"}, + {"schema-name", "invalid character"}, + {"", "schema cannot be empty"}, + {`"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: '%s' +slot_name: test_slot +tables: + - events +`, tt.pattern) + + conf, err := newPostgresCDCConfig().ParseYAML(yaml, nil) + require.NoError(t, err) + + mgr := service.MockResources() + license.InjectTestService(mgr) + + _, err = newPgStreamInput(conf, mgr) + if tt.errContains != "" { + require.Error(t, err) + assert.Contains(t, err.Error(), tt.errContains) + return + } + require.NoError(t, err) + }) + } +} diff --git a/internal/impl/postgresql/tests/schema_validation/validate_test.go b/internal/impl/postgresql/tests/schema_validation/validate_test.go deleted file mode 100644 index e45e4bf2f2..0000000000 --- a/internal/impl/postgresql/tests/schema_validation/validate_test.go +++ /dev/null @@ -1,147 +0,0 @@ -// Copyright 2024 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 schema_validation tests postgres_cdc schema pattern validation -// through the public service config API. No database connection is required: -// invalid schemas are rejected during stream construction (before any network -// I/O), so stream.Run returns synchronously for the invalid-schema cases. -package schema_validation_test - -import ( - "context" - "fmt" - "testing" - "time" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" - - _ "github.com/redpanda-data/benthos/v4/public/components/pure" // registers none tracer and other built-ins - "github.com/redpanda-data/benthos/v4/public/service" - - _ "github.com/redpanda-data/connect/v4/internal/impl/postgresql" // registers postgres_cdc - "github.com/redpanda-data/connect/v4/internal/license" -) - -// postgresStream builds a postgres_cdc stream with the given YAML schema value -// (the raw fragment that appears after "schema: " in YAML) and injects a test -// enterprise license. Returns the stream ready to run. -func postgresStream(t *testing.T, schemaYAML string) *service.Stream { - t.Helper() - yaml := fmt.Sprintf(` -postgres_cdc: - dsn: postgres://testuser:testpass@localhost:5432/testdb?sslmode=disable - schema: %s - slot_name: test_slot - tables: - - events -`, schemaYAML) - - sb := service.NewStreamBuilder() - require.NoError(t, sb.SetLoggerYAML(`level: ERROR`)) - require.NoError(t, sb.AddInputYAML(yaml)) - require.NoError(t, sb.AddBatchConsumerFunc(func(_ context.Context, _ service.MessageBatch) error { - return nil - })) - - stream, err := sb.Build() - require.NoError(t, err) - license.InjectTestService(stream.Resources()) - return stream -} - -// TestInvalidSchemaPatterns verifies that invalid schema values are rejected -// during stream construction — before any database connection is attempted. -// stream.Run returns synchronously (no goroutine needed) when the constructor -// fails. -func TestInvalidSchemaPatterns(t *testing.T) { - tests := []struct { - name string - schemaYAML string - wantErrMsg string - }{ - { - // Regression test: len("") == 2 used to pass the old `len(s) < 2` - // guard. Fixed to `len(s) < 3`. - name: "empty quoted identifier", - schemaYAML: `'""'`, - wantErrMsg: "invalid schema", - }, - { - name: "digit-first unquoted pattern", - schemaYAML: `"1abc"`, - wantErrMsg: "invalid schema", - }, - { - name: "unterminated quoted identifier", - schemaYAML: `'"unclosed'`, - wantErrMsg: "invalid schema", - }, - { - name: "hyphen in unquoted pattern", - schemaYAML: `schema-name`, - wantErrMsg: "invalid schema", - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - stream := postgresStream(t, tt.schemaYAML) - - // Run returns synchronously when the input constructor fails — - // no timeout context needed, but use one as a safety net. - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - err := stream.Run(ctx) - require.Error(t, err, "expected stream construction to fail") - assert.Contains(t, err.Error(), tt.wantErrMsg, - "error should indicate schema validation failure") - }) - } -} - -// TestValidSchemaPatterns verifies that valid schema values pass construction -// and are only rejected later (at DB-connect time). We run the stream briefly -// and confirm no "invalid schema" error surfaces. -func TestValidSchemaPatterns(t *testing.T) { - tests := []struct { - name string - schemaYAML string - }{ - {name: "exact unquoted schema", schemaYAML: `public`}, - {name: "glob pattern", schemaYAML: `tenant_*`}, - {name: "wildcard", schemaYAML: `"*"`}, - {name: "quoted exact identifier", schemaYAML: `'"MySchema"'`}, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - stream := postgresStream(t, tt.schemaYAML) - - ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) - defer cancel() - - errCh := make(chan error, 1) - go func() { errCh <- stream.Run(ctx) }() - - select { - case err := <-errCh: - // Stream stopped before timeout — must not be a schema error. - if err != nil { - assert.NotContains(t, err.Error(), "invalid schema", - "valid schema %q should not trigger schema validation error", tt.schemaYAML) - } - case <-ctx.Done(): - // Stream is still running after timeout — constructor succeeded, - // stream is attempting DB connection. This is the expected path. - _ = stream.StopWithin(2 * time.Second) - } - }) - } -} From 836f1cd6bf181bb0061ff78089e11343f40e1ed9 Mon Sep 17 00:00:00 2001 From: Joseph Woodward Date: Tue, 4 Aug 2026 13:26:13 +0100 Subject: [PATCH 18/20] postgres_cdc: clean up redundant comment --- .../pglogicalstream/schema_resolver.go | 27 ++++++++----------- 1 file changed, 11 insertions(+), 16 deletions(-) diff --git a/internal/impl/postgresql/pglogicalstream/schema_resolver.go b/internal/impl/postgresql/pglogicalstream/schema_resolver.go index 3859ccface..4737f7c4cb 100644 --- a/internal/impl/postgresql/pglogicalstream/schema_resolver.go +++ b/internal/impl/postgresql/pglogicalstream/schema_resolver.go @@ -18,22 +18,6 @@ import ( "github.com/redpanda-data/connect/v4/internal/impl/postgresql/pglogicalstream/sanitize" ) -// resolveSchemas expands a schema name or glob pattern into the set of -// quoted PostgreSQL identifiers that exist in the database. -// -// For unquoted patterns (e.g. "tenant_*") the pattern is matched -// case-insensitively against information_schema.schemata using LIKE, because -// PostgreSQL folds unquoted identifiers to lower-case at creation time. -// -// For quoted identifiers (e.g. `"MySchema"`) an exact case-sensitive lookup -// is performed. -// -// System schemas (pg_* and information_schema) are always excluded so that -// wildcard patterns like "*" do not attempt to replicate catalog tables. -// -// Returns an error if the query fails. Returns (nil, nil) if no schemas match. -// The caller is responsible for treating an empty result as an error. - // schemaPatternToLike converts a schema name or glob pattern into the LIKE // pattern used by resolveSchemas. Extracted for unit testing. // @@ -58,6 +42,17 @@ func schemaPatternToLike(pattern string) (string, error) { // warn the user instead of silently dropping schemas they expected to be // included, e.g. `"tenant_*"` matching a schema the configured role can't // see yet. +// +// For unquoted patterns (e.g. "tenant_*") the pattern is matched +// case-insensitively via LIKE, because PostgreSQL folds unquoted identifiers +// to lower-case at creation time. For quoted identifiers (e.g. `"MySchema"`) +// an exact case-sensitive lookup is performed. System schemas (pg_* and +// information_schema) are always excluded so that wildcard patterns like "*" +// do not attempt to replicate catalog tables. +// +// Returned schema names are quoted PostgreSQL identifiers. Returns an error +// if either query fails; returns a nil visibleSchemas slice (with a nil err) +// if no schemas match — callers should treat that as an error condition. func resolveSchemas(ctx context.Context, conn *pgconn.PgConn, pattern string) (visibleSchemas, inaccessibleSchemas []string, err error) { likePattern, err := schemaPatternToLike(pattern) if err != nil { From 620c9da5c1509ada217260b9c255cbf4af373ed2 Mon Sep 17 00:00:00 2001 From: Joseph Woodward Date: Tue, 4 Aug 2026 13:34:44 +0100 Subject: [PATCH 19/20] postgres_cdc: normalie test structure --- internal/impl/postgresql/integration_test.go | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/internal/impl/postgresql/integration_test.go b/internal/impl/postgresql/integration_test.go index e183f0f517..26217596a5 100644 --- a/internal/impl/postgresql/integration_test.go +++ b/internal/impl/postgresql/integration_test.go @@ -1697,7 +1697,11 @@ postgres_cdc: stream, err := sb.Build() require.NoError(t, err) license.InjectTestService(stream.Resources()) - go func() { _ = stream.Run(t.Context()) }() + 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. @@ -1814,7 +1818,11 @@ postgres_cdc: stream, err := sb.Build() require.NoError(t, err) license.InjectTestService(stream.Resources()) - go func() { _ = stream.Run(t.Context()) }() + 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. From d276c1e743008386e00bf580079f0fce5560adc2 Mon Sep 17 00:00:00 2001 From: Joseph Woodward Date: Tue, 4 Aug 2026 13:46:46 +0100 Subject: [PATCH 20/20] postgres_cdc: t.Context() --- internal/impl/postgresql/integration_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/impl/postgresql/integration_test.go b/internal/impl/postgresql/integration_test.go index 26217596a5..e68882e7bc 100644 --- a/internal/impl/postgresql/integration_test.go +++ b/internal/impl/postgresql/integration_test.go @@ -1861,7 +1861,7 @@ tables: input, err := newPgStreamInput(conf, mgr) require.NoError(t, err) - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) defer cancel() // Bypass the benthos AsyncReader's infinite connect-retry loop by calling