sql: add column identity sequence options to schema changer

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

@@ -9,6 +9,7 @@ go_library(
         "alter_table_add_constraint.go",
         "alter_table_alter_column_drop_not_null.go",
         "alter_table_alter_column_drop_stored.go",
+        "alter_table_alter_column_identity.go",
         "alter_table_alter_column_set_default.go",
         "alter_table_alter_column_set_not_null.go",
         "alter_table_alter_column_set_on_update.go",

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

@@ -45,6 +45,7 @@ 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.AlterTableIdentity)(nil)):           {fn: alterTableAlterColumnIdentity, on: true, checks: isV261Active},
 }
 
 func init() {

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

@@ -0,0 +1,97 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package scbuildstmt
+
+import (
+	"fmt"
+	"reflect"
+
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/schemaexpr"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
+	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+)
+
+func alterTableAlterColumnIdentity(
+	b BuildCtx, tn *tree.TableName, tbl *scpb.Table, stmt tree.Statement, t *tree.AlterTableIdentity,
+) {
+	alterColumnPreChecks(b, tn, tbl, t.Column)
+
+	columnID := getColumnIDFromColumnName(b, tbl.TableID, t.Column, true /* required */)
+	// Block alters on system columns.
+	panicIfSystemColumn(mustRetrieveColumnElem(b, tbl.TableID, columnID), t.Column)
+
+	sequenceOwner := b.QueryByID(tbl.TableID).FilterSequenceOwner().Filter(func(_ scpb.Status, _ scpb.TargetStatus, e *scpb.SequenceOwner) bool {
+		return e.TableID == tbl.TableID && e.ColumnID == columnID
+	}).MustGetZeroOrOneElement()
+
+	if sequenceOwner == nil {
+		panic(pgerror.Newf(
+			pgcode.FeatureNotSupported,
+			"cannot alter identity of a non-sequence column %q", tree.ErrString(&t.Column)))
+	}
+
+	newOpts := descpb.TableDescriptor_SequenceOpts{
+		Increment: 1,
+	}
+	if err := schemaexpr.AssignSequenceOptions(&newOpts,
+		t.SeqOptions,
+		64,
+		true,
+		nil,
+	); err != nil {
+		panic(err)
+	}
+
+	defaultOpts := descpb.TableDescriptor_SequenceOpts{
+		Increment: 1,
+	}
+	if err := schemaexpr.AssignSequenceOptions(&defaultOpts,
+		nil,
+		64,
+		true,
+		nil,
+	); err != nil {
+		panic(err)
+	}
+
+	updateSequenceOption := func(key string, defaultValue, value interface{}) {
+		newSeqOption := scpb.SequenceOption{
+			SequenceID: sequenceOwner.SequenceID,
+			Key:        key,
+			Value:      fmt.Sprintf("%v", value),
+		}
+
+		oldSeqOption := b.QueryByID(sequenceOwner.SequenceID).FilterSequenceOption().Filter(func(current scpb.Status, target scpb.TargetStatus, e *scpb.SequenceOption) bool {
+			return e.Key == key
+		}).MustGetZeroOrOneElement()
+		if oldSeqOption != nil {
+			// Skip a noop update.
+			if oldSeqOption.Value == newSeqOption.Value {
+				return
+			}
+			b.Drop(oldSeqOption)
+		}
+
+		// Skip setting to default values.
+		if reflect.DeepEqual(defaultValue, value) {
+			return
+		}
+
+		b.Add(&newSeqOption)
+	}
+
+	updateSequenceOption(tree.SeqOptIncrement, defaultOpts.Increment, newOpts.Increment)
+	// updateSequenceOption(tree.SeqOptMinValue, defaultOpts.MinValue, newOpts.MinValue)
+	// updateSequenceOption(tree.SeqOptMaxValue, defaultOpts.MaxValue, newOpts.MaxValue)
+	// updateSequenceOption(tree.SeqOptStart, defaultOpts.Start, newOpts.Start)
+	// updateSequenceOption(tree.SeqOptVirtual, defaultOpts.Virtual, newOpts.Virtual)
+	// updateSequenceOption(tree.SeqOptCacheSession, defaultOpts.SessionCacheSize, newOpts.SessionCacheSize)
+	// updateSequenceOption(tree.SeqOptCacheNode, defaultOpts.NodeCacheSize, newOpts.NodeCacheSize)
+	// updateSequenceOption(tree.SeqOptAs, defaultOpts.AsIntegerType, newOpts.AsIntegerType)
+}

pkg/sql/schemachanger/scexec/scmutationexec/sequence.go

@@ -48,9 +48,7 @@ func (i *immediateVisitor) CreateSequenceDescriptor(
 	return nil
 }
 
-func (i *immediateVisitor) SetSequenceOptions(
-	ctx context.Context, op scop.SetSequenceOptions,
-) error {
+func (i *immediateVisitor) SetSequenceOption(ctx context.Context, op scop.SetSequenceOption) error {
 	sc, err := i.checkOutTable(ctx, op.SequenceID)
 	if err != nil {
 		return err
@@ -88,6 +86,13 @@ func (i *immediateVisitor) SetSequenceOptions(
 	return sequenceOptionMeta[op.Key].SetFunc(op.Value)
 }
 
+// TODO: implement
+func (i *immediateVisitor) UnsetSequenceOption(
+	ctx context.Context, op scop.UnsetSequenceOption,
+) error {
+	panic("unimplemented")
+}
+
 func (i *immediateVisitor) InitSequence(ctx context.Context, op scop.InitSequence) error {
 	sc, err := i.checkOutTable(ctx, op.SequenceID)
 	if err != nil {

pkg/sql/schemachanger/scop/immediate_mutation.go

@@ -1085,13 +1085,19 @@ type CreateSequenceDescriptor struct {
 	Temporary  bool
 }
 
-type SetSequenceOptions struct {
+type SetSequenceOption struct {
 	immediateMutationOp
 	SequenceID descpb.ID
 	Key        string
 	Value      string
 }
 
+type UnsetSequenceOption struct {
+	immediateMutationOp
+	SequenceID descpb.ID
+	Key        string
+}
+
 type InitSequence struct {
 	immediateMutationOp
 	SequenceID     descpb.ID