Feature/postgres cdc multi schema - #4589
Conversation
…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.
squiidz
left a comment
squiidz
left a comment
There was a problem hiding this comment.
Re-reviewed after the rebase — the critical issue from the earlier pass is resolved: stream_message.go and replication_message_decoders.go are no longer touched, so the duplicate CommitTs/Before/tuple-decoding reimplementation (which had drifted from what main already shipped in #4554/#4555) is gone. The diff is now correctly scoped to the new schema-glob + pg_schema + publication-batching work layered on top of main's existing commit_ts_ms/before support. 👍
A few remaining items from the original pass, still present:
-
gofmt: the struct literal in
input_pg_stream.goisn't gofmt-clean —DBSchemaPattern: schema, DBTables: tables,
gofmt -wwantsDBSchemaPattern: schema,/DBTables: tables,to stay column-aligned withRefreshAuthToken:in the same literal. This will fail agofmt -l/CI format check as-is. -
validateSchemaPattern's quoted-identifier check (input_pg_stream.go) reimplements a shallow prefix/suffix/length check instead of delegating tosanitize.UnquotePostgresIdentifier, which this PR already uses inschema_resolver.go. The existing helper correctly handles escaped double-quotes ("foo""bar"); the new check doesn't. -
No
CHANGELOG.mdentry — recentpostgres_cdcPRs (#4554, #4555, ...) all added one under "Unreleased". -
tests/current/(Taskfile/docker-compose/setup.sql/test_config.yaml) reads like a personal manual-test harness — worth confirming this is meant to be committed long-term vs. a scratch dir that should be dropped or renamed to something more descriptive. -
Worth a doc callout: schema-pattern resolution happens once at stream construction (
resolveSchemasruns at startup only). For the stated multi-tenant use case, a newtenant_cschema created after the pipeline starts won't be picked up until restart. Not a bug, but non-obvious given the motivating use case. -
Test coverage gap: no test for the "zero schemas matched" startup-error path (
no schemas found matching pattern %q).
None of these block correctness — (1) is the only one that will mechanically fail CI as-is.
Jeffail
left a comment
Jeffail
left a comment
There was a problem hiding this comment.
nice feature but theres a failure mode that i think takes down the whole input:
with a glob schema + explicit tables, logical_stream.go builds the cartesian product schemas × tables and every combo has to exist. tables is required so we always hit the FOR TABLE ... path in CreatePublication (runs at startup before snapshot). if any matched schema is missing a listed table:
- fresh run →
CREATE PUBLICATION ... FOR TABLE "tenant_x"."missing"errorsrelation does not exist - restart → same thing via the batched
ALTER PUBLICATION ... ADD TABLE
so one drifted/half-provisioned schema that merely matches the glob halts replication for every tenant, not just the bad one. and this is exactly the documented flow — new schemas need a restart to get picked up, so a freshly created tenant_* schema thats not fully migrated yet will kill the whole thing on the next restart.
what id like:
- resolve tables per-schema — skip (with a warn log) any listed table thats missing from a given matched schema, so drift degrades gracefully instead of taking everything down
- if thats too much for v1, at minimum document that every matched schema must contain every listed table + that restarting mid-provision will stop the input
nit: the PR description lists commit_ts_ms and before as features but those are already on main, this pr is just the multi-schema glob (changelog is correct though 👍)
rest looks solid — the O(n²)→O(n) publication reconcile + batched ALTER are a nice touch, and pg_schema sourcing from the WAL relation is right
…c_multi_schema # Conflicts: # CHANGELOG.md
Co-authored-by: Joseph Woodward <joseph.woodward@xeuse.com>
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.
f0d74cb to
0e662de
Compare
| 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) |
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, ignoring schema.
(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 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.
| 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."
4 participants
Multi-schema replication — schema field now accepts a glob pattern (e.g. tenant_*) in addition to an exact name. All
matching schemas are replicated through a single replication slot and publication, removing the need for one slot per
tenant.
pg_schema metadata — schema name of the originating table, useful for per-tenant routing downstream.
Bug fix — validateSchemaPattern accepted "" (empty quoted identifier, length 2) due to an off-by-one; corrected to
require at least one character inside the quotes.