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-12836-68c686438f31fcbaa376081af6a4536caa227f12 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 89a5598eb5e258f30248009572c58d0deba24d60 Author: Goutam Adwant <[email protected]> AuthorDate: Sun Oct 18 06:11:12 2026 +0000 [Fix][Connector-V2] Resolve late PostgreSQL CDC type names by search path (#12836) --- docs/en/connectors/source/PostgreSQL-CDC.md | 4 ++ docs/zh/connectors/source/PostgreSQL-CDC.md | 4 ++ .../connector/postgresql/TypeRegistry.java | 7 ++- .../seatunnel/cdc/postgres/PostgresCDCIT.java | 73 ++++++++++++++++++++++ 4 files changed, 87 insertions(+), 1 deletion(-) diff --git a/docs/en/connectors/source/PostgreSQL-CDC.md b/docs/en/connectors/source/PostgreSQL-CDC.md index caf02983f1..c8f1949a6e 100644 --- a/docs/en/connectors/source/PostgreSQL-CDC.md +++ b/docs/en/connectors/source/PostgreSQL-CDC.md @@ -27,6 +27,10 @@ Some PostgreSQL-compatible databases (for example HighGo) return the same physic 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. +On a type-name cache miss, an unqualified name shared by multiple schemas resolves to the type +visible on the source connection's `search_path`. This does not change cached names or add source +type namespaces to downstream `ADD COLUMN` statements. + ## 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 438b9ac78e..d5de69eaab 100644 --- a/docs/zh/connectors/source/PostgreSQL-CDC.md +++ b/docs/zh/connectors/source/PostgreSQL-CDC.md @@ -25,6 +25,10 @@ PostgreSQL CDC 连接器允许从 PostgreSQL 数据库读取快照数据和增 部分 PostgreSQL 兼容数据库（例如 HighGo）在 `information_schema.tables` 中会对同一张物理表返回多行重复记录。 连接器在表发现阶段会按完整的 `catalog.schema.table` 标识去重，这些重复行会被忽略，任务可正常启动。 +当类型名未命中缓存且多个 schema 中存在同名类型时，未限定 schema 的类型名会解析为源连接的 +`search_path` 中可见的类型。此改动不影响已缓存的类型名，也不会将源类型的 schema 传递到下游的 +`ADD COLUMN` 语句。 + ## 支持的数据源信息 | 数据源 | 支持的版本 | 驱动 | Url | Maven | diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/TypeRegistry.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/TypeRegistry.java index b4cdf55dcf..0af0e1aca2 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/TypeRegistry.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/TypeRegistry.java @@ -90,7 +90,12 @@ public class TypeRegistry { + ") e ON (t.oid = e.id) " + "WHERE n.nspname != 'pg_toast' and t.typtypmod != 0"; - private static final String SQL_NAME_LOOKUP = SQL_TYPES + " AND t.typname = ?"; + // Match PostgreSQL's unqualified type lookup when duplicate names exist across schemas. + // The OID tie-breaker keeps the fallback deterministic when no candidate is visible. + private static final String SQL_NAME_LOOKUP = + SQL_TYPES + + " AND t.typname = ?" + + " ORDER BY pg_catalog.pg_type_is_visible(t.oid) DESC, t.oid"; private static final String SQL_OID_LOOKUP = SQL_TYPES + " AND t.oid = ?"; diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCIT.java index 4123adbac8..106125c90e 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCIT.java @@ -59,7 +59,10 @@ import org.testcontainers.lifecycle.Startables; import org.testcontainers.utility.DockerImageName; import org.testcontainers.utility.DockerLoggerFactory; +import io.debezium.connector.postgresql.TypeRegistry; import io.debezium.connector.postgresql.connection.Lsn; +import io.debezium.connector.postgresql.connection.PostgresConnection; +import io.debezium.jdbc.JdbcConfiguration; import io.debezium.jdbc.JdbcConnection; import io.debezium.relational.TableId; import lombok.extern.slf4j.Slf4j; @@ -71,6 +74,7 @@ import java.nio.file.Files; import java.nio.file.Paths; import java.sql.Connection; import java.sql.DriverManager; +import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; @@ -1274,6 +1278,75 @@ public class PostgresCDCIT extends TestSuiteBase implements TestResource { POSTGRES_CONTAINER.getPassword()); } + @Test + public void shouldResolveNewTypesUsingPostgresSearchPath() throws SQLException { + JdbcConfiguration config = + JdbcConfiguration.create() + .withHostname(POSTGRES_CONTAINER.getHost()) + .withPort( + POSTGRES_CONTAINER.getMappedPort( + PostgreSQLContainer.POSTGRESQL_PORT)) + .withDatabase(POSTGRES_CONTAINER.getDatabaseName()) + .withUser(POSTGRES_CONTAINER.getUsername()) + .withPassword(POSTGRES_CONTAINER.getPassword()) + .build(); + + try (PostgresConnection connection = + new PostgresConnection(config, "type-registry-search-path-test")) { + // Create the types after priming so get(String) exercises SQL_NAME_LOOKUP. + TypeRegistry registry = new TypeRegistry(connection); + try (Statement statement = connection.connection().createStatement()) { + try { + statement.execute("CREATE SCHEMA type_lookup_a"); + statement.execute("CREATE SCHEMA type_lookup_b"); + statement.execute("CREATE TYPE type_lookup_b.late_mood AS ENUM ('b')"); + statement.execute("CREATE TYPE type_lookup_a.late_mood AS ENUM ('a')"); + statement.execute("SET search_path TO type_lookup_a, public"); + Assertions.assertEquals( + visibleTypeOid(connection.connection(), "late_mood"), + registry.get("late_mood").getOid()); + + statement.execute("CREATE TYPE type_lookup_a.reverse_mood AS ENUM ('a')"); + statement.execute("CREATE TYPE type_lookup_b.reverse_mood AS ENUM ('b')"); + statement.execute("SET search_path TO type_lookup_b, type_lookup_a, public"); + Assertions.assertEquals( + visibleTypeOid(connection.connection(), "reverse_mood"), + registry.get("reverse_mood").getOid()); + + statement.execute("CREATE TYPE type_lookup_b.hidden_mood AS ENUM ('b')"); + statement.execute("CREATE TYPE type_lookup_a.hidden_mood AS ENUM ('a')"); + statement.execute("SET search_path TO public"); + try (ResultSet resultSet = + statement.executeQuery( + "SELECT MIN(oid) FROM pg_catalog.pg_type WHERE typname = 'hidden_mood'")) { + Assertions.assertTrue(resultSet.next()); + // Keep the old fallback of returning a type even when none is visible. + Assertions.assertEquals( + resultSet.getInt(1), registry.get("hidden_mood").getOid()); + } + + int otherOid = + visibleTypeOid(connection.connection(), "type_lookup_a.reverse_mood"); + Assertions.assertEquals(otherOid, registry.get(otherOid).getOid()); + } finally { + statement.execute("DROP SCHEMA IF EXISTS type_lookup_a CASCADE"); + statement.execute("DROP SCHEMA IF EXISTS type_lookup_b CASCADE"); + } + } + } + } + + private int visibleTypeOid(Connection connection, String name) throws SQLException { + try (PreparedStatement statement = + connection.prepareStatement("SELECT to_regtype(?)::oid")) { + statement.setString(1, name); + try (ResultSet resultSet = statement.executeQuery()) { + Assertions.assertTrue(resultSet.next()); + return resultSet.getInt(1); + } + } + } + /** * The Postgres container is shared across all test methods in this class, so stale generated * replication slots must be dropped before the next CDC job starts.
