Resolve cross-schema complex column types by schema ID

dpi_bridge/dpi_obj.go

@@ -30,6 +30,7 @@ import (
 	"database/sql"
 	"encoding/hex"
 	"fmt"
+	"strconv"
 	"strings"
 	"unsafe"
 
@@ -250,10 +251,23 @@ func supportedArrayElement(sqlType int16) bool {
 
 func (conn *connHandle) columnObjectDesc(typeName string) (*objDescHandle, uintptr, error) {
 	typeName = strings.ToUpper(typeName)
+	owner := strings.ToUpper(conn.user)
+	if name, schemaID, tagged := strings.Cut(typeName, "@"); tagged {
+		id, err := strconv.Atoi(schemaID)
+		if err != nil || id <= 0 {
+			return nil, 0, fmt.Errorf("invalid type schema ID %q", schemaID)
+		}
+		if err := conn.db.QueryRow("SELECT OBJECT_NAME FROM ALL_OBJECTS WHERE OBJECT_ID=? AND OBJECT_TYPE='SCH'", id).Scan(&owner); err != nil {
+			return nil, 0, err
+		}
+		typeName = name
+	}
 	if typeName == "" || objectScalarType(typeName) != 0 {
 		return nil, 0, fmt.Errorf("not an object type")
 	}
-	owner := strings.ToUpper(conn.user)
+	if qualifiedOwner, qualifiedName, qualified := strings.Cut(typeName, "."); qualified {
+		owner, typeName = qualifiedOwner, qualifiedName
+	}
 	var oid int64
 	if err := conn.db.QueryRow("SELECT TYPE_OID FROM ALL_TYPES WHERE OWNER=? AND TYPE_NAME=?", owner, typeName).Scan(&oid); err != nil {
 		return nil, 0, err

dpi_bridge/third_party/chunanyong_dm/t.go

@@ -8,6 +8,7 @@ import (
 	"database/sql/driver"
 	"io"
 	"reflect"
+	"strconv"
 	"strings"
 )
 
@@ -351,6 +352,12 @@ func (innerRows *innerRows) ColumnTypeDatabaseTypeName(index int) string {
 		return ""
 	}
 	if column := innerRows.checkIndex(index); column != nil {
+		if isComplexType(int(column.colType), int(column.scale)) && column.typeDescriptor != nil &&
+			column.typeDescriptor.m_sqlName != nil && column.typeDescriptor.m_sqlName.m_schId > 0 {
+			// The column schema is the table owner, which may differ from the type owner.
+			// Pass the server's schema ID to the bridge for an ALL_OBJECTS lookup.
+			return column.typeName + "@" + strconv.Itoa(column.typeDescriptor.m_sqlName.m_schId)
+		}
 		return column.typeName
 	}
 	return ""

tests/integration/test_p1_array_types.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import os
 from decimal import Decimal
 
 import pytest
@@ -188,3 +189,64 @@ def test_varray_of_varray_keeps_decimal_elements(conn, table_name_factory, drop_
                     cleanup.execute(f"DROP TYPE {name}")
         conn.commit()
         cur.close()
+
+
+def test_cross_schema_varray_of_objects(conn, conn_params, table_name_factory):
+    admin_password = os.environ.get("DM_CI_ADMIN_PASSWORD")
+    if not admin_password:
+        pytest.skip("cross-schema type regression requires the admin test password")
+
+    object_type = table_name_factory("DMPY_SHARED_ITEM")
+    array_type = table_name_factory("DMPY_SHARED_ARRAY")
+    table = table_name_factory("DMPY_SHARED_TAB")
+    local_table = table_name_factory("DMPY_REF_SHARED")
+    values = [[Decimal("12345678901234567890.12345678"), "汉字"], None]
+    admin = dmPython.connect(**{**conn_params, "user": "SYSDBA", "password": admin_password})
+    created = []
+    local_created = []
+    try:
+        with admin.cursor() as cur:
+            cur.execute(f"CREATE TYPE {object_type} AS OBJECT (AMOUNT DECIMAL(30,8), LABEL VARCHAR(20))")
+            created.append(("TYPE", object_type))
+            cur.execute(f"CREATE TYPE {array_type} AS VARRAY(2) OF {object_type}")
+            created.append(("TYPE", array_type))
+            cur.execute(f"CREATE TABLE {table} (V {array_type})")
+            created.append(("TABLE", table))
+            cur.execute(f"GRANT EXECUTE ON {object_type} TO {conn_params['user']}")
+            cur.execute(f"GRANT EXECUTE ON {array_type} TO {conn_params['user']}")
+            cur.execute(f"GRANT SELECT, INSERT ON {table} TO {conn_params['user']}")
+        admin.commit()
+
+        # A same-named type in the reader schema must not shadow SYSDBA's type.
+        with conn.cursor() as cur:
+            cur.execute(f"CREATE TYPE {object_type} AS OBJECT (FLAG INTEGER)")
+            local_created.append(("TYPE", object_type))
+            cur.execute(f"CREATE TYPE {array_type} AS VARRAY(2) OF {object_type}")
+            local_created.append(("TYPE", array_type))
+            cur.execute(f"CREATE TABLE {local_table} (V SYSDBA.{array_type})")
+            local_created.append(("TABLE", local_table))
+        conn.commit()
+
+        value = dmPython.objectvar(conn, array_type, schema="SYSDBA")
+        assert value.type.schema == "SYSDBA"
+        value.setvalue(values)
+        with conn.cursor() as cur:
+            cur.execute(f"INSERT INTO SYSDBA.{table} VALUES (?)", (value,))
+            cur.execute(f"INSERT INTO {local_table} VALUES (?)", (value,))
+            conn.commit()
+            for source in (f"SYSDBA.{table}", local_table):
+                cur.execute(f"SELECT V FROM {source}")
+                fetched = cur.fetchone()[0]
+                assert fetched.type.schema == "SYSDBA"
+                assert fetched.getvalue() == values
+    finally:
+        conn.rollback()
+        with conn.cursor() as cur:
+            for kind, name in reversed(local_created):
+                cur.execute(f"DROP {kind} {name}")
+        conn.commit()
+        with admin.cursor() as cur:
+            for kind, name in reversed(created):
+                cur.execute(f"DROP {kind} {name}")
+        admin.commit()
+        admin.close()