dev-donghwan commented on code in PR #4618:
URL: https://github.com/apache/flink-cdc/pull/4618#discussion_r4553969937
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java:
##########
@@ -128,8 +128,10 @@ public List<DataChangeEvent>
deserializeDataChangeRecord(SourceRecord record) th
return Collections.singletonList(
DataChangeEvent.updateEvent(tableId, before, after,
meta));
}
- return Collections.singletonList(
- DataChangeEvent.updateEvent(tableId, null, after, meta));
+ // In upsert mode the before image may be unavailable (e.g.
PostgreSQL with
+ // REPLICA IDENTITY DEFAULT), so emit a REPLACE event which
carries the after
+ // image only instead of an UPDATE event with a null before image.
+ return
Collections.singletonList(DataChangeEvent.replaceEvent(tableId, after, meta));
Review Comment:
Thanks a lot for this, it helped me harden the change considerably. I added
fail-fast handling and probed the TOAST-able types with REPLICA IDENTITY
DEFAULT + pgoutput (snapshot a large value, then update another column or the
primary key). While adding the primary-key-change case you asked for, I also
found that a PK change is emitted as DELETE + INSERT, so the check now covers
INSERT events in upsert mode as well.
| Case | Before this change | Now |
|---|---|---|
| text, jsonb, hstore (json mode), bytea (bytes mode), non-key UPDATE |
REPLACE with `__debezium_unavailable_value` | job fails with a REPLICA IDENTITY
FULL hint |
| text, PK-changing UPDATE | INSERT with `__debezium_unavailable_value` |
job fails with a REPLICA IDENTITY FULL hint |
| text[], int[], bytea (base64 / hex mode) | REPLACE / INSERT with **NULL**
| still **NULL** (documented limitation) |
All rows are covered by ITCases in `PostgresPipelineITCase`. The null cases
come from Debezium 1.9.8, which has no placeholder for these types, so they
can't be told apart from real nulls in the deserializer. I've asked the
maintainers in the conversation tab how they'd like to handle them.
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactoryTest.java:
##########
@@ -363,6 +365,88 @@ public void testTableValidationWithOriginalBugScenario() {
.hasMessageContaining("Cannot find any table by the option
'tables'");
}
+ @Test
+ public void testChangelogModeDefaultsToAll() {
+ Map<String, String> options = new HashMap<>();
+ options.put(HOSTNAME.key(), POSTGRES_CONTAINER.getHost());
+ options.put(
+ PG_PORT.key(),
String.valueOf(POSTGRES_CONTAINER.getMappedPort(POSTGRESQL_PORT)));
+ options.put(USERNAME.key(), TEST_USER);
+ options.put(PASSWORD.key(), TEST_PASSWORD);
+ options.put(TABLES.key(), POSTGRES_CONTAINER.getDatabaseName() +
".inventory.prod\\.*");
+ options.put(SLOT_NAME.key(), slotName);
+ Factory.Context context = new
MockContext(Configuration.fromMap(options));
+
+ PostgresDataSourceFactory factory = new PostgresDataSourceFactory();
+ PostgresDataSource dataSource = (PostgresDataSource)
factory.createDataSource(context);
+
assertThat(dataSource.getChangelogMode()).isEqualTo(DebeziumChangelogMode.ALL);
+ }
+
+ @Test
+ public void testChangelogModeUpsert() {
+ Map<String, String> options = new HashMap<>();
+ options.put(HOSTNAME.key(), POSTGRES_CONTAINER.getHost());
+ options.put(
+ PG_PORT.key(),
String.valueOf(POSTGRES_CONTAINER.getMappedPort(POSTGRESQL_PORT)));
+ options.put(USERNAME.key(), TEST_USER);
+ options.put(PASSWORD.key(), TEST_PASSWORD);
+ options.put(TABLES.key(), POSTGRES_CONTAINER.getDatabaseName() +
".inventory.prod\\.*");
+ options.put(SLOT_NAME.key(), slotName);
+ options.put(CHANGELOG_MODE.key(), "upsert");
+ Factory.Context context = new
MockContext(Configuration.fromMap(options));
+
+ PostgresDataSourceFactory factory = new PostgresDataSourceFactory();
Review Comment:
Added `testUpsertModeWithReplicaIdentityDefault` in
`PostgresPipelineITCase`. It creates the source through
`PostgresDataSourceFactory` and, after the snapshot is observed, checks INSERT
/ UPDATE (REPLACE) / PK-changing UPDATE / DELETE on a REPLICA IDENTITY DEFAULT
table, which also covers your ITCase comment.
##########
docs/content/docs/connectors/pipeline-connectors/postgres.md:
##########
@@ -324,6 +324,18 @@ pipeline:
WARNING: Incremental/binlog phase: Rate limiting can cause the
connector to fall behind the upstream change stream, risking binlog/WAL purging
before the connector catches up (data loss for MySQL, replication slot issues
for PostgreSQL).
</td>
</tr>
+ <tr>
+ <td>changelog-mode</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">all</td>
+ <td>String</td>
+ <td>
+ The changelog mode used for encoding streaming changes.<br>
+ <code>all</code>: Emits UPDATE events carrying both the before and the
after image. This is the default mode and requires the table's <code>REPLICA
IDENTITY</code> to be <code>FULL</code>.<br>
+ <code>upsert</code>: Emits REPLACE events carrying only the after
image, describing idempotent updates on a key. It can be used for tables with
primary keys when <code>REPLICA IDENTITY FULL</code> is not an option.<br>
Review Comment:
Documented the current behavior: unchanged TOASTed string, JSON and binary
(bytes mode) values fail the source, while arrays and binary values in
base64/hex mode are emitted as null because Debezium 1.9.8 has no placeholder
for them. See the thread on `DebeziumEventDeserializationSchema` for details.
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java:
##########
@@ -211,8 +216,42 @@ public DataSource createDataSource(Context context) {
String metadataList = config.get(METADATA_LIST);
List<PostgreSQLReadableMetadata> readableMetadataList =
listReadableMetadata(metadataList);
- // Create a custom PostgresDataSource that passes the
includeDatabaseInTableId flag
- return new PostgresDataSource(configFactory, readableMetadataList);
+ DebeziumChangelogMode changelogMode = config.get(CHANGELOG_MODE);
+ if (changelogMode == DebeziumChangelogMode.UPSERT) {
+ Set<String> capturedTableSet = new HashSet<>(capturedTables);
+ List<TableId> capturedTableIds =
+ tableIds.stream()
+ .filter(tableId ->
capturedTableSet.contains(tableId.toString()))
+ .collect(Collectors.toList());
+ validateUpsertModePrimaryKeys(configFactory.create(0),
capturedTableIds);
+ }
+
+ return new PostgresDataSource(configFactory, readableMetadataList,
changelogMode);
+ }
+
+ /**
+ * Validates that all captured tables have a primary key, which upsert
changelog mode relies on
+ * to describe idempotent updates on a key. Mirrors the validation the SQL
connector performs in
+ * PostgreSQLTableFactory.
+ */
+ private static void validateUpsertModePrimaryKeys(
+ PostgresSourceConfig sourceConfig, List<TableId> capturedTableIds)
{
+ List<String> tablesWithoutPrimaryKey = new ArrayList<>();
+ try (PostgresConnection jdbc =
+
PostgresSchemaUtils.getPostgresDialect(sourceConfig).openJdbcConnection()) {
+ for (TableId tableId : capturedTableIds) {
+ Schema schema = PostgresSchemaUtils.getTableSchema(tableId,
sourceConfig, jdbc);
Review Comment:
Good point. Added `PostgresSchemaUtils#getTableSchemas`, which refreshes the
schema once and resolves each table via `tableFor`; the validation now uses it,
and `getTableSchema` shares the same helper. Also added
`testChangelogModeUpsertReportsOnlyTablesWithoutPrimaryKey` to cover the
multi-table path (a table with and a table without a primary key).
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSourceOptions.java:
##########
@@ -317,4 +318,15 @@ public class PostgresDataSourceOptions {
.defaultValue(-1d)
.withDescription(
"The maximum number of records emitted per second,
the default value: -1, means no rate limiting.");
+
+ @Experimental
+ public static final ConfigOption<DebeziumChangelogMode> CHANGELOG_MODE =
+ ConfigOptions.key("changelog-mode")
+ .enumType(DebeziumChangelogMode.class)
+ .defaultValue(DebeziumChangelogMode.ALL)
+ .withDescription(
+ "The changelog mode used for encoding streaming
changes.\n"
+ + "\"all\": Encodes changes as UPDATE
events carrying both the before and the after image. This is the default
mode.\n"
Review Comment:
Aligned the option description with the docs.
--
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]
