sql: use leased database descriptors, implement database schema changes

pkg/ccl/backupccl/restore_job.go

@@ -274,7 +274,7 @@ rangeLoop:
 func WriteDescriptors(
 	ctx context.Context,
 	txn *kv.Txn,
-	databases []*sqlbase.ImmutableDatabaseDescriptor,
+	databases []sqlbase.DatabaseDescriptor,
 	tables []sqlbase.TableDescriptor,
 	types []sqlbase.TypeDescriptor,
 	descCoverage tree.DescriptorCoverage,
@@ -285,13 +285,19 @@ func WriteDescriptors(
 	defer tracing.FinishSpan(span)
 	err := func() error {
 		b := txn.NewBatch()
-		wroteDBs := make(map[descpb.ID]*sqlbase.ImmutableDatabaseDescriptor)
-		for _, desc := range databases {
+		wroteDBs := make(map[descpb.ID]sqlbase.DatabaseDescriptor)
+		for i := range databases {
+			desc := databases[i]
 			// If the restore is not a full cluster restore we cannot know that
 			// the users on the restoring cluster match the ones that were on the
 			// cluster that was backed up. So we wipe the privileges on the database.
 			if descCoverage != tree.AllDescriptors {
-				desc.Privileges = descpb.NewDefaultPrivilegeDescriptor(security.AdminRole)
+				if mut, ok := desc.(*sqlbase.MutableDatabaseDescriptor); ok {
+					mut.Privileges = descpb.NewDefaultPrivilegeDescriptor(security.AdminRole)
+				} else {
+					log.Fatalf(ctx, "wrong type for table %d, %T, expected MutableTableDescriptor",
+						desc.GetID(), desc)
+				}
 			}
 			wroteDBs[desc.GetID()] = desc
 			if err := catalogkv.WriteNewDescToBatch(ctx, false /* kvTrace */, settings, b, keys.SystemSQLCodec, desc.GetID(), desc); err != nil {
@@ -653,7 +659,7 @@ func loadBackupSQLDescs(
 type restoreResumer struct {
 	job       *jobs.Job
 	settings  *cluster.Settings
-	databases []*sqlbase.ImmutableDatabaseDescriptor
+	databases []sqlbase.DatabaseDescriptor
 	tables    []sqlbase.TableDescriptor
 	// writtenTypes is the set of types that are restored from the backup into
 	// the database. Note that this is not always the set of types within the
@@ -736,7 +742,7 @@ func remapRelevantStatistics(
 func isDatabaseEmpty(
 	ctx context.Context,
 	db *kv.DB,
-	dbDesc *sqlbase.ImmutableDatabaseDescriptor,
+	dbDesc sqlbase.DatabaseDescriptor,
 	ignoredTables map[descpb.ID]struct{},
 ) (bool, error) {
 	var allDescs []sqlbase.Descriptor
@@ -766,7 +772,7 @@ func isDatabaseEmpty(
 func createImportingDescriptors(
 	ctx context.Context, p sql.PlanHookState, sqlDescs []sqlbase.Descriptor, r *restoreResumer,
 ) (
-	databases []*sqlbase.ImmutableDatabaseDescriptor,
+	databases []sqlbase.DatabaseDescriptor,
 	tables []sqlbase.TableDescriptor,
 	oldTableIDs []descpb.ID,
 	writtenTypes []sqlbase.TypeDescriptor,

pkg/ccl/backupccl/targets_test.go

@@ -48,7 +48,7 @@ func TestDescriptorsMatchingTargets(t *testing.T) {
 			return desc
 		}
 		mkDB := func(id descpb.ID, name string) *sqlbase.ImmutableDatabaseDescriptor {
-			return sqlbase.NewInitialDatabaseDescriptor(id, name, security.AdminRole)
+			return &sqlbase.NewInitialDatabaseDescriptor(id, name, security.AdminRole).ImmutableDatabaseDescriptor
 		}
 		mkTyp := func(desc typDesc) *sqlbase.ImmutableTypeDescriptor {
 			return sqlbase.NewImmutableTypeDescriptor(desc)

pkg/ccl/importccl/import_stmt.go

@@ -323,7 +323,7 @@ func importPlanHook(
 		} else {
 			// No target table means we're importing whatever we find into the session
 			// database, so it must exist.
-			dbDesc, err := p.ResolveUncachedDatabaseByName(ctx, p.SessionData().Database, true /*required*/)
+			dbDesc, err := p.ResolveImmutableDatabaseDescriptor(ctx, p.SessionData().Database, true /*required*/)
 			if err != nil {
 				return pgerror.Wrap(err, pgcode.UndefinedObject,
 					"could not resolve current database")

pkg/cli/dump_test.go

@@ -447,7 +447,7 @@ INSERT INTO public.t (i, j) VALUES
 
 	if out, err := c.RunWithCaptureArgs([]string{"dump", "d", "t", "--as-of", "2000-01-01 00:00:00"}); err != nil {
 		t.Fatal(err)
-	} else if !strings.Contains(out, `database d does not exist`) {
+	} else if !strings.Contains(out, `database "d" does not exist`) {
 		t.Fatalf("unexpected output: %s", out)
 	}
 }

pkg/kv/kvserver/reports/constraint_stats_report_test.go

@@ -760,7 +760,7 @@ func compileTestCase(tc baseReportTestCase) (compiledTestCase, error) {
 			}
 		}
 		sysCfgBuilder.addDBDesc(dbID,
-			sqlbase.NewInitialDatabaseDescriptor(descpb.ID(dbID), db.name, security.AdminRole))
+			&sqlbase.NewInitialDatabaseDescriptor(descpb.ID(dbID), db.name, security.AdminRole).ImmutableDatabaseDescriptor)
 
 		for _, table := range db.tables {
 			tableID := objectCounter

pkg/sql/catalog/accessors/physical_schema_accessors.go

@@ -19,7 +19,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
-	"github.com/cockroachdb/cockroach/pkg/sql/catalog/lease"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
 	"github.com/cockroachdb/errors"
@@ -55,6 +54,7 @@ func NewCachedAccessor(
 // CachedPhysicalAccessor adds a cache on top of any Accessor.
 type CachedPhysicalAccessor struct {
 	catalog.Accessor
+	// TODO (lucy): Rename.
 	tc *descs.Collection
 	// Used to avoid allocations.
 	tableName tree.TableName
@@ -71,38 +71,18 @@ func (a *CachedPhysicalAccessor) GetDatabaseDesc(
 	name string,
 	flags tree.DatabaseLookupFlags,
 ) (desc sqlbase.DatabaseDescriptor, err error) {
-	isSystemDB := name == sqlbase.SystemDatabaseName
-	if !(flags.AvoidCached || isSystemDB || lease.TestingTableLeasesAreDisabled()) {
-		refuseFurtherLookup, dbID, err := a.tc.GetUncommittedDatabaseID(name, flags.Required)
-		if refuseFurtherLookup || err != nil {
+	if flags.RequireMutable {
+		db, err := a.tc.GetMutableDatabaseDescriptor(ctx, txn, name, flags)
+		if db == nil {
 			return nil, err
 		}
-
-		if dbID != descpb.InvalidID {
-			// Some database ID was found in the list of uncommitted DB changes.
-			// Use that to get the descriptor.
-			desc, err := a.tc.DatabaseCache().GetDatabaseDescByID(ctx, txn, dbID)
-			if desc == nil && flags.Required {
-				return nil, sqlbase.NewUndefinedDatabaseError(name)
-			} else if desc == nil {
-				// NB: We must return the actual value nil here as a typed nil will not
-				// be easily detectable by the caller.
-				return nil, nil
-			}
-			return desc, err
-		}
-
-		// The database was not known in the uncommitted list. Have the db
-		// cache look it up by name for us.
-		desc, err := a.tc.DatabaseCache().GetDatabaseDesc(ctx, a.tc.LeaseManager().DB().Txn, name, flags.Required)
-		if desc == nil || err != nil {
-			return nil, err
-		}
-		return desc, nil
+		return db, err
 	}
-
-	// We avoided the cache. Go lower.
-	return a.Accessor.GetDatabaseDesc(ctx, txn, codec, name, flags)
+	typ, err := a.tc.GetDatabaseVersion(ctx, txn, name, flags)
+	if typ == nil {
+		return nil, err
+	}
+	return typ, err
 }
 
 // GetSchema implements the Accessor interface.

pkg/sql/catalog/catalogkv/physical_accessor.go

@@ -46,10 +46,11 @@ func (a UncachedPhysicalAccessor) GetDatabaseDesc(
 	flags tree.DatabaseLookupFlags,
 ) (desc sqlbase.DatabaseDescriptor, err error) {
 	if name == sqlbase.SystemDatabaseName {
-		// We can't return a direct reference to SystemDB, because the
-		// caller expects a private object that can be modified in-place.
-		sysDB := sqlbase.MakeSystemDatabaseDesc()
-		return sysDB, nil
+		if flags.RequireMutable {
+			return sqlbase.NewMutableExistingDatabaseDescriptor(
+				*sqlbase.MakeSystemDatabaseDesc().DatabaseDesc()), nil
+		}
+		return sqlbase.SystemDB, nil
 	}
 
 	found, descID, err := LookupDatabaseID(ctx, txn, codec, name)
@@ -64,14 +65,15 @@ func (a UncachedPhysicalAccessor) GetDatabaseDesc(
 
 	// NB: Take care to actually return nil here rather than a typed nil which
 	// will not compare to nil when wrapped in the returned interface.
-	desc, err = GetDatabaseDescByID(ctx, txn, codec, descID)
+	untypedDesc, err := GetAnyDescriptorByID(ctx, txn, codec, descID, Mutability(flags.RequireMutable))
 	if err != nil {
 		return nil, err
 	}
-	if desc == nil {
+	db, ok := untypedDesc.(sqlbase.DatabaseDescriptor)
+	if !ok {
 		return nil, nil
 	}
-	return desc, err
+	return db, nil
 }
 
 // GetSchema implements the Accessor interface.