@@ -0,0 +1,183 @@
+package pg
+
+import (
+	"context"
+	"testing"
+
+	"github.com/bytebase/omni/pg/ast"
+	"github.com/bytebase/omni/pg/catalog"
+
+	storepb "github.com/bytebase/bytebase/backend/generated-go/store"
+)
+
+// TestLoaderInstallsCompositeTypes proves the query-span catalog loader
+// installs composite types from metadata with real attributes, in dependency
+// order (aa_nested sorts before its dependency zz_base alphabetically), and
+// that tables typed with them install cleanly.
+func TestLoaderInstallsCompositeTypes(t *testing.T) {
+	meta := &storepb.DatabaseSchemaMetadata{
+		Schemas: []*storepb.SchemaMetadata{
+			{
+				Name: "public",
+				EnumTypes: []*storepb.EnumTypeMetadata{
+					{Name: "status", Values: []string{"a", "b"}},
+				},
+				CompositeTypes: []*storepb.CompositeTypeMetadata{
+					{
+						// References zz_base only through an array suffix.
+						Name: "aa_array_only",
+						Attributes: []*storepb.CompositeTypeAttribute{
+							{Name: "items", Type: "public.zz_base[]"},
+						},
+					},
+					{
+						Name: "aa_nested",
+						Attributes: []*storepb.CompositeTypeAttribute{
+							{Name: "home", Type: "public.zz_base"},
+							{Name: "s", Type: "public.status"},
+						},
+					},
+					{
+						Name: "zz_base",
+						Attributes: []*storepb.CompositeTypeAttribute{
+							{Name: "street", Type: "text"},
+							{Name: "city", Type: "character varying(50)"},
+						},
+					},
+				},
+				Tables: []*storepb.TableMetadata{
+					{
+						Name: "users",
+						Columns: []*storepb.ColumnMetadata{
+							{Name: "id", Type: "integer"},
+							{Name: "home", Type: "public.zz_base", Nullable: true},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	cat := catalog.New()
+	cat.SetSearchPath([]string{"public"})
+	loader := newCatalogLoader(cat, meta)
+	if err := loader.Load(context.Background()); err != nil {
+		t.Fatalf("Load: %v", err)
+	}
+
+	base := cat.GetRelation("public", "zz_base")
+	if base == nil {
+		t.Fatal("composite type zz_base not installed")
+	}
+	if len(base.Columns) != 2 {
+		t.Errorf("zz_base attribute count: got %d, want 2", len(base.Columns))
+	}
+	if base.Columns[0].Name != "street" {
+		t.Errorf("zz_base first attribute: got %q, want street", base.Columns[0].Name)
+	}
+	nested := cat.GetRelation("public", "aa_nested")
+	if nested == nil {
+		t.Fatal("composite type aa_nested not installed")
+	}
+	if len(nested.Columns) != 2 {
+		t.Errorf("aa_nested attribute count: got %d, want 2", len(nested.Columns))
+	}
+	arrayOnly := cat.GetRelation("public", "aa_array_only")
+	if arrayOnly == nil {
+		t.Fatal("composite type aa_array_only not installed")
+	}
+	if len(arrayOnly.Columns) != 1 || arrayOnly.Columns[0].Name != "items" {
+		t.Errorf("aa_array_only must install with its real attribute, got %d columns", len(arrayOnly.Columns))
+	}
+	if cat.GetRelation("public", "users") == nil {
+		t.Fatal("table with composite column not installed")
+	}
+}
+
+// TestLoaderCompositeFallbackPreservesAttributeNames forces the pseudo
+// fallback via a domain-typed attribute (domains are not loader objects) and
+// asserts the fallback keeps the metadata attribute names.
+func TestLoaderCompositeFallbackPreservesAttributeNames(t *testing.T) {
+	meta := &storepb.DatabaseSchemaMetadata{
+		Schemas: []*storepb.SchemaMetadata{
+			{
+				Name: "public",
+				CompositeTypes: []*storepb.CompositeTypeMetadata{
+					{
+						Name: "with_domain",
+						Attributes: []*storepb.CompositeTypeAttribute{
+							{Name: "p", Type: "public.pos_int"},
+							{Name: "note", Type: "text"},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	cat := catalog.New()
+	cat.SetSearchPath([]string{"public"})
+	loader := newCatalogLoader(cat, meta)
+	if err := loader.Load(context.Background()); err != nil {
+		t.Fatalf("Load: %v", err)
+	}
+
+	rel := cat.GetRelation("public", "with_domain")
+	if rel == nil {
+		t.Fatal("composite must fall back, not vanish")
+	}
+	if len(rel.Columns) != 2 {
+		t.Fatalf("fallback must preserve attribute names, got %d columns", len(rel.Columns))
+	}
+	if rel.Columns[0].Name != "p" || rel.Columns[1].Name != "note" {
+		t.Errorf("fallback attribute names: got %q, %q", rel.Columns[0].Name, rel.Columns[1].Name)
+	}
+}
+
+// TestRangeVarFallbackSkipsCompositeTypes proves the query-span fallback does
+// not expand a standalone composite type as a FROM source — PostgreSQL
+// rejects composite types as table sources.
+func TestRangeVarFallbackSkipsCompositeTypes(t *testing.T) {
+	meta := &storepb.DatabaseSchemaMetadata{
+		Schemas: []*storepb.SchemaMetadata{
+			{
+				Name: "public",
+				CompositeTypes: []*storepb.CompositeTypeMetadata{
+					{
+						Name: "addr",
+						Attributes: []*storepb.CompositeTypeAttribute{
+							{Name: "street", Type: "text"},
+						},
+					},
+				},
+				Tables: []*storepb.TableMetadata{
+					{
+						Name: "t",
+						Columns: []*storepb.ColumnMetadata{
+							{Name: "id", Type: "integer"},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	cat := catalog.New()
+	cat.SetSearchPath([]string{"public"})
+	loader := newCatalogLoader(cat, meta)
+	if err := loader.Load(context.Background()); err != nil {
+		t.Fatalf("Load: %v", err)
+	}
+
+	e := &omniQuerySpanExtractor{
+		cat:             cat,
+		searchPath:      []string{"public"},
+		defaultDatabase: "db",
+	}
+	if results := e.extractColumnsFromRangeVar(&ast.RangeVar{Schemaname: "public", Relname: "addr"}); results != nil {
+		t.Errorf("composite type must not expand as a FROM source, got %d columns", len(results))
+	}
+	if results := e.extractColumnsFromRangeVar(&ast.RangeVar{Schemaname: "public", Relname: "t"}); len(results) != 1 {
+		t.Errorf("real table must still expand, got %d columns", len(results))
+	}
+}
@@ -55,6 +55,7 @@ type objectKind int
 const (
 	kindSchema objectKind = iota
 	kindEnum
+	kindComposite
 	kindTable
 	kindView
 	kindMatView
@@ -71,11 +72,12 @@ type objectEntry struct {
 	name   string
 
 	// Exactly one of the following is populated based on kind.
-	enumMeta    *storepb.EnumTypeMetadata
-	tableMeta   *storepb.TableMetadata
-	viewMeta    *storepb.ViewMetadata
-	matViewMeta *storepb.MaterializedViewMetadata
-	funcMeta    *storepb.FunctionMetadata
+	enumMeta      *storepb.EnumTypeMetadata
+	compositeMeta *storepb.CompositeTypeMetadata
+	tableMeta     *storepb.TableMetadata
+	viewMeta      *storepb.ViewMetadata
+	matViewMeta   *storepb.MaterializedViewMetadata
+	funcMeta      *storepb.FunctionMetadata
 }
 
 // key returns a canonical kind-prefixed identifier used for dependency
@@ -89,7 +91,7 @@ func (e *objectEntry) key() string {
 	switch e.kind {
 	case kindSchema:
 		return "schema:" + e.schema
-	case kindEnum:
+	case kindEnum, kindComposite:
 		return "type:" + e.schema + "." + e.name
 	case kindTable, kindView, kindMatView:
 		return "rel:" + e.schema + "." + e.name
@@ -130,6 +132,8 @@ func kindLabel(k objectKind) string {
 		return "0schema"
 	case kindEnum:
 		return "1enum"
+	case kindComposite:
+		return "1composite"
 	case kindTable:
 		return "2table"
 	case kindView:
@@ -205,6 +209,14 @@ func (l *catalogLoader) collectObjects() []*objectEntry {
 				enumMeta: enum,
 			})
 		}
+		for _, composite := range sm.CompositeTypes {
+			out = append(out, &objectEntry{
+				kind:          kindComposite,
+				schema:        sm.Name,
+				name:          composite.Name,
+				compositeMeta: composite,
+			})
+		}
 		for _, tbl := range sm.Tables {
 			out = append(out, &objectEntry{
 				kind:      kindTable,
@@ -378,6 +390,18 @@ func buildDependencyEdges(objects []*objectEntry, index map[string]*objectEntry)
 					}
 				}
 			}
+		case kindComposite:
+			for _, attribute := range obj.compositeMeta.Attributes {
+				for _, ref := range extractUserTypeRefs(attribute.Type) {
+					if k := typeRefKey(ref); index[k] != nil {
+						deps = append(deps, k)
+					}
+					// An attribute may be typed with a table/view row type.
+					if k := relationKey(ref.Schema, ref.Name); index[k] != nil {
+						deps = append(deps, k)
+					}
+				}
+			}
 		case kindSchema, kindEnum:
 			// Schemas have no outgoing edges (they are the top of the graph).
 			// Enums depend only on their schema, already captured above.
@@ -562,6 +586,12 @@ func (l *catalogLoader) installReal(obj *objectEntry) error {
 		return err
 	case kindEnum:
 		return l.cat.DefineEnum(buildCreateEnumStmt(obj.schema, obj.enumMeta))
+	case kindComposite:
+		stmt, err := buildCompositeTypeStmt(obj.schema, obj.compositeMeta)
+		if err != nil {
+			return err
+		}
+		return l.cat.DefineCompositeType(stmt)
 	case kindTable:
 		stmt, err := buildCreateStmt(obj.schema, obj.tableMeta)
 		if err != nil {
@@ -602,6 +632,27 @@ func (l *catalogLoader) installPseudo(obj *objectEntry) error {
 		return errors.New("schema has no pseudo form")
 	case kindEnum:
 		return l.cat.DefineEnum(pseudoCreateEnumStmt(obj.schema, obj.name))
+	case kindComposite:
+		// Name-preserving, text-backed fallback when metadata attributes are
+		// known, so (col).field access still resolves; the "_broken" shape
+		// remains for composites with no attribute information at all.
+		if len(obj.compositeMeta.GetAttributes()) > 0 {
+			items := make([]ast.Node, 0, len(obj.compositeMeta.Attributes))
+			for _, attribute := range obj.compositeMeta.Attributes {
+				if attribute.Name == "" {
+					continue
+				}
+				items = append(items, &ast.ColumnDef{
+					Colname:  attribute.Name,
+					TypeName: pseudoTextTypeName(),
+				})
+			}
+			return l.cat.DefineCompositeType(&ast.CompositeTypeStmt{
+				Typevar:    &ast.RangeVar{Schemaname: obj.schema, Relname: obj.name},
+				Coldeflist: &ast.List{Items: items},
+			})
+		}
+		return l.cat.DefineCompositeType(pseudoCompositeTypeStmt(obj.schema, obj.name))
 	case kindTable:
 		cols := columnNamesFromTableMetadata(obj.tableMeta)
 		return l.cat.DefineRelation(pseudoCreateTableStmt(obj.schema, obj.name, cols), 'r')
 
@@ -28,6 +28,34 @@ func buildCreateEnumStmt(schema string, enum *storepb.EnumTypeMetadata) *ast.Cre
 	}
 }
 
+// buildCompositeTypeStmt translates CompositeTypeMetadata + schema into a
+// CompositeTypeStmt. Attribute types are parsed via typeNameFromString; an
+// error on any attribute propagates so the loader can pseudo the whole type.
+func buildCompositeTypeStmt(schema string, composite *storepb.CompositeTypeMetadata) (*ast.CompositeTypeStmt, error) {
+	cols := make([]ast.Node, 0, len(composite.Attributes))
+	for _, attribute := range composite.Attributes {
+		if attribute.Name == "" {
+			continue
+		}
+		if attribute.Type == "" {
+			return nil, errors.Errorf("attribute %q: empty type", attribute.Name)
+		}
+		tn, err := typeNameFromString(attribute.Type)
+		if err != nil {
+			return nil, errors.Wrapf(err, "attribute %q", attribute.Name)
+		}
+		cols = append(cols, &ast.ColumnDef{
+			Colname:    attribute.Name,
+			TypeName:   tn,
+			CollClause: qsCollateClauseFromReference(attribute.Collation),
+		})
+	}
+	return &ast.CompositeTypeStmt{
+		Typevar:    &ast.RangeVar{Schemaname: schema, Relname: composite.Name},
+		Coldeflist: &ast.List{Items: cols},
+	}, nil
+}
+
 // buildCreateStmt translates TableMetadata into a CreateStmt with real
 // column types. Column types are parsed via typeNameFromString; an error on
 // any single column propagates back so the loader can pseudo the whole
@@ -243,3 +271,21 @@ func splitTopLevelCommas(s string) []string {
 	parts = append(parts, strings.TrimSpace(s[start:]))
 	return parts
 }
+
+// qsCollateClauseFromReference converts a stored emit-ready collation
+// reference (quoted as needed, schema-qualified outside pg_catalog) into an
+// AST collate clause. Returns nil for no collation.
+func qsCollateClauseFromReference(reference string) *ast.CollateClause {
+	if reference == "" {
+		return nil
+	}
+	var items []ast.Node
+	if schemaName, collationName, ok := splitQualifiedName(reference); ok {
+		items = []ast.Node{&ast.String{Str: schemaName}, &ast.String{Str: collationName}}
+	} else {
+		name := strings.Trim(reference, `"`)
+		name = strings.ReplaceAll(name, `""`, `"`)
+		items = []ast.Node{&ast.String{Str: name}}
+	}
+	return &ast.CollateClause{Collname: &ast.List{Items: items}}
+}
@@ -48,9 +48,10 @@ func pseudoCreateDomainStmt(schema, name string) *ast.CreateDomainStmt {
 }
 
 // pseudoCompositeTypeStmt builds a composite type with a single text field
-// named "_broken". Without CompositeTypeMetadata in storepb we cannot
-// preserve real field names, so (col).field access against a pseudo composite
-// will fall through to extractFallbackColumns.
+// named "_broken". It is the fallback when a composite type's real
+// definition is absent from metadata or fails to install, so (col).field
+// access against a pseudo composite will fall through to
+// extractFallbackColumns.
 func pseudoCompositeTypeStmt(schema, name string) *ast.CompositeTypeStmt {
 	return &ast.CompositeTypeStmt{
 		Typevar: &ast.RangeVar{
 
@@ -91,6 +91,12 @@ func extractUserTypeRefs(typeStr string) []UserTypeRef {
 // the caller's job to split.
 func stripTypeModifiers(typeStr string) string {
 	s := strings.TrimSpace(typeStr)
+	// Array suffixes wrap the element type: "public.addr[]" refers to
+	// public.addr. Quoted identifiers end with a quote, so a genuine name
+	// containing "[]" is never clipped.
+	for strings.HasSuffix(s, "[]") {
+		s = strings.TrimSpace(strings.TrimSuffix(s, "[]"))
+	}
 	s = stripTimeZoneSuffix(s)
 	s = stripParens(s)
 	return strings.TrimSpace(s)
 
@@ -1064,6 +1064,11 @@ func (e *omniQuerySpanExtractor) extractColumnsFromRangeVar(rv *ast.RangeVar) []
 	if rel == nil {
 		return nil
 	}
+	// Standalone composite types share the relation namespace but are not
+	// FROM-able; PostgreSQL rejects them as table sources.
+	if rel.RelKind == 'c' {
+		return nil
+	}
 	var results []base.QuerySpanResult
 	for _, col := range rel.Columns {
 		colSet := make(base.SourceColumnSet)