This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 65793d174b [Feature][Connector-v2][Postgres-CDC] Allow replica
identity other than full (#10334)
65793d174b is described below
commit 65793d174bea1cd4afb7054e55f2d4e2e1038ff9
Author: I Putu Ariyasa <[email protected]>
AuthorDate: Tue May 26 21:14:10 2026 +0700
[Feature][Connector-v2][Postgres-CDC] Allow replica identity other than
full (#10334)
---
docs/en/connectors/source/PostgreSQL-CDC.md | 3 +-
docs/zh/connectors/source/PostgreSQL-CDC.md | 3 +-
.../config/PostgresIncrementalSourceOptions.java | 11 +++
.../cdc/postgres/source/PostgresDialect.java | 12 ++-
.../postgres/source/PostgresIncrementalSource.java | 15 ++-
.../source/PostgresIncrementalSourceFactory.java | 1 +
.../cdc/postgres/source/PostgresDialectTest.java | 108 +++++++++++++++++++++
7 files changed, 148 insertions(+), 5 deletions(-)
diff --git a/docs/en/connectors/source/PostgreSQL-CDC.md
b/docs/en/connectors/source/PostgreSQL-CDC.md
index 89424a4a61..742aadbe61 100644
--- a/docs/en/connectors/source/PostgreSQL-CDC.md
+++ b/docs/en/connectors/source/PostgreSQL-CDC.md
@@ -55,7 +55,7 @@ ALTER SYSTEM SET wal_level TO 'logical';
SELECT pg_reload_conf();
```
-2. Change the REPLICA policy of the specified table to FULL
+2. Change the REPLICA policy of the specified table to FULL, unless
`require-replica-identity-full` is set to `false`.
```sql
ALTER TABLE your_table_name REPLICA IDENTITY FULL;
@@ -112,6 +112,7 @@ ALTER TABLE your_table_name REPLICA IDENTITY FULL;
| split.allow-sampling | Boolean | No | true |
Whether to allow sampling-based sharding strategy. When set to false, the
system will fall back to unevenly-sized chunk splitting (iterative query
approach) regardless of the shard count. The default value is true. |
| exactly_once | Boolean | No | false |
Enable exactly once semantic.
[...]
| format | Enum | No | DEFAULT |
Optional output format for PostgreSQL CDC, valid enumerations are `DEFAULT`,
`COMPATIBLE_DEBEZIUM_JSON`.
[...]
+| require-replica-identity-full | Boolean | No | true |
Require the table to have REPLICA IDENTITY FULL. When set to false, allows
tables with other replica identity settings, but UPDATE/DELETE events may not
contain the previous state. This should only be used for append-only tables
(e.g., outbox pattern). Default is true for backward compatibility.
[...]
| debezium | Config | No | - |
Pass-through [Debezium's
properties](https://github.com/debezium/debezium/blob/v1.9.8.Final/documentation/modules/ROOT/pages/connectors/postgresql.adoc#connector-configuration-properties)
to Debezium Embedded Engine which is used to capture data changes from
PostgreSQL server.
[...]
| common-options | | no | - |
Source plugin common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details
[...]
diff --git a/docs/zh/connectors/source/PostgreSQL-CDC.md
b/docs/zh/connectors/source/PostgreSQL-CDC.md
index 9ca3f5df32..6da31a81c9 100644
--- a/docs/zh/connectors/source/PostgreSQL-CDC.md
+++ b/docs/zh/connectors/source/PostgreSQL-CDC.md
@@ -53,7 +53,7 @@ ALTER SYSTEM SET wal_level TO 'logical';
SELECT pg_reload_conf();
```
-2. 将指定表的 REPLICA 策略更改为 FULL
+2. 将指定表的 REPLICA 策略更改为 FULL，除非 `require-replica-identity-full` 设置为 `false`。
```sql
ALTER TABLE your_table_name REPLICA IDENTITY FULL;
@@ -110,6 +110,7 @@ ALTER TABLE your_table_name REPLICA IDENTITY FULL;
| split.allow-sampling | Boolean | 否 | true |
是否允许基于采样的分片策略。当设置为 false 时，无论预估分片数是否超过阈值，系统都将回退到非均匀分片方式（迭代查询方式）。默认值为 true。 |
| exactly_once | Boolean | 否 | false |
启用精确一次语义。
[...]
| format | Enum | 否 | DEFAULT |
PostgreSQL CDC 的可选输出格式，有效枚举为 `DEFAULT`、`COMPATIBLE_DEBEZIUM_JSON`。
[...]
+| require-replica-identity-full | Boolean | 否 | true |
要求表具有 REPLICA IDENTITY FULL。设置为 false 时，允许表使用其他副本标识设置，但 UPDATE/DELETE
事件可能不包含之前的状态。此选项仅应用于仅追加的表（例如 outbox 模式）。默认为 true 以保持向后兼容性。
[...]
| debezium | Config | 否 | - | 将
[Debezium
的属性](https://github.com/debezium/debezium/blob/v1.9.8.Final/documentation/modules/ROOT/pages/connectors/postgresql.adoc#connector-configuration-properties)
传递给用于捕获 PostgreSQL 服务器数据更改的 Debezium 嵌入式引擎。
[...]
| common-options | | 否 | - |
源插件的公共参数，请参阅 [源公共选项](../common-options/source-common-options.md) 获取详细信息。
[...]
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresIncrementalSourceOptions.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresIncrementalSourceOptions.java
index bf28d79938..330a31bf34 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresIncrementalSourceOptions.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresIncrementalSourceOptions.java
@@ -48,4 +48,15 @@ public class PostgresIncrementalSourceOptions extends
JdbcSourceOptions {
.listType()
.noDefaultValue()
.withDescription("Schema name of the database to
monitor.");
+
+ public static final Option<Boolean> REQUIRE_REPLICA_IDENTITY_FULL =
+ Options.key("require-replica-identity-full")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Require the table to have REPLICA IDENTITY FULL. "
+ + "When set to false, allows tables with
other replica identity settings, "
+ + "but UPDATE/DELETE events may not
contain the previous state. "
+ + "This should only be used for
append-only tables (e.g., outbox pattern). "
+ + "Default is true for backward
compatibility.");
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresDialect.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresDialect.java
index 921cb52518..0af26e71df 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresDialect.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresDialect.java
@@ -66,6 +66,7 @@ public class PostgresDialect implements JdbcDataSourceDialect
{
private PostgresWalFetchTask postgresWalFetchTask;
private final Map<TableId, CatalogTable> tableMap;
+ private boolean requireReplicaIdentityFull = true;
public PostgresDialect(
PostgresSourceConfigFactory configFactory, List<CatalogTable>
catalogTables) {
@@ -73,6 +74,14 @@ public class PostgresDialect implements
JdbcDataSourceDialect {
this.tableMap = CatalogTableUtils.convertTables(catalogTables);
}
+ protected PostgresDialect(
+ PostgresSourceConfigFactory configFactory,
+ List<CatalogTable> catalogTables,
+ boolean requireReplicaIdentityFull) {
+ this(configFactory, catalogTables);
+ this.requireReplicaIdentityFull = requireReplicaIdentityFull;
+ }
+
@Override
public String getName() {
return DatabaseIdentifier.POSTGRESQL;
@@ -121,7 +130,8 @@ public class PostgresDialect implements
JdbcDataSourceDialect {
for (TableId tableId : tableIds) {
ServerInfo.ReplicaIdentity replicaIdentity =
postgresConnection.readReplicaIdentityInfo(tableId);
- if (!ServerInfo.ReplicaIdentity.FULL.equals(replicaIdentity)) {
+ if (requireReplicaIdentityFull
+ &&
!ServerInfo.ReplicaIdentity.FULL.equals(replicaIdentity)) {
throw new SeaTunnelException(
String.format(
"Table %s does not have a full replica
identity, please execute: ALTER TABLE %s REPLICA IDENTITY FULL;",
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSource.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSource.java
index a9e10e9516..38074fae78 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSource.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSource.java
@@ -35,6 +35,7 @@ import
org.apache.seatunnel.connectors.cdc.debezium.DebeziumDeserializationSchem
import org.apache.seatunnel.connectors.cdc.debezium.DeserializeFormat;
import
org.apache.seatunnel.connectors.cdc.debezium.row.DebeziumJsonDeserializeSchema;
import
org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializeSchema;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.postgres.config.PostgresIncrementalSourceOptions;
import
org.apache.seatunnel.connectors.seatunnel.cdc.postgres.config.PostgresSourceConfigFactory;
import
org.apache.seatunnel.connectors.seatunnel.cdc.postgres.source.offset.LsnOffsetFactory;
import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions;
@@ -59,8 +60,12 @@ public class PostgresIncrementalSource<T> extends
IncrementalSource<T, JdbcSourc
static final String IDENTIFIER = "Postgres-CDC";
+ private final boolean requireReplicaIdentityFull;
+
public PostgresIncrementalSource(ReadonlyConfig options,
List<CatalogTable> catalogTables) {
super(options, catalogTables);
+ this.requireReplicaIdentityFull =
+
options.get(PostgresIncrementalSourceOptions.REQUIRE_REPLICA_IDENTITY_FULL);
}
@Override
@@ -115,7 +120,10 @@ public class PostgresIncrementalSource<T> extends
IncrementalSource<T, JdbcSourc
@Override
public DataSourceDialect<JdbcSourceConfig>
createDataSourceDialect(ReadonlyConfig config) {
- return new PostgresDialect((PostgresSourceConfigFactory)
configFactory, catalogTables);
+ return new PostgresDialect(
+ (PostgresSourceConfigFactory) configFactory,
+ catalogTables,
+ requireReplicaIdentityFull);
}
@Override
@@ -132,7 +140,10 @@ public class PostgresIncrementalSource<T> extends
IncrementalSource<T, JdbcSourc
private Map<TableId, Struct> tableChanges() {
JdbcSourceConfig jdbcSourceConfig = configFactory.create(0);
PostgresDialect dialect =
- new PostgresDialect((PostgresSourceConfigFactory)
configFactory, catalogTables);
+ new PostgresDialect(
+ (PostgresSourceConfigFactory) configFactory,
+ catalogTables,
+ requireReplicaIdentityFull);
List<TableId> discoverTables =
dialect.discoverDataCollections(jdbcSourceConfig);
SchemaNameAdjuster adjuster = SchemaNameAdjuster.create();
ConnectTableChangeSerializer connectTableChangeSerializer =
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceFactory.java
index ed1c52d349..d4aefa4139 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceFactory.java
@@ -67,6 +67,7 @@ public class PostgresIncrementalSourceFactory implements
TableSourceFactory {
JdbcSourceOptions.CONNECTION_POOL_SIZE,
PostgresIncrementalSourceOptions.DECODING_PLUGIN_NAME,
PostgresIncrementalSourceOptions.SLOT_NAME,
+
PostgresIncrementalSourceOptions.REQUIRE_REPLICA_IDENTITY_FULL,
JdbcSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND,
JdbcSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND,
JdbcSourceOptions.SAMPLE_SHARDING_THRESHOLD,
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresDialectTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresDialectTest.java
new file mode 100644
index 0000000000..5cd1649d38
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresDialectTest.java
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.cdc.postgres.source;
+
+import org.apache.seatunnel.common.utils.SeaTunnelException;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.postgres.config.PostgresSourceConfigFactory;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import io.debezium.connector.postgresql.connection.PostgresConnection;
+import io.debezium.connector.postgresql.connection.ServerInfo;
+import io.debezium.jdbc.JdbcConfiguration;
+import io.debezium.relational.TableId;
+
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.List;
+
+public class PostgresDialectTest {
+ private static class TestPostgresConnection extends PostgresConnection {
+
+ private final ServerInfo.ReplicaIdentity identity;
+
+ public TestPostgresConnection(ServerInfo.ReplicaIdentity identity) {
+ super(JdbcConfiguration.empty(), "test");
+ this.identity = identity;
+ }
+
+ @Override
+ public ServerInfo.ReplicaIdentity readReplicaIdentityInfo(TableId
tableId) {
+ return identity;
+ }
+ }
+
+ @Test
+ public void shouldThrowWhenNotFullAndRequired() throws SQLException {
+ PostgresSourceConfigFactory configFactory =
+ (PostgresSourceConfigFactory)
+ new PostgresSourceConfigFactory()
+ .hostname("localhost")
+ .username("user")
+ .password("password")
+ .databaseList("database");
+
+ List<TableId> tableIds =
+ Collections.singletonList(new TableId("catalog", "schema",
"table"));
+
+ PostgresConnection conn = new
TestPostgresConnection(ServerInfo.ReplicaIdentity.DEFAULT);
+ PostgresDialect dialect = new PostgresDialect(configFactory,
Collections.emptyList());
+
+ Assertions.assertThrows(
+ SeaTunnelException.class,
+ () -> dialect.checkAllTablesEnabledCapture(conn, tableIds));
+ }
+
+ @Test
+ public void shouldNotThrowWhenFullAndRequired() throws SQLException {
+ PostgresSourceConfigFactory configFactory =
+ (PostgresSourceConfigFactory)
+ new PostgresSourceConfigFactory()
+ .hostname("localhost")
+ .username("user")
+ .password("password")
+ .databaseList("database");
+
+ List<TableId> tableIds =
+ Collections.singletonList(new TableId("catalog", "schema",
"table"));
+
+ PostgresConnection conn = new
TestPostgresConnection(ServerInfo.ReplicaIdentity.FULL);
+ PostgresDialect dialect = new PostgresDialect(configFactory,
Collections.emptyList());
+ Assertions.assertDoesNotThrow(() ->
dialect.checkAllTablesEnabledCapture(conn, tableIds));
+ }
+
+ @Test
+ public void shouldNotThrowWhenNotFullAndNotRequired() throws SQLException {
+ PostgresSourceConfigFactory configFactory =
+ (PostgresSourceConfigFactory)
+ new PostgresSourceConfigFactory()
+ .hostname("localhost")
+ .username("user")
+ .password("password")
+ .databaseList("database");
+
+ List<TableId> tableIds =
+ Collections.singletonList(new TableId("catalog", "schema",
"table"));
+
+ PostgresDialect dialect =
+ new PostgresDialect(configFactory, Collections.emptyList(),
false);
+ PostgresConnection conn = new
TestPostgresConnection(ServerInfo.ReplicaIdentity.DEFAULT);
+ Assertions.assertDoesNotThrow(() ->
dialect.checkAllTablesEnabledCapture(conn, tableIds));
+ }
+}
