Merge pull request #155965 from jeffswenson/backport25.2-155809

release-25.2: jobs: clean up corrupted auto SQL stats compaction

Author: jeffswenson

pkg/jobs/registry.go

@@ -1178,10 +1178,14 @@ func (r *Registry) cleanupOldJobs(ctx context.Context, olderThan time.Time) erro
 		var done bool
 		var err error
 		done, maxID, err = r.cleanupOldJobsPage(ctx, olderThan, maxID, cleanupPageSize)
-		if err != nil || done {
+		if err != nil {
 			return err
 		}
+		if done {
+			break
+		}
 	}
+	return r.CleanupCorruptJobs(ctx)
 }
 
 // AbandonedJobInfoRowsCleanupQuery is used by the CLI command
@@ -1318,6 +1322,36 @@ func (r *Registry) cleanupOldJobsPage(
 	return !morePages, maxID, nil
 }
 
+const findCorruptJobsQuery = `
+SELECT id
+FROM system.jobs
+LEFT JOIN system.job_info ON system.jobs.id = system.job_info.job_id
+WHERE system.job_info.job_id IS NULL AND system.jobs.job_type = 'AUTO SQL STATS COMPACTION'
+`
+
+// CleanupCorruptJobs is a temporary cleanup function that deletes corrupt
+// `AUTO SQL STATS COMPACTION` jobs. This function exists to clean up after
+// #155165.
+//
+// TODO(jeffswenson): in a separate PR we should run this as a migration so we
+// can guarantee the issue was cleaned up as of a specific version.
+func (r *Registry) CleanupCorruptJobs(ctx context.Context) error {
+	return r.db.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+		datums, err := txn.QueryBuffered(ctx, "get-corrupt-jobs", txn.KV(), findCorruptJobsQuery)
+		if err != nil {
+			return errors.Wrap(err, "querying for broken sql activity stats compaction jobs")
+		}
+		for _, row := range datums {
+			id := jobspb.JobID(tree.MustBeDInt(row[0]))
+			log.Dev.Errorf(ctx, "resetting broken sql activity stats compaction job %d", id)
+			if err := r.deleteJob(ctx, txn, id); err != nil {
+				return errors.Wrapf(err, "deleting broken sql activity stats compaction job %d", id)
+			}
+		}
+		return nil
+	})
+}
+
 // DeleteTerminalJobByID deletes the given job ID if it is in a
 // terminal state. If it is is in a non-terminal state, an error is
 // returned. This API should not be used.
@@ -1334,36 +1368,40 @@ func (r *Registry) DeleteTerminalJobByID(ctx context.Context, id jobspb.JobID) e
 		state := State(*row[0].(*tree.DString))
 		switch state {
 		case StateSucceeded, StateCanceled, StateFailed:
-			_, err := txn.Exec(
-				ctx, "delete-job", txn.KV(), "DELETE FROM system.jobs WHERE id = $1", id,
-			)
+			return r.deleteJob(ctx, txn, id)
+		default:
+			return errors.Newf("job %d has non-terminal state: %q", id, state)
+		}
+	})
+}
+
+func (r *Registry) deleteJob(ctx context.Context, txn isql.Txn, id jobspb.JobID) error {
+	_, err := txn.Exec(
+		ctx, "delete-job", txn.KV(), "DELETE FROM system.jobs WHERE id = $1", id,
+	)
+	if err != nil {
+		return err
+	}
+	for i, tbl := range jobMetadataTables {
+		if i > 0 {
+			v, err := txn.GetSystemSchemaVersion(ctx)
 			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.V25_1_AddJobsTables.Version()) {
-						break
-					}
-				}
-
-				_, err = txn.Exec(
-					ctx, redact.RedactableString("delete-job-"+tbl), txn.KV(),
-					"DELETE FROM system."+tbl+" WHERE job_id = $1", id,
-				)
-				if err != nil {
-					return err
-				}
+			if v.Less(clusterversion.V25_1_AddJobsTables.Version()) {
+				break
 			}
-			return nil
-		default:
-			return errors.Newf("job %d has non-terminal state: %q", id, state)
 		}
-	})
+
+		_, err = txn.Exec(
+			ctx, redact.RedactableString("delete-job-"+tbl), txn.KV(),
+			"DELETE FROM system."+tbl+" WHERE job_id = $1", id,
+		)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
 }
 
 // PauseRequested marks the job with id as paused-requested using the specified txn (may be nil).

pkg/jobs/registry_test.go

@@ -864,7 +864,39 @@ func TestDeleteTerminalJobByID(t *testing.T) {
 		require.NoError(t, r.DeleteTerminalJobByID(ctx, j.ID()))
 		assertValidDeletion(j.ID())
 	})
+}
+
+func TestCleanupCorruptJobs(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	s, sqlDB, _ := serverutils.StartServer(t, base.TestServerArgs{})
+	defer s.Stopper().Stop(ctx)
+
+	db := sqlutils.MakeSQLRunner(sqlDB)
+	r := s.ApplicationLayer().JobRegistry().(*Registry)
 
+	// Create a SQL STATS COMPACTION job in running state with proper job_info records.
+	jobID := r.MakeJobID()
+	db.Exec(t, `INSERT INTO system.jobs (id, status, created, job_type) VALUES ($1, 'running', now(), $2)`, jobID, jobspb.TypeAutoSQLStatsCompaction.String())
+	db.Exec(t, `INSERT INTO system.job_info (job_id, info_key, value) VALUES ($1, 'legacy_payload', 'payload')`, jobID)
+
+	// Job has info records, so CleanupCorruptJobs should do nothing.
+	require.NoError(t, r.CleanupCorruptJobs(ctx))
+
+	var count int
+	db.QueryRow(t, "SELECT count(*) FROM system.jobs WHERE id = $1", jobID).Scan(&count)
+	require.Equal(t, 1, count)
+
+	// Delete the job_info records to corrupt the running job.
+	db.Exec(t, "DELETE FROM system.job_info WHERE job_id = $1", jobID)
+
+	// Now CleanupCorruptJobs should delete the corrupt job.
+	require.NoError(t, r.CleanupCorruptJobs(ctx))
+
+	db.QueryRow(t, "SELECT count(*) FROM system.jobs WHERE id = $1", jobID).Scan(&count)
+	require.Zero(t, count)
 }
 
 // TestRunWithoutLoop tests that Run calls will trigger the execution of a
@@ -1073,7 +1105,6 @@ func TestJobIdleness(t *testing.T) {
 			})
 		}
 	})
-
 }
 
 // TestDisablingJobAdoptionClearsClaimSessionID tests that jobs adopted by a

pkg/sql/compact_sql_stats.go

@@ -174,6 +174,7 @@ func (e *scheduledSQLStatsCompactionExecutor) ExecuteJob(
 ) error {
 	if err := e.createSQLStatsCompactionJob(ctx, cfg, sj, txn); err != nil {
 		e.metrics.NumFailed.Inc(1)
+		return err
 	}
 
 	e.metrics.NumStarted.Inc(1)