sql: disable primary key changes when a schema change is in progress

Fixes cockroachdb#45362.

Release note (sql change): This PR disables primary key changes
when a concurrent schema change is executing on the same table,
or if a schema change has been started on the same table in
the current transaction.

pkg/sql/alter_table.go

@@ -375,18 +375,29 @@ func (n *alterTableNode) startExec(params runParams) error {
 					"cannot create table and change it's primary key in the same transaction")
 			}
 
-			// Ensure that there is not another primary key change attempted within this transaction.
+			// Ensure that other schema changes on this table are not currently
+			// executing, and that other schema changes have not been performed
+			// in the current transaction.
 			currentMutationID := n.tableDesc.ClusterVersion.NextMutationID
 			for i := range n.tableDesc.Mutations {
-				if desc := n.tableDesc.Mutations[i].GetPrimaryKeySwap(); desc != nil {
-					if n.tableDesc.Mutations[i].MutationID == currentMutationID {
-						return unimplemented.NewWithIssue(
-							43376, "multiple primary key changes in the same transaction are unsupported")
-					}
-					if n.tableDesc.Mutations[i].MutationID < currentMutationID {
-						return pgerror.Newf(pgcode.ObjectNotInPrerequisiteState,
-							"table %s is currently undergoing a primary key change", n.tableDesc.Name)
+				mut := &n.tableDesc.Mutations[i]
+				if mut.MutationID == currentMutationID {
+					return unimplemented.NewWithIssuef(
+						45510, "cannot perform a primary key change on %s "+
+							"with other schema changes on %s in the same transaction", n.tableDesc.Name, n.tableDesc.Name)
+				}
+				if mut.MutationID < currentMutationID {
+					// We can handle indexes being deleted concurrently. We do this
+					// in order to not be blocked on index drops created by a previous
+					// primary key change. If we errored out when seeing a previous
+					// index drop, then users would see a confusing message that a
+					// schema change is in progress when it doesn't seem like one is.
+					// TODO (rohany): This feels like such a hack until (#45150) is fixed.
+					if mut.GetIndex() != nil && mut.Direction == sqlbase.DescriptorMutation_DROP {
+						continue
 					}
+					return unimplemented.NewWithIssuef(
+						45510, "table %s is currently undergoing a schema change", n.tableDesc.Name)
 				}
 			}
 
@@ -574,10 +585,9 @@ func (n *alterTableNode) startExec(params runParams) error {
 				}
 			}
 
+			// TODO (rohany): this loop will be unused until #45510 is resolved.
 			for i := range n.tableDesc.Mutations {
 				mut := &n.tableDesc.Mutations[i]
-				// TODO (rohany): It's unclear about what to do if there are other mutations within
-				//  this transaction too.
 				// If there is an index that is getting built right now that started in a previous txn, we
 				// need to potentially rebuild that index as well.
 				if idx := mut.GetIndex(); mut.MutationID < currentMutationID && idx != nil &&

pkg/sql/logictest/testdata/logic_test/alter_primary_key

@@ -661,6 +661,23 @@ t  CREATE TABLE t (
    FAMILY "primary" (x, rowid)
 )
 
+# Regression for #45362.
+statement ok
+DROP TABLE IF EXISTS t;
+CREATE TABLE t (x INT NOT NULL)
+
+statement ok
+BEGIN
+
+statement ok
+ALTER TABLE t ADD COLUMN y INT
+
+statement error pq: unimplemented: cannot perform a primary key change on t with other schema changes on t in the same transaction
+ALTER TABLE t ALTER PRIMARY KEY USING COLUMNS (x)
+
+statement ok
+ROLLBACK
+
 # Ensure that we cannot cancel the index cleanup jobs spawned by
 # a primary key change.
 statement ok
@@ -688,4 +705,3 @@ SELECT job_id FROM [SHOW JOBS] WHERE
 description = 'CLEANUP JOB for ''ALTER TABLE test.public.t ALTER PRIMARY KEY USING COLUMNS (y)''' AND
 status = 'running'
 ----
-

pkg/sql/schema_changer_test.go

@@ -2383,11 +2383,13 @@ func TestPrimaryKeyChangeWithPrecedingIndexCreation(t *testing.T) {
 	s, sqlDB, kvDB := serverutils.StartServer(t, params)
 	defer s.Stopper().Stop(ctx)
 
+	if _, err := sqlDB.Exec(`CREATE DATABASE t`); err != nil {
+		t.Fatal(err)
+	}
+
 	t.Run("create-index-before", func(t *testing.T) {
-		if _, err := sqlDB.Exec(`
-CREATE DATABASE t;
-CREATE TABLE t.test (k INT NOT NULL, v INT);
-`); err != nil {
+		t.Skip("unskip once #45510 is completed")
+		if _, err := sqlDB.Exec(`CREATE TABLE t.test (k INT NOT NULL, v INT)`); err != nil {
 			t.Fatal(err)
 		}
 		if err := bulkInsertIntoTable(sqlDB, maxValue); err != nil {
@@ -2460,7 +2462,7 @@ ALTER TABLE t.test ALTER PRIMARY KEY USING COLUMNS (v2)`); err != nil {
 		_, err := sqlDB.Exec(`
 SET experimental_enable_primary_key_changes = true;
 ALTER TABLE t.test ALTER PRIMARY KEY USING COLUMNS (k)`)
-		if !testutils.IsError(err, "pq: table test is currently undergoing a primary key change") {
+		if !testutils.IsError(err, "pq: unimplemented: table test is currently undergoing a schema change") {
 			t.Errorf("expected to concurrent primary key change to error, but got %+v", err)
 		}