sql: handle dropped schemas in crdb_internal.table_spans

Previously, crdb_internal.table_spans returned spans for all tables not
yet GC'd, including those from dropped tables. This could cause a lookup
failure when trying to resolve the schema descriptor of a dropped schema
during virtual table row construction.

This change updates the forEachTableDescFromDescriptors helper to
tolerate missing schema descriptors if the table is dropped. When the
includeDropped flag is set, such tables now return a nil schema
descriptor instead of causing an error.

Fixes #147717

Epic: none
Release note (bug fix): Fixed an error in crdb_internal.table_spans when
a table's schema had been dropped.

Author: spilchen

pkg/sql/information_schema.go

@@ -34,6 +34,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/semenumpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlerrors"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/sql/vtable"
 	"github.com/cockroachdb/cockroach/pkg/util/collatedstring"
@@ -2671,10 +2672,6 @@ func forEachTypeDesc(
 		if err != nil {
 			continue
 		}
-		sc, err := lCtx.getSchemaByID(typ.GetParentSchemaID())
-		if err != nil {
-			return err
-		}
 		canSeeDescriptor, err := userCanSeeDescriptor(
 			ctx, p, typ, dbDesc, false /* allowAdding */, false /* includeDropped */)
 		if err != nil {
@@ -2683,6 +2680,10 @@ func forEachTypeDesc(
 		if !canSeeDescriptor {
 			continue
 		}
+		sc, err := lCtx.getSchemaByID(typ.GetParentSchemaID())
+		if err != nil {
+			return err
+		}
 		if err := fn(ctx, dbDesc, sc, typ); err != nil {
 			return err
 		}
@@ -2809,9 +2810,16 @@ func forEachTableDescFromDescriptors(
 		}
 		var sc catalog.SchemaDescriptor
 		if parentExists {
-			sc, err = lCtx.getSchemaByID(table.GetParentSchemaID())
-			if err != nil && !table.IsTemporary() {
-				return err
+			// The schema may not exist if the table is temporary or belongs to a
+			// dropped schema. If the schema is dropped and we're configured to
+			// tolerate that (includeDropped is true), then the schema descriptor (sc)
+			// will intentionally remain nil.
+			schemaID := table.GetParentSchemaID()
+			if lCtx.hasSchemaWithID(schemaID) {
+				sc, err = lCtx.getSchemaByID(schemaID)
+				if err != nil {
+					return err
+				}
 			} else if table.IsTemporary() {
 				// Look up the schemas for this database if we discover that there is a
 				// missing temporary schema name. Temporary schemas have namespace
@@ -2836,6 +2844,8 @@ func forEachTableDescFromDescriptors(
 				if sc == nil {
 					sc = schemadesc.NewTemporarySchema(catconstants.PgTempSchemaName, table.GetParentSchemaID(), dbDesc.GetID())
 				}
+			} else if !opts.includeDropped {
+				return sqlerrors.NewUndefinedSchemaError(fmt.Sprintf("[%d]", schemaID))
 			}
 		}
 		if err := fn(ctx, tableDescContext{dbDesc, sc, table, lCtx}); err != nil {

pkg/sql/logictest/testdata/logic_test/crdb_internal

@@ -1675,7 +1675,10 @@ statement ok
 CREATE TABLE foo (a INT PRIMARY KEY, INDEX idx(a)); INSERT INTO foo VALUES(1);
 
 statement ok
-CREATE TABLE bar (a INT PRIMARY KEY, INDEX idx(a));
+CREATE SCHEMA droptest;
+
+statement ok
+CREATE TABLE droptest.bar (a INT PRIMARY KEY, INDEX idx(a));
 
 query TB rowsort
 SELECT name, dropped  
@@ -1686,7 +1689,7 @@ foo    false
 bar    false
 
 statement ok
-DROP TABLE bar
+DROP SCHEMA droptest CASCADE
 
 query TB rowsort
 SELECT name, dropped  
@@ -1744,13 +1747,13 @@ CREATE TYPE other_db.public.enum1 AS ENUM ('yo');
 query ITTITTT
 SELECT * FROM "".crdb_internal.create_type_statements WHERE descriptor_name = 'enum1' and database_name = 'other_db'
 ----
-124  other_db  public  154  enum1  CREATE TYPE public.enum1 AS ENUM ('yo')  {yo}
+124  other_db  public  155  enum1  CREATE TYPE public.enum1 AS ENUM ('yo')  {yo}
 
 # This uses the virtual index. descriptor_id is an int in the vtable.
 query ITTITTT
 SELECT * FROM "".crdb_internal.create_type_statements WHERE descriptor_id = (('other_db.public.enum1'::regtype::int) - 100000)
 ----
-124  other_db  public  154  enum1  CREATE TYPE public.enum1 AS ENUM ('yo')  {yo}
+124  other_db  public  155  enum1  CREATE TYPE public.enum1 AS ENUM ('yo')  {yo}
 
 # Repeat above two queries but apply only for the current database. We do this by omitting the "".
 # This one uses the full table scan.

pkg/sql/resolver.go

@@ -969,6 +969,16 @@ func (l *internalLookupCtx) getTypeByID(id descpb.ID) (catalog.TypeDescriptor, e
 	return typ, nil
 }
 
+// hasSchemaWithID reports whether a schema with the given ID exists in the
+// lookup context.
+func (l *internalLookupCtx) hasSchemaWithID(id descpb.ID) bool {
+	if id == keys.SystemPublicSchemaID {
+		return true
+	}
+	_, ok := l.schemaDescs[id]
+	return ok
+}
+
 func (l *internalLookupCtx) getSchemaByID(id descpb.ID) (catalog.SchemaDescriptor, error) {
 	sc, ok := l.schemaDescs[id]
 	if !ok {