sql: remove Txn() from JobExecContext

This removes Txn() from JobExecContext. This method would always
return a nil context, making it a bit error prone to actually use in
jobs.

Release note: None

pkg/ccl/backupccl/backup_job.go

@@ -473,16 +473,19 @@ func (b *backupResumer) Resume(ctx context.Context, execCtx interface{}) error {
 	// TODO(adityamaru: Break this code block into helper methods.
 	if details.URI == "" {
 		initialDetails := details
-		backupDetails, m, err := getBackupDetailAndManifest(
-			ctx, p.ExecCfg(), p.Txn(), details, p.User(), backupDest,
-		)
-		if err != nil {
+		if err := p.ExecCfg().DB.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
+			backupDetails, m, err := getBackupDetailAndManifest(
+				ctx, p.ExecCfg(), txn, details, p.User(), backupDest,
+			)
+			if err != nil {
+				return err
+			}
+			details = backupDetails
+			backupManifest = &m
+			return nil
+		}); err != nil {
 			return err
 		}
-		details = backupDetails
-		// Reset backupDetails so nobody accidentally uses it.
-		backupDetails = jobspb.BackupDetails{} //lint:ignore SA4006 intentionally clearing so no one uses this.
-		backupManifest = &m
 
 		// Now that we have resolved the details, and manifest, write a protected
 		// timestamp record on the backup's target spans/schema object.
@@ -497,7 +500,7 @@ func (b *backupResumer) Resume(ctx context.Context, execCtx interface{}) error {
 			if details.ProtectedTimestampRecord != nil {
 				if err := p.ExecCfg().DB.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
 					return protectTimestampForBackup(
-						ctx, p.ExecCfg(), txn, b.job.ID(), m, details,
+						ctx, p.ExecCfg(), txn, b.job.ID(), backupManifest, details,
 					)
 				}); err != nil {
 					return err
@@ -563,7 +566,7 @@ func (b *backupResumer) Resume(ctx context.Context, execCtx interface{}) error {
 		lic := utilccl.CheckEnterpriseEnabled(
 			p.ExecCfg().Settings, p.ExecCfg().NodeInfo.LogicalClusterID(), "",
 		) != nil
-		collectTelemetry(ctx, m, initialDetails, details, lic, b.job.ID())
+		collectTelemetry(ctx, backupManifest, initialDetails, details, lic, b.job.ID())
 	}
 
 	// For all backups, partitioned or not, the main BACKUP manifest is stored at

pkg/ccl/backupccl/backup_planning.go

@@ -861,7 +861,7 @@ func logAndSanitizeBackupDestinations(ctx context.Context, backupDestinations ..
 
 func collectTelemetry(
 	ctx context.Context,
-	backupManifest backuppb.BackupManifest,
+	backupManifest *backuppb.BackupManifest,
 	initialDetails, backupDetails jobspb.BackupDetails,
 	licensed bool,
 	jobID jobspb.JobID,
@@ -1156,7 +1156,7 @@ func getReintroducedSpans(
 	return tableSpans, nil
 }
 
-func getProtectedTimestampTargetForBackup(backupManifest backuppb.BackupManifest) *ptpb.Target {
+func getProtectedTimestampTargetForBackup(backupManifest *backuppb.BackupManifest) *ptpb.Target {
 	if backupManifest.DescriptorCoverage == tree.AllDescriptors {
 		return ptpb.MakeClusterTarget()
 	}
@@ -1195,7 +1195,7 @@ func protectTimestampForBackup(
 	execCfg *sql.ExecutorConfig,
 	txn *kv.Txn,
 	jobID jobspb.JobID,
-	backupManifest backuppb.BackupManifest,
+	backupManifest *backuppb.BackupManifest,
 	backupDetails jobspb.BackupDetails,
 ) error {
 	tsToProtect := backupManifest.EndTime

pkg/ccl/changefeedccl/cdceval/constraint.go

@@ -102,8 +102,5 @@ func constrainSpansBySelectClause(
 }
 
 func schemaTS(execCtx sql.JobExecContext) hlc.Timestamp {
-	if execCtx.Txn() != nil {
-		return execCtx.Txn().ReadTimestamp()
-	}
 	return execCtx.ExecCfg().Clock.Now()
 }

pkg/ccl/changefeedccl/cdceval/validation.go

@@ -38,7 +38,7 @@ import (
 // buffer using cdceval.AsStringUnredacted.
 func NormalizeAndValidateSelectForTarget(
 	ctx context.Context,
-	execCtx sql.JobExecContext,
+	execCtx sql.PlanHookState,
 	desc catalog.TableDescriptor,
 	target jobspb.ChangefeedTargetSpecification,
 	sc *tree.SelectClause,

pkg/ccl/changefeedccl/cdceval/validation_test.go

@@ -65,7 +65,7 @@ func TestNormalizeAndValidate(t *testing.T) {
 			SearchPath: sessiondata.DefaultSearchPath.GetPathArray(),
 		})
 	defer cleanup()
-	execCtx := p.(sql.JobExecContext)
+	execCtx := p.(sql.PlanHookState)
 
 	for _, tc := range []struct {
 		name         string
@@ -289,7 +289,7 @@ func TestSelectClauseRequiresPrev(t *testing.T) {
 			SearchPath: sessiondata.DefaultSearchPath.GetPathArray(),
 		})
 	defer cleanup()
-	execCtx := p.(sql.JobExecContext)
+	execCtx := p.(sql.PlanHookState)
 
 	for _, tc := range []struct {
 		name   string

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -897,7 +897,7 @@ func validateDetailsAndOptions(
 // This method modifies passed in select clause to reflect normalization step.
 func validateAndNormalizeChangefeedExpression(
 	ctx context.Context,
-	execCtx sql.JobExecContext,
+	execCtx sql.PlanHookState,
 	sc *tree.SelectClause,
 	descriptors map[tree.TablePattern]catalog.Descriptor,
 	targets []jobspb.ChangefeedTargetSpecification,
@@ -977,7 +977,7 @@ func (b *changefeedResumer) handleChangefeedError(
 		const errorFmt = "job failed (%v) but is being paused because of %s=%s"
 		errorMessage := fmt.Sprintf(errorFmt, changefeedErr,
 			changefeedbase.OptOnError, changefeedbase.OptOnErrorPause)
-		return b.job.PauseRequested(ctx, jobExec.Txn(), func(ctx context.Context,
+		return b.job.PauseRequested(ctx, nil /* txn */, func(ctx context.Context,
 			planHookState interface{}, txn *kv.Txn, progress *jobspb.Progress) error {
 			err := b.OnPauseRequest(ctx, jobExec, txn, progress)
 			if err != nil {

pkg/ccl/changefeedccl/changefeed_test.go

@@ -6861,7 +6861,7 @@ func normalizeCDCExpression(t *testing.T, execCfgI interface{}, exprStr string)
 		})
 	defer cleanup()
 
-	execCtx := p.(sql.JobExecContext)
+	execCtx := p.(sql.PlanHookState)
 	_, _, err = cdceval.NormalizeAndValidateSelectForTarget(
 		context.Background(), execCtx, desc, target, sc, false, false, false,
 	)

pkg/ccl/streamingccl/streamingest/stream_ingestion_job.go

@@ -355,8 +355,7 @@ func (s *streamIngestionResumer) handleResumeError(
 	// The ingestion job is paused but the producer job will keep
 	// running until it times out. Users can still resume ingestion before
 	// the producer job times out.
-	jobExecCtx := execCtx.(sql.JobExecContext)
-	return s.job.PauseRequested(resumeCtx, jobExecCtx.Txn(), func(ctx context.Context,
+	return s.job.PauseRequested(resumeCtx, nil /* txn */, func(ctx context.Context,
 		planHookState interface{}, txn *kv.Txn, progress *jobspb.Progress) error {
 		progress.RunningStatus = errorMessage
 		return nil

pkg/sql/job_exec_context.go

@@ -13,7 +13,6 @@ package sql
 import (
 	"context"
 
-	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/spanconfig"
@@ -70,7 +69,6 @@ func (e *plannerJobExecContext) MigrationJobDeps() upgrade.JobDeps {
 func (e *plannerJobExecContext) SpanConfigReconciler() spanconfig.Reconciler {
 	return e.p.SpanConfigReconciler()
 }
-func (e *plannerJobExecContext) Txn() *kv.Txn { return e.p.Txn() }
 
 // ConstrainPrimaryIndexSpanByExpr implements SpanConstrainer
 func (e *plannerJobExecContext) ConstrainPrimaryIndexSpanByExpr(
@@ -106,5 +104,4 @@ type JobExecContext interface {
 	User() username.SQLUsername
 	MigrationJobDeps() upgrade.JobDeps
 	SpanConfigReconciler() spanconfig.Reconciler
-	Txn() *kv.Txn
 }