jobs: move JobSchedulerEnv helper function

Move this simple function to avoid a dependency cycle in an upcoming
commit.

Release note: None

Author: rafiss

pkg/backup/alter_backup_schedule.go

@@ -49,7 +49,7 @@ func loadSchedules(
 	}
 
 	execCfg := p.ExecCfg()
-	env := sql.JobSchedulerEnv(execCfg.JobsKnobs())
+	env := jobs.JobSchedulerEnv(execCfg.JobsKnobs())
 	schedules := jobs.ScheduledJobTxn(p.InternalSQLTxn())
 	schedule, err := schedules.Load(ctx, env, scheduleID)
 	if err != nil {
@@ -434,7 +434,7 @@ func processFullBackupRecurrence(
 		return s, nil
 	}
 
-	env := sql.JobSchedulerEnv(p.ExecCfg().JobsKnobs())
+	env := jobs.JobSchedulerEnv(p.ExecCfg().JobsKnobs())
 	scheduledJobs := jobs.ScheduledJobTxn(p.InternalSQLTxn())
 	if fullBackupAlways {
 		if s.incJob == nil {
@@ -535,7 +535,7 @@ func validateFullIncrementalFrequencies(p sql.PlanHookState, s scheduleDetails)
 	if s.incJob == nil {
 		return nil
 	}
-	env := sql.JobSchedulerEnv(p.ExecCfg().JobsKnobs())
+	env := jobs.JobSchedulerEnv(p.ExecCfg().JobsKnobs())
 	now := env.Now()
 
 	fullFreq, err := frequencyFromCron(now, s.fullJob.ScheduleExpr())
@@ -594,7 +594,7 @@ func processInto(p sql.PlanHookState, spec *alterBackupScheduleSpec, s scheduleD
 	// so we can unpause incrementals. This mirrors the behavior of
 	// CREATE SCHEDULE FOR BACKUP.
 	if !incPaused {
-		env := sql.JobSchedulerEnv(p.ExecCfg().JobsKnobs())
+		env := jobs.JobSchedulerEnv(p.ExecCfg().JobsKnobs())
 		s.fullJob.SetNextRun(env.Now())
 	}
 
@@ -608,7 +608,7 @@ func processNextRunNow(
 		return nil
 	}
 
-	env := sql.JobSchedulerEnv(p.ExecCfg().JobsKnobs())
+	env := jobs.JobSchedulerEnv(p.ExecCfg().JobsKnobs())
 
 	// Trigger the full schedule, unless there is an inc schedule and the user did
 	// not explicitly specify the full.

pkg/backup/create_scheduled_backup.go

@@ -180,7 +180,7 @@ func doCreateBackupSchedules(
 		}
 	}
 
-	env := sql.JobSchedulerEnv(p.ExecCfg().JobsKnobs())
+	env := jobs.JobSchedulerEnv(p.ExecCfg().JobsKnobs())
 
 	// Evaluate incremental and full recurrence.
 	incRecurrence, err := schedulebase.ComputeScheduleRecurrence(env.Now(), eval.recurrence)

pkg/backup/restore_job.go

@@ -2922,7 +2922,7 @@ func (r *restoreResumer) dropDescriptors(
 	// immediately.
 	dropTime := int64(1)
 	scheduledJobs := jobs.ScheduledJobTxn(txn)
-	env := sql.JobSchedulerEnv(r.execCfg.JobsKnobs())
+	env := jobs.JobSchedulerEnv(r.execCfg.JobsKnobs())
 	for i := range mutableTables {
 		tableToDrop := mutableTables[i]
 		tablesToGC = append(tablesToGC, tableToDrop.ID)

pkg/ccl/changefeedccl/scheduled_changefeed.go

@@ -560,7 +560,7 @@ func doCreateChangefeedSchedule(
 	resultsCh chan<- tree.Datums,
 ) error {
 
-	env := sql.JobSchedulerEnv(p.ExecCfg().JobsKnobs())
+	env := jobs.JobSchedulerEnv(p.ExecCfg().JobsKnobs())
 
 	if knobs, ok := p.ExecCfg().DistSQLSrv.TestingKnobs.JobsTestingKnobs.(*jobs.TestingKnobs); ok {
 		if knobs.JobSchedulerEnv != nil {

pkg/jobs/scheduled_job.go

@@ -68,6 +68,14 @@ func NewScheduledJob(env scheduledjobs.JobSchedulerEnv) *ScheduledJob {
 	}
 }
 
+// JobSchedulerEnv returns JobSchedulerEnv.
+func JobSchedulerEnv(knobs *TestingKnobs) scheduledjobs.JobSchedulerEnv {
+	if knobs != nil && knobs.JobSchedulerEnv != nil {
+		return knobs.JobSchedulerEnv
+	}
+	return scheduledjobs.ProdJobSchedulerEnv
+}
+
 // scheduledJobNotFoundError is returned from load when the scheduled job does
 // not exist.
 type scheduledJobNotFoundError struct {

pkg/sql/alter_table.go

@@ -2161,7 +2161,7 @@ func handleTTLStorageParamChange(
 
 		// Update cron schedule if required.
 		if before.DeletionCron != after.DeletionCron {
-			env := JobSchedulerEnv(params.ExecCfg().JobsKnobs())
+			env := jobs.JobSchedulerEnv(params.ExecCfg().JobsKnobs())
 			schedules := jobs.ScheduledJobTxn(params.p.InternalSQLTxn())
 			s, err := schedules.Load(
 				params.ctx,

pkg/sql/catalog/externalcatalog/external_catalog.go

@@ -197,7 +197,7 @@ func DropIngestedExternalCatalog(
 	// immediately.
 	dropTime := int64(1)
 	scheduledJobs := jobs.ScheduledJobTxn(txn)
-	env := sql.JobSchedulerEnv(execCfg.JobsKnobs())
+	env := jobs.JobSchedulerEnv(execCfg.JobsKnobs())
 	for i := range mutableTables {
 		tableToDrop := mutableTables[i]
 		tablesToGC = append(tablesToGC, tableToDrop.ID)

pkg/sql/check.go

@@ -736,7 +736,7 @@ func (p *planner) validateTTLScheduledJobInTable(
 	ttl := tableDesc.GetRowLevelTTL()
 
 	execCfg := p.ExecCfg()
-	env := JobSchedulerEnv(execCfg.JobsKnobs())
+	env := jobs.JobSchedulerEnv(execCfg.JobsKnobs())
 
 	wrapError := func(origErr error) error {
 		return errors.WithHintf(

pkg/sql/control_schedules.go

@@ -43,17 +43,9 @@ func collectTelemetry(command tree.ScheduleCommand) {
 	}
 }
 
-// JobSchedulerEnv returns JobSchedulerEnv.
-func JobSchedulerEnv(knobs *jobs.TestingKnobs) scheduledjobs.JobSchedulerEnv {
-	if knobs != nil && knobs.JobSchedulerEnv != nil {
-		return knobs.JobSchedulerEnv
-	}
-	return scheduledjobs.ProdJobSchedulerEnv
-}
-
 // loadSchedule loads schedule information as the node user.
 func loadSchedule(params runParams, scheduleID tree.Datum) (*jobs.ScheduledJob, error) {
-	env := JobSchedulerEnv(params.ExecCfg().JobsKnobs())
+	env := jobs.JobSchedulerEnv(params.ExecCfg().JobsKnobs())
 	schedule := jobs.NewScheduledJob(env)
 
 	// Load schedule expression.  This is needed for resume command, but we
@@ -89,7 +81,7 @@ func loadSchedule(params runParams, scheduleID tree.Datum) (*jobs.ScheduledJob,
 func DeleteSchedule(
 	ctx context.Context, execCfg *ExecutorConfig, txn isql.Txn, scheduleID jobspb.ScheduleID,
 ) error {
-	env := JobSchedulerEnv(execCfg.JobsKnobs())
+	env := jobs.JobSchedulerEnv(execCfg.JobsKnobs())
 	_, err := txn.ExecEx(
 		ctx,
 		"delete-schedule",
@@ -168,7 +160,7 @@ func (n *controlSchedulesNode) startExec(params runParams) error {
 			if schedule.IsPaused() {
 				err = errors.Newf("cannot execute a paused schedule; use RESUME SCHEDULE instead")
 			} else {
-				env := JobSchedulerEnv(params.ExecCfg().JobsKnobs())
+				env := jobs.JobSchedulerEnv(params.ExecCfg().JobsKnobs())
 				schedule.SetNextRun(env.Now())
 				err = jobs.ScheduledJobTxn(params.p.InternalSQLTxn()).
 					Update(params.ctx, schedule)

pkg/sql/create_table.go

@@ -2664,7 +2664,7 @@ func CreateRowLevelTTLScheduledJob(
 	}
 
 	telemetry.Inc(sqltelemetry.RowLevelTTLCreated)
-	env := JobSchedulerEnv(knobs)
+	env := jobs.JobSchedulerEnv(knobs)
 	j, err := newRowLevelTTLScheduledJob(env, owner, tblDesc, clusterID, version)
 	if err != nil {
 		return nil, err