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-12435-1f4038b8965d6aa471fd24cd0a40a1cf6a7ea618 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 4648c9956ec88dec560a4af27cbd610568f47a60 Author: Palm <[email protected]> AuthorDate: Thu Sep 24 10:39:53 2026 +0000 [Fix][Connector-V2] Deduplicate table ids in PostgreSQL CDC table discovery to support HighGo (#12435) --- docs/en/connectors/source/PostgreSQL-CDC.md | 4 + docs/zh/connectors/source/PostgreSQL-CDC.md | 3 + .../cdc/postgres/utils/TableDiscoveryUtils.java | 11 +- .../postgres/utils/TableDiscoveryUtilsTest.java | 153 +++++++++++++++++++++ 4 files changed, 169 insertions(+), 2 deletions(-) diff --git a/docs/en/connectors/source/PostgreSQL-CDC.md b/docs/en/connectors/source/PostgreSQL-CDC.md index b5e7facc77..b5b45d83bf 100644 --- a/docs/en/connectors/source/PostgreSQL-CDC.md +++ b/docs/en/connectors/source/PostgreSQL-CDC.md @@ -23,6 +23,10 @@ import ChangeLog from '../changelog/connector-cdc-postgres.md'; The PostgreSQL CDC connector allows for reading snapshot data and incremental data from PostgreSQL databases. This document describes how to set up the PostgreSQL CDC connector. +Some PostgreSQL-compatible databases (for example HighGo) return the same physical table more than +once from `information_schema.tables`. The connector deduplicates discovered tables by their fully +qualified `catalog.schema.table` id, so these repeated rows are ignored and the job starts normally. + ## Supported DataSource Info | Datasource | Supported versions | Driver | Url | Maven | diff --git a/docs/zh/connectors/source/PostgreSQL-CDC.md b/docs/zh/connectors/source/PostgreSQL-CDC.md index 1f7c2855bc..732d6657f8 100644 --- a/docs/zh/connectors/source/PostgreSQL-CDC.md +++ b/docs/zh/connectors/source/PostgreSQL-CDC.md @@ -22,6 +22,9 @@ import ChangeLog from '../changelog/connector-cdc-postgres.md'; PostgreSQL CDC 连接器允许从 PostgreSQL 数据库读取快照数据和增量数据。本文档介绍如何配置 PostgreSQL CDC 连接器。 +部分 PostgreSQL 兼容数据库（例如 HighGo）在 `information_schema.tables` 中会对同一张物理表返回多行重复记录。 +连接器在表发现阶段会按完整的 `catalog.schema.table` 标识去重，这些重复行会被忽略，任务可正常启动。 + ## 支持的数据源信息 | 数据源 | 支持的版本 | 驱动 | Url | Maven | diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/TableDiscoveryUtils.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/TableDiscoveryUtils.java index 5528c09bf8..2b34c90bb0 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/TableDiscoveryUtils.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/TableDiscoveryUtils.java @@ -26,7 +26,9 @@ import io.debezium.relational.TableId; import java.sql.SQLException; import java.util.ArrayList; +import java.util.LinkedHashSet; import java.util.List; +import java.util.Set; public class TableDiscoveryUtils { private static final Logger LOG = LoggerFactory.getLogger(TableDiscoveryUtils.class); @@ -34,7 +36,12 @@ public class TableDiscoveryUtils { @SuppressWarnings("MagicNumber") public static List<TableId> listTables(JdbcConnection jdbc, RelationalTableFilters tableFilters) throws SQLException { - final List<TableId> capturedTableIds = new ArrayList<>(); + // Use a LinkedHashSet to deduplicate table ids. Some PostgreSQL-compatible databases + // (e.g. HighGo) return the same physical table several times from + // INFORMATION_SCHEMA.TABLES, and duplicated TableId would break the downstream + // Collectors.toMap() in PostgresIncrementalSource#tableChanges(). + // LinkedHashSet keeps the discovery order stable for standard PostgreSQL. + final Set<TableId> capturedTableIds = new LinkedHashSet<>(); // ------------------- // READ DATABASE NAMES // ------------------- @@ -86,6 +93,6 @@ public class TableDiscoveryUtils { e.getMessage()); } } - return capturedTableIds; + return new ArrayList<>(capturedTableIds); } } diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/TableDiscoveryUtilsTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/TableDiscoveryUtilsTest.java new file mode 100644 index 0000000000..f68724951d --- /dev/null +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/TableDiscoveryUtilsTest.java @@ -0,0 +1,153 @@ +/* + * 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.utils; + +import org.apache.seatunnel.connectors.seatunnel.cdc.postgres.config.PostgresSourceConfig; +import org.apache.seatunnel.connectors.seatunnel.cdc.postgres.config.PostgresSourceConfigFactory; + +import org.junit.jupiter.api.Test; + +import io.debezium.connector.postgresql.connection.PostgresConnection; +import io.debezium.jdbc.JdbcConfiguration; +import io.debezium.jdbc.JdbcConnection; +import io.debezium.relational.RelationalTableFilters; +import io.debezium.relational.TableId; + +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class TableDiscoveryUtilsTest { + + /** Simulates a JDBC connection whose INFORMATION_SCHEMA.TABLES returns the given rows. */ + private static class FakePostgresConnection extends PostgresConnection { + + private final List<String[]> tableRows; + + private FakePostgresConnection(List<String[]> tableRows) { + super(JdbcConfiguration.empty(), "table-discovery-test"); + this.tableRows = tableRows; + } + + @Override + public JdbcConnection query(String query, ResultSetConsumer consumer) throws SQLException { + if (query.startsWith("select datname")) { + consumer.accept(mockResultSet(Collections.singletonList(new String[] {"testdb"}))); + return this; + } + consumer.accept(mockResultSet(tableRows)); + return this; + } + } + + /** Builds a ResultSet mock backed by rows of {TABLE_CATALOG, TABLE_SCHEMA, TABLE_NAME}. */ + private static ResultSet mockResultSet(List<String[]> rows) throws SQLException { + ResultSet resultSet = mock(ResultSet.class); + AtomicInteger cursor = new AtomicInteger(-1); + when(resultSet.next()).thenAnswer(invocation -> cursor.incrementAndGet() < rows.size()); + when(resultSet.getString(anyInt())) + .thenAnswer( + invocation -> { + int column = invocation.getArgument(0); + return rows.get(cursor.get())[column - 1]; + }); + return resultSet; + } + + private static RelationalTableFilters tableFilters() { + PostgresSourceConfig config = + (PostgresSourceConfig) + new PostgresSourceConfigFactory() + .hostname("localhost") + .username("user") + .password("password") + .databaseList("testdb") + .create(0); + return config.getTableFilters(); + } + + /** + * HighGo returns the same BASE TABLE row six times from INFORMATION_SCHEMA.TABLES. The + * discovery result must contain a single TableId with the original catalog/schema/table + * identity preserved. + */ + @Test + public void shouldDeduplicateRepeatedCatalogRows() throws SQLException { + List<String[]> rows = new ArrayList<>(); + for (int i = 0; i < 6; i++) { + rows.add(new String[] {"highgo", "testdb", "test_a"}); + } + + List<TableId> tableIds = + TableDiscoveryUtils.listTables(new FakePostgresConnection(rows), tableFilters()); + + assertEquals(1, tableIds.size()); + assertEquals(new TableId("highgo", "testdb", "test_a"), tableIds.get(0)); + } + + /** An ordinary PostgreSQL catalog without duplicates must pass through unchanged, in order. */ + @Test + public void shouldPreserveDistinctTablesInOrder() throws SQLException { + List<String[]> rows = + Arrays.asList( + new String[] {"testdb", "public", "orders"}, + new String[] {"testdb", "public", "users"}, + new String[] {"testdb", "inventory", "shipments"}); + + List<TableId> tableIds = + TableDiscoveryUtils.listTables(new FakePostgresConnection(rows), tableFilters()); + + assertEquals( + Arrays.asList( + new TableId("testdb", "public", "orders"), + new TableId("testdb", "public", "users"), + new TableId("testdb", "inventory", "shipments")), + tableIds); + } + + /** Duplicates interleaved with distinct tables must collapse to their first occurrence. */ + @Test + public void shouldKeepFirstOccurrenceWhenDuplicatesInterleave() throws SQLException { + List<String[]> rows = + Arrays.asList( + new String[] {"testdb", "public", "orders"}, + new String[] {"highgo", "testdb", "test_a"}, + new String[] {"testdb", "public", "users"}, + new String[] {"highgo", "testdb", "test_a"}, + new String[] {"testdb", "public", "orders"}); + + List<TableId> tableIds = + TableDiscoveryUtils.listTables(new FakePostgresConnection(rows), tableFilters()); + + assertEquals( + Arrays.asList( + new TableId("testdb", "public", "orders"), + new TableId("highgo", "testdb", "test_a"), + new TableId("testdb", "public", "users")), + tableIds); + } +}
