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/2] 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/2] 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())); + } }