scbuild: allow multiple column renames

This includes new dependency rules so that we fully remove the old name
before allowing a new column to take on that name, and another one to
make sure that a name is available before we create a computed column
expression that uses that name.

Release note: None

Author: rafiss

pkg/ccl/changefeedccl/changefeed_test.go

@@ -10242,23 +10242,19 @@ func TestChangefeedPredicateWithSchemaChange(t *testing.T) {
 	}
 
 	type testCase struct {
-		name                string
-		disableSchemaLocked bool
-		createFeedStmt      string   // Create changefeed statement.
-		initialPayload      []string // Expected payload after create.
-		alterStmt           string   // Alter statement to execute.
-		afterAlterStmt      string   // Execute after alter statement.
-		expectErr           string   // Alter may result in changefeed terminating with error.
-		payload             []string // Expect the following payload after executing afterAlterStmt.
+		name           string
+		createFeedStmt string   // Create changefeed statement.
+		initialPayload []string // Expected payload after create.
+		alterStmt      string   // Alter statement to execute.
+		afterAlterStmt string   // Execute after alter statement.
+		expectErr      string   // Alter may result in changefeed terminating with error.
+		payload        []string // Expect the following payload after executing afterAlterStmt.
 	}
 
 	testFn := func(tc testCase) cdcTestFn {
 		return func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
 			sqlDB := sqlutils.MakeSQLRunner(s.DB)
 
-			if tc.disableSchemaLocked {
-				sqlDB.Exec(t, "SET create_table_with_schema_locked=false")
-			}
 			sqlDB.ExecMultiple(t, setupSQL...)
 			foo := feed(t, f, tc.createFeedStmt)
 			feedJob := foo.(cdctest.EnterpriseTestFeed)
@@ -10364,7 +10360,7 @@ func TestChangefeedPredicateWithSchemaChange(t *testing.T) {
 			},
 			alterStmt:      "ALTER TABLE foo RENAME COLUMN c TO c_new",
 			afterAlterStmt: "INSERT INTO foo (a, b) VALUES (3, 'tres')",
-			expectErr:      `column "c" does not exist`,
+			expectErr:      `(column "c" does not exist|could not identify column "c")`,
 		},
 		{
 			name:           "alter enum",
@@ -10402,37 +10398,33 @@ func TestChangefeedPredicateWithSchemaChange(t *testing.T) {
 		{
 			// Alter and rename a column. The changefeed expression does not
 			// explicitly involve the column in question (c) -- so, schema change works
-			// fine. Note: we get 2 backfill events -- one for each logical change
-			// (rename column, then add column).
-			name:                "add and rename column",
-			disableSchemaLocked: true, // legacy schema change
-			createFeedStmt:      "CREATE CHANGEFEED AS SELECT *, (cdc_prev).e as old_e FROM foo",
+			// fine. With the declarative schema changer, both operations are processed
+			// atomically, resulting in a single backfill event per row with the final state.
+			name:           "add and rename column",
+			createFeedStmt: "CREATE CHANGEFEED AS SELECT *, (cdc_prev).e as old_e FROM foo",
 			initialPayload: []string{
 				`foo: [1, "one"]->{"a": 1, "b": "one", "c": null, "e": "inactive", "old_e": null}`,
 				`foo: [2, "two"]->{"a": 2, "b": "two", "c": "c string", "e": "open", "old_e": null}`,
 			},
 			alterStmt: "ALTER TABLE foo RENAME COLUMN c to c_old, ADD COLUMN c int DEFAULT 42",
 			payload: []string{
 				`foo: [1, "one"]->{"a": 1, "b": "one", "c": 42, "c_old": null, "e": "inactive", "old_e": "inactive"}`,
-				`foo: [1, "one"]->{"a": 1, "b": "one", "c_old": null, "e": "inactive", "old_e": "inactive"}`,
 				`foo: [2, "two"]->{"a": 2, "b": "two", "c": 42, "c_old": "c string", "e": "open", "old_e": "open"}`,
-				`foo: [2, "two"]->{"a": 2, "b": "two", "c_old": "c string", "e": "open", "old_e": "open"}`,
 			},
 		},
 		{
 			// Alter and rename a column. The changefeed expression does
 			// explicitly involve the column in question (c) -- so we expect
 			// to get an error because as soon as the first rename goes through, column
 			// no longer exists.
-			name:                "add and rename column error",
-			disableSchemaLocked: true, // legacy schema change
-			createFeedStmt:      "CREATE CHANGEFEED AS SELECT c, (cdc_prev).c AS prev_c FROM foo",
+			name:           "add and rename column error",
+			createFeedStmt: "CREATE CHANGEFEED AS SELECT c, (cdc_prev).c AS prev_c FROM foo",
 			initialPayload: []string{
 				`foo: [1, "one"]->{"c": null, "prev_c": null}`,
 				`foo: [2, "two"]->{"c": "c string", "prev_c": null}`,
 			},
 			alterStmt: "ALTER TABLE foo RENAME COLUMN c to c_old, ADD COLUMN c int DEFAULT 42",
-			expectErr: `column "c" does not exist`,
+			expectErr: `(column "c" does not exist|could not identify column "c")`,
 		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {

pkg/ccl/schemachangerccl/sctestbackupccl/backup_base_generated_test.go

@@ -22,10 +22,12 @@ checking for feature: ALTER TABLE
 increment telemetry for sql.schema.alter_table
 increment telemetry for sql.schema.alter_table.drop_column
 increment telemetry for sql.schema.alter_table.add_column
+increment telemetry for sql.schema.qualifcation.unique
 increment telemetry for sql.schema.qualifcation.default_expr
 increment telemetry for sql.schema.new_column_type.int8
 increment telemetry for sql.schema.alter_table.drop_column
 increment telemetry for sql.schema.alter_table.add_column
+increment telemetry for sql.schema.qualifcation.unique
 increment telemetry for sql.schema.qualifcation.default_expr
 increment telemetry for sql.schema.new_column_type.int8
 write *eventpb.AlterTable to event log:

pkg/ccl/schemachangerccl/testdata/end_to_end/add_column_multiple_regional_by_row/add_column_multiple_regional_by_row.side_effects

@@ -1221,9 +1221,9 @@ WHERE feature_name IN (
   'sql.schema.new_column.qualification.unique'
 )
 ----
-sql.schema.new_column.qualification.unique
 sql.schema.new_column.qualification.computed
 sql.schema.new_column.qualification.default_expr
+sql.schema.new_column.qualification.unique
 
 statement ok
 DROP TABLE telemetry_test

pkg/sql/logictest/testdata/logic_test/alter_table

@@ -228,3 +228,80 @@ column_name
 NewCamelCase
 SnakeCase
 decimal_value
+
+# Test renaming a column, adding a new column with the old name, and altering
+# the primary key to use the renamed column in a single statement.
+subtest rename_add_alter_primary_key
+
+statement ok
+CREATE TABLE rename_add_alter_pk_tbl (a INT PRIMARY KEY, b INT, FAMILY f (a, b));
+
+statement ok
+INSERT INTO rename_add_alter_pk_tbl VALUES (1, 10), (2, 20);
+
+skipif config local-legacy-schema-changer local-mixed-25.2 local-mixed-25.3 local-mixed-25.4
+statement ok
+ALTER TABLE rename_add_alter_pk_tbl RENAME COLUMN a TO b_old,
+              ADD COLUMN a INT NOT NULL DEFAULT 0,
+              ALTER PRIMARY KEY USING COLUMNS (b_old);
+
+skipif config local-legacy-schema-changer local-mixed-25.2 local-mixed-25.3 local-mixed-25.4
+query III colnames,rowsort
+SELECT b_old, b, a FROM rename_add_alter_pk_tbl;
+----
+b_old  b   a
+1      10  0
+2      20  0
+
+skipif config local-legacy-schema-changer local-mixed-25.2 local-mixed-25.3 local-mixed-25.4
+query TTBITTTBBBF colnames,rowsort
+SHOW INDEXES FROM rename_add_alter_pk_tbl;
+----
+table_name             index_name                  non_unique  seq_in_index  column_name  definition  direction  storing  implicit  visible  visibility
+rename_add_alter_pk_tbl  rename_add_alter_pk_tbl_pkey  false       1             b_old        b_old       ASC        false    false     true     1
+rename_add_alter_pk_tbl  rename_add_alter_pk_tbl_pkey  false       2             b            b           N/A        true     false     true     1
+rename_add_alter_pk_tbl  rename_add_alter_pk_tbl_pkey  false       3             a            a           N/A        true     false     true     1
+
+skipif config local-legacy-schema-changer local-mixed-25.2 local-mixed-25.3 local-mixed-25.4
+query T
+SELECT create_statement FROM [SHOW CREATE TABLE rename_add_alter_pk_tbl];
+----
+CREATE TABLE public.rename_add_alter_pk_tbl (
+  b_old INT8 NOT NULL,
+  b INT8 NULL,
+  a INT8 NOT NULL DEFAULT 0:::INT8,
+  CONSTRAINT rename_add_alter_pk_tbl_pkey PRIMARY KEY (b_old ASC),
+  FAMILY f (b_old, b, a)
+) WITH (schema_locked = true);
+
+skipif config local-legacy-schema-changer local-mixed-25.2 local-mixed-25.3 local-mixed-25.4
+statement ok
+UPDATE rename_add_alter_pk_tbl SET a = 10 WHERE b_old = 2;
+
+skipif config local-legacy-schema-changer local-mixed-25.2 local-mixed-25.3 local-mixed-25.4
+statement ok
+ALTER TABLE rename_add_alter_pk_tbl RENAME COLUMN b_old TO b_orig,
+              ADD COLUMN b_old INT NOT NULL DEFAULT 100,
+              ALTER PRIMARY KEY USING COLUMNS (a);
+
+skipif config local-legacy-schema-changer local-mixed-25.2 local-mixed-25.3 local-mixed-25.4
+query IIII colnames,rowsort
+SELECT b_old, b_orig, a, b FROM rename_add_alter_pk_tbl;
+----
+b_old  b_orig  a   b
+100    1       0   10
+100    2       10  20
+
+skipif config local-legacy-schema-changer local-mixed-25.2 local-mixed-25.3 local-mixed-25.4
+query T
+SELECT create_statement FROM [SHOW CREATE TABLE rename_add_alter_pk_tbl];
+----
+CREATE TABLE public.rename_add_alter_pk_tbl (
+  b_orig INT8 NOT NULL,
+  b INT8 NULL,
+  a INT8 NOT NULL DEFAULT 0:::INT8,
+  b_old INT8 NOT NULL DEFAULT 100:::INT8,
+  CONSTRAINT rename_add_alter_pk_tbl_pkey PRIMARY KEY (a ASC),
+  UNIQUE INDEX rename_add_alter_pk_tbl_b_orig_key (b_orig ASC),
+  FAMILY f (b_orig, b, a, b_old)
+) WITH (schema_locked = true);

pkg/sql/logictest/testdata/logic_test/rename_column

@@ -1120,10 +1120,12 @@ subtest referencing_mutations
 statement ok
 CREATE TABLE t_ref (i INT PRIMARY KEY) WITH (schema_locked = false);
 
+onlyif config local-legacy-schema-changer local-mixed-25.2 local-mixed-25.3
 statement error pgcode 0A000 virtual computed column "k" referencing columns \("j"\) added in the current transaction
 ALTER TABLE t_ref ADD COLUMN j INT NOT NULL DEFAULT 42,
    ADD COLUMN k INT AS (i+j) VIRTUAL;
 
+onlyif config local-legacy-schema-changer local-mixed-25.2 local-mixed-25.3
 statement error pgcode 0A000 virtual computed column "l" referencing columns \("j", "k"\) added in the current transaction
 BEGIN TRANSACTION ISOLATION LEVEL SERIALIZABLE;
     SET LOCAL autocommit_before_ddl = false;
@@ -1135,6 +1137,12 @@ COMMIT;
 statement ok
 ROLLBACK;
 
+# The declarative schema changer allows this.
+skipif config local-legacy-schema-changer local-mixed-25.2 local-mixed-25.3 local-mixed-25.4
+statement ok
+ALTER TABLE t_ref ADD COLUMN j INT NOT NULL DEFAULT 42,
+   ADD COLUMN k INT AS (i+j) VIRTUAL;
+
 # Test that adding virtual computed columns to tables which have been created
 # in the current transaction is fine.
 

pkg/sql/logictest/testdata/logic_test/virtual_columns

@@ -567,6 +567,14 @@ func TestAlterTableDMLInjection(t *testing.T) {
 			},
 			schemaChange: "ALTER TABLE tbl DROP CONSTRAINT c, DROP COLUMN i",
 		},
+		{
+			desc: "verify that names can be swapped atomically",
+			setup: []string{
+				"ALTER TABLE tbl ADD COLUMN i INT DEFAULT 11, ADD COLUMN j INT DEFAULT 22",
+			},
+			schemaChange: "ALTER TABLE tbl RENAME COLUMN i TO i_old, RENAME COLUMN j TO i",
+			query:        "SELECT i FROM tbl LIMIT 1",
+		},
 	}
 
 	ctx := context.Background()

pkg/sql/schemachanger/dml_injection_test.go

@@ -50,7 +50,13 @@ func alterTableAddColumn(
 		RequiredPrivilege:   privilege.CREATE,
 	})
 	_, colTargetStatus, col := scpb.FindColumn(elts)
-	columnAlreadyExists := col != nil && colTargetStatus != scpb.ToAbsent
+	_, colNameTargetStatus, colName := scpb.FindColumnName(elts)
+	// A column name already exists if both the Column and ColumnName elements exist
+	// and are not transitioning to ABSENT. When a column is being renamed, the
+	// Column remains but the old ColumnName transitions to ABSENT, freeing up the
+	// name for reuse.
+	columnAlreadyExists := col != nil && colTargetStatus != scpb.ToAbsent &&
+		colName != nil && colNameTargetStatus != scpb.ToAbsent
 	// If the column exists and IF NOT EXISTS is specified, continue parsing
 	// to ensure there are no other errors before treating the operation as a no-op.
 	if columnAlreadyExists && !t.IfNotExists {
@@ -119,6 +125,9 @@ func alterTableAddColumn(
 		unique:  d.Unique.IsUnique,
 		notNull: !desc.Nullable,
 	}
+	if spec.unique {
+		b.IncrementSchemaChangeAddColumnQualificationCounter("unique")
+	}
 
 	idx := cdd.PrimaryKeyOrUniqueIndexDescriptor
 	isRBR := b.QueryByID(tbl.TableID).FilterTableLocalityRegionalByRow().MustGetZeroOrOneElement() != nil

pkg/sql/schemachanger/scbuild/internal/scbuildstmt/alter_table_add_column.go

@@ -6,6 +6,7 @@
 package scbuildstmt
 
 import (
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
 	"github.com/cockroachdb/cockroach/pkg/sql/privilege"
@@ -24,7 +25,7 @@ func alterTableRenameColumn(
 	n *tree.AlterTable,
 	t *tree.AlterTableRenameColumn,
 ) {
-	if len(n.Cmds) > 1 {
+	if len(n.Cmds) > 1 && !b.EvalCtx().Settings.Version.IsActive(b, clusterversion.V26_1) {
 		panic(scerrors.NotImplementedError(n))
 	}
 	alterColumnPreChecks(b, tn, tbl, t.Column)
@@ -123,16 +124,10 @@ func checkNewColumnNameConflicts(b BuildCtx, tableID catid.DescID, oldName, newN
 					"column name %q conflicts with a system column name",
 					tree.ErrString(&newName)))
 			}
-			if current == scpb.Status_PUBLIC {
-				if target == scpb.ToAbsent {
-					panic(pgerror.Newf(pgcode.ObjectNotInPrerequisiteState, "column %q being dropped, try again later", tree.ErrString(&newName)))
-				}
+			if current == scpb.Status_PUBLIC && target != scpb.ToAbsent {
 				tableName := b.QueryByID(tableID).FilterNamespace().MustGetOneElement()
 				panic(sqlerrors.NewColumnAlreadyExistsInRelationError(tree.ErrString(&newName), tableName.Name))
 			}
-			if current == scpb.Status_ABSENT && target == scpb.ToPublic {
-				panic(pgerror.Newf(pgcode.ObjectNotInPrerequisiteState, "column %q being dropped, try again later", tree.ErrString(&newName)))
-			}
 		}
 	})
 }