ttl: add checkpointing to TTL job

Introduce a progress tracker for TTL jobs that supports checkpointing.
Previously, if a TTL job restarted, it would reprocess spans that had
already been completed. With this change, completed spans are stored
in the job record so that restarts can skip over them, avoiding
duplicate work.

Fixes #140514
Epic: none

Release note (performance improvement): TTL jobs now checkpoint their
progress, allowing them to resume without reprocessing already
completed spans after a restart.

Author: spilchen

pkg/jobs/jobspb/jobs.proto

@@ -1321,8 +1321,10 @@ message RowLevelTTLDetails {
 }
 
 message RowLevelTTLProgress {
-
-  // JobDeletedRowCount is the number of rows deleted by TTL job so far.
+  // JobDeletedRowCount is the number of rows deleted by TTL job so far. If
+  // UseCheckpointing is true, then this is the deleted row count across all
+  // restarts of the job. If that is false, this is the delete row count in
+  // the current running of the job.
   int64 job_deleted_row_count = 1;
 
   // ProcessorProgresses is the progress per DistSQL processor.
@@ -1331,12 +1333,18 @@ message RowLevelTTLProgress {
   // UseDistSQL is no longer used in v23.1+ as all TTL jobs are using DistSQL.
   reserved 3;
 
-  // JobTotalSpanCount is the number of spans for the entire TTL job.
-  int64 job_total_span_count = 4;
+  // JobTotalSpanCount is the number of spans for the entire TTL job. This is
+  // deprecated and not used if UseCheckpointing is true.
+  int64 job_total_span_count = 4 [deprecated = true];
 
   // JobProcessedSpanCount is the number of spans that have been processed by
-  // the TTL job so far.
-  int64 job_processed_span_count = 5;
+  // the TTL job so far. This is deprecated and not used if UseCheckpointing is
+  // true.
+  int64 job_processed_span_count = 5 [deprecated = true];
+
+  // UseCheckpointing is true if the TTL job will use checkpointing for progress
+  // tracking.
+  bool use_checkpointing = 6;
 }
 
 message RowLevelTTLProcessorProgress {
@@ -1354,11 +1362,13 @@ message RowLevelTTLProcessorProgress {
   // DeletedRowCount is the number of rows deleted by this DistSQL processor.
   int64 deleted_row_count = 3;
 
-  // TotalSpanCount is the total number of spans assigned to the DistSQL processor.
-  int64 total_span_count = 4;
+  // TotalSpanCount is the total number of spans assigned to the DistSQL
+  // processor. This is deprecated and not used if using checkpointing.
+  int64 total_span_count = 4 [deprecated = true];
 
-  // ProcessedSpanCount is the number of spans already processed.
-  int64 processed_span_count = 6;
+  // ProcessedSpanCount is the number of spans already processed. This is
+  // deprecated when using checkpointing.
+  int64 processed_span_count = 6 [deprecated = true];
 
   // ProcessorConcurrency is the number parallel tasks the processor will do at once.
   int64 processor_concurrency = 5;

pkg/sql/spanutils/query_bounds.go

@@ -40,6 +40,9 @@ type QueryBounds struct {
 	// span's end key is exclusive because the end bounds are based on the first
 	// row < Span.EndKey.
 	End tree.Datums
+	// Span is the original span that these query bounds were derived from.
+	// This preserves the source span information for reference.
+	Span roachpb.Span
 }
 
 var (
@@ -110,6 +113,7 @@ func SpanToQueryBounds(
 	if err != nil {
 		return bounds, false, errors.Wrapf(err, "decode endKeyValues error on %+v", endKeyValues)
 	}
+	bounds.Span = span
 	return bounds, true, nil
 }
 

pkg/sql/ttl/ttljob/BUILD.bazel

@@ -15,6 +15,7 @@ go_library(
         "//pkg/base",
         "//pkg/jobs",
         "//pkg/jobs/joberror",
+        "//pkg/jobs/jobfrontier",
         "//pkg/jobs/jobspb",
         "//pkg/kv",
         "//pkg/roachpb",
@@ -53,12 +54,14 @@ go_library(
         "//pkg/util/protoutil",
         "//pkg/util/quotapool",
         "//pkg/util/retry",
+        "//pkg/util/span",
         "//pkg/util/syncutil",
         "//pkg/util/timeutil",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_redact//:redact",
         "@com_github_gogo_protobuf//types",
         "@com_github_prometheus_client_model//go",
+        "@org_golang_x_sync//errgroup",
     ],
 )
 
@@ -83,6 +86,7 @@ go_test(
         "//pkg/ccl/kvccl/kvtenantccl",
         "//pkg/clusterversion",
         "//pkg/jobs",
+        "//pkg/jobs/jobfrontier",
         "//pkg/jobs/jobspb",
         "//pkg/jobs/jobstest",
         "//pkg/keys",