From 48d85f2507ce95280f58522b653a24261f1bdeb4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mehmet=20Can=20=C5=9Eakiro=C4=9Flu?= Date: Thu, 4 Jun 2026 15:25:51 +0300 Subject: [PATCH 1/3] Support changelog-mode=upsert + null-tolerant DataChangeEventSerializer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FLINK-38647 reports a NullPointerException from the Postgres Pipeline source when a captured table has REPLICA IDENTITY DEFAULT and an UPDATE arrives that does not change the primary key. Root cause: PostgresDataSource hardcodes DebeziumChangelogMode.ALL, which makes the deserializer extract a before-image that is null under DEFAULT replication, NPE'ing in DebeziumSchemaDataTypeInference.inferStruct. This change fixes the issue from two angles: 1. Connector side — expose 'changelog-mode' YAML option on the Postgres Pipeline connector (mirrors the legacy SQL connector). Accepts 'all' (default, current behaviour) or 'upsert'. In upsert mode the source emits UPDATE events with before == null and only after populated, so the pipeline runs cleanly under REPLICA IDENTITY DEFAULT without requiring FULL (which roughly multiplies WAL volume per UPDATE by column count). 2. Runtime side — make DataChangeEventSerializer null-tolerant. Three changes: - copy() null-guards each recordDataSerializer.copy() call so the chained CopyingChainingOutput.pushToOperator path stops NPE'ing on null before/after. - serialize() writes 2 leading boolean presence flags before conditionally writing each record. Required because the previous serialize() already skipped writing null fields, but deserialize() always tried to read two records back — the wire format itself couldn't roundtrip a null-before UPDATE. - deserialize() reads the 2 flag bytes and skips the corresponding read when absent. Wire format change is symmetric but breaking vs current 3.6.x. Production checkpoints/savepoints written by older versions cannot be restored by the new code without a snapshot-version-aware deserialize path; see the PR description for the proposed compat strategy (bumping DataChangeEventSerializerSnapshot.CURRENT_VERSION to 2 and reading old format when version == 1). Happy to add that in a follow-up commit on this PR — wanted reviewers' opinion on the strategy first. Signed-off-by: Mehmet Can Şakiroğlu --- .../factory/PostgresDataSourceFactory.java | 21 +++++- .../postgres/source/PostgresDataSource.java | 13 +++- .../source/PostgresDataSourceOptions.java | 13 ++++ .../event/DataChangeEventSerializer.java | 65 ++++++++++--------- 4 files changed, 76 insertions(+), 36 deletions(-) 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..e10345b954b 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 @@ -34,6 +34,7 @@ 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; @@ -54,6 +55,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 +205,22 @@ 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); + String changelogModeRaw = config.get(CHANGELOG_MODE); + DebeziumChangelogMode changelogMode; + switch (changelogModeRaw.toLowerCase()) { + case "upsert": + changelogMode = DebeziumChangelogMode.UPSERT; + break; + case "all": + changelogMode = DebeziumChangelogMode.ALL; + break; + default: + throw new IllegalArgumentException( + "Invalid value for option 'changelog-mode'. Supported: [all, upsert]. Got: " + + changelogModeRaw); + } + + return new PostgresDataSource(configFactory, readableMetadataList, changelogMode); } private List listReadableMetadata(String metadataList) { @@ -266,6 +282,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..aa1457c0333 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); 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..f6d67980c48 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 @@ -281,4 +281,17 @@ 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") + .stringType() + .defaultValue("all") + .withDescription( + "The changelog mode used for encoding streaming changes. " + + "Supported values: 'all' (default) emits both before-image and after-image " + + "for UPDATE events; 'upsert' emits only the after-image (before is null), " + + "which lets the pipeline work with REPLICA IDENTITY DEFAULT on the source — " + + "avoiding the WAL overhead of REPLICA IDENTITY FULL and sidestepping the " + + "NullPointerException reported in FLINK-38647."); } diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializer.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializer.java index 182650c3ad5..228006b07a9 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializer.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializer.java @@ -62,10 +62,17 @@ public void serialize(DataChangeEvent event, DataOutputView target) throws IOExc opSerializer.serialize(event.op(), target); tableIdSerializer.serialize(event.tableId(), target); - if (event.before() != null) { + // Write explicit presence flags so deserialize can recover null before/after. + // Required when changelog-mode=upsert (FLINK-38647) emits UPDATE events with + // before == null. Adds 2 bytes per event but makes the wire format symmetric. + boolean beforePresent = event.before() != null; + boolean afterPresent = event.after() != null; + target.writeBoolean(beforePresent); + target.writeBoolean(afterPresent); + if (beforePresent) { recordDataSerializer.serialize(event.before(), target); } - if (event.after() != null) { + if (afterPresent) { recordDataSerializer.serialize(event.after(), target); } metaSerializer.serialize(event.meta(), target); @@ -76,28 +83,26 @@ public DataChangeEvent deserialize(DataInputView source) throws IOException { OperationType op = opSerializer.deserialize(source); TableId tableId = tableIdSerializer.deserialize(source); + boolean beforePresent = source.readBoolean(); + boolean afterPresent = source.readBoolean(); + org.apache.flink.cdc.common.data.RecordData before = + beforePresent ? recordDataSerializer.deserialize(source) : null; + org.apache.flink.cdc.common.data.RecordData after = + afterPresent ? recordDataSerializer.deserialize(source) : null; + switch (op) { case DELETE: return DataChangeEvent.deleteEvent( - tableId, - recordDataSerializer.deserialize(source), - metaSerializer.deserialize(source)); + tableId, before, metaSerializer.deserialize(source)); case INSERT: return DataChangeEvent.insertEvent( - tableId, - recordDataSerializer.deserialize(source), - metaSerializer.deserialize(source)); + tableId, after, metaSerializer.deserialize(source)); case UPDATE: return DataChangeEvent.updateEvent( - tableId, - recordDataSerializer.deserialize(source), - recordDataSerializer.deserialize(source), - metaSerializer.deserialize(source)); + tableId, before, after, metaSerializer.deserialize(source)); case REPLACE: return DataChangeEvent.replaceEvent( - tableId, - recordDataSerializer.deserialize(source), - metaSerializer.deserialize(source)); + tableId, after, metaSerializer.deserialize(source)); default: throw new IllegalArgumentException("Unsupported data change event: " + op); } @@ -112,28 +117,24 @@ public DataChangeEvent deserialize(DataChangeEvent reuse, DataInputView source) @Override public DataChangeEvent copy(DataChangeEvent from) { OperationType op = from.op(); + // Null-guard the before/after copy so UPDATE.before=null (from upsert mode + // under REPLICA IDENTITY DEFAULT, FLINK-38647) doesn't NPE in + // CopyingChainingOutput. Stock code assumed both fields were always present. + org.apache.flink.cdc.common.data.RecordData before = + from.before() != null ? recordDataSerializer.copy(from.before()) : null; + org.apache.flink.cdc.common.data.RecordData after = + from.after() != null ? recordDataSerializer.copy(from.after()) : null; + TableId tableId = tableIdSerializer.copy(from.tableId()); + Map meta = metaSerializer.copy(from.meta()); switch (op) { case DELETE: - return DataChangeEvent.deleteEvent( - tableIdSerializer.copy(from.tableId()), - recordDataSerializer.copy(from.before()), - metaSerializer.copy(from.meta())); + return DataChangeEvent.deleteEvent(tableId, before, meta); case INSERT: - return DataChangeEvent.insertEvent( - tableIdSerializer.copy(from.tableId()), - recordDataSerializer.copy(from.after()), - metaSerializer.copy(from.meta())); + return DataChangeEvent.insertEvent(tableId, after, meta); case UPDATE: - return DataChangeEvent.updateEvent( - tableIdSerializer.copy(from.tableId()), - recordDataSerializer.copy(from.before()), - recordDataSerializer.copy(from.after()), - metaSerializer.copy(from.meta())); + return DataChangeEvent.updateEvent(tableId, before, after, meta); case REPLACE: - return DataChangeEvent.replaceEvent( - tableIdSerializer.copy(from.tableId()), - recordDataSerializer.copy(from.after()), - metaSerializer.copy(from.meta())); + return DataChangeEvent.replaceEvent(tableId, after, meta); default: throw new IllegalArgumentException("Unsupported data change event: " + op); } From de6e5036d4fa7d931adf28b8f750989f48f81a5c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mehmet=20Can=20=C5=9Eakiro=C4=9Flu?= Date: Mon, 6 Jul 2026 12:58:38 +0300 Subject: [PATCH 2/3] Address review comments: enumType changelog-mode, upsert PK validation, versioned deserialize, tests - PostgresDataSourceOptions: define 'changelog-mode' via enumType(DebeziumChangelogMode.class), aligning with the SQL connector; invalid values now fail at option validation. - PostgresDataSourceFactory: drop the string switch; validate that all captured tables have a primary key when changelog-mode=upsert, mirroring PostgreSQLTableFactory. - DataChangeEventSerializer: introduce CURRENT_VERSION = 2 and a version-aware deserialize(int, DataInputView) that still reads the pre-flags op-dependent layout, following the PhysicalColumnSerializer/SchemaSerializer convention. DataChangeEvent is never persisted in checkpoint state, so the snapshot stays a SimpleTypeSerializerSnapshot; the versioned overload keeps the old layout readable if event bytes are ever embedded in versioned state. - Tests: null-before UPDATE events in DataChangeEventSerializerTest test data (also exercised via EventSerializerTest), legacy-version and unknown-version deserialize tests, an upsert-mode roundtrip test, and factory tests for option parsing and PK validation. --- .../factory/PostgresDataSourceFactory.java | 49 ++++++-- .../postgres/source/PostgresDataSource.java | 5 + .../source/PostgresDataSourceOptions.java | 17 ++- .../PostgresDataSourceFactoryTest.java | 84 +++++++++++++ .../event/DataChangeEventSerializer.java | 114 +++++++++++++----- .../event/DataChangeEventSerializerTest.java | 111 ++++++++++++++--- 6 files changed, 311 insertions(+), 69 deletions(-) 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 e10345b954b..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,6 +32,7 @@ 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; @@ -38,6 +40,7 @@ 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; @@ -205,24 +208,44 @@ public DataSource createDataSource(Context context) { String metadataList = config.get(METADATA_LIST); List readableMetadataList = listReadableMetadata(metadataList); - String changelogModeRaw = config.get(CHANGELOG_MODE); - DebeziumChangelogMode changelogMode; - switch (changelogModeRaw.toLowerCase()) { - case "upsert": - changelogMode = DebeziumChangelogMode.UPSERT; - break; - case "all": - changelogMode = DebeziumChangelogMode.ALL; - break; - default: - throw new IllegalArgumentException( - "Invalid value for option 'changelog-mode'. Supported: [all, upsert]. Got: " - + changelogModeRaw); + 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) { if (StringUtils.isNullOrWhitespaceOnly(metadataList)) { return new ArrayList<>(); 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 aa1457c0333..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 @@ -108,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 f6d67980c48..7ee608f1843 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; @@ -283,15 +284,13 @@ public class PostgresDataSourceOptions { "Whether to infer CDC column types when processing pgoutput Relation messages."); @Experimental - public static final ConfigOption CHANGELOG_MODE = + public static final ConfigOption CHANGELOG_MODE = ConfigOptions.key("changelog-mode") - .stringType() - .defaultValue("all") + .enumType(DebeziumChangelogMode.class) + .defaultValue(DebeziumChangelogMode.ALL) .withDescription( - "The changelog mode used for encoding streaming changes. " - + "Supported values: 'all' (default) emits both before-image and after-image " - + "for UPDATE events; 'upsert' emits only the after-image (before is null), " - + "which lets the pipeline work with REPLICA IDENTITY DEFAULT on the source — " - + "avoiding the WAL overhead of REPLICA IDENTITY FULL and sidestepping the " - + "NullPointerException reported in FLINK-38647."); + "The changelog mode used for encoding streaming changes.\n" + + "\"all\": Encodes changes as retract stream using all RowKinds. This is the default mode.\n" + + "\"upsert\": Encodes changes as upsert stream that describes 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-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializer.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializer.java index 228006b07a9..5609184be25 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializer.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializer.java @@ -20,6 +20,7 @@ import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; +import org.apache.flink.cdc.common.data.RecordData; import org.apache.flink.cdc.common.event.DataChangeEvent; import org.apache.flink.cdc.common.event.OperationType; import org.apache.flink.cdc.common.event.TableId; @@ -44,6 +45,8 @@ public class DataChangeEventSerializer extends TypeSerializerSingleton> metaSerializer = new NullableSerializerWrapper<>( @@ -62,9 +65,8 @@ public void serialize(DataChangeEvent event, DataOutputView target) throws IOExc opSerializer.serialize(event.op(), target); tableIdSerializer.serialize(event.tableId(), target); - // Write explicit presence flags so deserialize can recover null before/after. - // Required when changelog-mode=upsert (FLINK-38647) emits UPDATE events with - // before == null. Adds 2 bytes per event but makes the wire format symmetric. + // Explicit presence flags: before may be null for UPDATE events in upsert changelog + // mode (FLINK-38647), so record presence cannot be derived from the operation type. boolean beforePresent = event.before() != null; boolean afterPresent = event.after() != null; target.writeBoolean(beforePresent); @@ -80,31 +82,81 @@ public void serialize(DataChangeEvent event, DataOutputView target) throws IOExc @Override public DataChangeEvent deserialize(DataInputView source) throws IOException { - OperationType op = opSerializer.deserialize(source); - TableId tableId = tableIdSerializer.deserialize(source); - - boolean beforePresent = source.readBoolean(); - boolean afterPresent = source.readBoolean(); - org.apache.flink.cdc.common.data.RecordData before = - beforePresent ? recordDataSerializer.deserialize(source) : null; - org.apache.flink.cdc.common.data.RecordData after = - afterPresent ? recordDataSerializer.deserialize(source) : null; + return deserialize(CURRENT_VERSION, source); + } - switch (op) { - case DELETE: - return DataChangeEvent.deleteEvent( - tableId, before, metaSerializer.deserialize(source)); - case INSERT: - return DataChangeEvent.insertEvent( - tableId, after, metaSerializer.deserialize(source)); - case UPDATE: - return DataChangeEvent.updateEvent( - tableId, before, after, metaSerializer.deserialize(source)); - case REPLACE: - return DataChangeEvent.replaceEvent( - tableId, after, metaSerializer.deserialize(source)); + /** + * De-serializes a {@link DataChangeEvent} written with the given serialization version. + * Versions 0 and 1 derived record presence from the operation type; version 2 writes explicit + * presence flags so UPDATE events with a null before image (upsert changelog mode, FLINK-38647) + * round-trip correctly. + */ + public DataChangeEvent deserialize(int version, DataInputView source) throws IOException { + switch (version) { + case 0: + case 1: + { + OperationType op = opSerializer.deserialize(source); + TableId tableId = tableIdSerializer.deserialize(source); + switch (op) { + case DELETE: + return DataChangeEvent.deleteEvent( + tableId, + recordDataSerializer.deserialize(source), + metaSerializer.deserialize(source)); + case INSERT: + return DataChangeEvent.insertEvent( + tableId, + recordDataSerializer.deserialize(source), + metaSerializer.deserialize(source)); + case UPDATE: + return DataChangeEvent.updateEvent( + tableId, + recordDataSerializer.deserialize(source), + recordDataSerializer.deserialize(source), + metaSerializer.deserialize(source)); + case REPLACE: + return DataChangeEvent.replaceEvent( + tableId, + recordDataSerializer.deserialize(source), + metaSerializer.deserialize(source)); + default: + throw new IllegalArgumentException( + "Unsupported data change event: " + op); + } + } + case 2: + { + OperationType op = opSerializer.deserialize(source); + TableId tableId = tableIdSerializer.deserialize(source); + + boolean beforePresent = source.readBoolean(); + boolean afterPresent = source.readBoolean(); + RecordData before = + beforePresent ? recordDataSerializer.deserialize(source) : null; + RecordData after = + afterPresent ? recordDataSerializer.deserialize(source) : null; + + switch (op) { + case DELETE: + return DataChangeEvent.deleteEvent( + tableId, before, metaSerializer.deserialize(source)); + case INSERT: + return DataChangeEvent.insertEvent( + tableId, after, metaSerializer.deserialize(source)); + case UPDATE: + return DataChangeEvent.updateEvent( + tableId, before, after, metaSerializer.deserialize(source)); + case REPLACE: + return DataChangeEvent.replaceEvent( + tableId, after, metaSerializer.deserialize(source)); + default: + throw new IllegalArgumentException( + "Unsupported data change event: " + op); + } + } default: - throw new IllegalArgumentException("Unsupported data change event: " + op); + throw new IOException("Unrecognized serialization version " + version); } } @@ -117,13 +169,9 @@ public DataChangeEvent deserialize(DataChangeEvent reuse, DataInputView source) @Override public DataChangeEvent copy(DataChangeEvent from) { OperationType op = from.op(); - // Null-guard the before/after copy so UPDATE.before=null (from upsert mode - // under REPLICA IDENTITY DEFAULT, FLINK-38647) doesn't NPE in - // CopyingChainingOutput. Stock code assumed both fields were always present. - org.apache.flink.cdc.common.data.RecordData before = - from.before() != null ? recordDataSerializer.copy(from.before()) : null; - org.apache.flink.cdc.common.data.RecordData after = - from.after() != null ? recordDataSerializer.copy(from.after()) : null; + // before may be null for UPDATE events in upsert changelog mode (FLINK-38647). + RecordData before = from.before() != null ? recordDataSerializer.copy(from.before()) : null; + RecordData after = from.after() != null ? recordDataSerializer.copy(from.after()) : null; TableId tableId = tableIdSerializer.copy(from.tableId()); Map meta = metaSerializer.copy(from.meta()); switch (op) { diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializerTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializerTest.java index 8f25467b025..4025efc608c 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializerTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializerTest.java @@ -21,15 +21,30 @@ import org.apache.flink.cdc.common.data.RecordData; import org.apache.flink.cdc.common.data.binary.BinaryStringData; import org.apache.flink.cdc.common.event.DataChangeEvent; +import org.apache.flink.cdc.common.event.OperationType; import org.apache.flink.cdc.common.event.TableId; import org.apache.flink.cdc.common.types.DataTypes; import org.apache.flink.cdc.common.types.RowType; +import org.apache.flink.cdc.runtime.serializer.EnumSerializer; +import org.apache.flink.cdc.runtime.serializer.MapSerializer; +import org.apache.flink.cdc.runtime.serializer.NullableSerializerWrapper; import org.apache.flink.cdc.runtime.serializer.SerializerTestBase; +import org.apache.flink.cdc.runtime.serializer.StringSerializer; +import org.apache.flink.cdc.runtime.serializer.TableIdSerializer; +import org.apache.flink.cdc.runtime.serializer.data.RecordDataSerializer; import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator; +import org.apache.flink.core.memory.DataInputDeserializer; +import org.apache.flink.core.memory.DataOutputSerializer; +import org.junit.jupiter.api.Test; + +import java.io.IOException; import java.util.HashMap; import java.util.Map; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + /** A test for the {@link DataChangeEventSerializer}. */ class DataChangeEventSerializerTest extends SerializerTestBase { @Override @@ -52,19 +67,8 @@ protected DataChangeEvent[] getTestData() { Map meta = new HashMap<>(); meta.put("option", "meta1"); - BinaryRecordDataGenerator generator = - new BinaryRecordDataGenerator( - RowType.of(DataTypes.BIGINT(), DataTypes.STRING(), DataTypes.STRING())); - RecordData before = - generator.generate( - new Object[] { - 1L, - BinaryStringData.fromString("test"), - BinaryStringData.fromString("comment") - }); - RecordData after = - generator.generate( - new Object[] {1L, null, BinaryStringData.fromString("updateComment")}); + RecordData before = generateBefore(); + RecordData after = generateAfter(); return new DataChangeEvent[] { DataChangeEvent.insertEvent(TableId.tableId("table"), after), DataChangeEvent.insertEvent(TableId.tableId("table"), after, meta), @@ -75,7 +79,86 @@ protected DataChangeEvent[] getTestData() { DataChangeEvent.updateEvent( TableId.tableId("namespace", "schema", "table"), before, after), DataChangeEvent.updateEvent( - TableId.tableId("namespace", "schema", "table"), before, after, meta) + TableId.tableId("namespace", "schema", "table"), before, after, meta), + // UPDATE with a null before image, as emitted in upsert changelog mode + // (FLINK-38647). + DataChangeEvent.updateEvent( + TableId.tableId("namespace", "schema", "table"), null, after), + DataChangeEvent.updateEvent( + TableId.tableId("namespace", "schema", "table"), null, after, meta) }; } + + @Test + void testDeserializeLegacyVersionDerivesPresenceFromOperationType() throws IOException { + Map meta = new HashMap<>(); + meta.put("option", "meta1"); + RecordData before = generateBefore(); + RecordData after = generateAfter(); + TableId tableId = TableId.tableId("schema", "table"); + + // Pre-version-2 layout: op, tableId, records without presence flags, meta. + EnumSerializer opSerializer = new EnumSerializer<>(OperationType.class); + TypeSerializer> metaSerializer = + new NullableSerializerWrapper<>( + new MapSerializer<>(StringSerializer.INSTANCE, StringSerializer.INSTANCE)); + DataOutputSerializer out = new DataOutputSerializer(256); + opSerializer.serialize(OperationType.UPDATE, out); + TableIdSerializer.INSTANCE.serialize(tableId, out); + RecordDataSerializer.INSTANCE.serialize(before, out); + RecordDataSerializer.INSTANCE.serialize(after, out); + metaSerializer.serialize(meta, out); + + DataChangeEvent event = + DataChangeEventSerializer.INSTANCE.deserialize( + 1, new DataInputDeserializer(out.getCopyOfBuffer())); + + assertThat(event).isEqualTo(DataChangeEvent.updateEvent(tableId, before, after, meta)); + } + + @Test + void testDeserializeUnknownVersion() { + assertThatThrownBy( + () -> + DataChangeEventSerializer.INSTANCE.deserialize( + 3, new DataInputDeserializer(new byte[0]))) + .isInstanceOf(IOException.class) + .hasMessageContaining("Unrecognized serialization version"); + } + + @Test + void testUpsertModeUpdateEventRoundTrip() throws IOException { + RecordData after = generateAfter(); + DataChangeEvent event = + DataChangeEvent.updateEvent(TableId.tableId("schema", "table"), null, after); + + DataOutputSerializer out = new DataOutputSerializer(256); + DataChangeEventSerializer.INSTANCE.serialize(event, out); + DataChangeEvent deserialized = + DataChangeEventSerializer.INSTANCE.deserialize( + new DataInputDeserializer(out.getCopyOfBuffer())); + + assertThat(deserialized).isEqualTo(event); + assertThat(deserialized.before()).isNull(); + } + + private static RecordData generateBefore() { + return generator() + .generate( + new Object[] { + 1L, + BinaryStringData.fromString("test"), + BinaryStringData.fromString("comment") + }); + } + + private static RecordData generateAfter() { + return generator() + .generate(new Object[] {1L, null, BinaryStringData.fromString("updateComment")}); + } + + private static BinaryRecordDataGenerator generator() { + return new BinaryRecordDataGenerator( + RowType.of(DataTypes.BIGINT(), DataTypes.STRING(), DataTypes.STRING())); + } } From 5fa5011abbf1e48e5a15ee764f7e6fbd99971c6e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EB=8F=99=ED=99=98?= <66408194+dev-donghwan@users.noreply.github.com> Date: Mon, 13 Jul 2026 19:30:08 +0900 Subject: [PATCH 3/3] [FLINK-38647] Emit REPLACE events in upsert changelog mode instead of null-before UPDATE events (#1) An UPDATE event with a null before image breaks the implicit contract that runtime operators and sinks rely on. Instead of making them null-tolerant, the source now emits REPLACE events (which carry the after image only) in upsert changelog mode. REPLACE is already handled by the event serializer and all pipeline sinks, so the null-tolerant DataChangeEventSerializer changes are reverted and no sink changes are needed. --- .../pipeline-connectors/postgres.md | 12 ++ .../pipeline-connectors/postgres.md | 12 ++ .../source/PostgresDataSourceOptions.java | 4 +- .../source/PostgresEventDeserializerTest.java | 134 +++++++++++++++++ .../DebeziumEventDeserializationSchema.java | 6 +- .../event/DataChangeEventSerializer.java | 137 ++++++------------ .../event/DataChangeEventSerializerTest.java | 111 ++------------ 7 files changed, 222 insertions(+), 194 deletions(-) create mode 100644 flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresEventDeserializerTest.java 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/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 7ee608f1843..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 @@ -290,7 +290,7 @@ public class PostgresDataSourceOptions { .defaultValue(DebeziumChangelogMode.ALL) .withDescription( "The changelog mode used for encoding streaming changes.\n" - + "\"all\": Encodes changes as retract stream using all RowKinds. This is the default mode.\n" - + "\"upsert\": Encodes changes as upsert stream that describes idempotent updates on a key. " + + "\"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/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(); diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializer.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializer.java index 5609184be25..182650c3ad5 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializer.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializer.java @@ -20,7 +20,6 @@ import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; -import org.apache.flink.cdc.common.data.RecordData; import org.apache.flink.cdc.common.event.DataChangeEvent; import org.apache.flink.cdc.common.event.OperationType; import org.apache.flink.cdc.common.event.TableId; @@ -45,8 +44,6 @@ public class DataChangeEventSerializer extends TypeSerializerSingleton> metaSerializer = new NullableSerializerWrapper<>( @@ -65,16 +62,10 @@ public void serialize(DataChangeEvent event, DataOutputView target) throws IOExc opSerializer.serialize(event.op(), target); tableIdSerializer.serialize(event.tableId(), target); - // Explicit presence flags: before may be null for UPDATE events in upsert changelog - // mode (FLINK-38647), so record presence cannot be derived from the operation type. - boolean beforePresent = event.before() != null; - boolean afterPresent = event.after() != null; - target.writeBoolean(beforePresent); - target.writeBoolean(afterPresent); - if (beforePresent) { + if (event.before() != null) { recordDataSerializer.serialize(event.before(), target); } - if (afterPresent) { + if (event.after() != null) { recordDataSerializer.serialize(event.after(), target); } metaSerializer.serialize(event.meta(), target); @@ -82,81 +73,33 @@ public void serialize(DataChangeEvent event, DataOutputView target) throws IOExc @Override public DataChangeEvent deserialize(DataInputView source) throws IOException { - return deserialize(CURRENT_VERSION, source); - } + OperationType op = opSerializer.deserialize(source); + TableId tableId = tableIdSerializer.deserialize(source); - /** - * De-serializes a {@link DataChangeEvent} written with the given serialization version. - * Versions 0 and 1 derived record presence from the operation type; version 2 writes explicit - * presence flags so UPDATE events with a null before image (upsert changelog mode, FLINK-38647) - * round-trip correctly. - */ - public DataChangeEvent deserialize(int version, DataInputView source) throws IOException { - switch (version) { - case 0: - case 1: - { - OperationType op = opSerializer.deserialize(source); - TableId tableId = tableIdSerializer.deserialize(source); - switch (op) { - case DELETE: - return DataChangeEvent.deleteEvent( - tableId, - recordDataSerializer.deserialize(source), - metaSerializer.deserialize(source)); - case INSERT: - return DataChangeEvent.insertEvent( - tableId, - recordDataSerializer.deserialize(source), - metaSerializer.deserialize(source)); - case UPDATE: - return DataChangeEvent.updateEvent( - tableId, - recordDataSerializer.deserialize(source), - recordDataSerializer.deserialize(source), - metaSerializer.deserialize(source)); - case REPLACE: - return DataChangeEvent.replaceEvent( - tableId, - recordDataSerializer.deserialize(source), - metaSerializer.deserialize(source)); - default: - throw new IllegalArgumentException( - "Unsupported data change event: " + op); - } - } - case 2: - { - OperationType op = opSerializer.deserialize(source); - TableId tableId = tableIdSerializer.deserialize(source); - - boolean beforePresent = source.readBoolean(); - boolean afterPresent = source.readBoolean(); - RecordData before = - beforePresent ? recordDataSerializer.deserialize(source) : null; - RecordData after = - afterPresent ? recordDataSerializer.deserialize(source) : null; - - switch (op) { - case DELETE: - return DataChangeEvent.deleteEvent( - tableId, before, metaSerializer.deserialize(source)); - case INSERT: - return DataChangeEvent.insertEvent( - tableId, after, metaSerializer.deserialize(source)); - case UPDATE: - return DataChangeEvent.updateEvent( - tableId, before, after, metaSerializer.deserialize(source)); - case REPLACE: - return DataChangeEvent.replaceEvent( - tableId, after, metaSerializer.deserialize(source)); - default: - throw new IllegalArgumentException( - "Unsupported data change event: " + op); - } - } + switch (op) { + case DELETE: + return DataChangeEvent.deleteEvent( + tableId, + recordDataSerializer.deserialize(source), + metaSerializer.deserialize(source)); + case INSERT: + return DataChangeEvent.insertEvent( + tableId, + recordDataSerializer.deserialize(source), + metaSerializer.deserialize(source)); + case UPDATE: + return DataChangeEvent.updateEvent( + tableId, + recordDataSerializer.deserialize(source), + recordDataSerializer.deserialize(source), + metaSerializer.deserialize(source)); + case REPLACE: + return DataChangeEvent.replaceEvent( + tableId, + recordDataSerializer.deserialize(source), + metaSerializer.deserialize(source)); default: - throw new IOException("Unrecognized serialization version " + version); + throw new IllegalArgumentException("Unsupported data change event: " + op); } } @@ -169,20 +112,28 @@ public DataChangeEvent deserialize(DataChangeEvent reuse, DataInputView source) @Override public DataChangeEvent copy(DataChangeEvent from) { OperationType op = from.op(); - // before may be null for UPDATE events in upsert changelog mode (FLINK-38647). - RecordData before = from.before() != null ? recordDataSerializer.copy(from.before()) : null; - RecordData after = from.after() != null ? recordDataSerializer.copy(from.after()) : null; - TableId tableId = tableIdSerializer.copy(from.tableId()); - Map meta = metaSerializer.copy(from.meta()); switch (op) { case DELETE: - return DataChangeEvent.deleteEvent(tableId, before, meta); + return DataChangeEvent.deleteEvent( + tableIdSerializer.copy(from.tableId()), + recordDataSerializer.copy(from.before()), + metaSerializer.copy(from.meta())); case INSERT: - return DataChangeEvent.insertEvent(tableId, after, meta); + return DataChangeEvent.insertEvent( + tableIdSerializer.copy(from.tableId()), + recordDataSerializer.copy(from.after()), + metaSerializer.copy(from.meta())); case UPDATE: - return DataChangeEvent.updateEvent(tableId, before, after, meta); + return DataChangeEvent.updateEvent( + tableIdSerializer.copy(from.tableId()), + recordDataSerializer.copy(from.before()), + recordDataSerializer.copy(from.after()), + metaSerializer.copy(from.meta())); case REPLACE: - return DataChangeEvent.replaceEvent(tableId, after, meta); + return DataChangeEvent.replaceEvent( + tableIdSerializer.copy(from.tableId()), + recordDataSerializer.copy(from.after()), + metaSerializer.copy(from.meta())); default: throw new IllegalArgumentException("Unsupported data change event: " + op); } diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializerTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializerTest.java index 4025efc608c..8f25467b025 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializerTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/event/DataChangeEventSerializerTest.java @@ -21,30 +21,15 @@ import org.apache.flink.cdc.common.data.RecordData; import org.apache.flink.cdc.common.data.binary.BinaryStringData; import org.apache.flink.cdc.common.event.DataChangeEvent; -import org.apache.flink.cdc.common.event.OperationType; import org.apache.flink.cdc.common.event.TableId; import org.apache.flink.cdc.common.types.DataTypes; import org.apache.flink.cdc.common.types.RowType; -import org.apache.flink.cdc.runtime.serializer.EnumSerializer; -import org.apache.flink.cdc.runtime.serializer.MapSerializer; -import org.apache.flink.cdc.runtime.serializer.NullableSerializerWrapper; import org.apache.flink.cdc.runtime.serializer.SerializerTestBase; -import org.apache.flink.cdc.runtime.serializer.StringSerializer; -import org.apache.flink.cdc.runtime.serializer.TableIdSerializer; -import org.apache.flink.cdc.runtime.serializer.data.RecordDataSerializer; import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator; -import org.apache.flink.core.memory.DataInputDeserializer; -import org.apache.flink.core.memory.DataOutputSerializer; -import org.junit.jupiter.api.Test; - -import java.io.IOException; import java.util.HashMap; import java.util.Map; -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; - /** A test for the {@link DataChangeEventSerializer}. */ class DataChangeEventSerializerTest extends SerializerTestBase { @Override @@ -67,8 +52,19 @@ protected DataChangeEvent[] getTestData() { Map meta = new HashMap<>(); meta.put("option", "meta1"); - RecordData before = generateBefore(); - RecordData after = generateAfter(); + BinaryRecordDataGenerator generator = + new BinaryRecordDataGenerator( + RowType.of(DataTypes.BIGINT(), DataTypes.STRING(), DataTypes.STRING())); + RecordData before = + generator.generate( + new Object[] { + 1L, + BinaryStringData.fromString("test"), + BinaryStringData.fromString("comment") + }); + RecordData after = + generator.generate( + new Object[] {1L, null, BinaryStringData.fromString("updateComment")}); return new DataChangeEvent[] { DataChangeEvent.insertEvent(TableId.tableId("table"), after), DataChangeEvent.insertEvent(TableId.tableId("table"), after, meta), @@ -79,86 +75,7 @@ protected DataChangeEvent[] getTestData() { DataChangeEvent.updateEvent( TableId.tableId("namespace", "schema", "table"), before, after), DataChangeEvent.updateEvent( - TableId.tableId("namespace", "schema", "table"), before, after, meta), - // UPDATE with a null before image, as emitted in upsert changelog mode - // (FLINK-38647). - DataChangeEvent.updateEvent( - TableId.tableId("namespace", "schema", "table"), null, after), - DataChangeEvent.updateEvent( - TableId.tableId("namespace", "schema", "table"), null, after, meta) + TableId.tableId("namespace", "schema", "table"), before, after, meta) }; } - - @Test - void testDeserializeLegacyVersionDerivesPresenceFromOperationType() throws IOException { - Map meta = new HashMap<>(); - meta.put("option", "meta1"); - RecordData before = generateBefore(); - RecordData after = generateAfter(); - TableId tableId = TableId.tableId("schema", "table"); - - // Pre-version-2 layout: op, tableId, records without presence flags, meta. - EnumSerializer opSerializer = new EnumSerializer<>(OperationType.class); - TypeSerializer> metaSerializer = - new NullableSerializerWrapper<>( - new MapSerializer<>(StringSerializer.INSTANCE, StringSerializer.INSTANCE)); - DataOutputSerializer out = new DataOutputSerializer(256); - opSerializer.serialize(OperationType.UPDATE, out); - TableIdSerializer.INSTANCE.serialize(tableId, out); - RecordDataSerializer.INSTANCE.serialize(before, out); - RecordDataSerializer.INSTANCE.serialize(after, out); - metaSerializer.serialize(meta, out); - - DataChangeEvent event = - DataChangeEventSerializer.INSTANCE.deserialize( - 1, new DataInputDeserializer(out.getCopyOfBuffer())); - - assertThat(event).isEqualTo(DataChangeEvent.updateEvent(tableId, before, after, meta)); - } - - @Test - void testDeserializeUnknownVersion() { - assertThatThrownBy( - () -> - DataChangeEventSerializer.INSTANCE.deserialize( - 3, new DataInputDeserializer(new byte[0]))) - .isInstanceOf(IOException.class) - .hasMessageContaining("Unrecognized serialization version"); - } - - @Test - void testUpsertModeUpdateEventRoundTrip() throws IOException { - RecordData after = generateAfter(); - DataChangeEvent event = - DataChangeEvent.updateEvent(TableId.tableId("schema", "table"), null, after); - - DataOutputSerializer out = new DataOutputSerializer(256); - DataChangeEventSerializer.INSTANCE.serialize(event, out); - DataChangeEvent deserialized = - DataChangeEventSerializer.INSTANCE.deserialize( - new DataInputDeserializer(out.getCopyOfBuffer())); - - assertThat(deserialized).isEqualTo(event); - assertThat(deserialized.before()).isNull(); - } - - private static RecordData generateBefore() { - return generator() - .generate( - new Object[] { - 1L, - BinaryStringData.fromString("test"), - BinaryStringData.fromString("comment") - }); - } - - private static RecordData generateAfter() { - return generator() - .generate(new Object[] {1L, null, BinaryStringData.fromString("updateComment")}); - } - - private static BinaryRecordDataGenerator generator() { - return new BinaryRecordDataGenerator( - RowType.of(DataTypes.BIGINT(), DataTypes.STRING(), DataTypes.STRING())); - } }