Merge #44783

44783: sql: avoid lookup on unsupported schema names r=solongordon a=otan

Refs: #44733

In `bbef8f9`, we removed the "bypass" where if schema names were not
PublicSchema names, we quickly exit. However, with custom schemas (well,
only temporary schemas for now), we can no longer do this. Since we only
support temporary schemas and public schema for now when doing these
lookups, this PR adds the bypass back in such that performance is
currently not degraded.

No release note as nothing here is publically available yet.

Release note: None

Co-authored-by: Oliver Tan <otan@cockroachlabs.com>

pkg/sql/table.go

@@ -158,6 +158,15 @@ type dbCacheSubscriber interface {
 	waitForCacheState(cond func(*databaseCache) bool)
 }
 
+// isSupportedSchemaName returns whether this schema name is supported.
+// TODO(sqlexec): this should be deleted when we use custom schemas.
+// However, this introduces an extra lookup for cases where `<database>.<table>`
+// is looked up.
+// See #44733.
+func isSupportedSchemaName(n tree.Name) bool {
+	return n == tree.PublicSchemaName || strings.HasPrefix(string(n), "pg_temp")
+}
+
 // getMutableTableDescriptor returns a mutable table descriptor.
 //
 // If flags.required is false, getMutableTableDescriptor() will gracefully
@@ -170,6 +179,10 @@ func (tc *TableCollection) getMutableTableDescriptor(
 		log.Infof(ctx, "reading mutable descriptor on table '%s'", tn)
 	}
 
+	if !isSupportedSchemaName(tn.SchemaName) {
+		return nil, nil
+	}
+
 	refuseFurtherLookup, dbID, err := tc.getUncommittedDatabaseID(tn.Catalog(), flags.Required)
 	if refuseFurtherLookup || err != nil {
 		return nil, err
@@ -230,6 +243,10 @@ func (tc *TableCollection) getTableVersion(
 		log.Infof(ctx, "planner acquiring lease on table '%s'", tn)
 	}
 
+	if !isSupportedSchemaName(tn.SchemaName) {
+		return nil, nil
+	}
+
 	refuseFurtherLookup, dbID, err := tc.getUncommittedDatabaseID(tn.Catalog(), flags.Required)
 	if refuseFurtherLookup || err != nil {
 		return nil, err

pkg/sql/table_test.go

@@ -15,12 +15,33 @@ import (
 	"reflect"
 	"testing"
 
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/stretchr/testify/assert"
 )
 
+func TestIsSupportedSchemaName(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	testCases := []struct {
+		name  string
+		valid bool
+	}{
+		{"db_name", false},
+		{"public", true},
+		{"pg_temp", true},
+		{"pg_temp_1234_1", true},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.valid, isSupportedSchemaName(tree.Name(tc.name)))
+		})
+	}
+}
+
 func TestMakeTableDescColumns(t *testing.T) {
 	defer leaktest.AfterTest(t)()