opt: track distinct object references in metadata

Prior to this commit, all referenced object names in a query, including
tables, UDTs, and UDFs, were appended to slices in a memo's metadata. These
names are later re-resolved to determine if a memo is stale. Because
this list could contain duplicate entries, the same name could be
re-resolved multiple times during the staleness check.

Now, the distinct set of reference object names are maintained,
eliminating duplicate object resolution. The
`tree.UnresolvedObjectNameSet` type has been added to facilitate this.

Fixes #153800

Release note: None

Author: mgartner

pkg/sql/opt/metadata.go

@@ -132,7 +132,7 @@ type Metadata struct {
 	// objectRefsByName stores each unique name that the query uses to reference
 	// each object. It is needed because changes to the search path may change
 	// which object a given name refers to; for example, switching the database.
-	objectRefsByName map[cat.StableID][]*tree.UnresolvedObjectName
+	objectRefsByName map[cat.StableID]tree.UnresolvedObjectNameSet
 
 	// privileges stores the privileges needed to access each object that the
 	// query depends on.
@@ -204,7 +204,7 @@ func (md *Metadata) Init() {
 
 	objectRefsByName := md.objectRefsByName
 	if objectRefsByName == nil {
-		objectRefsByName = make(map[cat.StableID][]*tree.UnresolvedObjectName)
+		objectRefsByName = make(map[cat.StableID]tree.UnresolvedObjectNameSet)
 	}
 	for id := range md.objectRefsByName {
 		delete(md.objectRefsByName, id)
@@ -306,10 +306,12 @@ func (md *Metadata) CopyFrom(from *Metadata, copyScalarFn func(Expr) Expr) {
 
 	for id, names := range from.objectRefsByName {
 		if md.objectRefsByName == nil {
-			md.objectRefsByName = make(map[cat.StableID][]*tree.UnresolvedObjectName)
+			md.objectRefsByName = make(map[cat.StableID]tree.UnresolvedObjectNameSet)
+		}
+		newNames := tree.MakeUnresolvedObjectNameSet(names.Len())
+		for i, n := 0, names.Len(); i < n; i++ {
+			newNames.Add(names.Get(i))
 		}
-		newNames := make([]*tree.UnresolvedObjectName, len(names))
-		copy(newNames, names)
 		md.objectRefsByName[id] = newNames
 	}
 
@@ -371,7 +373,9 @@ func (md *Metadata) AddDependency(name MDDepName, ds cat.DataSource, priv privil
 	md.privileges[id] = md.privileges[id] | (1 << priv)
 	if name.byID == 0 {
 		// This data source was referenced by name.
-		md.objectRefsByName[id] = append(md.objectRefsByName[id], name.byName.ToUnresolvedObjectName())
+		names := md.objectRefsByName[id]
+		names.Add(name.byName.ToUnresolvedObjectName())
+		md.objectRefsByName[id] = names
 	}
 }
 
@@ -465,8 +469,8 @@ func (md *Metadata) CheckDependencies(
 		var toCheck cat.DataSource
 		if names, ok := md.objectRefsByName[id]; ok {
 			// The data source was referenced by name at least once.
-			for _, name := range names {
-				tableName := name.ToTableName()
+			for i, n := 0, names.Len(); i < n; i++ {
+				tableName := names.Get(i).ToTableName()
 				toCheck, _, err = optCatalog.ResolveDataSource(ctx, cat.Flags{}, &tableName)
 				if err != nil || !dataSource.Equals(toCheck) {
 					return false, maybeSwallowMetadataResolveErr(err)
@@ -485,8 +489,8 @@ func (md *Metadata) CheckDependencies(
 	for _, typ := range md.AllUserDefinedTypes() {
 		id := cat.StableID(catid.UserDefinedOIDToID(typ.Oid()))
 		if names, ok := md.objectRefsByName[id]; ok {
-			for _, name := range names {
-				toCheck, err := optCatalog.ResolveType(ctx, name)
+			for i, n := 0, names.Len(); i < n; i++ {
+				toCheck, err := optCatalog.ResolveType(ctx, names.Get(i))
 				if err != nil || typ.Oid() != toCheck.Oid() ||
 					typ.TypeMeta.Version != toCheck.TypeMeta.Version {
 					return false, maybeSwallowMetadataResolveErr(err)
@@ -505,7 +509,8 @@ func (md *Metadata) CheckDependencies(
 	for id, dep := range md.routineDeps {
 		overload := dep.overload
 		if names, ok := md.objectRefsByName[id]; ok {
-			for _, name := range names {
+			for i, n := 0, names.Len(); i < n; i++ {
+				name := names.Get(i)
 				definition, err := optCatalog.ResolveFunction(
 					ctx, tree.MakeUnresolvedFunctionName(name.ToUnresolvedName()),
 					&evalCtx.SessionData().SearchPath,
@@ -677,7 +682,9 @@ func (md *Metadata) AddUserDefinedType(typ *types.T, name *tree.UnresolvedObject
 	}
 	if name != nil {
 		id := cat.StableID(catid.UserDefinedOIDToID(typ.Oid()))
-		md.objectRefsByName[id] = append(md.objectRefsByName[id], name)
+		names := md.objectRefsByName[id]
+		names.Add(name)
+		md.objectRefsByName[id] = names
 	}
 }
 
@@ -707,7 +714,9 @@ func (md *Metadata) AddUserDefinedRoutine(
 		invocationTypes: invocationTypes,
 	}
 	if name != nil {
-		md.objectRefsByName[id] = append(md.objectRefsByName[id], name)
+		names := md.objectRefsByName[id]
+		names.Add(name)
+		md.objectRefsByName[id] = names
 	}
 }
 
@@ -1171,7 +1180,7 @@ func (md *Metadata) TestingRoutineDepsEqual(other *Metadata) bool {
 }
 
 // TestingObjectRefsByName exposes the objectRefsByName for testing.
-func (md *Metadata) TestingObjectRefsByName() map[cat.StableID][]*tree.UnresolvedObjectName {
+func (md *Metadata) TestingObjectRefsByName() map[cat.StableID]tree.UnresolvedObjectNameSet {
 	return md.objectRefsByName
 }
 

pkg/sql/opt/metadata_test.go

@@ -182,8 +182,8 @@ func TestMetadata(t *testing.T) {
 	newNamesByID, oldNamesByID := mdNew.TestingObjectRefsByName(), md.TestingObjectRefsByName()
 	for id, names := range oldNamesByID {
 		newNames := newNamesByID[id]
-		for i, name := range names {
-			if newNames[i] != name {
+		for i, n := 0, names.Len(); i < n; i++ {
+			if newNames.Get(i) != names.Get(i) {
 				t.Fatalf("expected object name to be copied")
 			}
 		}

pkg/sql/sem/tree/BUILD.bazel

@@ -222,6 +222,7 @@ go_test(
         "main_test.go",
         "name_part_test.go",
         "name_resolution_test.go",
+        "object_name_test.go",
         "operators_test.go",
         "overload_test.go",
         "parse_array_test.go",

pkg/sql/sem/tree/object_name.go

@@ -335,6 +335,61 @@ func (u *UnresolvedObjectName) HasExplicitCatalog() bool {
 	return u.NumParts >= 3
 }
 
+// equals returns true if the unresolved object name is equal to the given name,
+// that is, they have the same number of parts and all parts are equal.
+// Annotations are ignored.
+func (u *UnresolvedObjectName) equals(other *UnresolvedObjectName) bool {
+	if u.NumParts != other.NumParts {
+		return false
+	}
+	for i := 0; i < u.NumParts; i++ {
+		if u.Parts[i] != other.Parts[i] {
+			return false
+		}
+	}
+	return true
+}
+
+// UnresolvedObjectNameSet is a set of distinct unresolved object names. Two
+// unresolved object names are considered non-distinct if they have the same
+// number of parts and all parts are equal. Annotations are ignored.
+//
+// UnresolvedObjectNameSet is designed for small sets. Add is has O(n)
+// complexity where n is the number of names in the set and all the names in the
+// set must be examined, using Len and Get, to test for containment in the set.
+type UnresolvedObjectNameSet struct {
+	names []*UnresolvedObjectName
+}
+
+// MakeUnresolvedObjectNameSet creates an UnresolvedObjectNameSet with the
+// given initial capacity.
+func MakeUnresolvedObjectNameSet(cap int) UnresolvedObjectNameSet {
+	return UnresolvedObjectNameSet{
+		names: make([]*UnresolvedObjectName, 0, cap),
+	}
+}
+
+// Add adds the given name to the set. No-op if the name is already in the set.
+// The complexity is O(n) where n = u.Len().
+func (u *UnresolvedObjectNameSet) Add(name *UnresolvedObjectName) {
+	for _, n := range u.names {
+		if n.equals(name) {
+			return
+		}
+	}
+	u.names = append(u.names, name)
+}
+
+// Len returns the number of names in the set.
+func (u *UnresolvedObjectNameSet) Len() int {
+	return len(u.names)
+}
+
+// Get returns the i-th name in the set. Panics if i is out of bounds.
+func (u *UnresolvedObjectNameSet) Get(i int) *UnresolvedObjectName {
+	return u.names[i]
+}
+
 // UnresolvedRoutineName is an unresolved function or procedure name. The two
 // implementations of this interface are used to differentiate between the two
 // types of routines for things like error messages.

pkg/sql/sem/tree/object_name_test.go

@@ -0,0 +1,84 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package tree
+
+import (
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+)
+
+func TestUnresolvedObjectNameSet(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	name := func(parts ...string) *UnresolvedObjectName {
+		var partsArr [3]string
+		copy(partsArr[:], parts)
+		n, err := NewUnresolvedObjectName(len(parts), partsArr, NoAnnotation)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return n
+	}
+
+	testCases := []struct {
+		add         *UnresolvedObjectName
+		expectedLen int
+	}{
+		{name("foo"), 1},
+		{name("bar"), 2},
+		{name("foo", "bar"), 3},
+		{name("bar", "foo"), 4},
+		{name("foo", "bar", "baz"), 5},
+		{name("foo", "bar"), 5},
+		{name("bar"), 5},
+		{name("foo"), 5},
+		{name("baz"), 6},
+	}
+
+	var s UnresolvedObjectNameSet
+	if s.Len() != 0 {
+		t.Error("expected set to be empty")
+	}
+
+	// Add each test case to the set and check the length.
+	for _, tc := range testCases {
+		s.Add(tc.add)
+		if l := s.Len(); l != tc.expectedLen {
+			t.Errorf("after adding %v, expected length of %d, got %d", tc.add, tc.expectedLen, l)
+		}
+	}
+
+	// Every name should be in the set.
+	for _, tc := range testCases {
+		inSet := false
+		for i := 0; i < s.Len(); i++ {
+			if s.Get(i).equals(tc.add) {
+				inSet = true
+				break
+			}
+		}
+		if !inSet {
+			t.Errorf("expected %v to be in the set", tc.add)
+		}
+	}
+
+	// There should be no names in the set that were never added.
+	for i := 0; i < s.Len(); i++ {
+		added := false
+		for _, tc := range testCases {
+			if s.Get(i).equals(tc.add) {
+				added = true
+				break
+			}
+		}
+		if !added {
+			t.Errorf("%v was never added to the set", s.Get(i))
+		}
+	}
+}