dev-donghwan opened a new pull request, #4618: URL: https://github.com/apache/flink-cdc/pull/4618
## What is the purpose of this pull request? [FLINK-38647](https://issues.apache.org/jira/browse/FLINK-38647) This continues #4437 by @cansakiroglu, rebased onto the latest master with the original commits and authorship kept. The Postgres pipeline source always reads UPDATE events in `all` changelog mode, so it needs the before image. For tables with `REPLICA IDENTITY DEFAULT`, PostgreSQL does not send one, and the pipeline fails. #4439 made this failure message clearer, but such tables still cannot be captured unless `REPLICA IDENTITY FULL` is set, which increases WAL volume on every UPDATE. This PR adds a `changelog-mode` option (`all` / `upsert`) to the Postgres pipeline source, the same option the SQL connector already has. In `upsert` mode the source emits `REPLACE` events that carry only the after image, as @yuxiqian suggested in #4437. Runtime operators and sinks already handle `REPLACE`, so this needs no changes to the serializer or the sinks. @yuxiqian FLINK-38647 has been resolved by #4439. Should I reopen it for this change, or file a new ticket? ## Brief change log - `PostgresDataSourceOptions`: add `changelog-mode` (`enumType(DebeziumChangelogMode.class)`, default `all`, `@Experimental`). - `PostgresDataSourceFactory`: when `changelog-mode` is `upsert`, check that all captured tables have a primary key, as `PostgreSQLTableFactory` does. - `PostgresDataSource`: pass the changelog mode to the deserializer instead of hardcoding `DebeziumChangelogMode.ALL`. - `DebeziumEventDeserializationSchema`: in `upsert` mode, emit `REPLACE` instead of `UPDATE` with a null before image. - Docs (en/zh): describe `changelog-mode`. Commits: - c0105a31a, ef081885a (@cansakiroglu): the `changelog-mode` option, primary key check, and factory tests. The null-tolerant `DataChangeEventSerializer` changes in these commits were reverted in the last commit. - 6423e3acd: emit `REPLACE` in `upsert` mode, and revert the serializer changes. --- ## Verifying this change This change added tests and can be verified as follows: - `PostgresDataSourceFactoryTest`: option parsing and the primary key check in upsert mode. - `PostgresEventDeserializerTest`: an UPDATE becomes `UPDATE` in `all` mode and `REPLACE` with `before == null` in `upsert` mode. The test added by #4439 is kept. - `PostgresDataSourceFactoryTest` and `PostgresEventDeserializerTest` pass (20/20). - Manually tested PG (`REPLICA IDENTITY DEFAULT`) → Paimon end to end with `changelog-mode: upsert`, before rebasing. ## Documentation - Does this pull request introduce a new feature? yes - If yes, how is the feature documented? docs --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) Generated-by: Claude Code (Claude Opus 5.5) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
