diff --git a/CHANGELOG.md b/CHANGELOG.md
index da4aa7f33f..51e5da7716 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/docs/modules/components/pages/inputs/postgres_cdc.adoc b/docs/modules/components/pages/inputs/postgres_cdc.adoc
index f490b9be8c..849adbd8ae 100644
--- a/docs/modules/components/pages/inputs/postgres_cdc.adoc
+++ b/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.
 
 
diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go
index 2e1cd15b6b..5cecc17457 100644
--- a/internal/impl/postgresql/input_pg_stream.go
+++ b/internal/impl/postgresql/input_pg_stream.go
@@ -15,6 +15,7 @@ import (
 	"errors"
 	"fmt"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -26,6 +27,7 @@ import (
 
 	"github.com/redpanda-data/connect/v4/internal/asyncroutine"
 	"github.com/redpanda-data/connect/v4/internal/impl/postgresql/pglogicalstream"
+	"github.com/redpanda-data/connect/v4/internal/impl/postgresql/pglogicalstream/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)
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/integration_test.go b/internal/impl/postgresql/integration_test.go
index 52d3b1e71c..e68882e7bc 100644
--- a/internal/impl/postgresql/integration_test.go
+++ b/internal/impl/postgresql/integration_test.go
@@ -1249,38 +1249,44 @@ postgres_cdc:
 		outBatches,
 		[]any{
 			map[string]any{
-				"operation": "read",
-				"table":     "FlightsCompositePK",
+				"operation":       "read",
+				"table":           "FlightsCompositePK",
+				"database_schema": "public",
 			},
 			map[string]any{
-				"operation": "read",
-				"table":     "flights",
+				"operation":       "read",
+				"table":           "flights",
+				"database_schema": "public",
 			},
 			map[string]any{
-				"operation":    "insert",
-				"table":        "FlightsCompositePK",
-				"lsn":          "XXX/XXX",
-				"commit_ts_ms": "SET",
+				"operation":       "insert",
+				"table":           "FlightsCompositePK",
+				"lsn":             "XXX/XXX",
+				"commit_ts_ms":    "SET",
+				"database_schema": "public",
 			},
 			map[string]any{
-				"operation":    "insert",
-				"table":        "flights",
-				"lsn":          "XXX/XXX",
-				"commit_ts_ms": "SET",
+				"operation":       "insert",
+				"table":           "flights",
+				"lsn":             "XXX/XXX",
+				"commit_ts_ms":    "SET",
+				"database_schema": "public",
 			},
 			map[string]any{
-				"operation":    "update",
-				"table":        "flights",
-				"lsn":          "XXX/XXX",
-				"commit_ts_ms": "SET",
-				"before":       "SET",
+				"operation":       "update",
+				"table":           "flights",
+				"lsn":             "XXX/XXX",
+				"commit_ts_ms":    "SET",
+				"before":          "SET",
+				"database_schema": "public",
 			},
 			map[string]any{
-				"operation":    "delete",
-				"table":        "flights",
-				"lsn":          "XXX/XXX",
-				"commit_ts_ms": "SET",
-				"before":       "SET",
+				"operation":       "delete",
+				"table":           "flights",
+				"lsn":             "XXX/XXX",
+				"commit_ts_ms":    "SET",
+				"before":          "SET",
+				"database_schema": "public",
 			},
 		},
 	)
@@ -1628,3 +1634,241 @@ postgres_cdc:
 	}
 	assert.Equal(t, "STRING", byName["extra"], "new 'extra' column should have type STRING")
 }
+
+func TestIntegrationMultiSchemaSnapshotAndCDC(t *testing.T) {
+	integration.CheckSkip(t)
+	databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16")
+	require.NoError(t, err)
+
+	// Two tenant schemas with the same table name, replicated on a single slot.
+	for _, schema := range []string{"tenant_a", "tenant_b"} {
+		_, err = db.Exec(fmt.Sprintf("CREATE SCHEMA %s", schema))
+		require.NoError(t, err)
+		_, err = db.Exec(fmt.Sprintf(
+			"CREATE TABLE %s.events (id SERIAL PRIMARY KEY, name TEXT)", schema))
+		require.NoError(t, err)
+	}
+
+	// Pre-load snapshot data: 2 rows in tenant_a, 1 in tenant_b.
+	_, err = db.Exec("INSERT INTO tenant_a.events (name) VALUES ('alice'), ('bob')")
+	require.NoError(t, err)
+	_, err = db.Exec("INSERT INTO tenant_b.events (name) VALUES ('carol')")
+	require.NoError(t, err)
+
+	type msgMeta struct {
+		dbSchema  string
+		table     string
+		operation string
+		lsn       string
+	}
+
+	var (
+		mu        sync.Mutex
+		collected []msgMeta
+	)
+
+	tmpl := fmt.Sprintf(`
+postgres_cdc:
+    dsn: %s
+    slot_name: multi_schema_test_slot
+    stream_snapshot: true
+    schema: tenant_*
+    tables:
+      - events
+`, databaseURL)
+
+	sb := service.NewStreamBuilder()
+	require.NoError(t, sb.SetLoggerYAML(`level: WARN`))
+	require.NoError(t, sb.AddInputYAML(tmpl))
+	require.NoError(t, sb.AddBatchConsumerFunc(func(_ context.Context, batch service.MessageBatch) error {
+		mu.Lock()
+		defer mu.Unlock()
+		for _, msg := range batch {
+			m := msgMeta{}
+			m.dbSchema, _ = msg.MetaGet("database_schema")
+			m.table, _ = msg.MetaGet("table")
+			m.operation, _ = msg.MetaGet("operation")
+			m.lsn, _ = msg.MetaGet("lsn")
+			collected = append(collected, m)
+		}
+		return nil
+	}))
+
+	stream, err := sb.Build()
+	require.NoError(t, err)
+	license.InjectTestService(stream.Resources())
+	go func() {
+		if err := stream.Run(t.Context()); err != nil && !errors.Is(err, context.Canceled) {
+			t.Error(err)
+		}
+	}()
+	t.Cleanup(func() { require.NoError(t, stream.StopWithin(10*time.Second)) })
+
+	// Wait for all 3 snapshot rows.
+	assert.Eventually(t, func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return len(collected) >= 3
+	}, 30*time.Second, 100*time.Millisecond, "timed out waiting for snapshot rows")
+
+	// Insert CDC rows.
+	_, err = db.Exec("INSERT INTO tenant_a.events (name) VALUES ('dave')")
+	require.NoError(t, err)
+	_, err = db.Exec("INSERT INTO tenant_b.events (name) VALUES ('eve')")
+	require.NoError(t, err)
+
+	// Wait for 2 CDC rows (total 5).
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		mu.Lock()
+		defer mu.Unlock()
+		assert.Len(c, collected, 5)
+	}, 30*time.Second, 100*time.Millisecond, "timed out waiting for CDC rows")
+
+	mu.Lock()
+	defer mu.Unlock()
+
+	var snapshots, cdcMsgs []msgMeta
+	for _, m := range collected {
+		if m.operation == "read" {
+			snapshots = append(snapshots, m)
+		} else {
+			cdcMsgs = append(cdcMsgs, m)
+		}
+	}
+
+	// Snapshot assertions.
+	require.Len(t, snapshots, 3)
+	snapshotSchemas := make(map[string]int)
+	for _, m := range snapshots {
+		assert.Equal(t, "events", m.table, "snapshot: table should be bare name without schema prefix")
+		assert.Empty(t, m.lsn, "snapshot rows have no LSN")
+		snapshotSchemas[m.dbSchema]++
+	}
+	assert.Equal(t, 2, snapshotSchemas["tenant_a"], "expected 2 snapshot rows from tenant_a")
+	assert.Equal(t, 1, snapshotSchemas["tenant_b"], "expected 1 snapshot row from tenant_b")
+
+	// CDC assertions.
+	require.Len(t, cdcMsgs, 2)
+	cdcSchemas := make(map[string]int)
+	for _, m := range cdcMsgs {
+		assert.Equal(t, "insert", m.operation)
+		assert.Equal(t, "events", m.table)
+		assert.NotEmpty(t, m.lsn, "CDC rows must have an LSN")
+		cdcSchemas[m.dbSchema]++
+	}
+	assert.Equal(t, 1, cdcSchemas["tenant_a"], "expected 1 CDC row from tenant_a")
+	assert.Equal(t, 1, cdcSchemas["tenant_b"], "expected 1 CDC row from tenant_b")
+}
+
+func TestIntegrationMultiSchemaMissingTableDegradesGracefully(t *testing.T) {
+	integration.CheckSkip(t)
+	databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16")
+	require.NoError(t, err)
+
+	// tenant_a is fully provisioned with the "events" table; tenant_b matches
+	// the schema glob but is missing it (e.g. still being migrated). Before
+	// this fix, CreatePublication's FOR TABLE clause would reference the
+	// non-existent tenant_b.events relation and fail publication setup for
+	// every matched schema, not just the drifted one.
+	_, err = db.Exec("CREATE SCHEMA tenant_a")
+	require.NoError(t, err)
+	_, err = db.Exec("CREATE TABLE tenant_a.events (id SERIAL PRIMARY KEY, name TEXT)")
+	require.NoError(t, err)
+	_, err = db.Exec("CREATE SCHEMA tenant_b")
+	require.NoError(t, err)
+
+	_, err = db.Exec("INSERT INTO tenant_a.events (name) VALUES ('alice')")
+	require.NoError(t, err)
+
+	type msgMeta struct {
+		dbSchema string
+		table    string
+	}
+
+	var (
+		mu        sync.Mutex
+		collected []msgMeta
+	)
+
+	tmpl := fmt.Sprintf(`
+postgres_cdc:
+    dsn: %s
+    slot_name: missing_table_degrade_slot
+    stream_snapshot: true
+    schema: tenant_*
+    tables:
+      - events
+`, databaseURL)
+
+	sb := service.NewStreamBuilder()
+	require.NoError(t, sb.SetLoggerYAML(`level: WARN`))
+	require.NoError(t, sb.AddInputYAML(tmpl))
+	require.NoError(t, sb.AddBatchConsumerFunc(func(_ context.Context, batch service.MessageBatch) error {
+		mu.Lock()
+		defer mu.Unlock()
+		for _, msg := range batch {
+			m := msgMeta{}
+			m.dbSchema, _ = msg.MetaGet("database_schema")
+			m.table, _ = msg.MetaGet("table")
+			collected = append(collected, m)
+		}
+		return nil
+	}))
+
+	stream, err := sb.Build()
+	require.NoError(t, err)
+	license.InjectTestService(stream.Resources())
+	go func() {
+		if err := stream.Run(t.Context()); err != nil && !errors.Is(err, context.Canceled) {
+			t.Error(err)
+		}
+	}()
+	t.Cleanup(func() { require.NoError(t, stream.StopWithin(10*time.Second)) })
+
+	// tenant_a should keep streaming even though tenant_b is missing the table.
+	assert.Eventually(t, func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return len(collected) >= 1
+	}, 30*time.Second, 100*time.Millisecond, "timed out waiting for tenant_a snapshot row; a missing table in tenant_b should not block replication")
+
+	mu.Lock()
+	defer mu.Unlock()
+	require.Len(t, collected, 1)
+	assert.Equal(t, "tenant_a", collected[0].dbSchema)
+	assert.Equal(t, "events", collected[0].table)
+}
+
+func TestIntegrationNoSchemasMatchedReturnsError(t *testing.T) {
+	integration.CheckSkip(t)
+	databaseURL, _, err := ResourceWithPostgreSQLVersion(t, "16")
+	require.NoError(t, err)
+
+	tmpl := fmt.Sprintf(`
+dsn: %s
+slot_name: no_schema_match_slot
+schema: nonexistent_schema_zzz_*
+tables:
+  - events
+`, databaseURL)
+
+	conf, err := newPostgresCDCConfig().ParseYAML(tmpl, nil)
+	require.NoError(t, err)
+
+	mgr := service.MockResources()
+	license.InjectTestService(mgr)
+
+	input, err := newPgStreamInput(conf, mgr)
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+	defer cancel()
+
+	// 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")
+}
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 13267f3083..b7cf77d571 100644
--- a/internal/impl/postgresql/pglogicalstream/logical_stream.go
+++ b/internal/impl/postgresql/pglogicalstream/logical_stream.go
@@ -98,18 +98,43 @@ func NewPgStream(ctx context.Context, config *Config) (*Stream, error) {
 		return nil, err
 	}
 
-	schema, err := sanitize.NormalizePostgresIdentifier(config.DBSchema)
+	schemas, inaccessibleSchemas, 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(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)
+	}
+	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 {
+		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 {
diff --git a/internal/impl/postgresql/pglogicalstream/pglogrepl.go b/internal/impl/postgresql/pglogicalstream/pglogrepl.go
index f92e222d0d..69385f753f 100644
--- a/internal/impl/postgresql/pglogicalstream/pglogrepl.go
+++ b/internal/impl/postgresql/pglogicalstream/pglogrepl.go
@@ -23,7 +23,6 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
-	"slices"
 	"strconv"
 	"strings"
 	"time"
@@ -335,41 +334,66 @@ func CreatePublication(ctx context.Context, conn *pgconn.PgConn, publicationName
 		return nil
 	}
 
-	tablesToRemoveFromPublication := []TableFQN{}
-	tablesToAddToPublication := []TableFQN{}
-	for _, table := range tables {
-		if !slices.Contains(pubTables, table) {
-			tablesToAddToPublication = append(tablesToAddToPublication, table)
-		}
+	// Build sets for O(1) lookup — avoids O(n²) slices.Contains when reconciling
+	// large publication table lists (e.g. 100 schemas × 5 tables = 500 entries).
+	wantSet := make(map[TableFQN]struct{}, len(tables))
+	for _, t := range tables {
+		wantSet[t] = struct{}{}
+	}
+	haveSet := make(map[TableFQN]struct{}, len(pubTables))
+	for _, t := range pubTables {
+		haveSet[t] = struct{}{}
 	}
 
-	for _, table := range pubTables {
-		if !slices.Contains(tables, table) {
-			tablesToRemoveFromPublication = append(tablesToRemoveFromPublication, table)
+	var tablesToAdd, tablesToRemove []TableFQN
+	for _, t := range tables {
+		if _, ok := haveSet[t]; !ok {
+			tablesToAdd = append(tablesToAdd, t)
+		}
+	}
+	for _, t := range pubTables {
+		if _, ok := wantSet[t]; !ok {
+			tablesToRemove = append(tablesToRemove, t)
 		}
 	}
 
-	// remove tables from publication
-	for _, dropTable := range tablesToRemoveFromPublication {
-		sq, err := sanitize.SQLQuery(fmt.Sprintf(`ALTER PUBLICATION %s DROP TABLE %s;`, publicationName, dropTable.String()))
+	// Batch DROP: single ALTER statement for all removed tables.
+	if len(tablesToRemove) > 0 {
+		var sb strings.Builder
+		fmt.Fprintf(&sb, "ALTER PUBLICATION %s DROP TABLE ", publicationName)
+		for i, t := range tablesToRemove {
+			if i > 0 {
+				sb.WriteString(", ")
+			}
+			sb.WriteString(t.String())
+		}
+		sb.WriteByte(';')
+		sq, err := sanitize.SQLQuery(sb.String())
 		if err != nil {
-			return fmt.Errorf("sanitizing drop table query: %w", err)
+			return fmt.Errorf("sanitizing drop tables query: %w", err)
 		}
-		result = conn.Exec(ctx, sq)
-		if _, err := result.ReadAll(); err != nil {
-			return fmt.Errorf("removing table from publication: %w", err)
+		if _, err := conn.Exec(ctx, sq).ReadAll(); err != nil {
+			return fmt.Errorf("removing tables from publication: %w", err)
 		}
 	}
 
-	// add tables to publication
-	for _, addTable := range tablesToAddToPublication {
-		sq, err := sanitize.SQLQuery(fmt.Sprintf("ALTER PUBLICATION %s ADD TABLE %s;", publicationName, addTable.String()))
+	// Batch ADD: single ALTER statement for all new tables.
+	if len(tablesToAdd) > 0 {
+		var sb strings.Builder
+		fmt.Fprintf(&sb, "ALTER PUBLICATION %s ADD TABLE ", publicationName)
+		for i, t := range tablesToAdd {
+			if i > 0 {
+				sb.WriteString(", ")
+			}
+			sb.WriteString(t.String())
+		}
+		sb.WriteByte(';')
+		sq, err := sanitize.SQLQuery(sb.String())
 		if err != nil {
-			return fmt.Errorf("sanitizing add table query: %w", err)
+			return fmt.Errorf("sanitizing add tables query: %w", err)
 		}
-		result = conn.Exec(ctx, sq)
-		if _, err := result.ReadAll(); err != nil {
-			return fmt.Errorf("adding table to publication: %w", err)
+		if _, err := conn.Exec(ctx, sq).ReadAll(); err != nil {
+			return fmt.Errorf("adding tables to publication: %w", err)
 		}
 	}
 
diff --git a/internal/impl/postgresql/pglogicalstream/sanitize/sanitize.go b/internal/impl/postgresql/pglogicalstream/sanitize/sanitize.go
index febdb3311f..c3826b0037 100644
--- a/internal/impl/postgresql/pglogicalstream/sanitize/sanitize.go
+++ b/internal/impl/postgresql/pglogicalstream/sanitize/sanitize.go
@@ -384,7 +384,7 @@ func QuotePostgresIdentifier(name string) string {
 // UnquotePostgresIdentifier returns the valid unescaped identifier.
 func UnquotePostgresIdentifier(quoted string) (string, error) {
 	var output strings.Builder
-	if !strings.HasPrefix(quoted, `"`) || !strings.HasSuffix(quoted, `"`) || len(quoted) < 2 {
+	if !strings.HasPrefix(quoted, `"`) || !strings.HasSuffix(quoted, `"`) || len(quoted) < 3 {
 		return "", errors.New("missing quotes for identifier")
 	}
 	unquoted := quoted[1 : len(quoted)-1]
diff --git a/internal/impl/postgresql/pglogicalstream/schema_resolver.go b/internal/impl/postgresql/pglogicalstream/schema_resolver.go
new file mode 100644
index 0000000000..4737f7c4cb
--- /dev/null
+++ b/internal/impl/postgresql/pglogicalstream/schema_resolver.go
@@ -0,0 +1,200 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/v4/blob/main/licenses/rcl.md
+
+package pglogicalstream
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/jackc/pgx/v5/pgconn"
+
+	"github.com/redpanda-data/connect/v4/internal/impl/postgresql/pglogicalstream/sanitize"
+)
+
+// 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
+}
+
+// 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.
+//
+// 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 {
+		return nil, 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, nil, fmt.Errorf("building schema resolution query: %w", err)
+	}
+
+	results, err := conn.Exec(ctx, q).ReadAll()
+	if err != nil {
+		return nil, nil, fmt.Errorf("querying schemas matching %q: %w", pattern, err)
+	}
+
+	visible := map[string]struct{}{}
+	var schemas []string
+	if len(results) > 0 {
+		for _, row := range results[0].Rows {
+			name := string(row[0])
+			visible[name] = struct{}{}
+			// QuotePostgresIdentifier preserves the exact stored name (including
+			// case for case-sensitive schemas), unlike NormalizePostgresIdentifier
+			// which would incorrectly fold to lower-case.
+			schemas = append(schemas, sanitize.QuotePostgresIdentifier(name))
+		}
+	}
+
+	// pg_namespace 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
+// 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.
+//
+// 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_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)
+}
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))
+		})
+	}
+}
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..512f8373fa
--- /dev/null
+++ b/internal/impl/postgresql/tests/current/setup.sql
@@ -0,0 +1,33 @@
+-- Multi-schema CDC test setup
+-- Tests: schema glob (tenant_*), database_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..9e40b5a1c1
--- /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    = @database_schema
+        let lsn       = @lsn
+        let ts_ms     = @commit_ts_ms
+        let before    = @before
+
+        root = {
+          "operation":    $op,
+          "database_schema": $schema,
+          "table":        $tbl,
+          "payload":      this,
+          "lsn":          if $lsn != null { $lsn } else { null },
+          "commit_ts_ms": if $ts_ms != null { $ts_ms } else { null },
+          "before":       if $before != null { $before.string().parse_json() } else { null },
+        }
+
+output:
+  stdout:
+    codec: lines
+
+logger:
+  level: INFO
+
+metrics:
+  none: {}
diff --git a/internal/plugins/cdctest/cdc_conformance_test.go b/internal/plugins/cdctest/cdc_conformance_test.go
index cd5eac4de5..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.
@@ -93,11 +100,11 @@ var knownNonConformant = map[string]map[string]string{
 		"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",
+		"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",
 	},
 }
 
@@ -179,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)
 		}
 	}