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 2 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
2 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
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
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 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.");
}
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