sql: move validation of primary key existence to TableCollection

To support `ADD/DROP PRIMARY KEY` operations within a single
transaction, we had added a step in the `connExecutor` to verify that
all tables modified in the transaction have a primary key before we
commit the transaction. This was done by examining `SchemaChanger`s
queued in the transaction in `extraTxnState`. To ease the process of
removing `SchemaChanger`s from `extraTxnState` as part of moving them to
jobs, this PR moves the validation implementation to a method on
`TableCollection` that examines all modified tables.

Release note: None

pkg/sql/conn_executor_exec.go

@@ -614,11 +614,9 @@ func (ex *connExecutor) commitSQLTransactionInternal(
 ) (ev fsm.Event, payload fsm.EventPayload, ok bool) {
 	ex.clearSavepoints()
 
-	for _, sc := range ex.extraTxnState.schemaChangers.schemaChangers {
-		if err := sc.validateTablePrimaryKeys(ctx, ex.state.mu.txn); err != nil {
-			ev, payload = ex.makeErrEvent(err, stmt)
-			return ev, payload, false
-		}
+	if err := ex.extraTxnState.tables.validatePrimaryKeys(); err != nil {
+		ev, payload = ex.makeErrEvent(err, stmt)
+		return ev, payload, false
 	}
 
 	if err := ex.checkTableTwoVersionInvariant(ctx); err != nil {

pkg/sql/schema_changer.go

@@ -1953,21 +1953,6 @@ func (sc *SchemaChanger) reverseMutation(
 	return mutation, columns
 }
 
-// validateTablePrimaryKeys ensures that the table this schema changer
-// references has a primary key. This is to avoid usage of the DROP PRIMARY KEY
-// command without an ADD PRIMARY KEY command before the end of the transaction.
-func (sc *SchemaChanger) validateTablePrimaryKeys(ctx context.Context, txn *client.Txn) error {
-	table, err := sqlbase.GetMutableTableDescFromID(ctx, txn, sc.tableID)
-	if err != nil {
-		return err
-	}
-	if !table.HasPrimaryKey() {
-		return errors.Errorf(
-			"primary key of table %s dropped without subsequent addition of new primary key", table.Name)
-	}
-	return nil
-}
-
 // TestingSchemaChangerCollection is an exported (for testing) version of
 // schemaChangerCollection.
 // TODO(andrei): get rid of this type once we can have tests internal to the sql

pkg/sql/table.go

@@ -796,6 +796,23 @@ type tableCollectionModifier interface {
 	copyModifiedSchema(to *TableCollection)
 }
 
+// validatePrimaryKeys verifies that all tables modified in the transaction have
+// an enabled primary key after potentially undergoing DROP PRIMARY KEY, which
+// is required to be followed by ADD PRIMARY KEY.
+func (tc *TableCollection) validatePrimaryKeys() error {
+	modifiedTables := tc.getTablesWithNewVersion()
+	for i := range modifiedTables {
+		table := tc.getUncommittedTableByID(modifiedTables[i].id).MutableTableDescriptor
+		if !table.HasPrimaryKey() {
+			return errors.Errorf(
+				"primary key of table %s dropped without subsequent addition of new primary key",
+				table.Name,
+			)
+		}
+	}
+	return nil
+}
+
 // createOrUpdateSchemaChangeJob finalizes the current mutations in the table
 // descriptor. If a schema change job in the system.jobs table has not been
 // created for mutations in the current transaction, one is created. The