postgres_cdc: multi-schema support

[josephwoodward]

This change adds multi-schema support to the PostgreSQL CDC connector via two new config fields (schema_include and schema_exclude). The two new fields take glob patterns, allowing users to select a range of tables across schemas.

Examples:

## replicate all "events" tables across schemas matching tenant_*
postgres_cdc:
    schema_include: tenant_*
    tables:
      - events

## replicate all "events" tables across schemas matching tenant_* excluding tenant_b
postgres_cdc:
    schema_include: tenant_*
    schema_exclude:
      - tenant_b
    tables:
      - events

## replicate all tables across schemas matching tenant_*
postgres_cdc:
    schema_include: tenant_*
    tables: []

Proof of Work

You can see in the below demonstration I create a config to include all schemas with the exception of tenant_b and replicate the users table. I run it once then create a new schema (tenant_c) which automatically gets picked up on the second run.

[claude]

Assertion can never fail — this test does not verify what its name claims.

The substring being searched for is fieldSchema + " is set", i.e. "schema is set". The warning the code actually emits is:

Field 'schema_include' configured, ignoring field 'schema' configuration

That string never contains "schema is set", so assert.NotContains passes unconditionally — the test would still be green if the warning were emitted for a default-valued schema. The same defect is present in TestSchemaExplicitlySetToDefaultAlongsideSchemaIncludeDoesNotWarn at line 188.

Suggested fix: assert against a substring that actually appears in the emitted warning (e.g. "ignoring field '"+fieldSchema+"'"), or better, have the production code expose the warning text as a constant/format string that both the code and the tests reference, so the two can't drift.

Ref: warning emitted at

connect/internal/impl/postgresql/input_pg_stream.go

Lines 369 to 373 in 7e38870

	}
	if err = validateSchemaPattern(schemaInclude); err != nil {
	return nil, fmt.Errorf("invalid schema_include: %w", err)
	}
	// Normalize unquoted patterns to lower-case: PostgreSQL folds unquoted

— see the test patterns guidance on assertions.

[claude]

Making tables optional removes the only guard against accidentally replicating the whole database.

Before this PR tables was a required field (the regenerated docs confirm it: tables: [] # No default (required) → tables: []). With .Optional().Default([]string{}), a config that simply omits tables now lints clean and, when schema_include is not set, falls through to CreatePublication's FOR ALL TABLES branch — replicating every table in every schema of the database and silently making stream_snapshot a no-op:

• newPgStreamInput performs no emptiness check on tables (input_pg_stream.go#L404-L408)
• CreatePublication builds FOR ALL TABLES whenever len(tables) == 0 (pglogrepl.go#L295-L308)

Previously that outcome required the user to explicitly write tables: []; now an omitted key produces it silently. That conflicts with CONTRIBUTING.md §1.2.4 ("Strongly lints and validates user-provided configuration, clearly telling users of any problems"), and it is a behaviour change not mentioned in the CHANGELOG.

tables only needs to be optional when schema_include is set (for auto-discovery). Suggested fix: keep the field optional at the spec level but reject an unset tables in newPgStreamInput when schema_include is empty — using conf.Contains(fieldTables) to distinguish "omitted" from an explicit tables: [] — so the FOR-ALL-TABLES path still requires a deliberate opt-in.

[claude]

These two negative assertions are vacuous — they can never fail.

The production warning is emitted as:

mgr.Logger().Warnf("Field '%s' configured, ignoring field '%s' configuration", fieldSchemaInclude, fieldSchema)

(input_pg_stream.go#L488-L491), which renders as Field 'schema_include' configured, ignoring field 'schema' configuration. It never contains the substring schema is set, so assert.NotContains(t, m, fieldSchema+" is set") holds for every message the connector could ever log — including the warning these tests are meant to prove is absent.

The same vacuous assertion is repeated in TestSchemaExplicitlySetToDefaultAlongsideSchemaIncludeDoesNotWarn (line 188).

Both tests should assert against the substring the code actually emits — e.g. mirror the positive check in TestSchemaIgnoredWhenSchemaIncludeSet (line 127) and assert that no message contains both fieldSchema and fieldSchemaInclude. As written, a regression that started warning on the default schema: public would ship undetected, which is exactly the behaviour the doc comment on line 160-167 claims is being pinned.

Per the test patterns, tests must actually exercise the asserted behaviour.

[claude]

Narrowing an existing FOR ALL TABLES publication now fails with a raw, unactionable PostgreSQL error, and this is a path existing users will hit on upgrade.

Before this change, forAllTables && len(pubTables) == 0 was always true for a FOR ALL TABLES publication, so the reconcile was a silent no-op. With the new condition, any caller that supplies a non-empty table list against an existing FOR ALL TABLES publication falls through to the ALTER PUBLICATION ... ADD TABLE batch below (L383-L400), which PostgreSQL rejects. The new test TestIntegrationCreatePublicationNarrowingFromForAllTablesFailsLoudly confirms this is the intended outcome.

The concrete upgrade scenario: a pipeline currently running with tables: [] (so pglog_stream_<slot> exists as FOR ALL TABLES) that adds schema_include on the same slot_name. Auto-discovery in logical_stream.go#L138-L149 makes tables non-empty, so Connect fails with adding tables to publication: ERROR: publication "pglog_stream_x" is defined as FOR ALL TABLES and the framework retries forever with no hint about the cause or the fix.

Suggest detecting forAllTables && len(tables) > 0 here and returning an error that names the publication and states the remedy (recreate the publication, or use a different slot_name), rather than letting the bare driver error surface. CONTRIBUTING §1.2.4 requires clearly telling users about configuration problems, and §3.2.2 lists difficult-to-diagnose errors as an anti-pattern. This transition is also not covered in the new schema_include / tables docs (§1.2.3).

[josephwoodward]

This is a pre-existing bug, I need to think about this part some more.

[claude]

Vacuous assertion — this test can never fail.

The substring being asserted against is fieldSchema + " is set", i.e. "schema is set". The production warning this test is guarding is emitted as:

mgr.Logger().Warnf("Field '%s' configured, ignoring field '%s' configuration", fieldSchemaInclude, fieldSchema)

(input_pg_stream.go#L371-L373) — which renders as Field 'schema_include' configured, ignoring field 'schema' configuration. That string never contains schema is set, so the NotContains check passes regardless of whether the warning is logged. The same dead assertion appears again at line 188 in TestSchemaExplicitlySetToDefaultAlongsideSchemaIncludeDoesNotWarn.

Both tests would still pass if the schema != defaultSchema guard on the warning were removed entirely, so neither actually covers the behaviour its doc comment claims to document.

Suggested fix: assert against the substring the code really emits — mirror the positive test at lines 125-131 by scanning for a message containing fieldSchemaInclude and "ignoring field" and asserting none was found. Note that the positive test's own predicate (strings.Contains(m, fieldSchema) && strings.Contains(m, fieldSchemaInclude)) is also weaker than intended, since "schema" is a substring of "schema_include".

Per CONTRIBUTING.md §1.3.2, tests must prove the behaviour they claim to cover.

[claude]

Undocumented data-loss window: auto-discovery publishes leaf partitions only, so partitions created after connect are never replicated.

Auto-discovery filters to RelKindOrdinaryTable, so a partitioned table enters the publication as its current leaf partitions rather than the parent. CreatePublication then issues CREATE PUBLICATION ... FOR TABLE <leaf1>, <leaf2> (pglogrepl.go#L296-L322).

Publishing the parent makes future partitions members automatically; publishing individual leaves does not. Table resolution only re-runs inside NewPgStream, i.e. on connect/reconnect, so for a healthy long-lived stream the set is frozen at connect time. Failure scenario: schema_include: tenant_* with tables left empty over a monthly-partitioned orders table — every row written to the orders_2026_09 partition created after the pipeline started is silently dropped, with no warning, until the pipeline happens to reconnect.

The tables field docs cover the metadata consequence of leaf-level discovery ("the table metadata ... will be the partition's name") but say nothing about newly-created partitions being missed, so an operator has no way to know they need to list the parent explicitly or restart after each partition rollover.

Per CONTRIBUTING.md §1.2.3 known limitations and edge cases must be documented. Suggested fix: either state this limitation explicitly in the tables field description alongside the existing partition note, or auto-discover the partitioned parent for publication purposes while continuing to snapshot only its leaves (the TABLESAMPLE/double-count concerns cited in ResolveExistingTables apply to snapshot planning, not to publication membership).