diff --git a/docs/content.zh/docs/connectors/pipeline-connectors/postgres.md b/docs/content.zh/docs/connectors/pipeline-connectors/postgres.md index 02fb272cf43..b35f5477787 100644 --- a/docs/content.zh/docs/connectors/pipeline-connectors/postgres.md +++ b/docs/content.zh/docs/connectors/pipeline-connectors/postgres.md @@ -294,6 +294,18 @@ pipeline: 默认值为 false。 + + changelog-mode + optional + all + String + + 用于编码流式变更的 changelog 模式。
+ all：发送同时携带变更前镜像和变更后镜像的 UPDATE 事件。这是默认模式，要求表的 REPLICA IDENTITY 为 FULL。
+ upsert：发送仅携带变更后镜像的 REPLACE 事件，描述基于主键的幂等更新。适用于无法开启 REPLICA IDENTITY FULL 的主键表。
+ 实验性选项，默认值为 all。 + + diff --git a/docs/content/docs/connectors/pipeline-connectors/postgres.md b/docs/content/docs/connectors/pipeline-connectors/postgres.md index 1deb246f50a..6a02a1c99ab 100644 --- a/docs/content/docs/connectors/pipeline-connectors/postgres.md +++ b/docs/content/docs/connectors/pipeline-connectors/postgres.md @@ -286,6 +286,18 @@ pipeline: Defaults to false. + + changelog-mode + optional + all + String + + The changelog mode used for encoding streaming changes.
+ all: Emits UPDATE events carrying both the before and the after image. This is the default mode and requires the table's REPLICA IDENTITY to be FULL.
+ upsert: Emits REPLACE events carrying only the after image, describing idempotent updates on a key. It can be used for tables with primary keys when REPLICA IDENTITY FULL is not an option.
+ Experimental option, defaults to all. + + diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java index 93d8657cf56..805d2f0084b 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java @@ -24,6 +24,7 @@ import org.apache.flink.cdc.common.factories.DataSourceFactory; import org.apache.flink.cdc.common.factories.Factory; import org.apache.flink.cdc.common.factories.FactoryHelper; +import org.apache.flink.cdc.common.schema.Schema; import org.apache.flink.cdc.common.schema.Selectors; import org.apache.flink.cdc.common.source.DataSource; import org.apache.flink.cdc.common.utils.StringUtils; @@ -31,12 +32,15 @@ import org.apache.flink.cdc.connectors.base.options.StartupOptions; import org.apache.flink.cdc.connectors.postgres.source.PostgresDataSource; import org.apache.flink.cdc.connectors.postgres.source.PostgresSourceBuilder; +import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfig; import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfigFactory; import org.apache.flink.cdc.connectors.postgres.table.PostgreSQLReadableMetadata; import org.apache.flink.cdc.connectors.postgres.utils.PostgresSchemaUtils; +import org.apache.flink.cdc.debezium.table.DebeziumChangelogMode; import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.data.RowData; +import io.debezium.connector.postgresql.connection.PostgresConnection; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -54,6 +58,7 @@ import java.util.stream.Collectors; import static org.apache.flink.cdc.connectors.base.utils.ObjectUtils.doubleCompare; +import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.CHANGELOG_MODE; import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND; import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND; import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.CHUNK_META_GROUP_SIZE; @@ -203,8 +208,42 @@ public DataSource createDataSource(Context context) { String metadataList = config.get(METADATA_LIST); List 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 capturedTableSet = new HashSet<>(capturedTables); + List 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 capturedTableIds) { + List tablesWithoutPrimaryKey = new ArrayList<>(); + try (PostgresConnection jdbc = + PostgresSchemaUtils.getPostgresDialect(sourceConfig).openJdbcConnection()) { + for (TableId tableId : capturedTableIds) { + Schema schema = PostgresSchemaUtils.getTableSchema(tableId, sourceConfig, jdbc); + if (schema.primaryKeys().isEmpty()) { + tablesWithoutPrimaryKey.add(tableId.toString()); + } + } + } + if (!tablesWithoutPrimaryKey.isEmpty()) { + throw new ValidationException( + String.format( + "Primary key must be present when '%s' is 'upsert', but these tables have no primary key: %s.", + CHANGELOG_MODE.key(), tablesWithoutPrimaryKey)); + } } private List listReadableMetadata(String metadataList) { @@ -266,6 +305,7 @@ public Set> optionalOptions() { options.add(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED); options.add(TABLE_ID_INCLUDE_DATABASE); options.add(SCHEMA_CHANGE_ENABLED); + options.add(CHANGELOG_MODE); return options; } diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSource.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSource.java index c3055198aa0..34740fc0468 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSource.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSource.java @@ -51,17 +51,26 @@ public class PostgresDataSource implements DataSource { private final PostgresSourceConfig postgresSourceConfig; private final List readableMetadataList; + private final DebeziumChangelogMode changelogMode; public PostgresDataSource(PostgresSourceConfigFactory configFactory) { - this(configFactory, new ArrayList<>()); + this(configFactory, new ArrayList<>(), DebeziumChangelogMode.ALL); } public PostgresDataSource( PostgresSourceConfigFactory configFactory, List readableMetadataList) { + this(configFactory, readableMetadataList, DebeziumChangelogMode.ALL); + } + + public PostgresDataSource( + PostgresSourceConfigFactory configFactory, + List readableMetadataList, + DebeziumChangelogMode changelogMode) { this.configFactory = configFactory; this.postgresSourceConfig = configFactory.create(0); this.readableMetadataList = readableMetadataList; + this.changelogMode = changelogMode; } @Override @@ -70,7 +79,7 @@ public EventSourceProvider getEventSourceProvider() { boolean includeDatabaseInTableId = postgresSourceConfig.isIncludeDatabaseInTableId(); DebeziumEventDeserializationSchema deserializer = new PostgresEventDeserializer( - DebeziumChangelogMode.ALL, + changelogMode, readableMetadataList, includeDatabaseInTableId, databaseName); @@ -99,6 +108,11 @@ public PostgresSourceConfig getPostgresSourceConfig() { return postgresSourceConfig; } + @VisibleForTesting + public DebeziumChangelogMode getChangelogMode() { + return changelogMode; + } + @Override public SupportedMetadataColumn[] supportedMetadataColumns() { return new SupportedMetadataColumn[] { diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSourceOptions.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSourceOptions.java index 95cc823d211..f49841c95b3 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSourceOptions.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSourceOptions.java @@ -21,6 +21,7 @@ import org.apache.flink.cdc.common.annotation.PublicEvolving; import org.apache.flink.cdc.common.configuration.ConfigOption; import org.apache.flink.cdc.common.configuration.ConfigOptions; +import org.apache.flink.cdc.debezium.table.DebeziumChangelogMode; import java.time.Duration; @@ -281,4 +282,15 @@ public class PostgresDataSourceOptions { .defaultValue(false) .withDescription( "Whether to infer CDC column types when processing pgoutput Relation messages."); + + @Experimental + public static final ConfigOption 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" + + "\"upsert\": Encodes changes as REPLACE events carrying only the after image, describing idempotent updates on a key. " + + "It can be used for tables with primary keys when replica identity FULL is not an option."); } diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactoryTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactoryTest.java index c3cce43bc97..36adf1cdbe3 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactoryTest.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactoryTest.java @@ -28,6 +28,7 @@ import org.apache.flink.cdc.connectors.postgres.source.PostgresDataSource; import org.apache.flink.cdc.connectors.postgres.source.SchemaNameMetadataColumn; import org.apache.flink.cdc.connectors.postgres.source.TableNameMetadataColumn; +import org.apache.flink.cdc.debezium.table.DebeziumChangelogMode; import org.apache.flink.table.api.ValidationException; import org.junit.jupiter.api.AfterEach; @@ -45,6 +46,7 @@ import java.util.Map; import java.util.stream.Collectors; +import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.CHANGELOG_MODE; import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.HOSTNAME; import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.PASSWORD; import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.PG_PORT; @@ -310,6 +312,88 @@ public void testTableValidationWithOriginalBugScenario() { .hasMessageContaining("Cannot find any table by the option 'tables'"); } + @Test + public void testChangelogModeDefaultsToAll() { + Map 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 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(); + PostgresDataSource dataSource = (PostgresDataSource) factory.createDataSource(context); + assertThat(dataSource.getChangelogMode()).isEqualTo(DebeziumChangelogMode.UPSERT); + } + + @Test + public void testInvalidChangelogMode() { + Map 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(), "invalid"); + Factory.Context context = new MockContext(Configuration.fromMap(options)); + + PostgresDataSourceFactory factory = new PostgresDataSourceFactory(); + assertThatThrownBy(() -> factory.createDataSource(context)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Could not parse value 'invalid' for key 'changelog-mode'"); + } + + @Test + public void testChangelogModeUpsertRequiresPrimaryKey() throws SQLException { + try (Connection connection = + PostgresTestBase.getJdbcConnection( + POSTGRES_CONTAINER, POSTGRES_CONTAINER.getDatabaseName()); + Statement statement = connection.createStatement()) { + statement.execute("CREATE TABLE inventory.no_pk_table (id INTEGER, name VARCHAR(64))"); + } + + Map 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.no_pk_table"); + options.put(SLOT_NAME.key(), slotName); + options.put(CHANGELOG_MODE.key(), "upsert"); + Factory.Context context = new MockContext(Configuration.fromMap(options)); + + PostgresDataSourceFactory factory = new PostgresDataSourceFactory(); + assertThatThrownBy(() -> factory.createDataSource(context)) + .isInstanceOf(ValidationException.class) + .hasMessageContaining( + "Primary key must be present when 'changelog-mode' is 'upsert'") + .hasMessageContaining("inventory.no_pk_table"); + } + @Test public void testSupportedMetadataColumns() { Map options = new HashMap<>(); diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresEventDeserializerTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresEventDeserializerTest.java new file mode 100644 index 00000000000..207bbb40a26 --- /dev/null +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresEventDeserializerTest.java @@ -0,0 +1,134 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.postgres.source; + +import org.apache.flink.cdc.common.event.DataChangeEvent; +import org.apache.flink.cdc.common.event.Event; +import org.apache.flink.cdc.common.event.OperationType; +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.debezium.table.DebeziumChangelogMode; +import org.apache.flink.util.Collector; + +import io.debezium.data.Envelope; +import org.apache.kafka.connect.data.Schema; +import org.apache.kafka.connect.data.SchemaBuilder; +import org.apache.kafka.connect.data.Struct; +import org.apache.kafka.connect.source.SourceRecord; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link PostgresEventDeserializer} data change event handling. */ +class PostgresEventDeserializerTest { + + private static final Schema ROW_SCHEMA = + SchemaBuilder.struct() + .name("server1.inventory.products.Value") + .field("id", Schema.INT32_SCHEMA) + .field("name", Schema.OPTIONAL_STRING_SCHEMA) + .optional() + .build(); + + private static final Schema VALUE_SCHEMA = + SchemaBuilder.struct() + .name("server1.inventory.products.Envelope") + .field(Envelope.FieldName.BEFORE, ROW_SCHEMA) + .field(Envelope.FieldName.AFTER, ROW_SCHEMA) + .field(Envelope.FieldName.OPERATION, Schema.STRING_SCHEMA) + .build(); + + @Test + void testUpdateEmitsUpdateEventInAllMode() throws Exception { + PostgresEventDeserializer deserializer = + new PostgresEventDeserializer(DebeziumChangelogMode.ALL); + + List 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 { + PostgresEventDeserializer deserializer = + new PostgresEventDeserializer(DebeziumChangelogMode.UPSERT); + + // With REPLICA IDENTITY DEFAULT the before image is unavailable for UPDATE events. + List events = deserialize(deserializer, buildUpdateRecord(null, row(1, "scooter"))); + + assertThat(events).hasSize(1); + DataChangeEvent event = (DataChangeEvent) events.get(0); + assertThat(event.op()).isEqualTo(OperationType.REPLACE); + assertThat(event.tableId()).isEqualTo(TableId.tableId("inventory", "products")); + assertThat(event.before()).isNull(); + assertThat(event.after().getInt(0)).isEqualTo(1); + assertThat(event.after().getString(1).toString()).isEqualTo("scooter"); + } + + private static List deserialize( + PostgresEventDeserializer deserializer, SourceRecord record) throws Exception { + List events = new ArrayList<>(); + deserializer.deserialize( + record, + new Collector() { + @Override + public void collect(Event event) { + events.add(event); + } + + @Override + public void close() {} + }); + return events; + } + + private static Struct row(int id, String name) { + return new Struct(ROW_SCHEMA).put("id", id).put("name", name); + } + + private static SourceRecord buildUpdateRecord(Struct before, Struct after) { + Struct value = + new Struct(VALUE_SCHEMA) + .put(Envelope.FieldName.OPERATION, Envelope.Operation.UPDATE.code()) + .put(Envelope.FieldName.AFTER, after); + if (before != null) { + value.put(Envelope.FieldName.BEFORE, before); + } + return new SourceRecord( + Collections.singletonMap("server", "server1"), + Collections.singletonMap("lsn", 1L), + "server1.inventory.products", + null, + null, + null, + VALUE_SCHEMA, + value); + } +} diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java index 77897efb067..40e801284fe 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java +++ b/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 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)); } else { LOG.trace("Received {} operation, skip", op); return Collections.emptyList();