This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12547-fe066b54f4922da312917e8031ed4d0eadadde70 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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(); + } +}
