jobs: make GC of old jobs txn'l

A single logical job is now made up of many rows stored in many different tables, meaning

that deletion of a job from the system should imply complete deletion of all of its
rows in all of these tables. Previously we were not using a single txn to delete from
all of the tables, meaning a job could have its row deleted from e.g. system.jobs but
not from job_status, orphaning the status row. These orphan rows were not particularly
problematic but are cruft which could build up over time as nothing would remove them
once the main record in system.jobs was removed.

Instead, we now identify all expired jobs in system jobs which we will delete using one
txn, then delete each job in a txn that deletes all rows for that job in all tables.

Release note: none.
Epic: CRDB-51121.
Fixes: #147552.

Author: dt

pkg/jobs/adopt.go

@@ -37,6 +37,10 @@ const (
 		`'` + string(StatePauseRequested) + `', ` +
 		`'` + string(StateReverting) + `'`
 
+	terminalStateList = `'` + string(StateFailed) + `', ` +
+		`'` + string(StateCanceled) + `', ` +
+		`'` + string(StateSucceeded) + `'`
+
 	claimableStateTupleString = `(` + claimableStateList + `)`
 
 	nonTerminalStateList = claimableStateList + `, ` +

pkg/jobs/registry.go

@@ -29,7 +29,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlliveness"
-	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/cidr"
 	"github.com/cockroachdb/cockroach/pkg/util/envutil"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
@@ -1175,26 +1174,6 @@ FROM system.job_info
 WHERE written < $1 AND job_id NOT IN (SELECT id FROM system.jobs)
 LIMIT $2`
 
-// The ordering is important as we keep track of the maximum ID we've seen.
-const expiredJobsQueryWithJobInfoTable = `
-WITH
-latestpayload AS (
-    SELECT job_id, value
-    FROM system.job_info AS payload
-    WHERE job_id > $2 AND info_key = 'legacy_payload'
-    ORDER BY written desc
-),
-jobpage AS (
-    SELECT id, status
-    FROM system.jobs
-    WHERE (created < $1) and (id > $2)
-    ORDER BY id
-    LIMIT $3
-)
-SELECT distinct (id), latestpayload.value AS payload, status
-FROM jobpage AS j
-INNER JOIN latestpayload ON j.id = latestpayload.job_id`
-
 // jobMetadataTables are all of the tables that have rows storing additional
 // attributes or data about jobs beyond the core job record in system.jobs. All
 // of these tables identity the job which own rows in them using a "job_id"
@@ -1209,86 +1188,31 @@ var jobMetadataTables = []string{"job_info", "job_progress", "job_progress_histo
 func (r *Registry) cleanupOldJobsPage(
 	ctx context.Context, olderThan time.Time, minID jobspb.JobID, pageSize int,
 ) (done bool, maxID jobspb.JobID, retErr error) {
-	query := expiredJobsQueryWithJobInfoTable
+	query := "SELECT id, status, finished, $1 FROM system.jobs WHERE status in (" + terminalStateList + ") AND (finished < $1) and (id >= $2) ORDER BY id LIMIT $3"
 
 	it, err := r.db.Executor().QueryIterator(ctx, "gc-jobs", nil, /* txn */
 		query, olderThan, minID, pageSize)
 	if err != nil {
 		return false, 0, err
 	}
-	// We have to make sure to close the iterator since we might return from the
-	// for loop early (before Next() returns false).
-	defer func() { retErr = errors.CombineErrors(retErr, it.Close()) }()
-	toDelete := tree.NewDArray(types.Int)
-	oldMicros := timeutil.ToUnixMicros(olderThan)
-
+	var candidates []jobspb.JobID
 	var ok bool
-	var numRows int
 	for ok, err = it.Next(ctx); ok; ok, err = it.Next(ctx) {
-		numRows++
-		row := it.Cur()
-		payload, err := UnmarshalPayload(row[1])
-		if err != nil {
-			return false, 0, err
-		}
-		remove := false
-		switch State(*row[2].(*tree.DString)) {
-		case StateSucceeded, StateCanceled, StateFailed:
-			remove = payload.FinishedMicros < oldMicros
-		}
-		if remove {
-			toDelete.Array = append(toDelete.Array, row[0])
-		}
-	}
-	if err != nil {
-		return false, 0, err
+		candidates = append(candidates, jobspb.JobID(tree.MustBeDInt(it.Cur()[0])))
 	}
-	if numRows == 0 {
-		return true, 0, nil
+	err = errors.CombineErrors(err, it.Close())
+
+	if err != nil || len(candidates) == 0 {
+		return len(candidates) == 0, 0, err
 	}
 
-	log.VEventf(ctx, 2, "read potentially expired jobs: %d", numRows)
-	if len(toDelete.Array) > 0 {
-		log.VEventf(ctx, 2, "attempting to clean up %d expired job records", len(toDelete.Array))
-		const stmt = `DELETE FROM system.jobs WHERE id = ANY($1)`
-		nDeleted, err := r.db.Executor().Exec(
-			ctx, "gc-jobs", nil /* txn */, stmt, toDelete,
-		)
-		if err != nil {
-			log.Warningf(ctx, "error cleaning up %d jobs: %v", len(toDelete.Array), err)
-			return false, 0, errors.Wrap(err, "deleting old jobs")
-		}
-
-		counts := make(map[string]int)
-		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.
-				deleted, err = txn.Exec(ctx, redact.RedactableString("gc-job-"+tbl), txn.KV(),
-					"DELETE FROM system."+tbl+" WHERE job_id = ANY($1)", toDelete,
-				)
-				if err != nil {
-					return err
-				}
-				return nil
-			}); err != nil {
-				return false, 0, errors.Wrapf(err, "deleting old job metadata from %s", tbl)
-			}
-			counts[tbl] = deleted
-		}
-		if nDeleted > 0 {
-			log.Infof(ctx, "cleaned up %d expired job records (%d infos, %d progresses, %d progress_hists, %d statuses, %d messages)",
-				nDeleted, counts["job_info"], counts["job_progress"], counts["job_progress_history"], counts["job_status"], counts["job_message"])
+	log.VEventf(ctx, 2, "found %d expired jobs", len(candidates))
+	for _, id := range candidates {
+		if err := r.DeleteTerminalJobByID(ctx, id); err != nil {
+			return false, 0, err
 		}
 	}
-	// If we got as many rows as we asked for, there might be more.
-	morePages := numRows == pageSize
-	// Track the highest ID we encounter, so it can serve as the bottom of the
-	// next page.
-	lastRow := it.Cur()
-	maxID = jobspb.JobID(*(lastRow[0].(*tree.DInt)))
-	return !morePages, maxID, nil
+	return len(candidates) < pageSize, candidates[len(candidates)-1], nil
 }
 
 // DeleteTerminalJobByID deletes the given job ID if it is in a

pkg/jobs/registry_test.go

@@ -196,9 +196,14 @@ INSERT INTO t."%s" VALUES('a', 'foo');
 			t.Fatal(err)
 		}
 
+		finishedOpt := &finished
+		if finished == (time.Time{}) {
+			finishedOpt = nil
+		}
+
 		var id jobspb.JobID
 		db.QueryRow(t,
-			`INSERT INTO system.jobs (status, created, job_type) VALUES ($1, $2, 'SCHEMA CHANGE') RETURNING id`, state, created).Scan(&id)
+			`INSERT INTO system.jobs (status, created, job_type, finished) VALUES ($1, $2, 'SCHEMA CHANGE', $3) RETURNING id`, state, created, finishedOpt).Scan(&id)
 		db.Exec(t, `INSERT INTO system.job_info (job_id, info_key, value) VALUES ($1, $2, $3)`, id, GetLegacyPayloadKey(), payload)
 		db.Exec(t, `INSERT INTO system.job_info (job_id, info_key, value) VALUES ($1, $2, $3)`, id, GetLegacyProgressKey(), progress)
 		return strconv.Itoa(int(id))
@@ -251,8 +256,7 @@ INSERT INTO t."%s" VALUES('a', 'foo');
 			if err := s.JobRegistry().(*Registry).cleanupOldJobs(ctx, ts.Add(time.Minute*-10)); err != nil {
 				t.Fatal(err)
 			}
-			db.CheckQueryResults(t, selectJobsQuery, [][]string{
-				{oldRunningJob}, {oldRevertFailedJob}, {newRunningJob}, {newRevertFailedJob}})
+			db.CheckQueryResults(t, selectJobsQuery, [][]string{{oldRunningJob}, {oldRevertFailedJob}, {newRunningJob}, {newRevertFailedJob}})
 
 			// Delete the revert failed, and running jobs for the next run of the
 			// test.
@@ -300,7 +304,7 @@ func TestRegistryGCPagination(t *testing.T) {
 		require.NoError(t, err)
 		var jobID jobspb.JobID
 		db.QueryRow(t,
-			`INSERT INTO system.jobs (status, created) VALUES ($1, $2) RETURNING id`,
+			`INSERT INTO system.jobs (status, created, finished) VALUES ($1, $2, $2::timestamptz) RETURNING id`,
 			StateCanceled, timeutil.Now().Add(-time.Hour)).Scan(&jobID)
 		db.Exec(t, `INSERT INTO system.job_info (job_id, info_key, value) VALUES ($1, $2, $3)`,
 			jobID, GetLegacyPayloadKey(), payload)