sql: alter column identity with the declarative schema changer

Adds support for the `ALTER COLUMN ... [RESTART | SET [CACHE | CYCLE |
... ]` identity operations in the declarative schema changer.

Fixes: #142914
Epic: CRDB-31283

Release note (sql change): The `ALTER COLUMN ...` sequence identity
commands are run by the declarative schema changer.

Release note (bug fix): Simultaneously restarting the identity sequence
and changing the increment sets the correct initial value.

Author: bghal

pkg/sql/alter_sequence.go

@@ -148,7 +148,7 @@ func alterSequenceImpl(
 	//
 	// The code below handles the second case.
 
-	if opts.Increment < 0 && (oldIncrement != seqDesc.SequenceOpts.Increment || oldMinValue != seqDesc.SequenceOpts.MinValue) {
+	if opts.Increment < 0 && (oldIncrement != seqDesc.SequenceOpts.Increment || seqDesc.SequenceOpts.MinValue < oldMinValue) {
 		// Only get the sequence value from KV if it's needed.
 		sequenceVal, err := getSequenceValue()
 		if err != nil {

pkg/sql/catalog/schemaexpr/sequence_options.go

@@ -274,3 +274,21 @@ func AssignSequenceOptions(
 	}
 	return nil
 }
+
+// DefaultSequenceOptions is a helper that returns the default sequence options.
+// It panics on error.
+func DefaultSequenceOptions() descpb.TableDescriptor_SequenceOpts {
+	defaultOpts := descpb.TableDescriptor_SequenceOpts{
+		Increment: 1,
+	}
+	if err := AssignSequenceOptions(&defaultOpts,
+		nil, /* optsNode */
+		64,
+		true, /* setDefaults */
+		nil,  /* existingTypes */
+	); err != nil {
+		panic(err)
+	}
+
+	return defaultOpts
+}

pkg/sql/logictest/testdata/logic_test/alter_table

@@ -4368,16 +4368,20 @@ SET create_table_with_schema_locked=false
 statement ok
 CREATE TABLE t_alter_identity (id SERIAL PRIMARY KEY, a int GENERATED ALWAYS AS IDENTITY, b int GENERATED ALWAYS AS IDENTITY, z int);
 
+skipif config local-mixed-25.4 local-mixed-25.3 local-mixed-25.2
 statement error pq: INCREMENT must not be zero
 ALTER TABLE t_alter_identity ALTER COLUMN b SET INCREMENT 0;
 
+skipif config local-mixed-25.4 local-mixed-25.3 local-mixed-25.2
 query error pq: START value \(11\) cannot be greater than MAXVALUE \(10\)
 ALTER TABLE t_alter_identity ALTER COLUMN b SET MAXVALUE 10 SET START WITH 11;
 
+skipif config local-mixed-25.4 local-mixed-25.3 local-mixed-25.2
 query error pq: START value \(5\) cannot be less than MINVALUE \(10\)
 ALTER TABLE t_alter_identity ALTER COLUMN b SET MINVALUE 10 SET START WITH 5;
 
-query error pq: column "z" of relation "t_alter_identity" is not an identity column
+skipif config local-mixed-25.4 local-mixed-25.3 local-mixed-25.2
+query error pq: cannot alter identity of a non-sequence column "z"
 ALTER TABLE t_alter_identity ALTER COLUMN z SET START WITH 5;
 
 # Verify sequence options are implemented on the identity
@@ -4401,22 +4405,91 @@ statement ok
 INSERT INTO t_alter_identity DEFAULT VALUES;
 
 statement ok
-ALTER TABLE t_alter_identity ALTER COLUMN b SET MAXVALUE 40 RESTART WITH 20 SET CACHE 5 SET INCREMENT BY 2
+ALTER TABLE t_alter_identity ALTER COLUMN b SET MAXVALUE 40 RESTART WITH 20 SET CACHE 5 SET INCREMENT BY 2;
 
 statement ok
 INSERT INTO t_alter_identity DEFAULT VALUES;
 
 statement ok
 INSERT INTO t_alter_identity DEFAULT VALUES;
 
+skipif config local-mixed-25.4 local-mixed-25.3 local-mixed-25.2
 query I rowsort
 SELECT b from t_alter_identity ORDER BY b;
 ----
 1
 5
 9
-18
 20
+22
+
+# TODO(21564): Enable this test when `ALTER COLUMN` on options updates the value of a sequence
+# subtest issue-52552
+# 
+# statement ok
+# CREATE TABLE test_52552_asc (c1 int GENERATED ALWAYS AS IDENTITY);
+# ALTER TABLE test_52552_asc ALTER COLUMN c1 SET INCREMENT 3 SET MINVALUE 1 SET MAXVALUE 12;
+# ALTER TABLE test_52552_asc ALTER COLUMN c1 SET INCREMENT 8 SET MINVALUE 1 SET MAXVALUE 12;
+# INSERT INTO test_52552_asc DEFAULT VALUES;
+# 
+# query I
+# SELECT c1 FROM test_52552_asc
+# ----
+# 8
+# 
+# statement error pq: reached maximum value of sequence "test_52552_asc_c1_seq" \(12\)
+# INSERT INTO test_52552_asc DEFAULT VALUES;
+# 
+# statement error pq: reached maximum value of sequence "test_52552_asc_c1_seq" \(12\)
+# INSERT INTO test_52552_asc DEFAULT VALUES;
+# 
+# statement ok
+# ALTER TABLE test_52552_asc ALTER COLUMN c1 SET NO MAXVALUE;
+# INSERT INTO test_52552_asc DEFAULT VALUES;
+# 
+# query I nosort
+# SELECT c1 FROM test_52552_asc
+# ----
+# 8
+# 32
+# 
+# statement ok
+# CREATE TABLE test_52552_desc (c1 int GENERATED ALWAYS AS IDENTITY);
+# ALTER TABLE test_52552_desc ALTER COLUMN c1 SET INCREMENT -5 SET MINVALUE 1 SET MAXVALUE 12;
+# ALTER TABLE test_52552_desc ALTER COLUMN c1 SET INCREMENT -8 SET MINVALUE 1 SET MAXVALUE 12;
+# INSERT INTO test_52552_desc DEFAULT VALUES;
+# 
+# query I
+# SELECT c1 FROM test_52552_desc
+# ----
+# 8
+# 
+# statement error pq: reached minimum value of sequence "test_52552_desc_seq" \(1\)
+# INSERT INTO test_52552_desc DEFAULT VALUES;
+# 
+# statement error pq: reached minimum value of sequence "test_52552_desc_seq" \(1\)
+# INSERT INTO test_52552_desc DEFAULT VALUES;
+# 
+# statement ok
+# ALTER TABLE test_52552_desc ALTER COLUMN c1 SET NO MINVALUE;
+# INSERT INTO test_52552_desc DEFAULT VALUES;
+# 
+# query I nosort
+# SELECT c1 FROM test_52552_desc
+# ----
+# -4
+# 
+# statement ok
+# CREATE TABLE test_52552_start INCREMENT BY 3 MINVALUE 1 MAXVALUE 100 START 50;
+# ALTER SEQUENCE test_52552_start INCREMENT BY 8;
+# INSERT INTO test_52552_desc DEFAULT VALUES;
+# 
+# query I
+# SELECT nextval('test_52552_start');
+# ----
+# 50
+# 
+# subtest end
 
 statement ok
 CREATE TABLE t_identity_drop (a int GENERATED ALWAYS AS IDENTITY (START WITH 10), b int GENERATED BY DEFAULT AS IDENTITY);

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

@@ -10,6 +10,7 @@ go_library(
         "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_identity.go",
         "alter_table_alter_column_set_default.go",
         "alter_table_alter_column_set_identity.go",
         "alter_table_alter_column_set_not_null.go",

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

@@ -47,6 +47,7 @@ var supportedAlterTableStatements = map[reflect.Type]supportedAlterTableCommand{
 	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},
+	reflect.TypeOf((*tree.AlterTableIdentity)(nil)):           {fn: alterTableAlterColumnIdentity, on: true, checks: isV261Active},
 }
 
 func init() {

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

@@ -293,7 +293,7 @@ func alterTableAddColumn(
 			seqOptions = *ptr
 		}
 		// Versions from 26.1 GeneratedAsIdentity will have a separate element for
-		// GeneratedAsIdentity. Older versios store it in the column element.
+		// GeneratedAsIdentity. Older versions store it in the column element.
 		if spec.colType.ElementCreationMetadata.In_26_1OrLater {
 			spec.generatedAsID = &scpb.ColumnGeneratedAsIdentity{
 				TableID:        tbl.TableID,

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

@@ -0,0 +1,226 @@
+// 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"
+	"strconv"
+	"strings"
+
+	"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.
+	column := mustRetrieveColumnElem(b, tbl.TableID, columnID)
+	panicIfSystemColumn(column, 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)))
+	}
+
+	// Get the current sequence options from elements and defaults.
+	currentOpts := schemaexpr.DefaultSequenceOptions()
+	b.QueryByID(sequenceOwner.SequenceID).FilterSequenceOption().ForEach(func(_ scpb.Status, _ scpb.TargetStatus, e *scpb.SequenceOption) {
+		var err error
+		switch name := e.Key; name {
+		case tree.SeqOptAs:
+			currentOpts.AsIntegerType = e.Value
+		case tree.SeqOptCacheNode:
+			currentOpts.NodeCacheSize, err = strconv.ParseInt(e.Value, 10, 64)
+			if err != nil {
+				panic(pgerror.Wrapf(err, pgcode.Internal, "invalid sequence option value for %q", name))
+			}
+		case tree.SeqOptCacheSession:
+			currentOpts.SessionCacheSize, err = strconv.ParseInt(e.Value, 10, 64)
+			if err != nil {
+				panic(pgerror.Wrapf(err, pgcode.Internal, "invalid sequence option value for %q", name))
+			}
+		case tree.SeqOptIncrement:
+			currentOpts.Increment, err = strconv.ParseInt(e.Value, 10, 64)
+			if err != nil {
+				panic(pgerror.Wrapf(err, pgcode.Internal, "invalid sequence option value for %q", name))
+			}
+		case tree.SeqOptMinValue:
+			currentOpts.MinValue, err = strconv.ParseInt(e.Value, 10, 64)
+			if err != nil {
+				panic(pgerror.Wrapf(err, pgcode.Internal, "invalid sequence option value for %q", name))
+			}
+		case tree.SeqOptMaxValue:
+			currentOpts.MaxValue, err = strconv.ParseInt(e.Value, 10, 64)
+			if err != nil {
+				panic(pgerror.Wrapf(err, pgcode.Internal, "invalid sequence option value for %q", name))
+			}
+		case tree.SeqOptStart:
+			currentOpts.Start, err = strconv.ParseInt(e.Value, 10, 64)
+			if err != nil {
+				panic(pgerror.Wrapf(err, pgcode.Internal, "invalid sequence option value for %q", name))
+			}
+		case tree.SeqOptVirtual:
+			currentOpts.Virtual, err = strconv.ParseBool(e.Value)
+			if err != nil {
+				panic(pgerror.Wrapf(err, pgcode.Internal, "invalid sequence option value for %q", name))
+			}
+		default:
+			panic(pgerror.Newf(pgcode.Internal, "unexpected sequence option %q", name))
+		}
+	})
+
+	// And the final state for the sequence options.
+	updatedOpts := currentOpts
+	if err := schemaexpr.AssignSequenceOptions(&updatedOpts,
+		t.SeqOptions,
+		64,
+		false, /* setDefaults */
+		nil,   /* existingTypes */
+	); err != nil {
+		panic(pgerror.Wrap(
+			err,
+			pgcode.FeatureNotSupported,
+			"", /* message */
+		))
+	}
+
+	defaultOpts := schemaexpr.DefaultSequenceOptions()
+
+	updateElement := 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(_ scpb.Status, _ 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)
+	}
+
+	var updateSequenceValue bool
+	var restartWith int64
+	var useRestartWith bool
+	for _, opt := range t.SeqOptions {
+		switch name := opt.Name; name {
+		case tree.SeqOptAs:
+			updateElement(name, defaultOpts.AsIntegerType, updatedOpts.AsIntegerType)
+		case tree.SeqOptCacheNode:
+			updateElement(name, defaultOpts.NodeCacheSize, updatedOpts.NodeCacheSize)
+		case tree.SeqOptCacheSession:
+			updateElement(name, defaultOpts.SessionCacheSize, updatedOpts.SessionCacheSize)
+		case tree.SeqOptIncrement:
+			updateElement(name, defaultOpts.Increment, updatedOpts.Increment)
+			updateSequenceValue = true
+		case tree.SeqOptMinValue:
+			updateElement(name, defaultOpts.MinValue, updatedOpts.MinValue)
+			updateSequenceValue = true
+		case tree.SeqOptMaxValue:
+			updateElement(name, defaultOpts.MaxValue, updatedOpts.MaxValue)
+			updateSequenceValue = true
+		case tree.SeqOptStart:
+			updateElement(name, defaultOpts.Start, updatedOpts.Start)
+			updateSequenceValue = true
+		case tree.SeqOptVirtual:
+			updateElement(name, defaultOpts.Virtual, updatedOpts.Virtual)
+		case tree.SeqOptRestart:
+			useRestartWith = true
+			if opt.IntVal != nil {
+				restartWith = *opt.IntVal
+			} else {
+				restartWith = updatedOpts.Start
+			}
+		default:
+			panic(fmt.Sprintf("unexpected sequence option: %q", name))
+		}
+	}
+
+	if useRestartWith {
+		restartSeqOption := scpb.SequenceOption{
+			SequenceID: sequenceOwner.SequenceID,
+			Key:        tree.SeqOptRestart,
+			Value:      fmt.Sprintf("%v", restartWith-updatedOpts.Increment),
+		}
+		b.AddTransient(&restartSeqOption)
+	} else if updateSequenceValue {
+		// TODO(21564): Update the sequence value when changes in options expose implementation.
+	}
+
+	var seqOptions tree.SequenceOptions
+	if updatedOpts.SessionCacheSize != defaultOpts.SessionCacheSize {
+		seqOptions = append(seqOptions, tree.SequenceOption{Name: tree.SeqOptCacheSession, IntVal: &updatedOpts.SessionCacheSize})
+	}
+	if updatedOpts.NodeCacheSize != defaultOpts.NodeCacheSize {
+		seqOptions = append(seqOptions, tree.SequenceOption{Name: tree.SeqOptCacheNode, IntVal: &updatedOpts.NodeCacheSize})
+	}
+	if updatedOpts.MinValue != defaultOpts.MinValue {
+		seqOptions = append(seqOptions, tree.SequenceOption{Name: tree.SeqOptMinValue, IntVal: &updatedOpts.MinValue})
+	}
+	if updatedOpts.MaxValue != defaultOpts.MaxValue {
+		seqOptions = append(seqOptions, tree.SequenceOption{Name: tree.SeqOptMaxValue, IntVal: &updatedOpts.MaxValue})
+	}
+	if updatedOpts.Increment != defaultOpts.Increment {
+		seqOptions = append(seqOptions, tree.SequenceOption{Name: tree.SeqOptIncrement, IntVal: &updatedOpts.Increment})
+	}
+	if updatedOpts.Start != defaultOpts.Start {
+		seqOptions = append(seqOptions, tree.SequenceOption{Name: tree.SeqOptStart, IntVal: &updatedOpts.Start})
+	}
+	if updatedOpts.Virtual != defaultOpts.Virtual {
+		seqOptions = append(seqOptions, tree.SequenceOption{Name: tree.SeqOptVirtual})
+	}
+	seqOptionsString := strings.TrimSpace(tree.Serialize(&seqOptions))
+
+	columnType := b.QueryByID(tbl.TableID).FilterColumnType().Filter(func(_ scpb.Status, _ scpb.TargetStatus, e *scpb.ColumnType) bool {
+		return e.ColumnID == sequenceOwner.ColumnID
+	}).MustGetOneElement()
+
+	if columnType.ElementCreationMetadata.In_26_1OrLater {
+		oldColumnGeneratedAsIdentity := b.QueryByID(tbl.TableID).FilterColumnGeneratedAsIdentity().Filter(func(_ scpb.Status, _ scpb.TargetStatus, e *scpb.ColumnGeneratedAsIdentity) bool {
+			return e.ColumnID == sequenceOwner.ColumnID
+		}).MustGetOneElement()
+
+		if oldColumnGeneratedAsIdentity.SequenceOption != seqOptionsString {
+			b.Drop(oldColumnGeneratedAsIdentity)
+			newColumnGeneratedAsIdentity := *oldColumnGeneratedAsIdentity
+			newColumnGeneratedAsIdentity.SequenceOption = seqOptionsString
+			b.Add(&newColumnGeneratedAsIdentity)
+		}
+	} else {
+		column.GeneratedAsIdentitySequenceOption = seqOptionsString
+		if column.GeneratedAsIdentitySequenceOption != seqOptionsString {
+			b.Drop(column)
+			newColumn := *column
+			newColumn.GeneratedAsIdentitySequenceOption = seqOptionsString
+			b.Add(&newColumn)
+		}
+	}
+}