catalog,descpb: deprecate draining_names fields and methods

This refactoring commit is the first step to removing these fields and
their associated methods.

See issue cockroachdb#54562.

Release note: None

pkg/ccl/backupccl/restore_planning.go

@@ -1026,15 +1026,16 @@ func rewriteDatabaseDescs(databases []*dbdesc.Mutable, descriptorRewrites DescRe
 
 		// Rewrite the name-to-ID mapping for the database's child schemas.
 		newSchemas := make(map[string]descpb.DatabaseDescriptor_SchemaInfo)
-		for schemaName, schemaInfo := range db.Schemas {
-			if schemaInfo.Dropped {
-				continue
-			}
-			rewrite, ok := descriptorRewrites[schemaInfo.ID]
+		err := db.ForEachNonDroppedSchema(func(id descpb.ID, name string) error {
+			rewrite, ok := descriptorRewrites[id]
 			if !ok {
 				return errors.Errorf("missing rewrite for schema %d", db.ID)
 			}
-			newSchemas[schemaName] = descpb.DatabaseDescriptor_SchemaInfo{ID: rewrite.ID}
+			newSchemas[name] = descpb.DatabaseDescriptor_SchemaInfo{ID: rewrite.ID}
+			return nil
+		})
+		if err != nil {
+			return err
 		}
 		db.Schemas = newSchemas
 	}

pkg/ccl/importccl/import_job.go

@@ -601,7 +601,7 @@ func (r *importResumer) prepareSchemasForIngestion(
 
 		// Update the parent database with this schema information.
 		dbDesc.Schemas[newMutableSchemaDescriptor.Name] =
-			descpb.DatabaseDescriptor_SchemaInfo{ID: newMutableSchemaDescriptor.ID, Dropped: false}
+			descpb.DatabaseDescriptor_SchemaInfo{ID: newMutableSchemaDescriptor.ID}
 
 		schemaMetadata.schemaRewrites[desc.Desc.ID] = &jobspb.RestoreDetails_DescriptorRewrite{
 			ID: id,
@@ -1485,9 +1485,12 @@ func (r *importResumer) dropSchemas(
 			return nil, errors.Newf("unable to resolve schema desc with ID %d", schema.Desc.ID)
 		}
 
-		schemaDesc.DrainingNames = append(schemaDesc.DrainingNames,
-			descpb.NameInfo{ParentID: details.ParentID, ParentSchemaID: keys.RootNamespaceID,
-				Name: schemaDesc.Name})
+		//lint:ignore SA1019 deprecated method call is OK
+		schemaDesc.AddDrainingName(descpb.NameInfo{
+			ParentID:       details.ParentID,
+			ParentSchemaID: keys.RootNamespaceID,
+			Name:           schemaDesc.Name,
+		})
 
 		// Update the parent database with information about the dropped schema.
 		if dbDesc.Schemas == nil {

pkg/sql/alter_schema.go

@@ -14,6 +14,7 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/security"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catalogkeys"
@@ -209,6 +210,11 @@ func (p *planner) renameSchema(
 
 	// Set the new name for the descriptor.
 	oldName := desc.Name
+	desc.AddDrainingName(descpb.NameInfo{
+		ParentID:       desc.ParentID,
+		ParentSchemaID: keys.RootNamespaceID,
+		Name:           desc.Name,
+	})
 	desc.SetName(newName)
 
 	// Write a new namespace entry for the new name.
@@ -247,10 +253,7 @@ func (p *planner) renameSchema(
 		Dropped: true,
 	}
 	// Create an entry for the new schema name.
-	db.Schemas[newName] = descpb.DatabaseDescriptor_SchemaInfo{
-		ID:      desc.ID,
-		Dropped: false,
-	}
+	db.Schemas[newName] = descpb.DatabaseDescriptor_SchemaInfo{ID: desc.ID}
 	if err := p.writeNonDropDatabaseChange(
 		ctx, db,
 		fmt.Sprintf("updating parent database %s for %s", db.GetName(), jobDesc),

pkg/sql/backfill.go

@@ -1988,16 +1988,20 @@ func (sc *SchemaChanger) truncateAndBackfillColumns(
 func runSchemaChangesInTxn(
 	ctx context.Context, planner *planner, tableDesc *tabledesc.Mutable, traceKV bool,
 ) error {
-	if len(tableDesc.DrainingNames) > 0 {
+	// TODO(postamar): remove if-block in 22.2
+	//lint:ignore SA1019 removal of deprecated method call scheduled for 22.2
+	if len(tableDesc.GetDrainingNames()) > 0 {
 		// Reclaim all the old names. Leave the data and descriptor
 		// cleanup for later.
-		for _, drain := range tableDesc.DrainingNames {
+		//lint:ignore SA1019 removal of deprecated method call scheduled for 22.2
+		for _, drain := range tableDesc.GetDrainingNames() {
 			key := catalogkeys.EncodeNameKey(planner.ExecCfg().Codec, drain)
 			if err := planner.Txn().Del(ctx, key); err != nil {
 				return err
 			}
 		}
-		tableDesc.DrainingNames = nil
+		//lint:ignore SA1019 removal of deprecated method call scheduled for 22.2
+		tableDesc.SetDrainingNames(nil)
 	}
 
 	if tableDesc.Dropped() {

pkg/sql/catalog/catalog.go

@@ -26,8 +26,15 @@ type MutableDescriptor interface {
 	// descriptor should increment the version on the mutable copy from the
 	// outset.
 	MaybeIncrementVersion()
+
 	// SetDrainingNames sets the draining names for the descriptor.
-	SetDrainingNames([]descpb.NameInfo)
+	//
+	// TODO(postamar): remove SetDrainingNames method in 22.2
+	SetDrainingNames([]descpb.NameInfo) // Deprecated
+	// AddDrainingName adds a draining name to the descriptor.
+	//
+	// TODO(postamar): remove AddDrainingName method in 22.2
+	AddDrainingName(descpb.NameInfo) // Deprecated
 
 	// Accessors for the original state of the descriptor prior to the mutations.
 	OriginalName() string

pkg/sql/catalog/dbdesc/database_desc.go

@@ -81,6 +81,8 @@ func (desc *immutable) DatabaseDesc() *descpb.DatabaseDescriptor {
 }
 
 // SetDrainingNames implements the MutableDescriptor interface.
+//
+// Deprecated: Do not use.
 func (desc *Mutable) SetDrainingNames(names []descpb.NameInfo) {
 	desc.DrainingNames = names
 }
@@ -183,6 +185,22 @@ func (desc *immutable) ForEachSchemaInfo(
 	return nil
 }
 
+// ForEachNonDroppedSchema implements the DatabaseDescriptor interface.
+func (desc *immutable) ForEachNonDroppedSchema(f func(id descpb.ID, name string) error) error {
+	for name, info := range desc.Schemas {
+		if info.Dropped {
+			continue
+		}
+		if err := f(info.ID, name); err != nil {
+			if iterutil.Done(err) {
+				return nil
+			}
+			return err
+		}
+	}
+	return nil
+}
+
 // GetSchemaID implements the DatabaseDescriptor interface.
 func (desc *immutable) GetSchemaID(name string) descpb.ID {
 	info := desc.Schemas[name]
@@ -371,6 +389,8 @@ func (desc *Mutable) SetOffline(reason string) {
 
 // AddDrainingName adds a draining name to the DatabaseDescriptor's slice of
 // draining names.
+//
+// Deprecated: Do not use.
 func (desc *Mutable) AddDrainingName(name descpb.NameInfo) {
 	desc.DrainingNames = append(desc.DrainingNames, name)
 }

pkg/sql/catalog/dbdesc/database_test.go

@@ -143,7 +143,7 @@ func TestValidateCrossDatabaseReferences(t *testing.T) {
 				ID:   51,
 				Name: "db1",
 				Schemas: map[string]descpb.DatabaseDescriptor_SchemaInfo{
-					"schema1": {ID: 52, Dropped: false},
+					"schema1": {ID: 52},
 				},
 			},
 			schemaDescs: []descpb.SchemaDescriptor{
@@ -169,7 +169,7 @@ func TestValidateCrossDatabaseReferences(t *testing.T) {
 				ID:   51,
 				Name: "db1",
 				Schemas: map[string]descpb.DatabaseDescriptor_SchemaInfo{
-					"schema1": {ID: 500, Dropped: false},
+					"schema1": {ID: 500},
 				},
 			},
 		},
@@ -179,7 +179,7 @@ func TestValidateCrossDatabaseReferences(t *testing.T) {
 				ID:   51,
 				Name: "db1",
 				Schemas: map[string]descpb.DatabaseDescriptor_SchemaInfo{
-					"schema1": {ID: 52, Dropped: false},
+					"schema1": {ID: 52},
 				},
 			},
 			schemaDescs: []descpb.SchemaDescriptor{