jobs: remove dead legacy branches

This is a simple cleanup of version conditionals that no longer serve any purpose
and the disused upgrades related to jobs in 25.1.

Release note: none.
Epic: none.

Author: dt

pkg/clusterversion/cockroach_versions.go

@@ -187,9 +187,6 @@ const (
 
 	TODO_Delete_V25_1_Start
 
-	// TODO_Delete_V25_1_AddJobsTables added new jobs tables.
-	TODO_Delete_V25_1_AddJobsTables
-
 	// TODO_Delete_V25_1_AddRangeForceFlushKey adds the RangeForceFlushKey, a replicated
 	// range-ID local key, which is written below raft.
 	TODO_Delete_V25_1_AddRangeForceFlushKey
@@ -203,19 +200,6 @@ const (
 	// that are part of the XA two-phase commit protocol.
 	TODO_Delete_V25_1_PreparedTransactionsTable
 
-	// TODO_Delete_V25_1_AddJobsColumns added new columns to system.jobs.
-	TODO_Delete_V25_1_AddJobsColumns
-
-	// TODO_Delete_V25_1_JobsWritesFence is an empty version that is used to add a "fence"
-	// between the column addition version and the backfill version. This allows
-	// the backfill version's upgrade step to make the assumption that all nodes
-	// will be writing to the new columns, since moving from fence to backfill can
-	// only start once no nodes are still on add-columnns.
-	TODO_Delete_V25_1_JobsWritesFence
-
-	// TODO_Delete_V25_1_JobsBackfill backfills the new jobs tables and columns.
-	TODO_Delete_V25_1_JobsBackfill
-
 	// V25_1 is CockroachDB v25.1. It's used for all v25.1.x patch releases.
 	V25_1
 
@@ -271,13 +255,9 @@ var versionTable = [numKeys]roachpb.Version{
 	// v25.1 versions. Internal versions must be even.
 	TODO_Delete_V25_1_Start: {Major: 24, Minor: 3, Internal: 2},
 
-	TODO_Delete_V25_1_AddJobsTables:             {Major: 24, Minor: 3, Internal: 4},
 	TODO_Delete_V25_1_AddRangeForceFlushKey:     {Major: 24, Minor: 3, Internal: 8},
 	TODO_Delete_V25_1_BatchStreamRPC:            {Major: 24, Minor: 3, Internal: 10},
 	TODO_Delete_V25_1_PreparedTransactionsTable: {Major: 24, Minor: 3, Internal: 12},
-	TODO_Delete_V25_1_AddJobsColumns:            {Major: 24, Minor: 3, Internal: 14},
-	TODO_Delete_V25_1_JobsWritesFence:           {Major: 24, Minor: 3, Internal: 16},
-	TODO_Delete_V25_1_JobsBackfill:              {Major: 24, Minor: 3, Internal: 18},
 
 	V25_1: {Major: 25, Minor: 1, Internal: 0},
 

pkg/jobs/registry.go

@@ -483,7 +483,7 @@ func batchJobInsertStmt(
 		return "", nil, nil, errors.NewAssertionErrorWithWrappedErrf(err, "failed to make timestamp for creation of job")
 	}
 	instanceID := r.ID()
-	columns := []string{`id`, `created`, `status`, `claim_session_id`, `claim_instance_id`, `job_type`}
+	columns := []string{`id`, `created`, `status`, `claim_session_id`, `claim_instance_id`, `job_type`, `owner`, `description`}
 	valueFns := map[string]func(*Job) (interface{}, error){
 		`id`:                func(job *Job) (interface{}, error) { return job.ID(), nil },
 		`created`:           func(job *Job) (interface{}, error) { return created, nil },
@@ -494,12 +494,8 @@ func batchJobInsertStmt(
 			payload := job.Payload()
 			return payload.Type().String(), nil
 		},
-	}
-
-	if schemaVersion.AtLeast(clusterversion.TODO_Delete_V25_1_AddJobsColumns.Version()) {
-		columns = append(columns, `owner`, `description`)
-		valueFns[`owner`] = func(job *Job) (interface{}, error) { return job.Payload().UsernameProto.Decode().Normalized(), nil }
-		valueFns[`description`] = func(job *Job) (interface{}, error) { return job.Payload().Description, nil }
+		`owner`:       func(job *Job) (interface{}, error) { return job.Payload().UsernameProto.Decode().Normalized(), nil },
+		`description`: func(job *Job) (interface{}, error) { return job.Payload().Description, nil },
 	}
 
 	appendValues := func(job *Job, vals *[]interface{}) (err error) {
@@ -590,15 +586,9 @@ func (r *Registry) CreateJobWithTxn(
 			return errors.NewAssertionErrorWithWrappedErrf(err, "failed to construct job created timestamp")
 		}
 
-		cols := []string{"id", "created", "status", "claim_session_id", "claim_instance_id", "job_type"}
-		vals := []interface{}{jobID, created, StateRunning, s.ID().UnsafeBytes(), r.ID(), jobType.String()}
-		v, err := txn.GetSystemSchemaVersion(ctx)
-		if err != nil {
-			return err
-		}
-		if v.AtLeast(clusterversion.TODO_Delete_V25_1_AddJobsColumns.Version()) {
-			cols = append(cols, "owner", "description")
-			vals = append(vals, j.mu.payload.UsernameProto.Decode().Normalized(), j.mu.payload.Description)
+		cols := []string{"id", "created", "status", "claim_session_id", "claim_instance_id", "job_type", "owner", "description"}
+		vals := []interface{}{
+			jobID, created, StateRunning, s.ID().UnsafeBytes(), r.ID(), jobType.String(), j.mu.payload.UsernameProto.Decode().Normalized(), j.mu.payload.Description,
 		}
 
 		totalNumCols := len(cols)
@@ -730,18 +720,9 @@ func (r *Registry) CreateAdoptableJobWithTxn(
 		}
 		typ := j.mu.payload.Type().String()
 
-		cols := []string{"id", "created", "status", "created_by_type", "created_by_id", "job_type"}
-		placeholders := []string{"$1", "now() at time zone 'utc'", "$2", "$3", "$4", "$5"}
-		vals := []interface{}{jobID, StateRunning, createdByType, createdByID, typ}
-		v, err := txn.GetSystemSchemaVersion(ctx)
-		if err != nil {
-			return err
-		}
-		if v.AtLeast(clusterversion.TODO_Delete_V25_1_AddJobsColumns.Version()) {
-			cols = append(cols, "owner", "description")
-			placeholders = append(placeholders, "$6", "$7")
-			vals = append(vals, j.mu.payload.UsernameProto.Decode().Normalized(), j.mu.payload.Description)
-		}
+		cols := []string{"id", "created", "status", "created_by_type", "created_by_id", "job_type", "owner", "description"}
+		placeholders := []string{"$1", "now() at time zone 'utc'", "$2", "$3", "$4", "$5", "$6", "$7"}
+		vals := []interface{}{jobID, StateRunning, createdByType, createdByID, typ, j.mu.payload.UsernameProto.Decode().Normalized(), j.mu.payload.Description}
 
 		// Insert the job row, but do not set a `claim_session_id`. By not
 		// setting the claim, the job can be adopted by any node and will
@@ -1275,20 +1256,11 @@ func (r *Registry) cleanupOldJobsPage(
 		}
 
 		counts := make(map[string]int)
-		for i, tbl := range jobMetadataTables {
+		for _, tbl := range jobMetadataTables {
 			var deleted int
 			if err := r.db.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
 				// Tables other than job_info -- the 0th -- are only present if the txn is
 				// running at a version that includes them.
-				if i > 0 {
-					v, err := txn.GetSystemSchemaVersion(ctx)
-					if err != nil {
-						return err
-					}
-					if v.Less(clusterversion.TODO_Delete_V25_1_AddJobsTables.Version()) {
-						return nil
-					}
-				}
 				deleted, err = txn.Exec(ctx, redact.RedactableString("gc-job-"+tbl), txn.KV(),
 					"DELETE FROM system."+tbl+" WHERE job_id = ANY($1)", toDelete,
 				)
@@ -1337,17 +1309,7 @@ func (r *Registry) DeleteTerminalJobByID(ctx context.Context, id jobspb.JobID) e
 			if err != nil {
 				return err
 			}
-			for i, tbl := range jobMetadataTables {
-				if i > 0 {
-					v, err := txn.GetSystemSchemaVersion(ctx)
-					if err != nil {
-						return err
-					}
-					if v.Less(clusterversion.TODO_Delete_V25_1_AddJobsTables.Version()) {
-						break
-					}
-				}
-
+			for _, tbl := range jobMetadataTables {
 				_, err = txn.Exec(
 					ctx, redact.RedactableString("delete-job-"+tbl), txn.KV(),
 					"DELETE FROM system."+tbl+" WHERE job_id = $1", id,

pkg/jobs/update.go

@@ -12,7 +12,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
@@ -281,96 +280,88 @@ WHERE id = $1
 		}
 	}
 
-	v, err := u.txn.GetSystemSchemaVersion(ctx)
-	if err != nil {
-		return err
-	}
-	if v.AtLeast(clusterversion.TODO_Delete_V25_1_AddJobsTables.Version()) {
-		if ju.md.State != "" && ju.md.State != state {
-			if err := j.Messages().Record(ctx, u.txn, "state", string(ju.md.State)); err != nil {
+	if ju.md.State != "" && ju.md.State != state {
+		if err := j.Messages().Record(ctx, u.txn, "state", string(ju.md.State)); err != nil {
+			return err
+		}
+		// If we are changing state, we should clear out the status, unless
+		// we are about to set it to something instead.
+		if progress == nil || progress.StatusMessage == "" {
+			if err := j.StatusStorage().Clear(ctx, u.txn); err != nil {
 				return err
 			}
-			// If we are changing state, we should clear out the status, unless
-			// we are about to set it to something instead.
-			if progress == nil || progress.StatusMessage == "" {
-				if err := j.StatusStorage().Clear(ctx, u.txn); err != nil {
-					return err
-				}
-			}
 		}
+	}
 
-		if progress != nil {
-			var ts hlc.Timestamp
-			if hwm := progress.GetHighWater(); hwm != nil {
-				ts = *hwm
-			}
+	if progress != nil {
+		var ts hlc.Timestamp
+		if hwm := progress.GetHighWater(); hwm != nil {
+			ts = *hwm
+		}
 
-			if err := j.ProgressStorage().Set(ctx, u.txn, float64(progress.GetFractionCompleted()), ts); err != nil {
-				return err
-			}
+		if err := j.ProgressStorage().Set(ctx, u.txn, float64(progress.GetFractionCompleted()), ts); err != nil {
+			return err
+		}
 
-			if progress.StatusMessage != beforeProgress.StatusMessage {
-				if err := j.StatusStorage().Set(ctx, u.txn, progress.StatusMessage); err != nil {
-					return err
-				}
+		if progress.StatusMessage != beforeProgress.StatusMessage {
+			if err := j.StatusStorage().Set(ctx, u.txn, progress.StatusMessage); err != nil {
+				return err
 			}
+		}
 
-			if progress.TraceID != beforeProgress.TraceID {
-				if err := j.Messages().Record(ctx, u.txn, "trace-id", fmt.Sprintf("%d", progress.TraceID)); err != nil {
-					return err
-				}
+		if progress.TraceID != beforeProgress.TraceID {
+			if err := j.Messages().Record(ctx, u.txn, "trace-id", fmt.Sprintf("%d", progress.TraceID)); err != nil {
+				return err
 			}
 		}
 	}
-	if v.AtLeast(clusterversion.TODO_Delete_V25_1_AddJobsColumns.Version()) {
 
-		vals := []interface{}{j.ID()}
+	vals := []interface{}{j.ID()}
 
-		var update strings.Builder
+	var update strings.Builder
 
-		if payloadBytes != nil {
-			if beforePayload.Description != payload.Description {
-				if update.Len() > 0 {
-					update.WriteString(", ")
-				}
-				vals = append(vals, payload.Description)
-				fmt.Fprintf(&update, "description = $%d", len(vals))
+	if payloadBytes != nil {
+		if beforePayload.Description != payload.Description {
+			if update.Len() > 0 {
+				update.WriteString(", ")
 			}
+			vals = append(vals, payload.Description)
+			fmt.Fprintf(&update, "description = $%d", len(vals))
+		}
 
-			if beforePayload.UsernameProto.Decode() != payload.UsernameProto.Decode() {
-				if update.Len() > 0 {
-					update.WriteString(", ")
-				}
-				vals = append(vals, payload.UsernameProto.Decode().Normalized())
-				fmt.Fprintf(&update, "owner = $%d", len(vals))
+		if beforePayload.UsernameProto.Decode() != payload.UsernameProto.Decode() {
+			if update.Len() > 0 {
+				update.WriteString(", ")
 			}
+			vals = append(vals, payload.UsernameProto.Decode().Normalized())
+			fmt.Fprintf(&update, "owner = $%d", len(vals))
+		}
 
-			if beforePayload.Error != payload.Error {
-				if update.Len() > 0 {
-					update.WriteString(", ")
-				}
-				vals = append(vals, payload.Error)
-				fmt.Fprintf(&update, "error_msg = $%d", len(vals))
+		if beforePayload.Error != payload.Error {
+			if update.Len() > 0 {
+				update.WriteString(", ")
 			}
+			vals = append(vals, payload.Error)
+			fmt.Fprintf(&update, "error_msg = $%d", len(vals))
+		}
 
-			if beforePayload.FinishedMicros != payload.FinishedMicros {
-				if update.Len() > 0 {
-					update.WriteString(", ")
-				}
-				vals = append(vals, time.UnixMicro(payload.FinishedMicros))
-				fmt.Fprintf(&update, "finished = $%d", len(vals))
+		if beforePayload.FinishedMicros != payload.FinishedMicros {
+			if update.Len() > 0 {
+				update.WriteString(", ")
 			}
-
+			vals = append(vals, time.UnixMicro(payload.FinishedMicros))
+			fmt.Fprintf(&update, "finished = $%d", len(vals))
 		}
-		if len(vals) > 1 {
-			stmt := fmt.Sprintf("UPDATE system.jobs SET %s WHERE id = $1", update.String())
-			if _, err := u.txn.ExecEx(
-				ctx, "job-update-row", u.txn.KV(),
-				sessiondata.NodeUserSessionDataOverride,
-				stmt, vals...,
-			); err != nil {
-				return err
-			}
+
+	}
+	if len(vals) > 1 {
+		stmt := fmt.Sprintf("UPDATE system.jobs SET %s WHERE id = $1", update.String())
+		if _, err := u.txn.ExecEx(
+			ctx, "job-update-row", u.txn.KV(),
+			sessiondata.NodeUserSessionDataOverride,
+			stmt, vals...,
+		); err != nil {
+			return err
 		}
 	}
 

pkg/upgrade/upgrades/BUILD.bazel

@@ -16,7 +16,6 @@ go_library(
         "permanent_upgrades.go",
         "schema_changes.go",
         "upgrades.go",
-        "v25_1_add_jobs_tables.go",
         "v25_1_prepared_transactions_table.go",
         "v25_2_add_sql_activity_flush_job.go",
         "v25_2_set_ui_default_timezone.go",
@@ -85,7 +84,6 @@ go_test(
         "schema_changes_external_test.go",
         "schema_changes_helpers_test.go",
         "upgrades_test.go",
-        "v25_1_add_jobs_tables_test.go",
         "v25_1_prepared_transactions_table_test.go",
         "v25_2_set_ui_default_timezone_test.go",
         "v25_3_add_event_log_column_and_index_test.go",
@@ -100,7 +98,6 @@ go_test(
         "//pkg/clusterversion",
         "//pkg/jobs",
         "//pkg/jobs/jobspb",
-        "//pkg/jobs/jobstest",
         "//pkg/keys",
         "//pkg/kv",
         "//pkg/roachpb",
@@ -130,7 +127,6 @@ go_test(
         "//pkg/sql/sem/tree",
         "//pkg/sql/types",
         "//pkg/testutils",
-        "//pkg/testutils/jobutils",
         "//pkg/testutils/serverutils",
         "//pkg/testutils/skip",
         "//pkg/testutils/sqlutils",

pkg/upgrade/upgrades/upgrades.go

@@ -61,35 +61,13 @@ var upgrades = []upgradebase.Upgrade{
 
 	newFirstUpgrade(clusterversion.TODO_Delete_V25_1_Start.Version()),
 
-	upgrade.NewTenantUpgrade(
-		"add new jobs tables",
-		clusterversion.TODO_Delete_V25_1_AddJobsTables.Version(),
-		upgrade.NoPrecondition,
-		addJobsTables,
-		upgrade.RestoreActionNotRequired("cluster restore does not restore the new field"),
-	),
-
 	upgrade.NewTenantUpgrade(
 		"create prepared_transactions table",
 		clusterversion.TODO_Delete_V25_1_PreparedTransactionsTable.Version(),
 		upgrade.NoPrecondition,
 		createPreparedTransactionsTable,
 		upgrade.RestoreActionNotRequired("cluster restore does not restore this table"),
 	),
-	upgrade.NewTenantUpgrade(
-		"add new jobs tables",
-		clusterversion.TODO_Delete_V25_1_AddJobsColumns.Version(),
-		upgrade.NoPrecondition,
-		addJobsColumns,
-		upgrade.RestoreActionNotRequired("cluster restore does not restore the new field"),
-	),
-	upgrade.NewTenantUpgrade(
-		"backfill new jobs tables",
-		clusterversion.TODO_Delete_V25_1_JobsBackfill.Version(),
-		upgrade.NoPrecondition,
-		backfillJobsTablesAndColumns,
-		upgrade.RestoreActionNotRequired("cluster restore does not restore jobs tables"),
-	),
 
 	newFirstUpgrade(clusterversion.TODO_Delete_V25_2_Start.Version()),