schemachanger: handle manual schema_locked setting in declarative

This storage parameter is special, so this requires logic to use the
pre-existing TableSchemaLocked element rather than the normal storage
param element.

As part of this, we partially revert
bc3f228 and refactor the check to be a
little simpler since we don't actually need to handle the mixed version
case now.

Release note: None

Author: rafiss

pkg/sql/alter_table.go

@@ -2519,10 +2519,8 @@ func (p *planner) checkSchemaChangeIsAllowed(
 		return nil
 	}
 	// Check if this schema change is on the allowed list, which will only
-	// be simple non-back filling schema changes. All commands except set/reset
-	// schema_locked are unsupported before 25.2
-	preventedBySchemaLocked := !p.ExecCfg().Settings.Version.IsActive(ctx, clusterversion.V25_3) &&
-		!tree.IsSetOrResetSchemaLocked(n)
+	// be simple non-back filling schema changes.
+	preventedBySchemaLocked := false
 	// These schema changes are allowed because the events generated will always
 	// be ignored by the schema_locked. The tableEventFilter (in CDC schemafeed)
 	// only cares about a limited number of events:

pkg/sql/logictest/testdata/logic_test/distsql_stats

@@ -1900,14 +1900,14 @@ ALTER TABLE greeting_stats INJECT STATISTICS '$stats'
 query T
 SELECT feature_name FROM crdb_internal.feature_usage
 WHERE feature_name in ('job.typedesc_schema_change.successful',
-'job.schema_change.successful',
+'job.new_schema_change.successful',
 'job.create_stats.successful',
 'job.auto_create_stats.successful') AND
 usage_count > 0
 ORDER BY feature_name DESC
 ----
 job.typedesc_schema_change.successful
-job.schema_change.successful
+job.new_schema_change.successful
 job.create_stats.successful
 job.auto_create_stats.successful
 

pkg/sql/logictest/testdata/logic_test/schema_locked

@@ -174,6 +174,71 @@ ALTER TABLE ref CONFIGURE ZONE USING num_replicas = 11;
 
 subtest end
 
+# Test SET/RESET schema_locked with varying session settings.
+subtest set_reset_schema_locked_session_settings
+
+statement ok
+CREATE TABLE t_sl (i INT PRIMARY KEY) WITH (schema_locked = false)
+
+# Test SET schema_locked=true
+statement ok
+ALTER TABLE t_sl SET (schema_locked=true)
+
+query TT
+SHOW CREATE TABLE t_sl
+----
+t_sl  CREATE TABLE public.t_sl (
+        i INT8 NOT NULL,
+        CONSTRAINT t_sl_pkey PRIMARY KEY (i ASC)
+      ) WITH (schema_locked = true);
+
+# Test SET schema_locked=false
+statement ok
+ALTER TABLE t_sl SET (schema_locked=false)
+
+query TT
+SHOW CREATE TABLE t_sl
+----
+t_sl  CREATE TABLE public.t_sl (
+        i INT8 NOT NULL,
+        CONSTRAINT t_sl_pkey PRIMARY KEY (i ASC)
+      );
+
+statement ok
+SET create_table_with_schema_locked = 'true'
+
+# Test RESET schema_locked - should reset to session setting (true).
+statement ok
+ALTER TABLE t_sl RESET (schema_locked)
+
+query TT
+SHOW CREATE TABLE t_sl
+----
+t_sl  CREATE TABLE public.t_sl (
+        i INT8 NOT NULL,
+        CONSTRAINT t_sl_pkey PRIMARY KEY (i ASC)
+      ) WITH (schema_locked = true);
+
+statement ok
+SET create_table_with_schema_locked = 'false'
+
+# Test RESET schema_locked - should reset to session setting (false).
+statement ok
+ALTER TABLE t_sl RESET (schema_locked)
+
+query TT
+SHOW CREATE TABLE t_sl
+----
+t_sl  CREATE TABLE public.t_sl (
+        i INT8 NOT NULL,
+        CONSTRAINT t_sl_pkey PRIMARY KEY (i ASC)
+      );
+
+statement ok
+DROP TABLE t_sl
+
+subtest end
+
 
 # Validate schema_locked can be unset properly in add column txns.
 subtest regression_147993
@@ -228,15 +293,18 @@ subtest regression_150484
 statement ok
 CREATE TABLE t_150484 (n INT, m INT, FAMILY (n, m)) WITH (schema_locked=true);
 
-# Combining SET (schema_locked=false) with DROP COLUMN should fail.
+# Combining SET (schema_locked=false) with DROP COLUMN should fail in the legacy schema changer.
+onlyif config local-legacy-schema-changer
 statement error pgcode 57000 pq: this schema change is disallowed because table "t_150484" is locked
 ALTER TABLE t_150484 SET (schema_locked=false), DROP COLUMN n;
 
-# Combining RESET (schema_locked) with ADD COLUMN should fail.
+# Combining RESET (schema_locked) with ADD COLUMN should fail in the legacy schema changer.
+onlyif config local-legacy-schema-changer
 statement error pgcode 57000 pq: this schema change is disallowed because table "t_150484" is locked
 ALTER TABLE t_150484 RESET (schema_locked), ADD COLUMN k INT;
 
-# Verify the table is still locked and unchanged.
+# Verify the table is still locked and unchanged in the legacy schema changer.
+onlyif config local-legacy-schema-changer
 query TT
 SHOW CREATE TABLE t_150484
 ----
@@ -248,13 +316,42 @@ t_150484  CREATE TABLE public.t_150484 (
             FAMILY fam_0_n_m_rowid (n, m, rowid)
           ) WITH (schema_locked = true);
 
-# Confirm that SET (schema_locked=false) alone still works.
+# In the declarative schema changer, the above statements are allowed, as we
+# take care to always unlock the table first.
+skipif config local-legacy-schema-changer local-mixed-25.4
 statement ok
-ALTER TABLE t_150484 SET (schema_locked=false);
+ALTER TABLE t_150484 SET (schema_locked=false), DROP COLUMN n;
 
-# Now schema changes should work.
+skipif config local-legacy-schema-changer local-mixed-25.4
+query TT
+SHOW CREATE TABLE t_150484
+----
+t_150484  CREATE TABLE public.t_150484 (
+            m INT8 NULL,
+            rowid INT8 NOT VISIBLE NOT NULL DEFAULT unique_rowid(),
+            CONSTRAINT t_150484_pkey PRIMARY KEY (rowid ASC),
+            FAMILY fam_0_n_m_rowid (m, rowid)
+          );
+
+skipif config local-legacy-schema-changer local-mixed-25.4
+statement ok
+ALTER TABLE t_150484 SET (schema_locked=true)
+
+skipif config local-legacy-schema-changer local-mixed-25.4
 statement ok
-ALTER TABLE t_150484 DROP COLUMN n;
+ALTER TABLE t_150484 RESET (schema_locked), ADD COLUMN k INT;
+
+skipif config local-legacy-schema-changer local-mixed-25.4
+query TT
+SHOW CREATE TABLE t_150484
+----
+t_150484  CREATE TABLE public.t_150484 (
+            m INT8 NULL,
+            rowid INT8 NOT VISIBLE NOT NULL DEFAULT unique_rowid(),
+            k INT8 NULL,
+            CONSTRAINT t_150484_pkey PRIMARY KEY (rowid ASC),
+            FAMILY fam_0_n_m_rowid (m, rowid, k)
+          );
 
 statement ok
 DROP TABLE t_150484;

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

@@ -6,6 +6,7 @@
 package scbuildstmt
 
 import (
+	"strconv"
 	"strings"
 
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catpb"
@@ -30,10 +31,6 @@ func validateStorageParamKey(t tree.NodeFormatter, key string) {
 	if key == catpb.RBRUsingConstraintTableSettingName {
 		panic(scerrors.NotImplementedErrorf(t, "infer_rbr_region_col_using_constraint not implemented yet"))
 	}
-	if key == "schema_locked" {
-		// schema_locked has a dedicated element and will be handled differently.
-		panic(scerrors.NotImplementedErrorf(t, "schema_locked not implemented yet"))
-	}
 }
 
 // AlterTableSetStorageParams implements ALTER TABLE ... SET {storage_param} in the declarative schema changer.
@@ -60,6 +57,13 @@ func AlterTableSetStorageParams(
 		}
 		key := param.Key
 		validateStorageParamKey(t, key)
+
+		// schema_locked uses a dedicated TableSchemaLocked element.
+		if key == "schema_locked" {
+			setSchemaLocked(b, tbl, val)
+			continue
+		}
+
 		// Do extra validation for exclude_data_from_backup
 		validateExcludeDataFromBackup(b, tbl, key)
 		currElem := b.QueryByID(tbl.TableID).FilterTableStorageParam().Filter(
@@ -149,6 +153,12 @@ func AlterTableResetStorageParams(
 			panic(err)
 		}
 
+		// schema_locked uses a dedicated TableSchemaLocked element.
+		if key == "schema_locked" {
+			setSchemaLocked(b, tbl, resetVal)
+			continue
+		}
+
 		// Find and drop the existing element.
 		currElem := b.QueryByID(tbl.TableID).FilterTableStorageParam().Filter(
 			func(_ scpb.Status, _ scpb.TargetStatus, e *scpb.TableStorageParam) bool {
@@ -170,3 +180,18 @@ func AlterTableResetStorageParams(
 		}
 	}
 }
+
+// setSchemaLocked sets the schema_locked storage parameter using the dedicated
+// TableSchemaLocked element. The val is parsed as a boolean; empty string is
+// treated as false.
+func setSchemaLocked(b BuildCtx, tbl *scpb.Table, val string) {
+	locked, _ := strconv.ParseBool(val)
+	currElem := b.QueryByID(tbl.TableID).FilterTableSchemaLocked().MustGetZeroOrOneElement()
+	if locked {
+		// Setting schema_locked=true means TableSchemaLocked should be PUBLIC.
+		b.Add(&scpb.TableSchemaLocked{TableID: tbl.TableID})
+	} else if currElem != nil {
+		// Setting schema_locked=false means TableSchemaLocked should be ABSENT.
+		b.Drop(currElem)
+	}
+}

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

@@ -10,7 +10,6 @@ import (
 	"sort"
 
 	"github.com/cockroachdb/cockroach/pkg/build"
-	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catenumpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catpb"
@@ -1191,13 +1190,10 @@ func checkTableSchemaChangePrerequisites(
 	schemaLocked := tableElements.FilterTableSchemaLocked().MustGetZeroOrOneElement()
 	// No-op by default unless schema_locked has been setup.
 	maybeCleanupSchemaLockedFn = func() {}
-	if schemaLocked != nil && !tree.IsSetOrResetSchemaLocked(n) {
-		// Before 25.2 we don't support auto-unsetting schema locked.
-		if !b.ClusterSettings().Version.IsActive(b, clusterversion.V25_2) {
-			ns := tableElements.FilterNamespace().MustGetOneElement()
-			panic(sqlerrors.NewSchemaChangeOnLockedTableErr(ns.Name))
-		}
-		// Unset schema_locked for the user.
+	hasSchemaLockedChange := tree.HasSetOrResetSchemaLocked(n)
+	if schemaLocked != nil && !hasSchemaLockedChange {
+		// If the user is not explicitly setting schema_locked, then we should
+		// use the auto-unset logic.
 		b.DropTransient(schemaLocked)
 		maybeCleanupSchemaLockedFn = func() {
 			maybeCleanupSchemaLocked(b, tableElements.FilterTable().MustGetOneElement().TableID)

pkg/sql/sem/tree/schema_helpers.go

@@ -7,34 +7,26 @@ package tree
 
 import "slices"
 
-// IsSetOrResetSchemaLocked returns true if `n` contains only commands to
-// set/reset storage parameters including "schema_locked", with no other
-// schema-changing commands. This prevents bypassing schema_locked protection
-// by combining SET (schema_locked=false) with other operations like DROP
-// COLUMN in the same ALTER TABLE statement.
-func IsSetOrResetSchemaLocked(n Statement) bool {
+// HasSetOrResetSchemaLocked checks if the statement contains a command to
+// set/reset the "schema_locked" storage parameter.
+func HasSetOrResetSchemaLocked(n Statement) (hasSchemaLockedChange bool) {
 	alterStmt, ok := n.(*AlterTable)
 	if !ok {
 		return false
 	}
-	hasSchemaLockedChange := false
 	for _, cmd := range alterStmt.Cmds {
 		switch cmd := cmd.(type) {
 		case *AlterTableSetStorageParams:
 			if cmd.StorageParams.GetVal("schema_locked") != nil {
-				hasSchemaLockedChange = true
+				return true
 			}
 		case *AlterTableResetStorageParams:
 			if slices.Contains(cmd.Params, "schema_locked") {
-				hasSchemaLockedChange = true
+				return true
 			}
-		default:
-			// Any other command type means this is not a pure storage parameter
-			// change, so we should not allow bypassing schema_locked checks.
-			return false
 		}
 	}
-	return hasSchemaLockedChange
+	return false
 }
 
 // IsAllowedLDRSchemaChange returns true if the schema change statement is