sql: introduce MutableTableDescriptor struct

Replace the type alias which made MutableTableDescriptor and
TableDescriptor equivalent types with a struct that embeds
TableDescriptor. This forces any write of a table descriptor for schema
changes to use this struct.

Release note: None

pkg/ccl/changefeedccl/avro_test.go

@@ -47,12 +47,12 @@ func parseTableDesc(createTableStmt string) (*sqlbase.TableDescriptor, error) {
 	st := cluster.MakeTestingClusterSettings()
 	const parentID = sqlbase.ID(keys.MaxReservedDescID + 1)
 	const tableID = sqlbase.ID(keys.MaxReservedDescID + 2)
-	tableDesc, err := importccl.MakeSimpleTableDescriptor(
+	mutDesc, err := importccl.MakeSimpleTableDescriptor(
 		ctx, st, createTable, parentID, tableID, importccl.NoFKs, hlc.UnixNano())
 	if err != nil {
 		return nil, err
 	}
-	return tableDesc, tableDesc.ValidateTable(st)
+	return mutDesc.TableDesc(), mutDesc.TableDesc().ValidateTable(st)
 }
 
 func parseValues(tableDesc *sqlbase.TableDescriptor, values string) ([]sqlbase.EncDatumRow, error) {

pkg/ccl/importccl/import_stmt.go

@@ -136,7 +136,7 @@ type fkHandler struct {
 // NoFKs is used by formats that do not support FKs.
 var NoFKs = fkHandler{resolver: make(fkResolver)}
 
-// MakeSimpleTableDescriptor creates a TableDescriptor from a CreateTable parse
+// MakeSimpleTableDescriptor creates a MutableTableDescriptor from a CreateTable parse
 // node without the full machinery. Many parts of the syntax are unsupported
 // (see the implementation and TestMakeSimpleTableDescriptorErrors for details),
 // but this is enough for our csv IMPORT and for some unit tests.
@@ -152,7 +152,7 @@ func MakeSimpleTableDescriptor(
 	tableID sqlbase.ID,
 	fks fkHandler,
 	walltime int64,
-) (*sqlbase.TableDescriptor, error) {
+) (*sqlbase.MutableTableDescriptor, error) {
 	create.HoistConstraints()
 	if create.IfNotExists {
 		return nil, pgerror.Unimplemented("import.if-no-exists", "unsupported IF NOT EXISTS")
@@ -204,7 +204,7 @@ func MakeSimpleTableDescriptor(
 		Context:  ctx,
 		Sequence: &importSequenceOperators{},
 	}
-	affected := make(map[sqlbase.ID]*sqlbase.TableDescriptor)
+	affected := make(map[sqlbase.ID]*sqlbase.MutableTableDescriptor)
 
 	tableDesc, err := sql.MakeTableDesc(
 		ctx,
@@ -223,7 +223,7 @@ func MakeSimpleTableDescriptor(
 	if err != nil {
 		return nil, err
 	}
-	if err := fixDescriptorFKState(&tableDesc); err != nil {
+	if err := fixDescriptorFKState(tableDesc.TableDesc()); err != nil {
 		return nil, err
 	}
 
@@ -293,7 +293,7 @@ func (so *importSequenceOperators) SetSequenceValue(
 	return errSequenceOperators
 }
 
-type fkResolver map[string]*sqlbase.TableDescriptor
+type fkResolver map[string]*sqlbase.MutableTableDescriptor
 
 var _ sql.SchemaResolver = fkResolver{}
 
@@ -772,7 +772,7 @@ func importPlanHook(
 			if err != nil {
 				return err
 			}
-			tableDescs = []*sqlbase.TableDescriptor{tbl}
+			tableDescs = []*sqlbase.TableDescriptor{tbl.TableDesc()}
 			descStr, err := importJobDescription(importStmt, create.Defs, files, opts)
 			if err != nil {
 				return err

pkg/ccl/importccl/import_stmt_test.go

@@ -1461,7 +1461,7 @@ func BenchmarkConvertRecord(b *testing.B) {
 		}
 	}()
 
-	c := &csvInputReader{recordCh: recordCh, tableDesc: tableDesc}
+	c := &csvInputReader{recordCh: recordCh, tableDesc: tableDesc.TableDesc()}
 	// start up workers.
 	for i := 0; i < runtime.NumCPU(); i++ {
 		group.Go(func() error {

pkg/ccl/importccl/load.go

@@ -157,7 +157,7 @@ func Load(
 			// rejected during restore.
 			st := cluster.MakeTestingClusterSettings()
 
-			affected := make(map[sqlbase.ID]*sqlbase.TableDescriptor)
+			affected := make(map[sqlbase.ID]*sqlbase.MutableTableDescriptor)
 			// A nil txn is safe because it is only used by sql.MakeTableDesc, which
 			// only uses txn for resolving FKs and interleaved tables, neither of which
 			// are present here. Ditto for the schema accessor.
@@ -170,7 +170,7 @@ func Load(
 				return backupccl.BackupDescriptor{}, errors.Wrap(err, "make table desc")
 			}
 
-			tableDesc = &desc
+			tableDesc = desc.TableDesc()
 			tableDescs[tableName] = tableDesc
 			backup.Descriptors = append(backup.Descriptors, sqlbase.Descriptor{
 				Union: &sqlbase.Descriptor_Table{Table: tableDesc},

pkg/ccl/importccl/read_import_mysql.go

@@ -370,8 +370,8 @@ func mysqlTableToCockroach(
 		if err != nil {
 			return nil, nil, err
 		}
-		seqDesc = &desc
-		fks.resolver[seqName] = seqDesc
+		seqDesc = desc.TableDesc()
+		fks.resolver[seqName] = &desc
 		id++
 	}
 
@@ -454,9 +454,9 @@ func mysqlTableToCockroach(
 	}
 	fks.resolver[desc.Name] = desc
 	if seqDesc != nil {
-		return []*sqlbase.TableDescriptor{seqDesc, desc}, fkDefs, nil
+		return []*sqlbase.TableDescriptor{seqDesc, desc.TableDesc()}, fkDefs, nil
 	}
-	return []*sqlbase.TableDescriptor{desc}, fkDefs, nil
+	return []*sqlbase.TableDescriptor{desc.TableDesc()}, fkDefs, nil
 }
 
 func mysqlActionToCockroach(action mysql.ReferenceAction) tree.ReferenceAction {
@@ -474,18 +474,18 @@ func mysqlActionToCockroach(action mysql.ReferenceAction) tree.ReferenceAction {
 }
 
 type delayedFK struct {
-	tbl *sqlbase.TableDescriptor
+	tbl *sqlbase.MutableTableDescriptor
 	def *tree.ForeignKeyConstraintTableDef
 }
 
 func addDelayedFKs(ctx context.Context, defs []delayedFK, resolver fkResolver) error {
 	for _, def := range defs {
 		if err := sql.ResolveFK(
-			ctx, nil, resolver, def.tbl, def.def, map[sqlbase.ID]*sqlbase.TableDescriptor{}, sqlbase.ConstraintValidity_Validated,
+			ctx, nil, resolver, def.tbl, def.def, map[sqlbase.ID]*sqlbase.MutableTableDescriptor{}, sqlbase.ConstraintValidity_Validated,
 		); err != nil {
 			return err
 		}
-		if err := fixDescriptorFKState(def.tbl); err != nil {
+		if err := fixDescriptorFKState(def.tbl.TableDesc()); err != nil {
 			return err
 		}
 		if err := def.tbl.AllocateIDs(); err != nil {

pkg/ccl/importccl/read_import_mysql_test.go

@@ -74,10 +74,10 @@ func descForTable(
 	if err != nil {
 		t.Fatalf("could not interpret %q: %v", create, err)
 	}
-	if err := fixDescriptorFKState(table); err != nil {
+	if err := fixDescriptorFKState(table.TableDesc()); err != nil {
 		t.Fatal(err)
 	}
-	return table
+	return table.TableDesc()
 }
 
 func TestMysqldumpDataReader(t *testing.T) {
@@ -173,7 +173,7 @@ func TestMysqldumpSchemaReader(t *testing.T) {
 	referencedSimple := descForTable(t, readFile(t, `simple.cockroach-schema.sql`), expectedParent, 52, NoFKs)
 	fks := fkHandler{
 		allowed:  true,
-		resolver: fkResolver(map[string]*sqlbase.TableDescriptor{referencedSimple.Name: referencedSimple}),
+		resolver: fkResolver(map[string]*sqlbase.MutableTableDescriptor{referencedSimple.Name: sql.NewMutableTableDescriptor(*referencedSimple)}),
 	}
 
 	t.Run("simple", func(t *testing.T) {

pkg/ccl/importccl/read_import_pgdump.go

@@ -245,9 +245,9 @@ func readPostgresCreateTable(
 					return nil, err
 				}
 				fks.resolver[desc.Name] = &desc
-				ret = append(ret, &desc)
+				ret = append(ret, desc.TableDesc())
 			}
-			backrefs := make(map[sqlbase.ID]*sqlbase.TableDescriptor)
+			backrefs := make(map[sqlbase.ID]*sqlbase.MutableTableDescriptor)
 			for _, create := range createTbl {
 				if create == nil {
 					continue
@@ -260,7 +260,7 @@ func readPostgresCreateTable(
 				}
 				fks.resolver[desc.Name] = desc
 				backrefs[desc.ID] = desc
-				ret = append(ret, desc)
+				ret = append(ret, desc.TableDesc())
 			}
 			for name, constraints := range tableFKs {
 				desc := fks.resolver[name]
@@ -272,7 +272,7 @@ func readPostgresCreateTable(
 						return nil, err
 					}
 				}
-				if err := fixDescriptorFKState(desc); err != nil {
+				if err := fixDescriptorFKState(desc.TableDesc()); err != nil {
 					return nil, err
 				}
 			}

pkg/ccl/partitionccl/partition_test.go

@@ -125,11 +125,12 @@ func (t *partitioningTest) parse() error {
 		}
 		st := cluster.MakeTestingClusterSettings()
 		const parentID, tableID = keys.MinUserDescID, keys.MinUserDescID + 1
-		t.parsed.tableDesc, err = importccl.MakeSimpleTableDescriptor(
+		mutDesc, err := importccl.MakeSimpleTableDescriptor(
 			ctx, st, createTable, parentID, tableID, importccl.NoFKs, hlc.UnixNano())
 		if err != nil {
 			return err
 		}
+		t.parsed.tableDesc = mutDesc.TableDesc()
 		if err := t.parsed.tableDesc.ValidateTable(st); err != nil {
 			return err
 		}

pkg/sql/alter_index.go

@@ -72,7 +72,7 @@ func (n *alterIndexNode) startExec(params runParams) error {
 			)
 			err = deleteRemovedPartitionZoneConfigs(
 				params.ctx, params.p.txn,
-				n.tableDesc, n.indexDesc,
+				n.tableDesc.TableDesc(), n.indexDesc,
 				&n.indexDesc.Partitioning, &partitioning,
 				params.extendedEvalCtx.ExecCfg,
 			)

pkg/sql/alter_table.go

@@ -386,7 +386,7 @@ func (n *alterTableNode) startExec(params runParams) error {
 			// Drop check constraints which reference the column.
 			validChecks := n.tableDesc.Checks[:0]
 			for _, check := range n.tableDesc.Checks {
-				if used, err := check.UsesColumn(n.tableDesc, col.ID); err != nil {
+				if used, err := check.UsesColumn(n.tableDesc.TableDesc(), col.ID); err != nil {
 					return err
 				} else if !used {
 					validChecks = append(validChecks, check)
@@ -479,7 +479,7 @@ func (n *alterTableNode) startExec(params runParams) error {
 				}
 				ck := n.tableDesc.Checks[idx]
 				if err := params.p.validateCheckExpr(
-					params.ctx, ck.Expr, &n.n.Table, n.tableDesc,
+					params.ctx, ck.Expr, &n.n.Table, n.tableDesc.TableDesc(),
 				); err != nil {
 					return err
 				}
@@ -503,7 +503,7 @@ func (n *alterTableNode) startExec(params runParams) error {
 				if err != nil {
 					panic(err)
 				}
-				if err := params.p.validateForeignKey(params.ctx, n.tableDesc, idx); err != nil {
+				if err := params.p.validateForeignKey(params.ctx, n.tableDesc.TableDesc(), idx); err != nil {
 					return err
 				}
 				idx.ForeignKey.Validity = sqlbase.ConstraintValidity_Validated
@@ -542,7 +542,7 @@ func (n *alterTableNode) startExec(params runParams) error {
 			)
 			err = deleteRemovedPartitionZoneConfigs(
 				params.ctx, params.p.txn,
-				n.tableDesc, &n.tableDesc.PrimaryIndex, &n.tableDesc.PrimaryIndex.Partitioning,
+				n.tableDesc.TableDesc(), &n.tableDesc.PrimaryIndex, &n.tableDesc.PrimaryIndex.Partitioning,
 				&partitioning, params.extendedEvalCtx.ExecCfg,
 			)
 			if err != nil {
@@ -552,7 +552,7 @@ func (n *alterTableNode) startExec(params runParams) error {
 
 		case *tree.AlterTableSetAudit:
 			var err error
-			descriptorChanged, err = params.p.setAuditMode(params.ctx, n.tableDesc, t.Mode)
+			descriptorChanged, err = params.p.setAuditMode(params.ctx, n.tableDesc.TableDesc(), t.Mode)
 			if err != nil {
 				return err
 			}
@@ -562,7 +562,7 @@ func (n *alterTableNode) startExec(params runParams) error {
 			if !ok {
 				return pgerror.NewAssertionErrorf("missing stats data")
 			}
-			if err := injectTableStats(params, n.tableDesc, sd); err != nil {
+			if err := injectTableStats(params, n.tableDesc.TableDesc(), sd); err != nil {
 				return err
 			}
 

pkg/sql/authorization.go

@@ -173,11 +173,12 @@ func (p *planner) MemberOfWithAdminOption(
 	ctx context.Context, member string,
 ) (map[string]bool, error) {
 	// Lookup table version.
-	tableDesc, _, err := p.PhysicalSchemaAccessor().GetObjectDesc(&roleMembersTableName,
+	objDesc, _, err := p.PhysicalSchemaAccessor().GetObjectDesc(&roleMembersTableName,
 		p.ObjectLookupFlags(ctx, true /*required*/))
 	if err != nil {
 		return nil, err
 	}
+	tableDesc := objDesc.TableDesc()
 	tableVersion := tableDesc.Version
 
 	// We loop in case the table version changes while we're looking up memberships.

pkg/sql/backfill.go

@@ -499,7 +499,7 @@ func runSchemaChangesInTxn(
 	tc *TableCollection,
 	execCfg *ExecutorConfig,
 	evalCtx *tree.EvalContext,
-	tableDesc *sqlbase.TableDescriptor,
+	tableDesc *sqlbase.MutableTableDescriptor,
 	traceKV bool,
 ) error {
 	if len(tableDesc.DrainingNames) > 0 {
@@ -531,13 +531,13 @@ func runSchemaChangesInTxn(
 				if doneColumnBackfill || !sqlbase.ColumnNeedsBackfill(m.GetColumn()) {
 					break
 				}
-				if err := columnBackfillInTxn(ctx, txn, tc, evalCtx, tableDesc, traceKV); err != nil {
+				if err := columnBackfillInTxn(ctx, txn, tc, evalCtx, tableDesc.TableDesc(), traceKV); err != nil {
 					return err
 				}
 				doneColumnBackfill = true
 
 			case *sqlbase.DescriptorMutation_Index:
-				if err := indexBackfillInTxn(ctx, txn, tableDesc, traceKV); err != nil {
+				if err := indexBackfillInTxn(ctx, txn, tableDesc.TableDesc(), traceKV); err != nil {
 					return err
 				}
 
@@ -552,13 +552,13 @@ func runSchemaChangesInTxn(
 				if doneColumnBackfill {
 					break
 				}
-				if err := columnBackfillInTxn(ctx, txn, tc, evalCtx, tableDesc, traceKV); err != nil {
+				if err := columnBackfillInTxn(ctx, txn, tc, evalCtx, tableDesc.TableDesc(), traceKV); err != nil {
 					return err
 				}
 				doneColumnBackfill = true
 
 			case *sqlbase.DescriptorMutation_Index:
-				if err := indexTruncateInTxn(ctx, txn, execCfg, tableDesc, traceKV); err != nil {
+				if err := indexTruncateInTxn(ctx, txn, execCfg, tableDesc.TableDesc(), traceKV); err != nil {
 					return err
 				}
 
@@ -616,7 +616,7 @@ func columnBackfillInTxn(
 				"table %s not created in the same transaction as id = %d", tableDesc.Name, k)
 		}
 		table := tc.getUncommittedTableByID(k)
-		otherTableDescs = append(otherTableDescs, *table)
+		otherTableDescs = append(otherTableDescs, table.TableDescriptor)
 	}
 	sp := tableDesc.PrimaryIndexSpan()
 	for sp.Key != nil {