This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 7c4f2d7720 [Fix][Connector-V2] Apply Postgres-CDC ADD COLUMN on the
first RELATION with pgjdbc 42.7.5+ (#12547)
7c4f2d7720 is described below
commit 7c4f2d77209fd0c8edcc42aff105625f3d2a1406
Author: Goutam Adwant <[email protected]>
AuthorDate: Fri Oct 9 09:26:13 2026 +0000
[Fix][Connector-V2] Apply Postgres-CDC ADD COLUMN on the first RELATION
with pgjdbc 42.7.5+ (#12547)
---
.../postgresql/RelationAwarePostgresSchema.java | 20 ++-
.../RelationAwarePostgresSchemaTest.java | 141 +++++++++++++++++++++
2 files changed, 160 insertions(+), 1 deletion(-)
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/RelationAwarePostgresSchema.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/RelationAwarePostgresSchema.java
index 6300dd8d6b..fef54bf08f 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/RelationAwarePostgresSchema.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/RelationAwarePostgresSchema.java
@@ -30,6 +30,7 @@ import java.util.function.Consumer;
/** A PostgreSQL schema that exposes pgoutput RELATION changes to SeaTunnel. */
public class RelationAwarePostgresSchema extends PostgresSchema {
+ private final String databaseName;
private Consumer<Table> relationChangeListener;
public RelationAwarePostgresSchema(
@@ -39,6 +40,7 @@ public class RelationAwarePostgresSchema extends
PostgresSchema {
TopicSelector<TableId> topicSelector,
PostgresValueConverter valueConverter) {
super(config, typeRegistry, defaultValueConverter, topicSelector,
valueConverter);
+ this.databaseName = config.databaseName();
}
public void setRelationChangeListener(Consumer<Table>
relationChangeListener) {
@@ -51,7 +53,7 @@ public class RelationAwarePostgresSchema extends
PostgresSchema {
*/
@Override
public void applySchemaChangesForTable(int relationId, Table table) {
- Table previousTable = tableFor(table.id());
+ Table previousTable = trackedTableFor(table.id());
super.applySchemaChangesForTable(relationId, table);
// Always forward streaming RELATION messages for known tables. After
recovery Debezium may
@@ -63,6 +65,22 @@ public class RelationAwarePostgresSchema extends
PostgresSchema {
}
}
+ /**
+ * pgoutput RELATION ids carry no catalog, but pgjdbc 42.7.5+ returns the
database name as
+ * TABLE_CAT, so tables loaded from JDBC metadata may be tracked as
database.schema.table.
+ */
+ private Table trackedTableFor(TableId relationTableId) {
+ Table table = tableFor(relationTableId);
+ if (table != null
+ || relationTableId.catalog() != null
+ || databaseName == null
+ || databaseName.isEmpty()) {
+ return table;
+ }
+ return tableFor(
+ new TableId(databaseName, relationTableId.schema(),
relationTableId.table()));
+ }
+
/** Compare two raw pgoutput RELATION schemas, including their primary-key
definitions. */
public static boolean hasSameRelationSchema(Table before, Table after) {
if (!Objects.equals(before.primaryKeyColumnNames(),
after.primaryKeyColumnNames())) {
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/io/debezium/connector/postgresql/RelationAwarePostgresSchemaTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/io/debezium/connector/postgresql/RelationAwarePostgresSchemaTest.java
new file mode 100644
index 0000000000..d44bd13818
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/io/debezium/connector/postgresql/RelationAwarePostgresSchemaTest.java
@@ -0,0 +1,141 @@
+/*
+ * 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 io.debezium.connector.postgresql;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import io.debezium.config.Configuration;
+import io.debezium.connector.postgresql.connection.PostgresConnection;
+import
io.debezium.connector.postgresql.connection.PostgresDefaultValueConverter;
+import io.debezium.relational.Column;
+import io.debezium.relational.Table;
+import io.debezium.relational.TableId;
+import io.debezium.relational.Tables;
+import io.debezium.schema.TopicSelector;
+
+import java.sql.Types;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+
+public class RelationAwarePostgresSchemaTest {
+
+ private static final String DATABASE = "traffic";
+
+ @Test
+ public void
testRelationNotifiesListenerWhenSchemaWasLoadedWithoutCatalog() throws
Exception {
+ // pgjdbc before 42.7.5 returns TABLE_CAT = NULL, so Debezium keys the
table without it.
+ List<Table> notified =
+ applyFirstRelation(new TableId(null, "public", "users"),
relationWithNewColumn());
+
+ Assertions.assertEquals(1, notified.size());
+ Assertions.assertEquals(2, notified.get(0).columns().size());
+ }
+
+ @Test
+ public void testRelationNotifiesListenerWhenSchemaWasLoadedWithCatalog()
throws Exception {
+ // pgjdbc 42.7.5+ returns TABLE_CAT = <database>, while pgoutput
RELATION ids have none.
+ List<Table> notified =
+ applyFirstRelation(
+ new TableId(DATABASE, "public", "users"),
relationWithNewColumn());
+
+ Assertions.assertEquals(1, notified.size());
+ Assertions.assertEquals(2, notified.get(0).columns().size());
+ }
+
+ @Test
+ public void testRelationForUnknownTableDoesNotNotifyListener() throws
Exception {
+ List<Table> notified =
+ applyFirstRelation(
+ new TableId(DATABASE, "public", "other"),
relationWithNewColumn());
+
+ Assertions.assertTrue(notified.isEmpty());
+ }
+
+ @Test
+ public void testRelationDoesNotMatchTableFromAnotherCatalog() throws
Exception {
+ List<Table> notified =
+ applyFirstRelation(
+ new TableId("other_db", "public", "users"),
relationWithNewColumn());
+
+ Assertions.assertTrue(notified.isEmpty());
+ }
+
+ private static List<Table> applyFirstRelation(TableId loadedTableId, Table
relation)
+ throws Exception {
+ RelationAwarePostgresSchema schema = newSchema();
+ PostgresConnection connection = mock(PostgresConnection.class);
+ doAnswer(
+ invocation -> {
+ Tables tables = invocation.getArgument(0);
+ tables.overwriteTable(table(loadedTableId, "id"));
+ return null;
+ })
+ .when(connection)
+ .readSchema(any(), any(), any(), any(), any(), anyBoolean());
+ schema.refresh(connection, false);
+
+ List<Table> notified = new ArrayList<>();
+ schema.setRelationChangeListener(notified::add);
+ schema.applySchemaChangesForTable(1, relation);
+ return notified;
+ }
+
+ private static RelationAwarePostgresSchema newSchema() {
+ PostgresConnectorConfig config =
+ new PostgresConnectorConfig(
+ Configuration.create()
+ .with("database.server.name", "seatunnel")
+ .with("database.dbname", DATABASE)
+ .build());
+ return new RelationAwarePostgresSchema(
+ config,
+ mock(TypeRegistry.class),
+ mock(PostgresDefaultValueConverter.class),
+ TopicSelector.defaultSelector(config, (id, prefix, delimiter)
-> id.toString()),
+ mock(PostgresValueConverter.class));
+ }
+
+ private static Table relationWithNewColumn() {
+ return table(new TableId(null, "public", "users"), "id", "extra");
+ }
+
+ private static Table table(TableId tableId, String... columnNames) {
+ List<Column> columns = new ArrayList<>();
+ for (int i = 0; i < columnNames.length; i++) {
+ columns.add(
+ Column.editor()
+ .name(columnNames[i])
+ .type("int4")
+ .jdbcType(Types.INTEGER)
+ .position(i + 1)
+ .optional(i != 0)
+ .create());
+ }
+ return Table.editor()
+ .tableId(tableId)
+ .addColumns(columns)
+ .setPrimaryKeyNames(columnNames[0])
+ .create();
+ }
+}
