sql: add ALTER TABLE ... SET/ADD IDENTITY to the declarative schema changer

This change adds support for `ALTER TABLE ... SET/ADD IDENTITY` to the
declarative schema changer. This schema change operation runs using the
legacy schema changer in versions older than 26.1.

Epic: CRDB-31283

Fixes #142918

Release note (sql change): `ALTER TABLE ... SET/ADD GENERATED AS IDENTITY`
is supported by the declerative schema changer in 26.1 or later.

Author: shghasemi

pkg/ccl/schemachangerccl/sctestbackupccl/backup_base_generated_test.go

@@ -7,9 +7,11 @@ go_library(
         "alter_table.go",
         "alter_table_add_column.go",
         "alter_table_add_constraint.go",
+        "alter_table_alter_column_add_identity.go",
         "alter_table_alter_column_drop_not_null.go",
         "alter_table_alter_column_drop_stored.go",
         "alter_table_alter_column_set_default.go",
+        "alter_table_alter_column_set_identity.go",
         "alter_table_alter_column_set_not_null.go",
         "alter_table_alter_column_set_on_update.go",
         "alter_table_alter_column_type.go",

pkg/sql/schemachanger/scbuild/internal/scbuildstmt/BUILD.bazel

@@ -45,6 +45,8 @@ var supportedAlterTableStatements = map[reflect.Type]supportedAlterTableCommand{
 	reflect.TypeOf((*tree.AlterTableRenameColumn)(nil)):       {fn: alterTableRenameColumn, on: true, checks: isV254Active},
 	reflect.TypeOf((*tree.AlterTableDropStored)(nil)):         {fn: alterTableDropStored, on: true, checks: isV261Active},
 	reflect.TypeOf((*tree.AlterTableRenameConstraint)(nil)):   {fn: alterTableRenameConstraint, on: true, checks: isV261Active},
+	reflect.TypeOf((*tree.AlterTableSetIdentity)(nil)):        {fn: alterTableSetIdentity, on: true, checks: isV261Active},
+	reflect.TypeOf((*tree.AlterTableAddIdentity)(nil)):        {fn: alterTableAddIdentity, on: true, checks: isV261Active},
 }
 
 func init() {

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

@@ -399,27 +399,17 @@ func alterTableAddColumnSerialOrGeneratedIdentity(
 		return catalog.UseUnorderedRowID(*d), nil
 	}
 
-	// Start with a fixed sequence number and find the first one
-	// that is free.
-	nameBase := tree.Name(tn.Table() + "_" + string(d.Name) + "_seq")
-	seqName := tree.NewTableNameWithSchema(
-		tn.CatalogName,
-		tn.SchemaName,
-		nameBase)
+	return alterTableCreateColumnSequence(b, d, tn, serialNormalizationMode, defType)
+}
 
-	for idx := 0; ; idx++ {
-		ers := b.ResolveRelation(seqName.ToUnresolvedObjectName(),
-			ResolveParams{
-				IsExistenceOptional: true,
-				RequiredPrivilege:   privilege.USAGE,
-				WithOffline:         true, // We search sequence with provided name, including offline ones.
-				ResolveTypes:        true, // Check for collisions with type names.
-			})
-		if ers.IsEmpty() {
-			break
-		}
-		seqName.ObjectName = tree.Name(fmt.Sprintf("%s%d", nameBase, idx))
-	}
+func alterTableCreateColumnSequence(
+	b BuildCtx,
+	d *tree.ColumnTableDef,
+	tn *tree.TableName,
+	serialNormalizationMode sessiondatapb.SerialNormalizationMode,
+	defType *types.T,
+) (newDef *tree.ColumnTableDef, colDefaultExpression *scpb.Expression) {
+	seqName := getNextAvailableSeqName(b, d.Name, tn)
 
 	seqOptions, err := catalog.SequenceOptionsFromNormalizationMode(serialNormalizationMode, b.ClusterSettings(), d, defType)
 	if err != nil {
@@ -452,6 +442,31 @@ func alterTableAddColumnSerialOrGeneratedIdentity(
 	}
 }
 
+func getNextAvailableSeqName(b BuildCtx, colName tree.Name, tn *tree.TableName) *tree.TableName {
+	// Start with a fixed sequence number and find the first one
+	// that is free.
+	nameBase := tree.Name(tn.Table() + "_" + string(colName) + "_seq")
+	seqName := tree.NewTableNameWithSchema(
+		tn.CatalogName,
+		tn.SchemaName,
+		nameBase)
+
+	for idx := 0; ; idx++ {
+		ers := b.ResolveRelation(seqName.ToUnresolvedObjectName(),
+			ResolveParams{
+				IsExistenceOptional: true,
+				RequiredPrivilege:   privilege.USAGE,
+				WithOffline:         true, // We search sequence with provided name, including offline ones.
+				ResolveTypes:        true, // Check for collisions with type names.
+			})
+		if ers.IsEmpty() {
+			break
+		}
+		seqName.ObjectName = tree.Name(fmt.Sprintf("%s%d", nameBase, idx))
+	}
+	return seqName
+}
+
 func columnNamesToIDs(b BuildCtx, tbl *scpb.Table) map[string]descpb.ColumnID {
 	tableElts := b.QueryByID(tbl.TableID)
 	namesToIDs := make(map[string]descpb.ColumnID)

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

@@ -0,0 +1,126 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package scbuildstmt
+
+import (
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
+	"github.com/cockroachdb/cockroach/pkg/sql/privilege"
+	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/sql/sessiondatapb"
+	"github.com/cockroachdb/cockroach/pkg/sql/types"
+)
+
+func alterTableAddIdentity(
+	b BuildCtx,
+	tn *tree.TableName,
+	tbl *scpb.Table,
+	stmt tree.Statement,
+	t *tree.AlterTableAddIdentity,
+) {
+	alterColumnPreChecks(b, tn, tbl, t.Column)
+	colElems := b.ResolveColumn(tbl.TableID, t.Column, ResolveParams{
+		RequiredPrivilege: privilege.CREATE,
+	})
+	colElem := colElems.FilterColumn().MustGetOneElement()
+	columnID := colElem.ColumnID
+	// Block alters on system columns.
+	panicIfSystemColumn(colElem, t.Column)
+	colTypeElem := mustRetrieveColumnTypeElem(b, tbl.TableID, columnID)
+	// Ensure that column is an integer
+	if colTypeElem.Type.InternalType.Family != types.IntFamily {
+		panic(pgerror.Newf(
+			pgcode.InvalidParameterValue,
+			"column %q of relation %q type must be an integer type", t.Column, tn.ObjectName))
+	}
+	// Ensure that column is not already an identity column
+	if isColumnGeneratedAsIdentity(b, tbl.TableID, columnID) {
+		panic(pgerror.Newf(
+			pgcode.ObjectNotInPrerequisiteState,
+			"column %q of relation %q is already an identity column", t.Column, tn.ObjectName))
+	}
+	// Ensure that column does not have a default expression
+	defaultExpr := retrieveColumnDefaultExpressionElem(b, tbl.TableID, columnID)
+	if defaultExpr != nil {
+		panic(pgerror.Newf(
+			pgcode.ObjectNotInPrerequisiteState,
+			"column %q of relation %q already has a default value", t.Column, tn.ObjectName))
+	}
+	// Ensure that column does not have a compute expression
+	computeExpr := retrieveColumnComputeExpression(b, tbl.TableID, columnID)
+	if computeExpr != nil {
+		panic(pgerror.Newf(
+			pgcode.ObjectNotInPrerequisiteState,
+			"column %q of relation %q already has a computed value", t.Column, tn.ObjectName))
+	}
+	// Ensure that column is declared as not null
+	columNotNull := b.QueryByID(tbl.TableID).FilterColumnNotNull().Filter(
+		func(current scpb.Status, target scpb.TargetStatus, e *scpb.ColumnNotNull) bool {
+			return e.ColumnID == columnID
+		}).MustGetZeroOrOneElement()
+	if columNotNull == nil {
+		panic(pgerror.Newf(
+			pgcode.ObjectNotInPrerequisiteState,
+			"column %q of relation %q must be declared NOT NULL before identity can be added",
+			t.Column, tn.ObjectName))
+	}
+	// Ensure that column does not have a on update expression
+	onUpdate := retrieveColumnOnUpdateExpressionElem(b, tbl.TableID, columnID)
+	if onUpdate != nil {
+		panic(pgerror.Newf(
+			pgcode.ObjectNotInPrerequisiteState,
+			"column %q of relation %q already has an update expression", t.Column, tn.ObjectName))
+	}
+	// Create column definition for identity column
+	q := []tree.NamedColumnQualification{{Qualification: t.Qualification}}
+	colDef, err := tree.NewColumnTableDef(t.Column, colTypeElem.Type, false /* isSerial */, q)
+	if err != nil {
+		panic(err)
+	}
+	// Step 1: Get type and sequence options
+	colDef.GeneratedIdentity.IsGeneratedAsIdentity = true
+	var identityType catpb.GeneratedAsIdentityType
+	switch q := (t.Qualification).(type) {
+	case *tree.GeneratedAlwaysAsIdentity:
+		identityType = catpb.GeneratedAsIdentityType_GENERATED_ALWAYS
+		colDef.GeneratedIdentity.GeneratedAsIdentityType = tree.GeneratedAlways
+		colDef.GeneratedIdentity.SeqOptions = q.SeqOptions
+	case *tree.GeneratedByDefAsIdentity:
+		identityType = catpb.GeneratedAsIdentityType_GENERATED_BY_DEFAULT
+		colDef.GeneratedIdentity.GeneratedAsIdentityType = tree.GeneratedByDefault
+		colDef.GeneratedIdentity.SeqOptions = q.SeqOptions
+	}
+	// Step 2. create a sequence and default expression
+	serialNormalizationMode := sessiondatapb.SerialUsesSQLSequences
+	colDef, expr := alterTableCreateColumnSequence(b, colDef, tn, serialNormalizationMode, colTypeElem.Type)
+	if expr == nil || len(expr.UsesSequenceIDs) != 1 {
+		panic(pgerror.Newf(
+			pgcode.AssertFailure,
+			"failed to create sequence for new identity column %q of relation %q",
+			t.Column, tn.ObjectName))
+	}
+	// Step 3. add the sequence owner
+	b.Add(&scpb.SequenceOwner{
+		SequenceID: expr.UsesSequenceIDs[0],
+		TableID:    tbl.TableID,
+		ColumnID:   columnID,
+	})
+	// Step 4. add the default expression
+	b.Add(&scpb.ColumnDefaultExpression{
+		TableID:    tbl.TableID,
+		ColumnID:   columnID,
+		Expression: *expr,
+	})
+	// Step 5. add the GeneratedAsIdentity element
+	b.Add(&scpb.ColumnGeneratedAsIdentity{
+		TableID:        tbl.TableID,
+		ColumnID:       columnID,
+		Type:           identityType,
+		SequenceOption: tree.Serialize(&colDef.GeneratedIdentity.SeqOptions),
+	})
+}

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

@@ -0,0 +1,53 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package scbuildstmt
+
+import (
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
+	"github.com/cockroachdb/cockroach/pkg/sql/privilege"
+	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+)
+
+func alterTableSetIdentity(
+	b BuildCtx,
+	tn *tree.TableName,
+	tbl *scpb.Table,
+	stmt tree.Statement,
+	t *tree.AlterTableSetIdentity,
+) {
+	alterColumnPreChecks(b, tn, tbl, t.Column)
+	colElem := b.ResolveColumn(tbl.TableID, t.Column, ResolveParams{
+		RequiredPrivilege: privilege.CREATE,
+	}).FilterColumn().MustGetOneElement()
+	columnID := colElem.ColumnID
+	// Block alters on system columns.
+	panicIfSystemColumn(colElem, t.Column)
+	oldIdentityElem := retrieveColumnGeneratedAsIdentityElem(b, tbl.TableID, columnID)
+	// Ensure that column is an identity column
+	if oldIdentityElem == nil {
+		panic(pgerror.Newf(
+			pgcode.ObjectNotInPrerequisiteState,
+			"column %q of relation %q is not an identity column", t.Column, tn.ObjectName))
+	}
+	// Create a copy of the old element and update the type if needed
+	newIdentityElem := *oldIdentityElem
+	switch t.GeneratedAsIdentityType {
+	case tree.GeneratedAlways:
+		newIdentityElem.Type = catpb.GeneratedAsIdentityType_GENERATED_ALWAYS
+	case tree.GeneratedByDefault:
+		newIdentityElem.Type = catpb.GeneratedAsIdentityType_GENERATED_BY_DEFAULT
+	}
+	// Skip operation if the new type is the same as the old one
+	if newIdentityElem.Type == oldIdentityElem.Type {
+		return
+	}
+	// Replace the old element with the new one
+	b.Drop(oldIdentityElem)
+	b.Add(&newIdentityElem)
+}