Skip to content

[FLINK-38647] Support changelog-mode=upsert + null-tolerance #4437

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
cansakiroglu wants to merge 3 commits into
base: master
Choose a base branch
from
Open

[FLINK-38647] Support changelog-mode=upsert + null-tolerance #4437

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
48d85f2
Support changelog-mode=upsert + null-tolerant DataChangeEventSerializer
cansakiroglu Jun 4, 2026
de6e503
Address review comments: enumType changelog-mode, upsert PK validatio…
cansakiroglu Jul 6, 2026
5fa5011
[FLINK-38647] Emit REPLACE events in upsert changelog mode instead of…
dev-donghwan Jul 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions docs/content.zh/docs/connectors/pipeline-connectors/postgres.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -294,6 +294,18 @@ pipeline:
默认值为 false。
</td>
</tr>
<tr>
<td>changelog-mode</td>
<td>optional</td>
<td style="word-wrap: break-word;">all</td>
<td>String</td>
<td>
用于编码流式变更的 changelog 模式。<br>
<code>all</code>：发送同时携带变更前镜像和变更后镜像的 UPDATE 事件。这是默认模式，要求表的 <code>REPLICA IDENTITY</code> 为 <code>FULL</code>。<br>
<code>upsert</code>：发送仅携带变更后镜像的 REPLACE 事件，描述基于主键的幂等更新。适用于无法开启 <code>REPLICA IDENTITY FULL</code> 的主键表。<br>
实验性选项，默认值为 all。
</td>
</tr>
</tbody>
</table>
</div>
Expand Down
12 changes: 12 additions & 0 deletions docs/content/docs/connectors/pipeline-connectors/postgres.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -286,6 +286,18 @@ pipeline:
Defaults to false.
</td>
</tr>
<tr>
<td>changelog-mode</td>
<td>optional</td>
<td style="word-wrap: break-word;">all</td>
<td>String</td>
<td>
The changelog mode used for encoding streaming changes.<br>
<code>all</code>: Emits UPDATE events carrying both the before and the after image. This is the default mode and requires the table's <code>REPLICA IDENTITY</code> to be <code>FULL</code>.<br>
<code>upsert</code>: Emits REPLACE events carrying only the after image, describing idempotent updates on a key. It can be used for tables with primary keys when <code>REPLICA IDENTITY FULL</code> is not an option.<br>
Experimental option, defaults to all.
</td>
</tr>
</tbody>
</table>
</div>
Expand Down
44 changes: 42 additions & 2 deletions ...main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,19 +24,23 @@
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;
import org.apache.flink.cdc.connectors.base.options.SourceOptions;
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;

Expand All @@ -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;
Expand Down Expand Up @@ -203,8 +208,42 @@ public DataSource createDataSource(Context context) {
String metadataList = config.get(METADATA_LIST);
List<PostgreSQLReadableMetadata> readableMetadataList = listReadableMetadata(metadataList);

// Create a custom PostgresDataSource that passes the includeDatabaseInTableId flag
return new PostgresDataSource(configFactory, readableMetadataList);
DebeziumChangelogMode changelogMode = config.get(CHANGELOG_MODE);
if (changelogMode == DebeziumChangelogMode.UPSERT) {
Set<String> capturedTableSet = new HashSet<>(capturedTables);
List<TableId> capturedTableIds =
tableIds.stream()
.filter(tableId -> capturedTableSet.contains(tableId.toString()))
.collect(Collectors.toList());
validateUpsertModePrimaryKeys(configFactory.create(0), capturedTableIds);
}

return new PostgresDataSource(configFactory, readableMetadataList, changelogMode);
}

/**
* Validates that all captured tables have a primary key, which upsert changelog mode relies on
* to describe idempotent updates on a key. Mirrors the validation the SQL connector performs in
* PostgreSQLTableFactory.
*/
private static void validateUpsertModePrimaryKeys(
PostgresSourceConfig sourceConfig, List<TableId> capturedTableIds) {
List<String> tablesWithoutPrimaryKey = new ArrayList<>();
try (PostgresConnection jdbc =
PostgresSchemaUtils.getPostgresDialect(sourceConfig).openJdbcConnection()) {
for (TableId tableId : capturedTableIds) {
Schema schema = PostgresSchemaUtils.getTableSchema(tableId, sourceConfig, jdbc);
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<PostgreSQLReadableMetadata> listReadableMetadata(String metadataList) {
Expand Down Expand Up @@ -266,6 +305,7 @@ public Set<ConfigOption<?>> 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;
}

Expand Down
18 changes: 16 additions & 2 deletions ...res/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,17 +51,26 @@ public class PostgresDataSource implements DataSource {
private final PostgresSourceConfig postgresSourceConfig;

private final List<PostgreSQLReadableMetadata> readableMetadataList;
private final DebeziumChangelogMode changelogMode;

public PostgresDataSource(PostgresSourceConfigFactory configFactory) {
this(configFactory, new ArrayList<>());
this(configFactory, new ArrayList<>(), DebeziumChangelogMode.ALL);
}

public PostgresDataSource(
PostgresSourceConfigFactory configFactory,
List<PostgreSQLReadableMetadata> readableMetadataList) {
this(configFactory, readableMetadataList, DebeziumChangelogMode.ALL);
}

public PostgresDataSource(
PostgresSourceConfigFactory configFactory,
List<PostgreSQLReadableMetadata> readableMetadataList,
DebeziumChangelogMode changelogMode) {
this.configFactory = configFactory;
this.postgresSourceConfig = configFactory.create(0);
this.readableMetadataList = readableMetadataList;
this.changelogMode = changelogMode;
}

@Override
Expand All @@ -70,7 +79,7 @@ public EventSourceProvider getEventSourceProvider() {
boolean includeDatabaseInTableId = postgresSourceConfig.isIncludeDatabaseInTableId();
DebeziumEventDeserializationSchema deserializer =
new PostgresEventDeserializer(
DebeziumChangelogMode.ALL,
changelogMode,
readableMetadataList,
includeDatabaseInTableId,
databaseName);
Expand Down Expand Up @@ -99,6 +108,11 @@ public PostgresSourceConfig getPostgresSourceConfig() {
return postgresSourceConfig;
}

@VisibleForTesting
public DebeziumChangelogMode getChangelogMode() {
return changelogMode;
}

@Override
public SupportedMetadataColumn[] supportedMetadataColumns() {
return new SupportedMetadataColumn[] {
Expand Down
12 changes: 12 additions & 0 deletions .../main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSourceOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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<DebeziumChangelogMode> CHANGELOG_MODE =
ConfigOptions.key("changelog-mode")
.enumType(DebeziumChangelogMode.class)
.defaultValue(DebeziumChangelogMode.ALL)
.withDescription(
"The changelog mode used for encoding streaming changes.\n"
+ "\"all\": Encodes changes as UPDATE events carrying both the before and the after image. This is the default mode.\n"
+ "\"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.");
}
84 changes: 84 additions & 0 deletions .../java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactoryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -310,6 +312,88 @@ public void testTableValidationWithOriginalBugScenario() {
.hasMessageContaining("Cannot find any table by the option 'tables'");
}

@Test
public void testChangelogModeDefaultsToAll() {
Map<String, String> options = new HashMap<>();
options.put(HOSTNAME.key(), POSTGRES_CONTAINER.getHost());
options.put(
PG_PORT.key(), String.valueOf(POSTGRES_CONTAINER.getMappedPort(POSTGRESQL_PORT)));
options.put(USERNAME.key(), TEST_USER);
options.put(PASSWORD.key(), TEST_PASSWORD);
options.put(TABLES.key(), POSTGRES_CONTAINER.getDatabaseName() + ".inventory.prod\\.*");
options.put(SLOT_NAME.key(), slotName);
Factory.Context context = new MockContext(Configuration.fromMap(options));

PostgresDataSourceFactory factory = new PostgresDataSourceFactory();
PostgresDataSource dataSource = (PostgresDataSource) factory.createDataSource(context);
assertThat(dataSource.getChangelogMode()).isEqualTo(DebeziumChangelogMode.ALL);
}

@Test
public void testChangelogModeUpsert() {
Map<String, String> options = new HashMap<>();
options.put(HOSTNAME.key(), POSTGRES_CONTAINER.getHost());
options.put(
PG_PORT.key(), String.valueOf(POSTGRES_CONTAINER.getMappedPort(POSTGRESQL_PORT)));
options.put(USERNAME.key(), TEST_USER);
options.put(PASSWORD.key(), TEST_PASSWORD);
options.put(TABLES.key(), POSTGRES_CONTAINER.getDatabaseName() + ".inventory.prod\\.*");
options.put(SLOT_NAME.key(), slotName);
options.put(CHANGELOG_MODE.key(), "upsert");
Factory.Context context = new MockContext(Configuration.fromMap(options));

PostgresDataSourceFactory factory = new PostgresDataSourceFactory();
PostgresDataSource dataSource = (PostgresDataSource) factory.createDataSource(context);
assertThat(dataSource.getChangelogMode()).isEqualTo(DebeziumChangelogMode.UPSERT);
}

@Test
public void testInvalidChangelogMode() {
Map<String, String> options = new HashMap<>();
options.put(HOSTNAME.key(), POSTGRES_CONTAINER.getHost());
options.put(
PG_PORT.key(), String.valueOf(POSTGRES_CONTAINER.getMappedPort(POSTGRESQL_PORT)));
options.put(USERNAME.key(), TEST_USER);
options.put(PASSWORD.key(), TEST_PASSWORD);
options.put(TABLES.key(), POSTGRES_CONTAINER.getDatabaseName() + ".inventory.prod\\.*");
options.put(SLOT_NAME.key(), slotName);
options.put(CHANGELOG_MODE.key(), "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<String, String> options = new HashMap<>();
options.put(HOSTNAME.key(), POSTGRES_CONTAINER.getHost());
options.put(
PG_PORT.key(), String.valueOf(POSTGRES_CONTAINER.getMappedPort(POSTGRESQL_PORT)));
options.put(USERNAME.key(), TEST_USER);
options.put(PASSWORD.key(), TEST_PASSWORD);
options.put(TABLES.key(), POSTGRES_CONTAINER.getDatabaseName() + ".inventory.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<String, String> options = new HashMap<>();
Expand Down
Loading