schemachanger: support ttl_job_cron parameter in declarative schema changer

Previously, modifying the ttl_job_cron storage parameter required falling
back to the legacy schema changer because the declarative schema changer
had no mechanism to update the TTL scheduled job's cron expression.

This commit adds support for ttl_job_cron by:

1. Adding UpdateTTLScheduleCron as a new deferred mutation operation that
    updates the cron expression on the existing TTL scheduled job.

2. Implementing findOldTTLCron in opgen to detect when the cron expression
    has changed by comparing the new RowLevelTTL element against the one
    being dropped, emitting UpdateTTLScheduleCron only when necessary.

3. Adding UpdateTTLScheduleCron to metadataUpdater which loads the
    scheduled job, updates its cron expression, and persists the change.

Release note: None

Author: rafiss

pkg/sql/descmetadata/BUILD.bazel

@@ -6,8 +6,10 @@ go_library(
     importpath = "github.com/cockroachdb/cockroach/pkg/sql/descmetadata",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/jobs",
         "//pkg/jobs/jobspb",
         "//pkg/keys",
+        "//pkg/scheduledjobs",
         "//pkg/settings",
         "//pkg/sql/catalog",
         "//pkg/sql/catalog/descpb",

pkg/sql/descmetadata/metadata_updater.go

@@ -9,8 +9,10 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
+	"github.com/cockroachdb/cockroach/pkg/scheduledjobs"
 	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
@@ -113,3 +115,19 @@ func (mu metadataUpdater) UpdateTTLScheduleLabel(
 	)
 	return err
 }
+
+// UpdateTTLScheduleCron implement scexec.DescriptorMetadataUpdater.
+func (mu metadataUpdater) UpdateTTLScheduleCron(
+	ctx context.Context, scheduleID jobspb.ScheduleID, cronExpr string,
+) error {
+	env := scheduledjobs.ProdJobSchedulerEnv
+	schedules := jobs.ScheduledJobTxn(mu.txn)
+	s, err := schedules.Load(ctx, env, scheduleID)
+	if err != nil {
+		return err
+	}
+	if err := s.SetScheduleAndNextRun(cronExpr); err != nil {
+		return err
+	}
+	return schedules.Update(ctx, s)
+}

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

@@ -27,9 +27,9 @@ import (
 func validateStorageParamKey(t tree.NodeFormatter, key string) {
 	loweredKey := strings.ToLower(key)
 	// These TTL params still require legacy schema changer because they
-	// affect column management, column dependencies, or schedule management.
+	// affect column management or column dependencies.
 	switch loweredKey {
-	case "ttl", "ttl_expire_after", "ttl_expiration_expression", "ttl_job_cron":
+	case "ttl", "ttl_expire_after", "ttl_expiration_expression":
 		panic(scerrors.NotImplementedErrorf(t, redact.Sprintf("%s not implemented yet", redact.SafeString(key))))
 	}
 	if loweredKey == catpb.RBRUsingConstraintTableSettingName {

pkg/sql/schemachanger/scdeps/sctestdeps/test_deps.go

@@ -1341,6 +1341,14 @@ func (s *TestState) UpdateTTLScheduleLabel(ctx context.Context, tbl catalog.Tabl
 	return nil
 }
 
+// UpdateTTLScheduleCron implements scexec.DescriptorMetadataUpdater
+func (s *TestState) UpdateTTLScheduleCron(
+	ctx context.Context, scheduleID jobspb.ScheduleID, cronExpr string,
+) error {
+	s.LogSideEffectf("update ttl schedule cron #%d to %q", scheduleID, cronExpr)
+	return nil
+}
+
 // DescriptorMetadataUpdater implement scexec.Dependencies.
 func (s *TestState) DescriptorMetadataUpdater(
 	ctx context.Context,

pkg/sql/schemachanger/scexec/dependencies.go

@@ -386,6 +386,9 @@ type DescriptorMetadataUpdater interface {
 	// UpdateTTLScheduleLabel updates the schedule_name for the TTL Scheduled Job
 	// of the given table.
 	UpdateTTLScheduleLabel(ctx context.Context, tbl catalog.TableDescriptor) error
+
+	// UpdateTTLScheduleCron updates the cron schedule for a TTL job.
+	UpdateTTLScheduleCron(ctx context.Context, scheduleID jobspb.ScheduleID, cronExpr string) error
 }
 
 type TemporarySchemaCreator interface {

pkg/sql/schemachanger/scexec/exec_deferred_mutation.go

@@ -29,6 +29,7 @@ type deferredState struct {
 	statsToRefresh               catalog.DescriptorIDSet
 	indexesToSplitAndScatter     []indexesToSplitAndScatter
 	ttlScheduleMetadataUpdates   []ttlScheduleMetadataUpdate
+	ttlScheduleCronUpdates       []ttlScheduleCronUpdate
 	gcJobs
 }
 
@@ -47,6 +48,11 @@ type ttlScheduleMetadataUpdate struct {
 	newName string
 }
 
+type ttlScheduleCronUpdate struct {
+	scheduleID  jobspb.ScheduleID
+	newCronExpr string
+}
+
 type schemaChangerJobUpdate struct {
 	isNonCancelable       bool
 	runningStatus         redact.RedactableString
@@ -92,6 +98,16 @@ func (s *deferredState) UpdateTTLScheduleMetadata(
 	return nil
 }
 
+func (s *deferredState) UpdateTTLScheduleCron(
+	ctx context.Context, scheduleID jobspb.ScheduleID, cronExpr string,
+) error {
+	s.ttlScheduleCronUpdates = append(s.ttlScheduleCronUpdates, ttlScheduleCronUpdate{
+		scheduleID:  scheduleID,
+		newCronExpr: cronExpr,
+	})
+	return nil
+}
+
 func (s *deferredState) AddNewSchemaChangerJob(
 	jobID jobspb.JobID,
 	stmts []scpb.Statement,
@@ -226,6 +242,11 @@ func (s *deferredState) exec(
 			return err
 		}
 	}
+	for _, cronUpdate := range s.ttlScheduleCronUpdates {
+		if err := m.UpdateTTLScheduleCron(ctx, cronUpdate.scheduleID, cronUpdate.newCronExpr); err != nil {
+			return err
+		}
+	}
 	for _, idx := range s.indexesToSplitAndScatter {
 		descs, err := c.MustReadImmutableDescriptors(ctx, idx.tableID)
 		if err != nil {

pkg/sql/schemachanger/scexec/executor_external_test.go

@@ -530,6 +530,12 @@ func (noopMetadataUpdater) UpdateTTLScheduleLabel(
 	return nil
 }
 
+func (u noopMetadataUpdater) UpdateTTLScheduleCron(
+	ctx context.Context, scheduleID jobspb.ScheduleID, cronExpr string,
+) error {
+	return nil
+}
+
 type noopTemporarySchemaCreator struct{}
 
 var _ scexec.TemporarySchemaCreator = noopTemporarySchemaCreator{}

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

@@ -159,4 +159,7 @@ type DeferredMutationStateUpdater interface {
 
 	// UpdateTTLScheduleMetadata updates the TTL schedule metadata for a table.
 	UpdateTTLScheduleMetadata(ctx context.Context, tableID descpb.ID, newName string) error
+
+	// UpdateTTLScheduleCron updates the cron expression for a TTL schedule.
+	UpdateTTLScheduleCron(ctx context.Context, scheduleID jobspb.ScheduleID, cronExpr string) error
 }

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

@@ -77,3 +77,9 @@ func (d *deferredVisitor) UpdateTTLScheduleMetadata(
 ) error {
 	return d.DeferredMutationStateUpdater.UpdateTTLScheduleMetadata(ctx, op.TableID, op.NewName)
 }
+
+func (d *deferredVisitor) UpdateTTLScheduleCron(
+	ctx context.Context, op scop.UpdateTTLScheduleCron,
+) error {
+	return d.DeferredMutationStateUpdater.UpdateTTLScheduleCron(ctx, op.ScheduleID, op.NewCronExpr)
+}

pkg/sql/schemachanger/scop/deferred_mutation.go

@@ -107,3 +107,11 @@ type UpdateTTLScheduleMetadata struct {
 	TableID descpb.ID
 	NewName string
 }
+
+// UpdateTTLScheduleCron updates the cron schedule expression for a TTL job.
+type UpdateTTLScheduleCron struct {
+	deferredMutationOp
+	TableID     descpb.ID
+	ScheduleID  jobspb.ScheduleID
+	NewCronExpr string
+}