Resolve cross-schema row types to STRUCT (not VARCHAR)

src/include/postgres_utils.hpp

@@ -20,6 +20,7 @@ class PostgresTransaction;
 struct PostgresTypeData {
 	int64_t type_modifier = 0;
 	string type_name;
+	string type_schema;
 	idx_t array_dimensions = 0;
 };
 

src/postgres_utils.cpp

@@ -143,6 +143,7 @@ LogicalType PostgresUtils::TypeToLogicalType(optional_ptr<PostgresTransaction> t
 		PostgresTypeData child_type_info;
 		child_type_info.type_name = pgtypename.substr(1);
 		child_type_info.type_modifier = type_info.type_modifier;
+		child_type_info.type_schema = type_info.type_schema;
 		PostgresType child_pg_type;
 		auto child_type = PostgresUtils::TypeToLogicalType(transaction, schema, child_type_info, child_pg_type);
 		// populate the child OID from the actual Postgres type name
@@ -244,8 +245,12 @@ LogicalType PostgresUtils::TypeToLogicalType(optional_ptr<PostgresTransaction> t
 		if (!context) {
 			throw InternalException("Context is destroyed!?");
 		}
-		auto entry = schema->GetEntry(CatalogTransaction(schema->ParentCatalog(), *context), CatalogType::TYPE_ENTRY,
-		                              pgtypename);
+		optional_ptr<SchemaCatalogEntry> lookup_schema =
+		    type_info.type_schema != schema->name
+		        ? schema->ParentCatalog().GetSchema(*context, type_info.type_schema, OnEntryNotFound::THROW_EXCEPTION)
+		        : schema.get();
+		auto entry = lookup_schema->GetEntry(CatalogTransaction(lookup_schema->ParentCatalog(), *context),
+		                                     CatalogType::TYPE_ENTRY, pgtypename);
 		if (!entry) {
 			// unsupported so fallback to varchar
 			postgres_type.info = PostgresTypeAnnotation::CAST_TO_VARCHAR;

src/storage/postgres_table_set.cpp

@@ -24,17 +24,18 @@ string PostgresTableSet::GetInitializeQuery(const string &schema, const string &
 SELECT pg_namespace.oid AS namespace_id, relname, relpages, attname,
     pg_type.typname type_name, atttypmod type_modifier, pg_attribute.attndims ndim,
     attnum, pg_attribute.attnotnull AS notnull, NULL constraint_id,
-    NULL constraint_type, NULL constraint_key
+    NULL constraint_type, NULL constraint_key, type_ns.nspname AS type_schema
 FROM pg_class
 JOIN pg_namespace ON relnamespace = pg_namespace.oid
 JOIN pg_attribute ON pg_class.oid=pg_attribute.attrelid
 JOIN pg_type ON atttypid=pg_type.oid
+JOIN pg_namespace type_ns ON pg_type.typnamespace = type_ns.oid
 WHERE attnum > 0 AND relkind IN ('r', 'v', 'm', 'f', 'p') ${CONDITION}
 UNION ALL
 SELECT pg_namespace.oid AS namespace_id, relname, NULL relpages, NULL attname, NULL type_name,
     NULL type_modifier, NULL ndim, NULL attnum, NULL AS notnull,
     pg_constraint.oid AS constraint_id, contype AS constraint_type,
-    conkey AS constraint_key
+    conkey AS constraint_key, NULL AS type_schema
 FROM pg_class
 JOIN pg_namespace ON relnamespace = pg_namespace.oid
 JOIN pg_constraint ON (pg_class.oid=pg_constraint.conrelid)
@@ -61,6 +62,10 @@ void PostgresTableSet::AddColumn(optional_ptr<PostgresTransaction> transaction,
 	type_info.type_modifier = result.GetInt64(row, column_index + 2);
 	type_info.array_dimensions = result.GetInt64(row, column_index + 3);
 	bool is_not_null = result.GetBool(row, column_index + 5);
+	idx_t type_schema_index = column_index + 9;
+	if (!result.IsNull(row, type_schema_index)) {
+		type_info.type_schema = result.GetString(row, type_schema_index);
+	}
 	string default_value;
 
 	PostgresType postgres_type;

src/storage/postgres_type_set.cpp

@@ -84,12 +84,14 @@ void PostgresTypeSet::InitializeEnums(PostgresTransaction &transaction, Postgres
 
 string PostgresTypeSet::GetInitializeCompositesQuery(const string &schema) {
 	string base_query = R"(
-SELECT n.oid, t.typrelid AS id, t.typname as type, pg_attribute.attname, sub_type.typname
+SELECT n.oid, t.typrelid AS id, t.typname as type, pg_attribute.attname, sub_type.typname,
+       sub_type_ns.nspname AS sub_type_schema
 FROM pg_type t
 JOIN pg_catalog.pg_namespace n ON n.oid = t.typnamespace
 JOIN pg_class ON pg_class.oid = t.typrelid
 JOIN pg_attribute ON attrelid=t.typrelid
 JOIN pg_type sub_type ON (pg_attribute.atttypid=sub_type.oid)
+JOIN pg_catalog.pg_namespace sub_type_ns ON sub_type_ns.oid = sub_type.typnamespace
 WHERE pg_class.relkind IN ('c', 'r', 'v', 'm', 'f', 'p')
 AND t.typtype='c'
 AND pg_attribute.attnum > 0
@@ -116,6 +118,7 @@ void PostgresTypeSet::CreateCompositeType(PostgresTransaction &transaction, Post
 		auto type_name = result.GetString(row, 3);
 		PostgresTypeData type_data;
 		type_data.type_name = result.GetString(row, 4);
+		type_data.type_schema = result.GetString(row, 5);
 		PostgresType child_type;
 		child_types.push_back(
 		    make_pair(type_name, PostgresUtils::TypeToLogicalType(&transaction, &schema, type_data, child_type)));

test/sql/storage/attach_types_table_row.test

@@ -71,3 +71,65 @@ SET pg_use_text_protocol=false;
 
 statement ok
 DROP SCHEMA IF EXISTS row_types CASCADE;
+
+statement ok
+DROP SCHEMA IF EXISTS row_types_a CASCADE;
+
+statement ok
+DROP SCHEMA IF EXISTS row_types_b CASCADE;
+
+statement ok
+CREATE SCHEMA row_types_a;
+
+statement ok
+CREATE SCHEMA row_types_b;
+
+statement ok
+CREATE TABLE row_types_a.widget(id INT, label VARCHAR);
+
+statement ok
+INSERT INTO row_types_a.widget VALUES (1, 'hello');
+
+# view in schema B whose column is row type of a relation in schema A
+statement ok
+CALL postgres_execute('s', 'CREATE VIEW row_types_b.widget_view AS SELECT w FROM row_types_a.widget AS w');
+
+# table in schema B whose column is declared with schema A's row type
+statement ok
+CALL postgres_execute('s', 'CREATE TABLE row_types_b.wrapper(id INT, payload row_types_a.widget)');
+
+statement ok
+CALL postgres_execute('s', 'INSERT INTO row_types_b.wrapper VALUES (1, ROW(2, ''nested''))');
+
+statement ok
+CALL pg_clear_cache();
+
+query II
+SELECT w.id, w.label FROM row_types_b.widget_view
+----
+1	hello
+
+query III
+SELECT id, payload.id, payload.label FROM row_types_b.wrapper
+----
+1	2	nested
+
+statement ok
+SET pg_use_text_protocol=true;
+
+statement ok
+CALL pg_clear_cache();
+
+query II
+SELECT w.id, w.label FROM row_types_b.widget_view
+----
+1	hello
+
+statement ok
+SET pg_use_text_protocol=false;
+
+statement ok
+DROP SCHEMA IF EXISTS row_types_a CASCADE;
+
+statement ok
+DROP SCHEMA IF EXISTS row_types_b CASCADE;