Feature/postgres cdc multi schema

CHANGELOG.md

@@ -69,6 +69,7 @@ All notable changes to this project will be documented in this file.
 
 ### 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))
 - aws_dynamodb_cdc: DynamoDB CDC now supports an optional checkpoint_namespace field, allowing multiple independent pipelines to share a single checkpoint table without overwriting each other's checkpoints. ([@squiidz](https://github.com/squiidz), [#4602](https://github.com/redpanda-data/connect/pull/4602))
 
 ### Fixed

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

@@ -170,7 +170,13 @@ 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.
+
+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`
@@ -182,12 +188,18 @@ The PostgreSQL schema from which to replicate data.
 schema: public
 
 schema: '"MyCaseSensitiveSchemaNeedingQuotes"'
+
+schema: tenant_*
+
+schema: '*'
 ```
 
 === `tables`
 
 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.
 
 

internal/impl/postgresql/input_pg_stream.go

@@ -15,6 +15,7 @@ import (
 	"errors"
 	"fmt"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -26,6 +27,7 @@ import (
 
 	"github.com/redpanda-data/connect/v4/internal/asyncroutine"
 	"github.com/redpanda-data/connect/v4/internal/impl/postgresql/pglogicalstream"
+	"github.com/redpanda-data/connect/v4/internal/impl/postgresql/pglogicalstream/sanitize"
 	"github.com/redpanda-data/connect/v4/internal/license"
 )
 
@@ -83,6 +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
+- 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
@@ -113,12 +116,20 @@ 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.
+
+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).
 			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.
+
 If left empty, the underlying PostgreSQL publication is created ` + "`FOR ALL TABLES`" + `, which replicates every table in every schema of the database, ignoring ` + "`" + fieldSchema + "`" + `. This also disables ` + "`" + fieldStreamSnapshot + "`" + `, since the initial snapshot is only planned for tables listed here.`).
 			Example([]string{"my_table_1", `"MyCaseSensitiveTableNeedingQuotes"`})).
 		Field(service.NewIntField(fieldCheckpointLimit).
@@ -262,6 +273,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
@@ -341,7 +361,7 @@ func newPgStreamInput(conf *service.ParsedConfig, mgr *service.Resources) (s ser
 			DBConfig:         pgConnConfig,
 			TLSConfig:        pgConnConfig.TLSConfig,
 			DBRawDSN:         dsn,
-			DBSchema:         schema,
+			DBSchemaPattern:  schema,
 			DBTables:         tables,
 			RefreshAuthToken: iamAuthTokenBuilder,
 
@@ -381,6 +401,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 _, err := sanitize.UnquotePostgresIdentifier(s); err != nil {
+			return fmt.Errorf("invalid quoted schema identifier: %w", err)
+		}
+		if strings.ContainsRune(s, '*') {
+			return errors.New("wildcard '*' is not allowed inside a quoted schema identifier")
+		}
+		return nil
+	}
+	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) {
@@ -535,6 +586,7 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher
 				}
 				batchMsg := service.NewMessage(mb)
 				batchMsg.MetaSet("table", msg.Table)
+				batchMsg.MetaSet("database_schema", msg.Schema)
 				batchMsg.MetaSet("operation", string(msg.Operation))
 				if msg.LSN != nil {
 					batchMsg.MetaSet("lsn", *msg.LSN)

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)
+		})
+	}
+}