sql: correctly identify and cleanup dependent objects in drop database

There were a few things going on here because of the way we filtered
tables during `DROP DATABASE CASCADE`. The intention was to filter
out objects that depended on other objects also being deleted (and would
therefore be deleted by the CASCADE nature of object drops). This
assumption is correct for table-view and view-view dependencies -- but
not for sequences. We also switched from individual schema change jobs
during a database drop to a single job which doesn't play well with
this filtering -- as no new jobs are queued, all objects that were
filtered would leave orphaned namespace/descriptor entries behind.
It's interesting to note that the filtering didn't directly cause this
issue, it just made the underlying issue more visible -- the single drop
database schema change job relies on knowing about all object
descriptors that need to be dropped upfront. Orphaned entries which
would have only occured for cross-database dependencies can occur in
the same database because of the filtering. This leads to why the fix
is what it is.

As part of this patch, I've tried to make the preperation steps for
dropping databases more explicit. First, we accumalate all objects
that need to be deleted. This includes objects that will be implicitly
deleted, such as owned sequences, dependent views (both in the database
being deleted and other databases that aren't being deleted). The
schema change job uses this new list to ensure entries are appropriately
cleaned up. We still perform the filter step as before to identify
objects which are the "root" of a drop and only call drop on these
objects. The only change here is that instead of accumulating dependent
objects, we explicitly accumulate cascading views.

Fixes #51782
Fixes #50997

Release note (bug fix): Before this change, we would leave orphaned
system.namespace/system.descriptor entries if we ran a
`DROP DATABASE CASCADE` and the database contained "dependency"
relations. For example, if the database included a view which
depended on a table in the database, dropping the database would result
in an orphaned entry for the view. Same thing for a sequence that was
used by a table in the database. (See #51782 for repro steps). This bug
is now fixed, and cleanup happens as expected.

pkg/sql/drop_database.go

@@ -32,10 +32,11 @@ import (
 )
 
 type dropDatabaseNode struct {
-	n               *tree.DropDatabase
-	dbDesc          *sqlbase.ImmutableDatabaseDescriptor
-	td              []toDelete
-	schemasToDelete []string
+	n                  *tree.DropDatabase
+	dbDesc             *sqlbase.ImmutableDatabaseDescriptor
+	td                 []toDelete
+	schemasToDelete    []string
+	allObjectsToDelete []*sqlbase.MutableTableDescriptor
 }
 
 // DropDatabase drops a database.
@@ -149,12 +150,17 @@ func (p *planner) DropDatabase(ctx context.Context, n *tree.DropDatabase) (planN
 		td = append(td, toDelete{&tbNames[i], tbDesc})
 	}
 
-	td, err = p.filterCascadedTables(ctx, td)
+	allObjectsToDelete, implicitDeleteMap, err := p.accumulateAllObjectsToDelete(ctx, td)
 	if err != nil {
 		return nil, err
 	}
 
-	return &dropDatabaseNode{n: n, dbDesc: dbDesc, td: td, schemasToDelete: schemasToDelete}, nil
+	return &dropDatabaseNode{
+		n:                  n,
+		dbDesc:             dbDesc,
+		td:                 filterImplicitlyDeletedObjects(td, implicitDeleteMap),
+		schemasToDelete:    schemasToDelete,
+		allObjectsToDelete: allObjectsToDelete}, nil
 }
 
 func (n *dropDatabaseNode) startExec(params runParams) error {
@@ -164,14 +170,12 @@ func (n *dropDatabaseNode) startExec(params runParams) error {
 	p := params.p
 	tbNameStrings := make([]string, 0, len(n.td))
 	droppedTableDetails := make([]jobspb.DroppedTableDetails, 0, len(n.td))
-	tableDescs := make([]*sqlbase.MutableTableDescriptor, 0, len(n.td))
 
-	for _, toDel := range n.td {
+	for _, delDesc := range n.allObjectsToDelete {
 		droppedTableDetails = append(droppedTableDetails, jobspb.DroppedTableDetails{
-			Name: toDel.tn.FQString(),
-			ID:   toDel.desc.ID,
+			Name: delDesc.Name,
+			ID:   delDesc.ID,
 		})
-		tableDescs = append(tableDescs, toDel.desc)
 	}
 	if err := p.createDropDatabaseJob(
 		ctx, n.dbDesc.GetID(), droppedTableDetails, tree.AsStringWithFQNames(n.n, params.Ann()),
@@ -230,7 +234,7 @@ func (n *dropDatabaseNode) startExec(params runParams) error {
 
 	// No job was created because no tables were dropped, so zone config can be
 	// immediately removed, if applicable.
-	if len(tableDescs) == 0 && params.ExecCfg().Codec.ForSystemTenant() {
+	if len(n.allObjectsToDelete) == 0 && params.ExecCfg().Codec.ForSystemTenant() {
 		zoneKeyPrefix := config.MakeZoneKeyPrefix(config.SystemTenantObjectID(n.dbDesc.GetID()))
 		if p.ExtendedEvalContext().Tracing.KVTracingEnabled() {
 			log.VEventf(ctx, 2, "DelRange %s", zoneKeyPrefix)
@@ -270,39 +274,102 @@ func (*dropDatabaseNode) Next(runParams) (bool, error) { return false, nil }
 func (*dropDatabaseNode) Close(context.Context)        {}
 func (*dropDatabaseNode) Values() tree.Datums          { return tree.Datums{} }
 
-// filterCascadedTables takes a list of table descriptors and removes any
-// descriptors from the list that are dependent on other descriptors in the
-// list (e.g. if view v1 depends on table t1, then v1 will be filtered from
-// the list).
-func (p *planner) filterCascadedTables(ctx context.Context, tables []toDelete) ([]toDelete, error) {
-	// Accumulate the set of all tables/views that will be deleted by cascade
-	// behavior so that we can filter them out of the list.
-	cascadedTables := make(map[sqlbase.ID]bool)
+// filterImplicitlyDeletedObjects takes a list of table descriptors and removes
+// any descriptor that will be implicitly deleted.
+func filterImplicitlyDeletedObjects(
+	tables []toDelete, implicitDeleteObjects map[sqlbase.ID]*MutableTableDescriptor,
+) []toDelete {
+	filteredDeleteList := make([]toDelete, 0, len(tables))
 	for _, toDel := range tables {
-		desc := toDel.desc
-		if err := p.accumulateDependentTables(ctx, cascadedTables, desc); err != nil {
-			return nil, err
+		if _, found := implicitDeleteObjects[toDel.desc.ID]; !found {
+			filteredDeleteList = append(filteredDeleteList, toDel)
 		}
 	}
-	filteredTableList := make([]toDelete, 0, len(tables))
-	for _, toDel := range tables {
-		if !cascadedTables[toDel.desc.ID] {
-			filteredTableList = append(filteredTableList, toDel)
+	return filteredDeleteList
+}
+
+// accumulateAllObjectsToDelete constructs a list of all the descriptors that
+// will be deleted as a side effect of deleting the given objects. Additional
+// objects may be deleted because of cascading views or sequence ownership. We
+// also return a map of objects that will be "implicitly" deleted so we can
+// filter on it later.
+func (p *planner) accumulateAllObjectsToDelete(
+	ctx context.Context, objects []toDelete,
+) ([]*MutableTableDescriptor, map[sqlbase.ID]*MutableTableDescriptor, error) {
+	implicitDeleteObjects := make(map[sqlbase.ID]*MutableTableDescriptor)
+	for _, toDel := range objects {
+		err := p.accumulateCascadingViews(ctx, implicitDeleteObjects, toDel.desc)
+		if err != nil {
+			return nil, nil, err
+		}
+		// Sequences owned by the table will also be implicitly deleted.
+		if toDel.desc.IsTable() {
+			err := p.accumulateOwnedSequences(ctx, implicitDeleteObjects, toDel.desc)
+			if err != nil {
+				return nil, nil, err
+			}
 		}
 	}
-	return filteredTableList, nil
+	allObjectsToDelete := make([]*MutableTableDescriptor, 0,
+		len(objects)+len(implicitDeleteObjects))
+	for _, desc := range implicitDeleteObjects {
+		allObjectsToDelete = append(allObjectsToDelete, desc)
+	}
+	for _, toDel := range objects {
+		if _, found := implicitDeleteObjects[toDel.desc.ID]; !found {
+			allObjectsToDelete = append(allObjectsToDelete, toDel.desc)
+		}
+	}
+	return allObjectsToDelete, implicitDeleteObjects, nil
+}
+
+// accumulateOwnedSequences finds all sequences that will be dropped as a result
+// of the table referenced by desc being dropped, and adds them to the
+// dependentObjects map.
+func (p *planner) accumulateOwnedSequences(
+	ctx context.Context,
+	dependentObjects map[sqlbase.ID]*MutableTableDescriptor,
+	desc *sqlbase.MutableTableDescriptor,
+) error {
+	for colID := range desc.GetColumns() {
+		for _, seqID := range desc.GetColumns()[colID].OwnsSequenceIds {
+			ownedSeqDesc, err := p.Descriptors().GetMutableTableVersionByID(ctx, seqID, p.txn)
+			if err != nil {
+				// Special case error swallowing for #50711 and #50781, which can
+				// cause columns to own sequences that have been dropped/do not
+				// exist.
+				if errors.Is(err, sqlbase.ErrDescriptorNotFound) {
+					log.Infof(ctx,
+						"swallowing error for owned sequence that was not found %s", err.Error())
+					continue
+				}
+				return err
+			}
+			dependentObjects[seqID] = ownedSeqDesc
+		}
+	}
+	return nil
 }
 
-func (p *planner) accumulateDependentTables(
-	ctx context.Context, dependentTables map[sqlbase.ID]bool, desc *sqlbase.MutableTableDescriptor,
+// accumulateCascadingViews finds all views that are to be deleted as part
+// of a drop database cascade. This is important as CRDB allows cross-database
+// references, which means this list can't be constructed by simply scanning
+// the namespace table.
+func (p *planner) accumulateCascadingViews(
+	ctx context.Context,
+	dependentObjects map[sqlbase.ID]*MutableTableDescriptor,
+	desc *sqlbase.MutableTableDescriptor,
 ) error {
 	for _, ref := range desc.DependedOnBy {
-		dependentTables[ref.ID] = true
 		dependentDesc, err := p.Descriptors().GetMutableTableVersionByID(ctx, ref.ID, p.txn)
 		if err != nil {
 			return err
 		}
-		if err := p.accumulateDependentTables(ctx, dependentTables, dependentDesc); err != nil {
+		if !dependentDesc.IsView() {
+			continue
+		}
+		dependentObjects[ref.ID] = dependentDesc
+		if err := p.accumulateCascadingViews(ctx, dependentObjects, dependentDesc); err != nil {
 			return err
 		}
 	}

pkg/sql/logictest/testdata/logic_test/drop_database

@@ -259,3 +259,81 @@ query TT
 SELECT job_type, status FROM [SHOW JOBS] WHERE description LIKE '%empty%'
 ----
 SCHEMA CHANGE       succeeded
+
+subtest bug_50997
+
+statement ok
+CREATE DATABASE db50997
+
+statement ok
+CREATE SEQUENCE db50997.seq50997
+
+statement ok
+CREATE SEQUENCE db50997.useq50997
+
+statement ok
+CREATE TABLE db50997.t50997(a INT DEFAULT nextval('db50997.seq50997'))
+
+statement ok
+CREATE TABLE db50997.t250997(a INT DEFAULT nextval('db50997.seq50997'))
+
+statement ok
+DROP DATABASE db50997 CASCADE
+
+query I
+SELECT count(*) FROM system.namespace WHERE name = 'seq50997'
+----
+0
+
+query I
+SELECT count(*) FROM system.namespace WHERE name = 'useq50997'
+----
+0
+
+# Similar to the test above, except now the "sequence use" relationship
+# is cross database. In this case the sequence should be simply unlinked
+# (and not dropped). This wasn't broken before, but it is nice to have a test
+# that checks things behave as expected.
+statement ok
+CREATE DATABASE db50997
+
+statement ok
+CREATE SEQUENCE seq50997
+
+statement ok
+CREATE TABLE db50997.t50997(a INT DEFAULT nextval('seq50997'))
+
+statement ok
+DROP DATABASE db50997 CASCADE
+
+query I
+SELECT count(*) FROM system.namespace WHERE name LIKE 'seq50997'
+----
+1
+
+subtest regression_51782
+
+statement ok
+CREATE DATABASE db_51782
+
+statement ok
+CREATE TABLE db_51782.t_51782(a int, b int);
+
+statement ok
+CREATE VIEW db_51782.v_51782 AS SELECT a, b FROM db_51782.t_51782
+
+statement ok
+CREATE VIEW db_51782.w_51782 AS SELECT a FROM db_51782.v_51782
+
+statement ok
+DROP DATABASE db_51782 CASCADE
+
+query I
+SELECT count(*) FROM system.namespace WHERE name LIKE 'v_51782'
+----
+0
+
+query I
+SELECT count(*) FROM system.namespace WHERE name LIKE 'w_51782'
+----
+0

pkg/sql/logictest/testdata/logic_test/sequences

@@ -1161,8 +1161,10 @@ CREATE SEQUENCE seq_50712 OWNED BY db_50712.t_50712.a
 statement ok
 DROP DATABASE db_50712 CASCADE
 
-statement error pq: relation "seq_50712" does not exist
-SELECT * FROM seq_50712
+query I
+SELECT count(*) FROM system.namespace WHERE name LIKE 'seq_50712'
+----
+0
 
 # Tests db drop.
 # Sequence: inside db

pkg/sql/sequence.go

@@ -530,6 +530,17 @@ func (p *planner) removeSequenceDependencies(
 		if err != nil {
 			return err
 		}
+		// If the sequence descriptor has been dropped, we do not need to unlink the
+		// dependency. This can happen during a `DROP DATABASE CASCADE` when both
+		// the table and sequence are objects in the database being dropped. If
+		// `dropImpl` is called on the sequence before the table, because CRDB
+		// doesn't implement CASCADE for sequences, the dependency to the
+		// table descriptor is not unlinked. This check prevents us from failing
+		// when trying to unlink a dependency that really shouldn't have existed
+		// at this point in the code to begin with.
+		if seqDesc.Dropped() {
+			continue
+		}
 		// Find the item in seqDesc.DependedOnBy which references tableDesc and col.
 		refTableIdx := -1
 		refColIdx := -1