opt: track UDT references by name in the Metadata

Previously, it was possible for invalid queries to be kept in the cache
after a schema change, since user-defined types were tracked only by OID.
This missed cases where the search path changed which object a name in
the query resolved to (or whether it resolved at all).

This patch fixes this behavior by tracking UDT references by name, similar
to what was already done for data sources. This ensures that the query
staleness check doesn't miss changes to the search path.

Fixes cockroachdb#96674

Release note (bug fix): Fixed a bug that could prevent a cached query with
a user-defined type reference from being invalidated even after a schema
change that should prevent the type from being resolved.

pkg/sql/logictest/testdata/logic_test/schema

@@ -943,3 +943,78 @@ statement ok
 USE test;
 DROP DATABASE d;
 DROP TABLE xy;
+
+# Regression tests for #96674.
+subtest alter_udt_schema
+
+# Renaming the schema should invalidate a schema-qualified UDT reference.
+statement ok
+CREATE SCHEMA sc;
+CREATE TYPE sc.t AS ENUM ('HELLO');
+
+query T
+SELECT 'HELLO'::sc.t;
+----
+HELLO
+
+statement ok
+ALTER SCHEMA sc RENAME TO sc1;
+
+query error pq: type "sc.t" does not exist
+SELECT 'HELLO'::sc.t;
+
+query T
+SELECT 'HELLO'::sc1.t;
+----
+HELLO
+
+statement ok
+DROP SCHEMA sc1 CASCADE;
+
+# Renaming the database should invalidate a database-qualified UDT reference.
+statement ok
+CREATE DATABASE d;
+USE d;
+CREATE TYPE d.t AS ENUM ('HELLO');
+
+query T
+SELECT 'HELLO'::d.t;
+----
+HELLO
+
+statement ok
+ALTER DATABASE d RENAME TO d1;
+USE d1;
+
+query error pq: type "d.t" does not exist
+SELECT 'HELLO'::d.t;
+
+query T
+SELECT 'HELLO'::d1.t;
+----
+HELLO
+
+statement ok
+USE test;
+DROP DATABASE d1 CASCADE;
+
+# Changing the current database should invalidate an unqualified UDT reference.
+statement ok
+CREATE TYPE t AS ENUM ('HELLO');
+
+query T
+SELECT 'HELLO'::t;
+----
+HELLO
+
+statement ok
+CREATE DATABASE d;
+USE d;
+
+query error pq: type "t" does not exist
+SELECT 'HELLO'::t;
+
+statement ok
+USE test;
+DROP DATABASE d;
+DROP TYPE t;

pkg/sql/opt/BUILD.bazel

@@ -24,6 +24,7 @@ go_library(
         "//pkg/server/telemetry",
         "//pkg/sql/catalog",
         "//pkg/sql/catalog/colinfo",
+        "//pkg/sql/catalog/typedesc",
         "//pkg/sql/opt/cat",
         "//pkg/sql/opt/partition",
         "//pkg/sql/pgwire/pgcode",

pkg/sql/opt/metadata.go

@@ -17,6 +17,7 @@ import (
 	"strings"
 
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/typedesc"
 	"github.com/cockroachdb/cockroach/pkg/sql/opt/cat"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
@@ -344,11 +345,23 @@ func (md *Metadata) CheckDependencies(
 		}
 	}
 
-	// Check that all of the user defined types present have not changed.
+	// Check that no referenced user defined types have changed.
 	for _, typ := range md.AllUserDefinedTypes() {
-		toCheck, err := optCatalog.ResolveTypeByOID(ctx, typ.Oid())
-		if err != nil || typ.TypeMeta.Version != toCheck.TypeMeta.Version {
-			return false, maybeSwallowMetadataResolveErr(err)
+		descID, _ := typedesc.UserDefinedTypeOIDToID(typ.Oid())
+		id := cat.StableID(descID)
+		if names, ok := md.objectRefsByName[id]; ok {
+			for _, name := range names {
+				toCheck, err := optCatalog.ResolveType(ctx, name)
+				if err != nil || typ.Oid() != toCheck.Oid() ||
+					typ.TypeMeta.Version != toCheck.TypeMeta.Version {
+					return false, maybeSwallowMetadataResolveErr(err)
+				}
+			}
+		} else {
+			toCheck, err := optCatalog.ResolveTypeByOID(ctx, typ.Oid())
+			if err != nil || typ.TypeMeta.Version != toCheck.TypeMeta.Version {
+				return false, maybeSwallowMetadataResolveErr(err)
+			}
 		}
 	}
 
@@ -414,7 +427,8 @@ func (md *Metadata) Schema(schID SchemaID) cat.Schema {
 }
 
 // AddUserDefinedType adds a user defined type to the metadata for this query.
-func (md *Metadata) AddUserDefinedType(typ *types.T) {
+// If the type was resolved by name, the name will be tracked as well.
+func (md *Metadata) AddUserDefinedType(typ *types.T, name *tree.UnresolvedObjectName) {
 	if !typ.UserDefined() {
 		return
 	}
@@ -425,6 +439,11 @@ func (md *Metadata) AddUserDefinedType(typ *types.T) {
 		md.userDefinedTypes[typ.Oid()] = struct{}{}
 		md.userDefinedTypesSlice = append(md.userDefinedTypesSlice, typ)
 	}
+	if name != nil {
+		descID, _ := typedesc.UserDefinedTypeOIDToID(typ.Oid())
+		id := cat.StableID(descID)
+		md.objectRefsByName[id] = append(md.objectRefsByName[id], name)
+	}
 }
 
 // AllUserDefinedTypes returns all user defined types contained in this query.

pkg/sql/opt/metadata_test.go

@@ -39,7 +39,7 @@ func TestMetadata(t *testing.T) {
 	tabID := md.AddTable(&testcat.Table{}, &tree.TableName{})
 	seqID := md.AddSequence(&testcat.Sequence{})
 	md.AddView(&testcat.View{})
-	md.AddUserDefinedType(types.MakeEnum(152100, 154180))
+	md.AddUserDefinedType(types.MakeEnum(152100, 154180), nil /* name */)
 
 	// Call Init and add objects from catalog, verifying that IDs have been reset.
 	testCat := testcat.New()
@@ -88,7 +88,7 @@ func TestMetadata(t *testing.T) {
 		t.Fatalf("unexpected views")
 	}
 
-	md.AddUserDefinedType(types.MakeEnum(151500, 152510))
+	md.AddUserDefinedType(types.MakeEnum(151500, 152510), nil /* name */)
 	if len(md.AllUserDefinedTypes()) != 1 {
 		fmt.Println(md)
 		t.Fatalf("unexpected types")

pkg/sql/opt/optbuilder/builder.go

@@ -474,7 +474,7 @@ func (o *optTrackingTypeResolver) ResolveType(
 	if err != nil {
 		return nil, err
 	}
-	o.metadata.AddUserDefinedType(typ)
+	o.metadata.AddUserDefinedType(typ, name)
 	return typ, nil
 }
 
@@ -486,6 +486,6 @@ func (o *optTrackingTypeResolver) ResolveTypeByOID(
 	if err != nil {
 		return nil, err
 	}
-	o.metadata.AddUserDefinedType(typ)
+	o.metadata.AddUserDefinedType(typ, nil /* name */)
 	return typ, nil
 }