sql: disallow other schema changes an in progress primary key change

Fixes cockroachdb#45363.

Release note (sql change): This commit disables the use of schema
changes when a primary key change is in progress. This includes
the following:
* running a primary key change in a transaction, and then starting
  another schema change in the same transaction.
* starting a primary key change on one connection, and then starting
  a schema change on the same table on another connection while
  the initial primary key change is currently executing.

pkg/sql/logictest/testdata/logic_test/alter_primary_key

@@ -592,6 +592,40 @@ INSERT INTO t VALUES (1, 2, 3);
 ALTER TABLE t ALTER PRIMARY KEY USING COLUMNS (z);
 UPDATE t SET y = 3 WHERE z = 3
 
+# Test for #45363.
+
+statement ok
+DROP TABLE IF EXISTS t;
+CREATE TABLE t (x INT PRIMARY KEY, y INT NOT NULL)
+
+statement ok
+BEGIN
+
+statement ok
+ALTER TABLE t ALTER PRIMARY KEY USING COLUMNS (y)
+
+statement error pq: unimplemented: cannot perform other schema changes in the same transaction as a primary key change
+CREATE INDEX ON t (y)
+
+statement ok
+ROLLBACK
+
+statement ok
+DROP TABLE IF EXISTS t;
+CREATE TABLE t (x INT PRIMARY KEY, y INT NOT NULL)
+
+statement ok
+BEGIN
+
+statement ok
+ALTER TABLE t ALTER PRIMARY KEY USING COLUMNS (y)
+
+statement error pq: unimplemented: cannot perform other schema changes in the same transaction as a primary key change
+ALTER TABLE t ADD COLUMN z INT
+
+statement ok
+ROLLBACK
+
 subtest add_pk_rowid
 # Tests for #45509.
 statement ok

pkg/sql/schema_changer_test.go

@@ -2484,6 +2484,64 @@ ALTER TABLE t.test ALTER PRIMARY KEY USING COLUMNS (k)`); err != nil {
 	})
 }
 
+// TestSchemaChangeWhileExecutingPrimaryKeyChange tests that other schema
+// changes cannot be queued while a primary key change is executing.
+func TestSchemaChangeWhileExecutingPrimaryKeyChange(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	ctx := context.Background()
+
+	backfillNotification := make(chan struct{})
+	waitBeforeContinuing := make(chan struct{})
+
+	params, _ := tests.CreateTestServerParams()
+	params.Knobs = base.TestingKnobs{
+		DistSQL: &execinfra.TestingKnobs{
+			RunBeforeBackfillChunk: func(_ roachpb.Span) error {
+				backfillNotification <- struct{}{}
+				<-waitBeforeContinuing
+				return nil
+			},
+		},
+	}
+
+	s, sqlDB, _ := serverutils.StartServer(t, params)
+	defer s.Stopper().Stop(ctx)
+
+	if _, err := sqlDB.Exec(`
+SET experimental_enable_primary_key_changes = true;
+CREATE DATABASE t;
+CREATE TABLE t.test (k INT NOT NULL, v INT);
+`); err != nil {
+		t.Fatal(err)
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		if _, err := sqlDB.Exec(`ALTER TABLE t.test ALTER PRIMARY KEY USING COLUMNS (k)`); err != nil {
+			t.Error(err)
+		}
+		wg.Done()
+	}()
+
+	<-backfillNotification
+
+	// Test that trying different schema changes results an error.
+	_, err := sqlDB.Exec(`ALTER TABLE t.test ADD COLUMN z INT`)
+	expected := "pq: unimplemented: cannot perform a schema change operation while a primary key change is in progress"
+	if !testutils.IsError(err, expected) {
+		t.Fatalf("expected to find error %s but found %+v", expected, err)
+	}
+
+	_, err = sqlDB.Exec(`CREATE INDEX ON t.test(v)`)
+	if !testutils.IsError(err, expected) {
+		t.Fatalf("expected to find error %s but found %+v", expected, err)
+	}
+
+	waitBeforeContinuing <- struct{}{}
+	wg.Wait()
+}
+
 // TestPrimaryKeyChangeWithOperations ensures that different operations succeed
 // while a primary key change is happening.
 func TestPrimaryKeyChangeWithOperations(t *testing.T) {

pkg/sql/sqlbase/structured.go

@@ -1851,6 +1851,31 @@ func (desc *TableDescriptor) ValidateTable() error {
 	// run again and mixed-version clusters always write "good" descriptors.
 	desc.Privileges.MaybeFixPrivileges(desc.GetID())
 
+	// Ensure that mutations cannot be queued if a primary key change has
+	// either been started in this transaction, or is currently in progress.
+	var alterPKMutation MutationID
+	var foundAlterPK bool
+	for _, m := range desc.Mutations {
+		// If we have seen an alter primary key mutation, then
+		// m we are considering right now is invalid.
+		if foundAlterPK {
+			if alterPKMutation == m.MutationID {
+				return unimplemented.NewWithIssue(
+					45615,
+					"cannot perform other schema changes in the same transaction as a primary key change",
+				)
+			}
+			return unimplemented.NewWithIssue(
+				45615,
+				"cannot perform a schema change operation while a primary key change is in progress",
+			)
+		}
+		if m.GetPrimaryKeySwap() != nil {
+			foundAlterPK = true
+			alterPKMutation = m.MutationID
+		}
+	}
+
 	// Validate the privilege descriptor.
 	return desc.Privileges.Validate(desc.GetID())
 }