Savonitar commented on code in PR #4618:
URL: https://github.com/apache/flink-cdc/pull/4618#discussion_r4530833606
##########
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:
Could we exercise `changelog-mode: upsert` through the factory-created
source provider, including a positive `REPLICA IDENTITY DEFAULT` update after
the snapshot is observed?
##########
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:
`getTableSchema` builds a new `PostgresSchema` and calls
`schema.refresh(connection, false)` on every call.
Did we consider other options, e.g. could we build the schema once and call
`tableFor(...)` in the loop?
##########
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:
Should we handle or reject unavailable TOAST fields before emitting a full
`REPLACE`, and cover this with a `DEFAULT`/`pgoutput` integration test?
For a primary-key table with an externally stored `TEXT` payload, snapshot
the row and then update only an integer counter.
[Debezium 1.9's TOAST
contract](https://debezium.io/documentation/reference/1.9/connectors/postgresql.html#postgresql-toasted-values)
says the unchanged payload is represented by an unavailable-value placeholder
under `REPLICA IDENTITY DEFAULT`.
##########
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:
How are unchanged TOAST columns handled in upsert mode?
##########
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:
nit: the docs say `all` requires `REPLICA IDENTITY FULL`, but this
description doesn't. Should the two match?
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresEventDeserializerTest.java:
##########
@@ -50,4 +77,80 @@ void testNullBeforeThrowsNpeWithReplicaIdentityHint() throws
Exception {
.hasMessageContaining("Before data is null")
.hasMessageContaining("REPLICA IDENTITY FULL");
}
+
+ @Test
+ void testUpdateEmitsUpdateEventInAllMode() throws Exception {
+ PostgresEventDeserializer deserializer =
+ new PostgresEventDeserializer(DebeziumChangelogMode.ALL);
+
+ List<Event> events =
+ deserialize(deserializer, buildUpdateRecord(row(1, "spare"),
row(1, "scooter")));
+
+ assertThat(events).hasSize(1);
+ DataChangeEvent event = (DataChangeEvent) events.get(0);
+ assertThat(event.op()).isEqualTo(OperationType.UPDATE);
+ assertThat(event.tableId()).isEqualTo(TableId.tableId("inventory",
"products"));
+ assertThat(event.before()).isNotNull();
+ assertThat(event.before().getInt(0)).isEqualTo(1);
+ assertThat(event.before().getString(1).toString()).isEqualTo("spare");
+ assertThat(event.after().getInt(0)).isEqualTo(1);
+ assertThat(event.after().getString(1).toString()).isEqualTo("scooter");
+ }
+
+ @Test
+ void testUpdateEmitsReplaceEventInUpsertMode() throws Exception {
Review Comment:
Could we also add an ITCase against a real table with `REPLICA IDENTITY
DEFAULT` in upsert mode (INSERT, UPDATE, DELETE, and an UPDATE that changes the
primary key)?
--
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]
