diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java index 93d8657cf56..805d2f0084b 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java @@ -24,6 +24,7 @@ import org.apache.flink.cdc.common.factories.DataSourceFactory; import org.apache.flink.cdc.common.factories.Factory; import org.apache.flink.cdc.common.factories.FactoryHelper; +import org.apache.flink.cdc.common.schema.Schema; import org.apache.flink.cdc.common.schema.Selectors; import org.apache.flink.cdc.common.source.DataSource; import org.apache.flink.cdc.common.utils.StringUtils; @@ -31,12 +32,15 @@ import org.apache.flink.cdc.connectors.base.options.StartupOptions; import org.apache.flink.cdc.connectors.postgres.source.PostgresDataSource; import org.apache.flink.cdc.connectors.postgres.source.PostgresSourceBuilder; +import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfig; import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfigFactory; import org.apache.flink.cdc.connectors.postgres.table.PostgreSQLReadableMetadata; import org.apache.flink.cdc.connectors.postgres.utils.PostgresSchemaUtils; +import org.apache.flink.cdc.debezium.table.DebeziumChangelogMode; import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.data.RowData; +import io.debezium.connector.postgresql.connection.PostgresConnection; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -54,6 +58,7 @@ import java.util.stream.Collectors; import static org.apache.flink.cdc.connectors.base.utils.ObjectUtils.doubleCompare; +import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.CHANGELOG_MODE; import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND; import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND; import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.CHUNK_META_GROUP_SIZE; @@ -203,8 +208,42 @@ public DataSource createDataSource(Context context) { String metadataList = config.get(METADATA_LIST); List readableMetadataList = listReadableMetadata(metadataList); - // Create a custom PostgresDataSource that passes the includeDatabaseInTableId flag - return new PostgresDataSource(configFactory, readableMetadataList); + DebeziumChangelogMode changelogMode = config.get(CHANGELOG_MODE); + if (changelogMode == DebeziumChangelogMode.UPSERT) { + Set capturedTableSet = new HashSet<>(capturedTables); + List capturedTableIds = + tableIds.stream() + .filter(tableId -> capturedTableSet.contains(tableId.toString())) + .collect(Collectors.toList()); + validateUpsertModePrimaryKeys(configFactory.create(0), capturedTableIds); + } + + return new PostgresDataSource(configFactory, readableMetadataList, changelogMode); + } + + /** + * Validates that all captured tables have a primary key, which upsert changelog mode relies on + * to describe idempotent updates on a key. Mirrors the validation the SQL connector performs in + * PostgreSQLTableFactory. + */ + private static void validateUpsertModePrimaryKeys( + PostgresSourceConfig sourceConfig, List capturedTableIds) { + List tablesWithoutPrimaryKey = new ArrayList<>(); + try (PostgresConnection jdbc = + PostgresSchemaUtils.getPostgresDialect(sourceConfig).openJdbcConnection()) { + for (TableId tableId : capturedTableIds) { + Schema schema = PostgresSchemaUtils.getTableSchema(tableId, sourceConfig, jdbc); + if (schema.primaryKeys().isEmpty()) { + tablesWithoutPrimaryKey.add(tableId.toString()); + } + } + } + if (!tablesWithoutPrimaryKey.isEmpty()) { + throw new ValidationException( + String.format( + "Primary key must be present when '%s' is 'upsert', but these tables have no primary key: %s.", + CHANGELOG_MODE.key(), tablesWithoutPrimaryKey)); + } } private List listReadableMetadata(String metadataList) { @@ -266,6 +305,7 @@ public Set> optionalOptions() { options.add(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED); options.add(TABLE_ID_INCLUDE_DATABASE); options.add(SCHEMA_CHANGE_ENABLED); + options.add(CHANGELOG_MODE); return options; } diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSource.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSource.java index c3055198aa0..34740fc0468 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSource.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSource.java @@ -51,17 +51,26 @@ public class PostgresDataSource implements DataSource { private final PostgresSourceConfig postgresSourceConfig; private final List readableMetadataList; + private final DebeziumChangelogMode changelogMode; public PostgresDataSource(PostgresSourceConfigFactory configFactory) { - this(configFactory, new ArrayList<>()); + this(configFactory, new ArrayList<>(), DebeziumChangelogMode.ALL); } public PostgresDataSource( PostgresSourceConfigFactory configFactory, List readableMetadataList) { + this(configFactory, readableMetadataList, DebeziumChangelogMode.ALL); + } + + public PostgresDataSource( + PostgresSourceConfigFactory configFactory, + List readableMetadataList, + DebeziumChangelogMode changelogMode) { this.configFactory = configFactory; this.postgresSourceConfig = configFactory.create(0); this.readableMetadataList = readableMetadataList; + this.changelogMode = changelogMode; } @Override @@ -70,7 +79,7 @@ public EventSourceProvider getEventSourceProvider() { boolean includeDatabaseInTableId = postgresSourceConfig.isIncludeDatabaseInTableId(); DebeziumEventDeserializationSchema deserializer = new PostgresEventDeserializer( - DebeziumChangelogMode.ALL, + changelogMode, readableMetadataList, includeDatabaseInTableId, databaseName); @@ -99,6 +108,11 @@ public PostgresSourceConfig getPostgresSourceConfig() { return postgresSourceConfig; } + @VisibleForTesting + public DebeziumChangelogMode getChangelogMode() { + return changelogMode; + } + @Override public SupportedMetadataColumn[] supportedMetadataColumns() { return new SupportedMetadataColumn[] { diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSourceOptions.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSourceOptions.java index 95cc823d211..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; @@ -281,4 +282,15 @@ public class PostgresDataSourceOptions { .defaultValue(false) .withDescription( "Whether to infer CDC column types when processing pgoutput Relation messages."); + + @Experimental + public static final ConfigOption CHANGELOG_MODE = + ConfigOptions.key("changelog-mode") + .enumType(DebeziumChangelogMode.class) + .defaultValue(DebeziumChangelogMode.ALL) + .withDescription( + "The changelog mode used for encoding streaming changes.\n" + + "\"all\": Encodes changes as 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 182650c3ad5..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,10 +65,16 @@ public void serialize(DataChangeEvent event, DataOutputView target) throws IOExc opSerializer.serialize(event.op(), target); tableIdSerializer.serialize(event.tableId(), target); - if (event.before() != null) { + // 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) { recordDataSerializer.serialize(event.before(), target); } - if (event.after() != null) { + if (afterPresent) { recordDataSerializer.serialize(event.after(), target); } metaSerializer.serialize(event.meta(), target); @@ -73,33 +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); + return deserialize(CURRENT_VERSION, 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)); + /** + * 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); } } @@ -112,28 +169,20 @@ 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( - 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); } 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())); + } }