schemachanger: handle basic TTL params

The builder was updated so that it modifies the scpb.RowLevelTTL element
for when handling TTL storage params. This is needed so that we can do
proper validation on the TTL params, since the validation logic checks
dependencies between different parameters.

The UpsertRowLevelTTL immediate mutation takes care of modifying the
descriptor. Logic was also needed to avoid dropping the TTL schedule if
we are modifying parameters.

Release note: None

Author: rafiss

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

@@ -25,14 +25,120 @@ import (
 // NotImplementedError if the param is not yet supported in the declarative
 // schema changer.
 func validateStorageParamKey(t tree.NodeFormatter, key string) {
-	if strings.HasPrefix(strings.ToLower(key), "ttl") {
+	loweredKey := strings.ToLower(key)
+	// These TTL params still require legacy schema changer because they
+	// affect column management, column dependencies, or schedule management.
+	switch loweredKey {
+	case "ttl", "ttl_expire_after", "ttl_expiration_expression", "ttl_job_cron":
 		panic(scerrors.NotImplementedErrorf(t, redact.Sprintf("%s not implemented yet", redact.SafeString(key))))
 	}
-	if key == catpb.RBRUsingConstraintTableSettingName {
+	if loweredKey == catpb.RBRUsingConstraintTableSettingName {
 		panic(scerrors.NotImplementedErrorf(t, "infer_rbr_region_col_using_constraint not implemented yet"))
 	}
 }
 
+// isTTLParam returns true if this is a TTL storage parameter.
+func isTTLParam(key string) bool {
+	loweredKey := strings.ToLower(key)
+	return strings.HasPrefix(loweredKey, "ttl")
+}
+
+// applyTTLStorageParamsSet processes TTL-related storage parameters for SET.
+// It returns:
+// - origElem: the original RowLevelTTL element (nil if table had no TTL).
+// - newElem: the updated RowLevelTTL element (nil if no TTL params were handled).
+// - changed: true if any TTL param was modified.
+func applyTTLStorageParamsSet(
+	b BuildCtx, tbl *scpb.Table, params tree.StorageParams,
+) (origElem *scpb.RowLevelTTL, newElem *scpb.RowLevelTTL, changed bool) {
+	origElem = b.QueryByID(tbl.TableID).FilterRowLevelTTL().MustGetZeroOrOneElement()
+
+	if len(params) == 0 {
+		return origElem, nil, false
+	}
+
+	// Build new TTL from original (or empty if none).
+	var newTTL catpb.RowLevelTTL
+	var origTTLExpr *scpb.Expression
+	if origElem != nil {
+		newTTL = origElem.RowLevelTTL
+		origTTLExpr = origElem.TTLExpr
+	}
+
+	setter := tablestorageparam.NewTTLSetter(&newTTL, false /* isNewObject */)
+	if err := storageparam.Set(
+		b,
+		b.SemaCtx(),
+		b.EvalCtx(),
+		params,
+		setter,
+	); err != nil {
+		panic(err)
+	}
+
+	// Construct new scpb.RowLevelTTL element with incremented SeqNum.
+	var seqNum uint32
+	if origElem != nil {
+		seqNum = origElem.SeqNum + 1
+	}
+	newElem = &scpb.RowLevelTTL{
+		TableID:     tbl.TableID,
+		RowLevelTTL: newTTL,
+		TTLExpr:     origTTLExpr,
+		SeqNum:      seqNum,
+	}
+	return origElem, newElem, true
+}
+
+// applyTTLStorageParamsReset processes TTL-related storage parameters for RESET.
+// It returns:
+// - origElem: the original RowLevelTTL element (nil if table had no TTL).
+// - newElem: the updated RowLevelTTL element (nil if no TTL params were handled).
+// - changed: true if any TTL param was modified.
+func applyTTLStorageParamsReset(
+	b BuildCtx, tbl *scpb.Table, params []string,
+) (origElem *scpb.RowLevelTTL, newElem *scpb.RowLevelTTL, changed bool) {
+	origElem = b.QueryByID(tbl.TableID).FilterRowLevelTTL().MustGetZeroOrOneElement()
+
+	if len(params) == 0 {
+		return origElem, nil, false
+	}
+
+	// Build new TTL from original (or empty if none).
+	var newTTL *catpb.RowLevelTTL
+	var origTTLExpr *scpb.Expression
+	if origElem != nil {
+		ttlCopy := origElem.RowLevelTTL
+		newTTL = &ttlCopy
+		origTTLExpr = origElem.TTLExpr
+	}
+
+	// Even if origElem is nil, we apply all the resets, since some of them
+	// have side effects like sending notices.
+	setter := tablestorageparam.NewTTLSetter(newTTL, false /* isNewObject */)
+	if err := storageparam.Reset(
+		b,
+		b.EvalCtx(),
+		params,
+		setter,
+	); err != nil {
+		panic(err)
+	}
+
+	if origElem == nil {
+		return nil, nil, false
+	}
+
+	// Construct new scpb.RowLevelTTL element with incremented SeqNum.
+	newElem = &scpb.RowLevelTTL{
+		TableID:     tbl.TableID,
+		RowLevelTTL: *newTTL,
+		TTLExpr:     origTTLExpr,
+		SeqNum:      origElem.SeqNum + 1,
+	}
+	return origElem, newElem, true
+}
+
 // AlterTableSetStorageParams implements ALTER TABLE ... SET {storage_param} in the declarative schema changer.
 func AlterTableSetStorageParams(
 	b BuildCtx,
@@ -41,22 +147,41 @@ func AlterTableSetStorageParams(
 	stmt tree.Statement,
 	t *tree.AlterTableSetStorageParams,
 ) {
+	var ttlParams, otherParams tree.StorageParams
+	for _, param := range t.StorageParams {
+		validateStorageParamKey(t, param.Key)
+		if isTTLParam(param.Key) {
+			ttlParams = append(ttlParams, param)
+		} else {
+			otherParams = append(otherParams, param)
+		}
+	}
+
+	// Handle TTL params first, using the RowLevelTTL element.
+	origTTL, newTTL, ttlChanged := applyTTLStorageParamsSet(b, tbl, ttlParams)
+	if ttlChanged {
+		if origTTL != nil {
+			b.Drop(origTTL)
+		}
+		b.Add(newTTL)
+	}
+
 	if err := storageparam.StorageParamPreChecks(
 		b,
 		b.EvalCtx(),
 		false, /* isNewObject */
-		t.StorageParams,
+		otherParams,
 		nil, /* resetParams */
 	); err != nil {
 		panic(err)
 	}
-	for _, param := range t.StorageParams {
+
+	for _, param := range otherParams {
+		key := param.Key
 		val, err := tablestorageparam.ParseAndValidate(b, b.SemaCtx(), b.EvalCtx(), param)
 		if err != nil {
 			panic(err) // tried to set an invalid value for param
 		}
-		key := param.Key
-		validateStorageParamKey(t, key)
 
 		// schema_locked uses a dedicated TableSchemaLocked element.
 		if key == "schema_locked" {
@@ -130,21 +255,40 @@ func AlterTableResetStorageParams(
 	stmt tree.Statement,
 	t *tree.AlterTableResetStorageParams,
 ) {
+	var ttlParams, otherParams []string
+	for _, param := range t.Params {
+		validateStorageParamKey(t, param)
+		if isTTLParam(param) {
+			ttlParams = append(ttlParams, param)
+		} else {
+			otherParams = append(otherParams, param)
+		}
+	}
+
+	// Handle TTL params first, using the RowLevelTTL element.
+	origTTL, newTTL, ttlChanged := applyTTLStorageParamsReset(b, tbl, ttlParams)
+	if ttlChanged {
+		if origTTL != nil {
+			b.Drop(origTTL)
+		}
+		b.Add(newTTL)
+	}
+
 	if err := storageparam.StorageParamPreChecks(
 		b,
 		b.EvalCtx(),
 		false, /* isNewObject */
 		nil,   /* setParams */
-		t.Params,
+		otherParams,
 	); err != nil {
 		panic(err)
 	}
-	for _, key := range t.Params {
+
+	for _, key := range otherParams {
 		// Validate the key is a known storage parameter.
 		if err := tablestorageparam.IsValidParamKey(key); err != nil {
 			panic(err)
 		}
-		validateStorageParamKey(t, key)
 
 		// Get the reset value for this param. Most of the time this is the
 		// zero value, but for some params it may be different.

pkg/sql/schemachanger/scdecomp/decomp.go

@@ -499,6 +499,10 @@ func (w *walkCtx) walkStorageParams(tbl catalog.TableDescriptor) {
 			// schema_locked is handled separately via the TableSchemaLocked element.
 			continue
 		}
+		if strings.HasPrefix(key, "ttl") {
+			// ttl parameters are handled separately via the RowLevelTTL element.
+			continue
+		}
 		w.ev(scpb.Status_PUBLIC, &scpb.TableStorageParam{
 			TableID: tableID,
 			Name:    key,

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

@@ -9,6 +9,7 @@ import (
 	"context"
 
 	"github.com/cockroachdb/cockroach/pkg/config/zonepb"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scop"
 	"github.com/cockroachdb/cockroach/pkg/sql/storageparam/tablestorageparam"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
@@ -54,6 +55,23 @@ func (i *immediateVisitor) ResetTableStorageParam(
 	return setter.ResetToZeroValue(ctx, op.Param.Name)
 }
 
+func (i *immediateVisitor) UpsertRowLevelTTL(ctx context.Context, op scop.UpsertRowLevelTTL) error {
+	tbl, err := i.checkOutTable(ctx, op.TableID)
+	if err != nil {
+		return err
+	}
+
+	if op.RowLevelTTL == (catpb.RowLevelTTL{}) {
+		tbl.RowLevelTTL = nil
+		return nil
+	}
+
+	// Make a copy of the RowLevelTTL so we can take its address.
+	ttl := op.RowLevelTTL
+	tbl.RowLevelTTL = &ttl
+	return nil
+}
+
 func (d *deferredVisitor) UpdateTTLScheduleMetadata(
 	ctx context.Context, op scop.UpdateTTLScheduleMetadata,
 ) error {

pkg/sql/schemachanger/scop/immediate_mutation.go

@@ -1271,3 +1271,11 @@ type ResetTableStorageParam struct {
 	immediateMutationOp
 	Param scpb.TableStorageParam
 }
+
+// UpsertRowLevelTTL sets the RowLevelTTL on a table descriptor.
+type UpsertRowLevelTTL struct {
+	immediateMutationOp
+	TableID     descpb.ID
+	RowLevelTTL catpb.RowLevelTTL
+	TTLExpr     *scpb.Expression
+}

pkg/sql/schemachanger/scop/immediate_mutation_visitor_generated.go

@@ -770,6 +770,10 @@ message RowLevelTTL {
   // ExpirationExpr is not set. This expression is used to establish column
   // dependencies.
   Expression ttl_expr = 3 [(gogoproto.customname) = "TTLExpr"];
+  // SeqNum differentiates different versions of the TTL configuration for the
+  // same table. When modifying TTL parameters, the old element is dropped and
+  // a new one with an incremented SeqNum is added.
+  uint32 seq_num = 4 [(gogoproto.customname) = "SeqNum"];
 }
 
 message ColumnName {

pkg/sql/schemachanger/scpb/elements.proto

@@ -15,8 +15,12 @@ func init() {
 		toPublic(
 			scpb.Status_ABSENT,
 			to(scpb.Status_PUBLIC,
-				emit(func(this *scpb.RowLevelTTL) *scop.NotImplemented {
-					return notImplemented(this)
+				emit(func(this *scpb.RowLevelTTL, md *opGenContext) *scop.UpsertRowLevelTTL {
+					return &scop.UpsertRowLevelTTL{
+						TableID:     this.TableID,
+						RowLevelTTL: this.RowLevelTTL,
+						TTLExpr:     this.TTLExpr,
+					}
 				}),
 			),
 		),
@@ -25,7 +29,10 @@ func init() {
 			to(scpb.Status_ABSENT,
 				// TODO(postamar): remove revertibility constraint when possible
 				revertible(false),
-				emit(func(this *scpb.RowLevelTTL) *scop.DeleteSchedule {
+				emit(func(this *scpb.RowLevelTTL, md *opGenContext) *scop.DeleteSchedule {
+					if ttlAppliedLater(this, md) {
+						return nil
+					}
 					return &scop.DeleteSchedule{
 						ScheduleID: this.RowLevelTTL.ScheduleID,
 					}
@@ -34,3 +41,26 @@ func init() {
 		),
 	)
 }
+
+// ttlAppliedLater returns true if there is another RowLevelTTL element
+// with the same ScheduleID and TableID, a higher SeqNum, and
+// a target status of PUBLIC. This indicates that the TTL schedule
+// will be applied later, so we should not delete the schedule yet.
+func ttlAppliedLater(this *scpb.RowLevelTTL, md *opGenContext) bool {
+	if this.RowLevelTTL.ScheduleID == 0 {
+		return false
+	}
+	for _, t := range md.Targets {
+		other, ok := t.Element().(*scpb.RowLevelTTL)
+		if !ok || other == this {
+			continue
+		}
+		if other.RowLevelTTL.ScheduleID == this.RowLevelTTL.ScheduleID &&
+			other.TableID == this.TableID &&
+			other.SeqNum > this.SeqNum &&
+			t.TargetStatus == scpb.Status_PUBLIC {
+			return true
+		}
+	}
+	return false
+}

pkg/sql/schemachanger/scplan/internal/opgen/opgen_row_level_ttl.go

@@ -31,4 +31,22 @@ func init() {
 			}
 		},
 	)
+
+	// This rule ensures that when modifying TTL configurations, the old TTL
+	// element (with lower SeqNum) is processed before the new TTL element (with
+	// higher SeqNum).
+	registerDepRule(
+		"ensure row level TTL configs are in increasing seqNum order",
+		scgraph.Precedence,
+		"later-seqNum", "earlier-seqNum",
+		func(from, to NodeVars) rel.Clauses {
+			return rel.Clauses{
+				from.Type((*scpb.RowLevelTTL)(nil)),
+				to.Type((*scpb.RowLevelTTL)(nil)),
+				JoinOnDescID(from, to, "table-id"),
+				FilterElements("SmallerSeqNumFirst", from, to, func(from, to *scpb.RowLevelTTL) bool {
+					return from.SeqNum < to.SeqNum
+				}),
+			}
+		})
 }