jw/postgres cdc multi schema cw - #4665
Conversation
josephwoodward
commented
Aug 4, 2026
josephwoodward
commented
- postgres_cdc: add multi-schema support
- postgres_cdc: reject empty quoted schema identifier and fix misleading godoc
- postgres_cdc: add tests/current/ Docker Compose + Taskfile manual test harness
- ** postgres_cdc: add commit_ts_ms and before metadata fields**
- postgres_cdc: fix lint and docs
- postgres_cdc: review fixes and test coverage
- postgres_cdc: fix tests
- test(cdctest): waive tigerbeetle_cdc conformance fields
- postgres_cdc: fix lint
- postgres_cdc: skip missing tables per-schema instead of failing whole publication
- postgres_cdc: fix lint
- Update internal/impl/postgresql/pglogicalstream/schema_resolver.go
- postgres_cdc: warn when schema pattern matches privilege-hidden schemas
- postgres_cdc: Address minor issues
…t 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.
…c_multi_schema # Conflicts: # CHANGELOG.md
Co-authored-by: Joseph Woodward <[email protected]>
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.
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| 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 { |
There was a problem hiding this comment.
Behaviour regression for the documented FOR ALL TABLES mode
This hard-fails startup whenever the pattern resolves to zero schemas, unconditionally. But when tables is left empty the connector documents schema as being ignored:
If left empty, the underlying PostgreSQL publication is created
FOR ALL TABLES, which replicates every table in every schema of the database, ignoringschema.
(see input_pg_stream.go L128-L134)
schema is a required field, so a user running in FOR ALL TABLES mode must set it to something. Previously any syntactically valid value worked; now a value that happens to match no existing schema aborts the pipeline with no schemas found matching pattern, even though the field has no effect in that mode. The code right below already special-cases len(normalizedTables) > 0, so the same guard is needed here — skip schema resolution (or downgrade to a warning) when config.DBTables is empty.
| 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}) | ||
| } |
There was a problem hiding this comment.
Silently swallows table-name typos for non-glob schema values
The skip-missing-table behaviour is applied to every resolved schema, including the single-schema case where schema is an exact name and no glob is involved. Concretely, with schema: public and tables: [flights, flght_legs] (typo), the pipeline used to fail loudly at publication creation; now it logs a warning and happily streams only flights. The user gets a silently incomplete pipeline.
This conflicts with CONTRIBUTING §1.2.4 ("Strongly lints and validates user-provided configuration, clearly telling users of any problems").
Suggested fix: only tolerate a missing table when the configured schema is actually a glob pattern (i.e. more than one schema was resolved / the pattern contains *); for an exact schema name keep returning a hard error naming the missing table.
| 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) | ||
| } |
There was a problem hiding this comment.
Narrows the accepted schema-name character set, breaking existing configs
This whitelist is ASCII-only, but the validation it replaces used unicode.IsLetter — see sanitize.NormalizePostgresIdentifier L443-L453, which was the only schema validation before this PR (NewPgStream normalised config.DBSchema with it).
So a pipeline that today runs with an unquoted non-ASCII schema name (PostgreSQL permits these — e.g. schema: café, schema: ünternehmen) now fails at startup with invalid character 'é' at position 3 in schema pattern "café". Existing quoted-identifier configs are unaffected, only unquoted ones.
Suggested fix: keep the wildcard/* handling but use unicode.IsLetter/unicode.IsDigit for the character class so the accepted set is a superset of what the previous validation allowed.
| return nil | ||
| } | ||
| for i, ch := range s { | ||
| if (ch >= 'a' && ch <= 'z') || (ch >= 'A' && ch <= 'Z') || (ch >= '0' && ch <= '9') || ch == '_' || ch == '*' { |
There was a problem hiding this comment.
Regression: non-ASCII schema names that worked before are now rejected at config time.
validateSchemaPattern replaces what NormalizePostgresIdentifier used to do for the schema field, but narrows the accepted character class from Unicode to ASCII. The old path used unicode.IsLetter/unicode.IsDigit:
connect/internal/impl/postgresql/pglogicalstream/sanitize/sanitize.go
Lines 443 to 453 in b0c39d2
PostgreSQL accepts non-ASCII letters in unquoted identifiers (folding them to lower case), so a config such as schema: münchen previously normalised fine and streamed. With this change the for i, ch := range s loop hits ü, and newPgStreamInput now fails with invalid schema: invalid character 'ü' at position 1 … — an existing pipeline stops starting after upgrade.
Suggested fix: accept unicode.IsLetter(ch) || unicode.IsDigit(ch) || ch == '_' || ch == '*' (and use unicode.IsLetter in the first-character check below) so the pattern validator is a superset of NormalizePostgresIdentifier plus *, rather than a subset of it. Worth a table case for a non-ASCII schema name too.
| } | ||
|
|
||
| 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'", |
There was a problem hiding this comment.
Schema existence is gated on a privilege-filtered view, turning previously-working configs into hard startup failures with a misleading error.
information_schema.schemata is privilege-filtered — this PR's own code acknowledges that by cross-checking pg_catalog.pg_namespace (schema_resolver.go#L94-L110). But the visible/hidden split is only used for a warning: when the pattern resolves only to hidden schemas, NewPgStream logs "…the configured role cannot see…" and then fails with no schemas found matching pattern %q (logical_stream.go#L105-L110) — i.e. the code knows the schema exists and still reports it as not found.
This is also a behaviour change for existing single-schema users: before this PR the schema name was only run through sanitize.NormalizePostgresIdentifier with no catalog lookup, so schema: public never depended on how the connecting role appears in information_schema.schemata. Roles that can replicate but aren't listed there (this view is restricted by ownership/privilege, and is stricter on the older PostgreSQL releases this connector supports — see integration_test.go#L697-L699, which covers 12–17) will now fail at connect time on a config that previously worked.
Suggested fix: resolve the schema set from pg_catalog.pg_namespace (already queried here) and use information_schema.schemata only to classify a match as inaccessible; and when every match is inaccessible, return an error that says the schema exists but the role lacks privileges, rather than "no schemas found". Per CONTRIBUTING.md §1.2.4 errors must clearly tell users what the problem is, and §3.2.2 lists poor/hard-to-diagnose error handling as an anti-pattern.
| } | ||
| return nil | ||
| } | ||
| for i, ch := range s { |
There was a problem hiding this comment.
validateSchemaPattern is stricter than the validation it replaces, so some previously-valid schema values are now rejected at config parse time.
The old validation path was sanitize.NormalizePostgresIdentifier, which accepts any unicode.IsLetter/unicode.IsDigit character plus _ and . (sanitize.go#L443-L453). This loop only accepts ASCII a-z, A-Z, 0-9, _ and *, so a legal unquoted PostgreSQL schema containing a non-ASCII letter (e.g. schema: crème) now fails startup with invalid schema: invalid character …, where it previously worked. The same applies to ..
Suggest mirroring NormalizePostgresIdentifier's character classes (unicode.IsLetter/unicode.IsDigit) and only adding * on top, so the glob validator is a strict superset of what was accepted before. Otherwise this is a silent breaking change for existing configs (CONTRIBUTING.md §3.1.4 — implementation is complete and correct with no known bugs).
ece86ef to
836f1cd
Compare
| 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) | ||
| } |
There was a problem hiding this comment.
Missing tables are now silently skipped even when schema is an exact name (behaviour regression + doc mismatch).
Before this change, a configured table that didn't exist reached CreatePublication's FOR TABLE clause and failed the connect with a hard error. Now any table not found in a schema is only warned about and dropped from tables, and the error at logical_stream.go#L136-L138 only fires when every table is missing.
Failure scenario: schema: public, tables: [orders, ordres] (typo). Previously the pipeline failed loudly with the offending relation named; now it starts, replicates orders, and the typo'd entry is only visible as a WARN line — silent partial data loss for a config error.
This also contradicts the documentation added in this PR, which scopes the leniency to glob patterns: "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)" (input_pg_stream.go#L133-L135).
Suggested fix: only tolerate a missing table when the pattern actually resolved to more than one schema (or when the pattern contains a wildcard), and keep the hard error for the single/exact-schema case. Note the same skip path also masks a privilege problem: information_schema.tables is privilege-filtered, so a table the role can't see is reported as "not found".
Ref: CONTRIBUTING.md §1.2.4 — "Strongly lints and validates user-provided configuration, clearly telling users of any problems."
| 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) | ||
| } |
There was a problem hiding this comment.
Regression: unquoted schema names containing non-ASCII letters are now rejected at startup.
This validator only accepts [a-zA-Z0-9_*], but the code path it replaces — sanitize.NormalizePostgresIdentifier, previously called on config.DBSchema in NewPgStream — accepts any unicode.IsLetter/unicode.IsDigit rune (plus .): see sanitize.go#L443-L457.
Failure scenario: an existing pipeline with schema: münchen (a legal unquoted PostgreSQL identifier that previously normalised to "münchen" and worked) now fails config construction with invalid schema: invalid character 'ü' at position 1 in schema pattern "münchen". The user has no way to know quoting is the workaround.
Suggested fix: mirror NormalizePostgresIdentifier's character classes here (unicode.IsLetter/unicode.IsDigit, plus _ and *) so the pattern validator is a superset of what was previously accepted, rather than a stricter ASCII-only rule.
| return nil | ||
| } | ||
| for i, ch := range s { | ||
| if (ch >= 'a' && ch <= 'z') || (ch >= 'A' && ch <= 'Z') || (ch >= '0' && ch <= '9') || ch == '_' || ch == '*' { |
There was a problem hiding this comment.
Schema validation is stricter than the identifier rules it replaces.
validateSchemaPattern takes over from the sanitize.NormalizePostgresIdentifier(config.DBSchema) call that NewPgStream used to make for this field, but its character class is ASCII-only (a-z, A-Z, 0-9, _, *), whereas NormalizePostgresIdentifier accepts any unicode.IsLetter/unicode.IsDigit rune:
connect/internal/impl/postgresql/pglogicalstream/sanitize/sanitize.go
Lines 442 to 453 in d276c1e
So a config that works today with an unquoted non-ASCII schema name — e.g. schema: münchen, a legal unquoted Postgres identifier that NormalizePostgresIdentifier folds and quotes fine — now fails at config-parse time with invalid schema: invalid character 'ü' at position 1. The same substitution also drops the MaxIdentifierLength (63 char) check that NormalizePostgresIdentifier enforced for this field.
Suggested fix: use unicode.IsLetter/unicode.IsDigit (plus _ and *) here and keep a length check — or strip the * wildcards and delegate the remainder to NormalizePostgresIdentifier so the two paths can't drift again.
Ref: CONTRIBUTING.md §3.1.4 (implementation complete and correct, no known bugs).
| } | ||
| 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) |
There was a problem hiding this comment.
A typo'd table name now degrades silently, even without a glob pattern.
Before this change the configured table list went straight into CreatePublication, so a table in tables that didn't exist failed loudly (relation "public.typo" does not exist) and the input never started. Now a missing table is skipped with a single WARN line, and the guard below only fires when every configured table is missing across every matched schema:
connect/internal/impl/postgresql/pglogicalstream/logical_stream.go
Lines 136 to 138 in d276c1e
So with schema: public and tables: [orders, ordres], the pipeline now starts happily and silently replicates only one of the two tables. Skipping per-schema is the right call for a glob that matched several schemas (what the commit and the new tables docs describe), but applying it unconditionally turns a hard config error into missing data for the far more common exact-schema case.
Suggested fix: only downgrade to a warning when the resolved schema set came from a wildcard pattern (or resolved to more than one schema); keep the hard failure when schema is an exact name.
Refs: CONTRIBUTING.md §1.2.4 (strongly lint and validate user-provided configuration, clearly telling users of any problems), §3.2.2 (difficult-to-diagnose bugs).
| 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 { |
There was a problem hiding this comment.
This hard failure also fires when schema is documented as ignored.
The tables field description updated in this PR states that an empty list creates the publication FOR ALL TABLES, "which replicates every table in every schema of the database, ignoring schema":
connect/internal/impl/postgresql/input_pg_stream.go
Lines 94 to 100 in d276c1e
But this check runs unconditionally. With tables: [], normalizedTables is empty, so the loop below adds nothing and schemas is never used for anything — yet a schema value that resolves to zero schemas now aborts startup with no schemas found matching pattern. schema is a required field with no default, so a user who intentionally left tables empty and set schema to a value they were told is ignored goes from a working pipeline to one that won't start. (The per-schema resolveExistingTables queries below are also wasted work in that case.)
Suggested fix: skip schema resolution and this check entirely when len(config.DBTables) == 0.
| go run ../../../../../cmd/redpanda-connect/main.go run \ | ||
| --set 'input.postgres_cdc.schema=""' \ | ||
| ./test_config.yaml 2>&1 | head -5 | ||
| echo "exit $?" |
There was a problem hiding this comment.
This smoke test can never fail.
$? after go run ... | head -5 is the exit status of head, not of go run, so this always prints exit 0 regardless of whether the binary rejected the schema. Combined with set +e on line 80, the test:invalid-schema task succeeds unconditionally — it can never detect the non-zero exit that the trailing comment on line 85 says it expects.
Suggested fix: check ${PIPESTATUS[0]} (and set an explicit bash shell), or drop the pipe and test the go run status directly, then fail the task when the status is zero or the output doesn't contain invalid schema.
| @@ -0,0 +1,36 @@ | |||
| services: | |||
There was a problem hiding this comment.
Committed local dev tooling that duplicates the automated tests in this same PR.
The tests/current/ directory (this compose file plus setup.sql, Taskfile.yaml, test_config.yaml) is a manual harness for exactly the scenario TestIntegrationMultiSchemaSnapshotAndCDC already covers automatically with testcontainers:
connect/internal/impl/postgresql/integration_test.go
Lines 1637 to 1646 in d276c1e
CONTRIBUTING.md §6 asks that no local tooling be committed, and every comparable harness in this repo lives under a directory whose name says what it's for — internal/impl/postgresql/bench/, internal/impl/iceberg/demo/, internal/impl/iceberg/e2e/ — rather than a tests/current/ path that conveys nothing and will read as stale the moment "current" isn't (§3.1.5, consistency with the rest of the codebase).
Suggested fix: drop the directory now that the behaviour is covered by the integration tests. If it's meant to stay, move it to a named directory with a short README explaining when a developer should reach for it instead of task test:integration -- postgresql.
| // 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{ |
There was a problem hiding this comment.
Out of scope for this PR.
The new conditionalConnectors escape hatch and the rewritten tigerbeetle_cdc waiver reasons below have nothing to do with postgres_cdc multi-schema support, and nothing else in this diff depends on them — postgres_cdc's own conformance entry is untouched.
CONTRIBUTING.md §3.1.1 asks that a PR stay within the scope agreed in the issue/PRD and that additional capabilities be proposed and reviewed separately; §3.3.2 asks for self-contained PRs. Suggested fix: split this into its own PR so the tigerbeetle waiver rationale ("pure streaming with no initial snapshot", progress_cache → checkpoint_cache) gets reviewed by the people who own that connector rather than landing under a postgres review.
2 participants
