[
https://issues.apache.org/jira/browse/FLINK-40512?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40512:
-----------------------------------
Labels: pull-request-available (was: )
> PostgreSQL source emits relation schema events for tables outside the
> configured table list
> -------------------------------------------------------------------------------------------
>
> Key: FLINK-40512
> URL: https://issues.apache.org/jira/browse/FLINK-40512
> Project: Flink
> Issue Type: New Feature
> Affects Versions: cdc-3.6.0
> Reporter: Hongshun Wang
> Priority: Major
> Labels: pull-request-available
> Fix For: cdc-3.7.0
>
>
> When a PostgreSQL source uses a publication covering all tables
> (`publication.autocreate.mode=all_tables` / `CREATE PUBLICATION ... FOR ALL
> TABLES`),
> the Fluss YAML connector may receive schema events for tables that are not
> included
> in the configured CDC table list.
> h3. Actual behavior
> PostgreSQL pgoutput sends Relation messages for all relations covered by the
> publication. Relation-derived schema records for excluded tables are emitted
> downstream and may be converted into schema change events, including
> CreateTableEvent, by PostgresPipelineRecordEmitter.
> As a result, the Fluss connector subscribes to and processes schemas belonging
> to unrelated tables.
> h3. Expected behavior
> Relation messages should still be consumed internally because they are
> required
> for pgoutput tuple decoding. However, relation-derived schema records for
> tables
> excluded by the configured table filter should not be emitted downstream.
> Signal events and schema events required for internal state recovery must
> remain
> unaffected.
> h3. Root cause
> IncrementalSourceStreamFetcher#shouldEmit applies table and monitored-split
> filtering only to data change records.
> For every non-data-change record, it currently returns true unconditionally:
> {code:java}
> // always send the schema change event and signal event
> // we need record them to state of Flink
> return true;
> {code}
> PostgreSQL Relation messages are represented as schema records. Therefore,
> schemas for unrelated tables bypass taskContext.getTableFilter(), even when
> those tables are not part of the configured CDC table list.
> PostgresPipelineRecordEmitter#handleSchemaChangeRecord may subsequently infer
> and emit schema change events for those tables.
> h3. Steps to reproduce
> 1. Create a PostgreSQL publication covering all tables:
> {code:sql}
> CREATE PUBLICATION my_publication FOR ALL TABLES;
> {code}
> 2. Configure a PostgreSQL Pipeline YAML source to capture only a subset of the
> database tables.
> 3. Configure Fluss as the sink and enable schema change processing.
> 4. Produce changes that cause pgoutput to send Relation messages for included
> and excluded tables.
> 5. Observe that schema events, such as CreateTableEvent, for excluded tables
> reach the downstream Fluss pipeline.
> h3. Relevant code
> *
> org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher#shouldEmit
> *
> org.apache.flink.cdc.connectors.postgres.source.reader.PostgresPipelineRecordEmitter#handleSchemaChangeRecord
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
