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 89a5598eb5 [Fix][Connector-V2] Resolve late PostgreSQL CDC type names
by search path (#12836)
89a5598eb5 is described below
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.
