sql/catalog: cache negative name lookups within a transaction

pkg/sql/catalog/descs/descriptor.go

@@ -714,6 +714,11 @@ func (tc *Collection) getNonVirtualDescriptorID(
 			}
 			return haltLookups, descpb.InvalidID, nil
 		}
+		// Short-circuit before the leased lookup, which would otherwise re-read
+		// system.namespace on every absent-name reference within the txn.
+		if tc.cr.IsNameKnownToNotExist(ni) {
+			return haltLookups, descpb.InvalidID, nil
+		}
 		return continueLookups, descpb.InvalidID, nil
 	}
 	lookupLeasedID := func() (continueOrHalt, descpb.ID, error) {

pkg/sql/catalog/internal/catkv/catalog_reader.go

@@ -47,6 +47,10 @@ type CatalogReader interface {
 	// exists no descriptor in storage with that ID.
 	IsDescIDKnownToNotExist(id, maybeParentID descpb.ID) bool
 
+	// IsNameKnownToNotExist returns true when we know that there definitely
+	// exists no namespace entry in storage for this name key.
+	IsNameKnownToNotExist(key descpb.NameInfo) bool
+
 	// Reset resets any state that the CatalogReader may hold.
 	Reset(ctx context.Context)
 
@@ -214,6 +218,11 @@ func (cr catalogReader) IsDescIDKnownToNotExist(_, _ descpb.ID) bool {
 	return false
 }
 
+// IsNameKnownToNotExist is part of the CatalogReader interface.
+func (cr catalogReader) IsNameKnownToNotExist(_ descpb.NameInfo) bool {
+	return false
+}
+
 // ScanAll is part of the CatalogReader interface.
 func (cr catalogReader) ScanAll(ctx context.Context, txn *kv.Txn) (nstree.Catalog, error) {
 	var mc nstree.MutableCatalog

pkg/sql/catalog/internal/catkv/catalog_reader_cached.go

@@ -127,6 +127,14 @@ func (c *cachedCatalogReader) IsNameInCache(key descpb.NameInfo) bool {
 	return c.cache.LookupNamespaceEntry(key) != nil
 }
 
+// IsNameKnownToNotExist is part of the CatalogReader interface.
+func (c *cachedCatalogReader) IsNameKnownToNotExist(key descpb.NameInfo) bool {
+	if c.cache.LookupNamespaceEntry(key) != nil {
+		return false
+	}
+	return c.hasScanAll || c.byNameState[key].hasGetNamespaceEntries
+}
+
 // IsDescIDKnownToNotExist is part of the CatalogReader interface.
 func (c *cachedCatalogReader) IsDescIDKnownToNotExist(id, maybeParentID descpb.ID) bool {
 	if c.cache.LookupDescriptor(id) != nil {

pkg/sql/catalog/internal/catkv/catalog_reader_test.go

@@ -125,6 +125,17 @@ func TestDataDriven(t *testing.T) {
 					}
 					return fmt.Sprintf("%v", ccr.IsDescIDKnownToNotExist(descpb.ID(id), descpb.ID(maybeParentID)))
 
+				case "is_name_known_to_not_exist":
+					var name string
+					var dbID, scID int
+					d.ScanArgs(t, "name_key", &dbID, &scID, &name)
+					ni := descpb.NameInfo{
+						ParentID:       descpb.ID(dbID),
+						ParentSchemaID: descpb.ID(scID),
+						Name:           name,
+					}
+					return fmt.Sprintf("%v", ccr.IsNameKnownToNotExist(ni))
+
 				case "scan_all":
 					q := func(ctx context.Context, txn *kv.Txn, cr catkv.CatalogReader) (nstree.Catalog, error) {
 						return cr.ScanAll(ctx, txn)

pkg/sql/catalog/internal/catkv/testdata/testdata_app

@@ -599,3 +599,38 @@ catalog: {}
 error: 'failed to verify keys for Scan: end key /Tenant/10/Table/3/1/111/2/1 must be greater than start /Tenant/10/Table/3/1/112/2/1'
 trace:
 - Scan Range /Table/3/1/112/2/1 /Table/3/1/111/2/1
+
+# Reset before exercising the per-txn negative name cache.
+reset
+----
+
+# Looking up a name that doesn't exist marks it as known-to-not-exist for
+# the remainder of this txn, so subsequent layers can short-circuit before
+# hitting storage.
+get_by_names name_key=(100,101,does_not_exist)
+----
+catalog: {}
+trace:
+- Get /NamespaceTable/30/1/100/101/"does_not_exist"/4/1
+
+is_name_known_to_not_exist name_key=(100,101,does_not_exist)
+----
+true
+
+# A different absent name we haven't looked up yet is still unknown.
+is_name_known_to_not_exist name_key=(100,101,also_missing)
+----
+false
+
+# A positive cache entry is not a negative one.
+is_name_known_to_not_exist name_key=(0,0,system)
+----
+false
+
+# Resetting wipes the per-txn negative cache.
+reset
+----
+
+is_name_known_to_not_exist name_key=(100,101,does_not_exist)
+----
+false

pkg/sql/catalog/internal/catkv/testdata/testdata_system

@@ -609,3 +609,38 @@ catalog: {}
 error: 'failed to verify keys for Scan: end key /Table/3/1/111/2/1 must be greater than start /Table/3/1/112/2/1'
 trace:
 - Scan Range /Table/3/1/112/2/1 /Table/3/1/111/2/1
+
+# Reset before exercising the per-txn negative name cache.
+reset
+----
+
+# Looking up a name that doesn't exist marks it as known-to-not-exist for
+# the remainder of this txn, so subsequent layers can short-circuit before
+# hitting storage.
+get_by_names name_key=(100,101,does_not_exist)
+----
+catalog: {}
+trace:
+- Get /NamespaceTable/30/1/100/101/"does_not_exist"/4/1
+
+is_name_known_to_not_exist name_key=(100,101,does_not_exist)
+----
+true
+
+# A different absent name we haven't looked up yet is still unknown.
+is_name_known_to_not_exist name_key=(100,101,also_missing)
+----
+false
+
+# A positive cache entry is not a negative one.
+is_name_known_to_not_exist name_key=(0,0,system)
+----
+false
+
+# Resetting wipes the per-txn negative cache.
+reset
+----
+
+is_name_known_to_not_exist name_key=(100,101,does_not_exist)
+----
+false