feat(gaussdb): 支持 GaussDB OID 别名兼容及运行时查询

- 新增 GAUSSDB_OID_ALIASES 字典，实现 Postgres OID 到 GaussDB 多别名映射
- 添加 is_compatible_oid 方法，实现对 OID 兼容性的判断支持别名
- 实现 get_oid_name 方法，根据 OID 获取对应的类型名称字符串
- 在 TypeInfo 中增加 fetch_runtime_oid，支持运行时查询类型 OID，兼容同步和异步连接
- 在 TypeInfo 中添加 get_compatible_oids，用于获取基础 OID 及其别名列表
- 编写测试用例，覆盖 OID 兼容性判断、OID 名称获取及运行时 OID 查询
- 测试中增加 type_code 兼容性校验，确保 GaussDB 返回的 OID 合法有效

Author: 陈云亮

gaussdb/gaussdb/_oids.py

@@ -122,3 +122,55 @@
 YEAR_OID = 1038
 
 # autogenerated: end
+
+
+# =====================================================
+# GaussDB OID 别名映射（PostgreSQL OID -> GaussDB 等效 OID 列表）
+# =====================================================
+
+GAUSSDB_OID_ALIASES: dict[int, list[int]] = {
+    # date 类型可能映射到多个 OID
+    DATE_OID: [DATE_OID, SMALLDATETIME_OID],
+    # timestamp 类型
+    TIMESTAMP_OID: [TIMESTAMP_OID, SMALLDATETIME_OID],
+    TIMESTAMPTZ_OID: [TIMESTAMPTZ_OID],
+    # 其他类型保持一对一映射
+}
+
+
+def is_compatible_oid(expected_oid: int, actual_oid: int) -> bool:
+    """
+    检查两个 OID 是否兼容
+
+    用于 GaussDB 场景下的类型比较，考虑 OID 别名。
+
+    Args:
+        expected_oid: 期望的 OID
+        actual_oid: 实际的 OID
+
+    Returns:
+        是否兼容
+    """
+    if expected_oid == actual_oid:
+        return True
+
+    # 检查别名映射
+    aliases = GAUSSDB_OID_ALIASES.get(expected_oid, [expected_oid])
+    return actual_oid in aliases
+
+
+def get_oid_name(oid: int) -> str:
+    """
+    获取 OID 对应的类型名称
+
+    Args:
+        oid: 类型 OID
+
+    Returns:
+        类型名称字符串
+    """
+    # 反向查找 OID 常量名
+    for name, value in globals().items():
+        if name.endswith("_OID") and value == oid:
+            return name.replace("_OID", "").lower()
+    return f"oid_{oid}"

gaussdb/gaussdb/_typeinfo.py

@@ -16,6 +16,7 @@
 from . import sql
 from .abc import AdaptContext, Query
 from .rows import dict_row
+from ._oids import GAUSSDB_OID_ALIASES
 from ._compat import TypeAlias, TypeVar
 from ._typemod import TypeModifier
 from ._encodings import conn_encoding
@@ -209,6 +210,56 @@ def get_precision(self, fmod: int) -> int | None:
     def get_scale(self, fmod: int) -> int | None:
         return self.typemod.get_scale(fmod)
 
+    @classmethod
+    def fetch_runtime_oid(cls, conn: Any, typename: str) -> int | None:
+        """
+        运行时获取类型 OID
+
+        从数据库查询正确的 OID，处理 GaussDB 差异。
+
+        Args:
+            conn: 数据库连接
+            typename: 类型名称
+
+        Returns:
+            类型 OID，查询失败返回 None
+        """
+        try:
+            from .connection import Connection
+
+            if isinstance(conn, Connection):
+                result = conn.execute(
+                    "SELECT oid FROM pg_type WHERE typname = %s", [typename]
+                ).fetchone()
+            else:
+                # AsyncConnection
+                import asyncio
+
+                async def _fetch():
+                    result = await conn.execute(
+                        "SELECT oid FROM pg_type WHERE typname = %s", [typename]
+                    )
+                    return await result.fetchone()
+
+                result = asyncio.run(_fetch())
+
+            return result[0] if result else None
+        except Exception:
+            return None
+
+    @classmethod
+    def get_compatible_oids(cls, base_oid: int) -> list[int]:
+        """
+        获取兼容的 OID 列表
+
+        Args:
+            base_oid: 基础 OID
+
+        Returns:
+            包含基础 OID 及其别名的列表
+        """
+        return GAUSSDB_OID_ALIASES.get(base_oid, [base_oid])
+
 
 class TypesRegistry:
     """

tests/test_gaussdb_dbapi20.py

@@ -161,3 +161,30 @@ def fake_connect(conninfo, *, timeout=0.0):
 def test_connect_badargs(monkeypatch, pgconn, args, kwargs, exctype):
     with pytest.raises(exctype):
         gaussdb.connect(*args, **kwargs)
+
+
+class TestTypeCode:
+    """type_code 兼容性测试"""
+
+    def test_type_code_comparison(self, conn):
+        """测试 type_code 比较"""
+        cur = conn.cursor()
+        cur.execute("select 1::int, 'hello'::text")
+
+        desc = cur.description
+
+        # 验证 type_code 是整数
+        for col in desc:
+            assert isinstance(col.type_code, int)
+            assert col.type_code > 0
+
+    def test_type_code_date(self, conn):
+        """测试日期类型 type_code"""
+        cur = conn.cursor()
+        cur.execute("select current_date")
+
+        type_code = cur.description[0].type_code
+
+        # GaussDB 可能返回不同的 OID
+        # 只要是有效的 OID 即可
+        assert type_code > 0, f"Invalid type_code: {type_code}"

tests/test_typeinfo.py

@@ -3,6 +3,14 @@
 import gaussdb
 from gaussdb import sql
 from gaussdb.pq import TransactionStatus
+from gaussdb._oids import (
+    DATE_OID,
+    GAUSSDB_OID_ALIASES,
+    SMALLDATETIME_OID,
+    TIMESTAMP_OID,
+    get_oid_name,
+    is_compatible_oid,
+)
 from gaussdb.types import TypeInfo
 from gaussdb.types.enum import EnumInfo
 from gaussdb.types.range import RangeInfo
@@ -208,3 +216,34 @@ def test_registry_isolated():
     print(f"orig={orig},tinfo={tinfo},r={r},tdummy={tdummy}")
     assert r[25] is r["dummy"] is tdummy
     assert orig[25] is r["text"] is tinfo
+
+
+class TestOidCompatibility:
+    """OID 兼容性测试"""
+
+    def test_same_oid_compatible(self):
+        """相同 OID 应兼容"""
+        assert is_compatible_oid(DATE_OID, DATE_OID)
+        assert is_compatible_oid(TIMESTAMP_OID, TIMESTAMP_OID)
+
+    def test_alias_oid_compatible(self):
+        """别名 OID 应兼容"""
+        # 如果 smalldatetime 是 date 的别名
+        if SMALLDATETIME_OID in GAUSSDB_OID_ALIASES.get(DATE_OID, []):
+            assert is_compatible_oid(DATE_OID, SMALLDATETIME_OID)
+
+    def test_different_oid_not_compatible(self):
+        """不同类型 OID 不兼容"""
+        assert not is_compatible_oid(DATE_OID, 23)  # int4
+
+    def test_get_oid_name(self):
+        """测试获取 OID 名称"""
+        assert get_oid_name(DATE_OID) == "date"
+        assert get_oid_name(TIMESTAMP_OID) == "timestamp"
+
+    def test_runtime_oid_fetch(self, conn):
+        """测试运行时 OID 查询"""
+        oid = TypeInfo.fetch_runtime_oid(conn, "date")
+        if oid is not None:
+            assert isinstance(oid, int)
+            assert oid > 0