sql/inspect: optimize progress updates to avoid unnecessary writes

Previously, INSPECT job progress flushes occurred unconditionally on
periodic intervals, writing to the jobs table and jobfrontier even when
no progress had changed.

This change adds conditional checks to both flushProgress and
flushCheckpointUpdate to skip writes when nothing has changed. For
fraction updates, we now compare against the persisted check counts.
For checkpoint updates, we track the raw count of received spans
(before merging) to detect new completions.

Release note: None

Informs: #154457
Epic: CRDB-30356

Author: spilchen

pkg/sql/inspect/progress.go

@@ -48,6 +48,11 @@ type inspectProgressTracker struct {
 		cachedProgress *jobspb.Progress
 		// completedSpans tracks all completed spans with automatic deduplication.
 		completedSpans roachpb.SpanGroup
+		// receivedSpanCount tracks the total number of spans received from metadata.
+		// This is used to determine if we have new spans to checkpoint.
+		receivedSpanCount int
+		// lastCheckpointedSpanCount tracks the span count at the last checkpoint write.
+		lastCheckpointedSpanCount int
 	}
 
 	// Goroutine management.
@@ -208,6 +213,7 @@ func (t *inspectProgressTracker) updateProgressCache(
 	// automatic deduplication and merging.
 	if len(meta.BulkProcessorProgress.CompletedSpans) > 0 {
 		t.mu.completedSpans.Add(meta.BulkProcessorProgress.CompletedSpans...)
+		t.mu.receivedSpanCount += len(meta.BulkProcessorProgress.CompletedSpans)
 	}
 
 	// Update check count.
@@ -299,13 +305,22 @@ func (t *inspectProgressTracker) flushProgress(ctx context.Context) error {
 		cachedInspectProgress = t.mu.cachedProgress.GetInspect()
 	}()
 
-	// Calculate fraction complete based on check counts from cached progress.
-	var fractionComplete float32
-	if cachedInspectProgress.JobTotalCheckCount > 0 {
-		fractionComplete = float32(cachedInspectProgress.JobCompletedCheckCount) / float32(cachedInspectProgress.JobTotalCheckCount)
-	}
-
 	return t.job.NoTxn().Update(ctx, func(_ isql.Txn, md jobs.JobMetadata, ju *jobs.JobUpdater) error {
+		// Only write if any part of the inspect progress has changed.
+		currentInspectProgress := md.Progress.GetInspect()
+		if currentInspectProgress != nil &&
+			currentInspectProgress.JobTotalCheckCount == cachedInspectProgress.JobTotalCheckCount &&
+			currentInspectProgress.JobCompletedCheckCount == cachedInspectProgress.JobCompletedCheckCount &&
+			md.Progress.GetProgress() != nil { // Ensure fraction complete has been initialized.
+			return nil
+		}
+
+		// Calculate fraction complete based on check counts from cached progress.
+		var fractionComplete float32
+		if cachedInspectProgress.JobTotalCheckCount > 0 {
+			fractionComplete = float32(cachedInspectProgress.JobCompletedCheckCount) / float32(cachedInspectProgress.JobTotalCheckCount)
+		}
+
 		newProgress := &jobspb.Progress{
 			Details: &jobspb.Progress_Inspect{
 				Inspect: cachedInspectProgress,
@@ -322,19 +337,25 @@ func (t *inspectProgressTracker) flushProgress(ctx context.Context) error {
 // flushCheckpointUpdate performs a progress update to include completed spans.
 // The completed spans are stored via jobfrontier.
 func (t *inspectProgressTracker) flushCheckpointUpdate(ctx context.Context) error {
+	if !t.hasUncheckpointedSpans() {
+		return nil
+	}
+
 	var completedSpans []roachpb.Span
+	var capturedReceivedCount int
 	func() {
 		t.mu.Lock()
 		defer t.mu.Unlock()
 		completedSpans = t.mu.completedSpans.Slice()
+		capturedReceivedCount = t.mu.receivedSpanCount
 	}()
 
 	// If no completed spans, nothing to store.
 	if len(completedSpans) == 0 {
 		return nil
 	}
 
-	return t.internalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+	err := t.internalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
 		// Create a frontier for the spans (using zero timestamps since INSPECT doesn't need timing).
 		frontier, err := span.MakeFrontier(completedSpans...)
 		if err != nil {
@@ -343,6 +364,20 @@ func (t *inspectProgressTracker) flushCheckpointUpdate(ctx context.Context) erro
 		defer frontier.Release()
 		return jobfrontier.Store(ctx, txn, t.jobID, inspectCompletedSpansKey, frontier)
 	})
+
+	// If the checkpoint write succeeded, update the last checkpointed span count.
+	// This prevents unnecessary future writes when no new spans have been completed.
+	if err == nil {
+		func() {
+			t.mu.Lock()
+			defer t.mu.Unlock()
+			if capturedReceivedCount > t.mu.lastCheckpointedSpanCount {
+				t.mu.lastCheckpointedSpanCount = capturedReceivedCount
+			}
+		}()
+	}
+
+	return err
 }
 
 // getCachedCheckCounts returns the current cached check counts (total and completed).
@@ -363,6 +398,14 @@ func (t *inspectProgressTracker) getCachedCheckCounts() (totalChecks int64, comp
 	return inspectProgress.JobTotalCheckCount, inspectProgress.JobCompletedCheckCount
 }
 
+// hasUncheckpointedSpans returns true if there are completed spans that have not been
+// checkpointed yet. This is used to determine if a checkpoint flush is needed.
+func (t *inspectProgressTracker) hasUncheckpointedSpans() bool {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return t.mu.receivedSpanCount > t.mu.lastCheckpointedSpanCount
+}
+
 // countApplicableChecks determines how many checks will actually run across all spans.
 // This provides accurate progress calculation by only counting checks that apply to each span.
 func countApplicableChecks(

pkg/sql/inspect/progress_test.go

@@ -7,7 +7,9 @@ package inspect
 
 import (
 	"context"
+	"math"
 	"testing"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
@@ -17,6 +19,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
+	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -437,3 +440,182 @@ func createProcessorProgressUpdate(
 
 	return meta, nil
 }
+
+func TestInspectProgressTracker_ProgressFlushConditions(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx, s, _, _, cleanup := setupProgressTestInfra(t)
+	defer cleanup()
+
+	const totalChecks = 1000
+
+	testCases := []struct {
+		name                      string
+		setupFunc                 func(t *testing.T, tracker *inspectProgressTracker, job *jobs.Job)
+		expectedFraction          float32
+		expectUncheckpointedSpans bool
+	}{
+		{
+			name: "initial state - no progress updates",
+			setupFunc: func(t *testing.T, tracker *inspectProgressTracker, job *jobs.Job) {
+				require.NoError(t, tracker.initJobProgress(ctx, totalChecks, 0))
+			},
+			expectedFraction:          0.0,
+			expectUncheckpointedSpans: false,
+		},
+		{
+			name: "check count updates without spans",
+			setupFunc: func(t *testing.T, tracker *inspectProgressTracker, job *jobs.Job) {
+				require.NoError(t, tracker.initJobProgress(ctx, totalChecks, 0))
+
+				// Send progress updates with check counts but no spans.
+				meta, err := createProcessorProgressUpdate(100, false, nil)
+				require.NoError(t, err)
+				_, err = tracker.updateProgressCache(meta)
+				require.NoError(t, err)
+			},
+			expectedFraction:          0.1,
+			expectUncheckpointedSpans: false,
+		},
+		{
+			name: "span updates trigger checkpoint need",
+			setupFunc: func(t *testing.T, tracker *inspectProgressTracker, job *jobs.Job) {
+				require.NoError(t, tracker.initJobProgress(ctx, totalChecks, 0))
+
+				// Send progress with completed spans.
+				spans := []roachpb.Span{{Key: roachpb.Key("a"), EndKey: roachpb.Key("m")}}
+				meta, err := createProcessorProgressUpdate(50, false, spans)
+				require.NoError(t, err)
+				_, err = tracker.updateProgressCache(meta)
+				require.NoError(t, err)
+			},
+			expectedFraction:          0.05,
+			expectUncheckpointedSpans: true,
+		},
+		{
+			name: "automatic flush clears uncheckpointed state",
+			setupFunc: func(t *testing.T, tracker *inspectProgressTracker, job *jobs.Job) {
+				require.NoError(t, tracker.initJobProgress(ctx, totalChecks, 0))
+
+				// Send progress with completed spans.
+				spans := []roachpb.Span{{Key: roachpb.Key("a"), EndKey: roachpb.Key("m")}}
+				meta, err := createProcessorProgressUpdate(100, false, spans)
+				require.NoError(t, err)
+				_, err = tracker.updateProgressCache(meta)
+				require.NoError(t, err)
+
+				// Wait for automatic checkpoint flush.
+				testutils.SucceedsSoon(t, func() error {
+					if tracker.hasUncheckpointedSpans() {
+						return errors.New("still has uncheckpointed spans")
+					}
+					return nil
+				})
+			},
+			expectedFraction:          0.1,
+			expectUncheckpointedSpans: false,
+		},
+		{
+			name: "multiple span updates merge and checkpoint",
+			setupFunc: func(t *testing.T, tracker *inspectProgressTracker, job *jobs.Job) {
+				require.NoError(t, tracker.initJobProgress(ctx, totalChecks, 0))
+
+				// Send multiple progress updates with different spans.
+				spans1 := []roachpb.Span{{Key: roachpb.Key("a"), EndKey: roachpb.Key("d")}}
+				meta1, err := createProcessorProgressUpdate(100, false, spans1)
+				require.NoError(t, err)
+				_, err = tracker.updateProgressCache(meta1)
+				require.NoError(t, err)
+
+				spans2 := []roachpb.Span{{Key: roachpb.Key("d"), EndKey: roachpb.Key("g")}}
+				meta2, err := createProcessorProgressUpdate(100, false, spans2)
+				require.NoError(t, err)
+				_, err = tracker.updateProgressCache(meta2)
+				require.NoError(t, err)
+
+				// Wait for checkpoint to complete.
+				testutils.SucceedsSoon(t, func() error {
+					if tracker.hasUncheckpointedSpans() {
+						return errors.New("still has uncheckpointed spans")
+					}
+					return nil
+				})
+			},
+			expectedFraction:          0.2,
+			expectUncheckpointedSpans: false,
+		},
+		{
+			name: "immediate flush on drained processor",
+			setupFunc: func(t *testing.T, tracker *inspectProgressTracker, job *jobs.Job) {
+				require.NoError(t, tracker.initJobProgress(ctx, totalChecks, 0))
+
+				// Send progress with drained=true to trigger immediate flush.
+				spans := []roachpb.Span{{Key: roachpb.Key("a"), EndKey: roachpb.Key("z")}}
+				meta, err := createProcessorProgressUpdate(500, true, spans)
+				require.NoError(t, err)
+				require.NoError(t, tracker.handleProgressUpdate(ctx, meta))
+
+				// Immediate flush should have happened, so no uncheckpointed spans.
+				require.False(t, tracker.hasUncheckpointedSpans())
+			},
+			expectedFraction:          0.5,
+			expectUncheckpointedSpans: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			// Create a fresh job for each test case.
+			record := jobs.Record{
+				Details:  jobspb.InspectDetails{},
+				Progress: jobspb.InspectProgress{},
+				Username: username.TestUserName(),
+			}
+
+			freshJob, err := s.JobRegistry().(*jobs.Registry).CreateJobWithTxn(ctx, record, s.JobRegistry().(*jobs.Registry).MakeJobID(), nil)
+			require.NoError(t, err)
+
+			freshTracker := newInspectProgressTracker(freshJob, &s.ClusterSettings().SV, s.InternalDB().(isql.DB))
+			defer freshTracker.terminateTracker()
+
+			// Override intervals for faster testing.
+			const fastCheckpointInterval = 10 * time.Millisecond
+			const fastFractionInterval = 5 * time.Millisecond
+			freshTracker.checkpointInterval = func() time.Duration { return fastCheckpointInterval }
+			freshTracker.fractionInterval = func() time.Duration { return fastFractionInterval }
+
+			// Run the test setup.
+			tc.setupFunc(t, freshTracker, freshJob)
+
+			// Verify uncheckpointed spans state.
+			require.Equal(t, tc.expectUncheckpointedSpans, freshTracker.hasUncheckpointedSpans(),
+				"unexpected uncheckpointed spans state")
+
+			// Verify fraction complete.
+			progress := freshJob.Progress()
+			fractionCompleted, ok := progress.Progress.(*jobspb.Progress_FractionCompleted)
+			require.True(t, ok, "progress should be FractionCompleted type")
+			if tc.expectedFraction == 0.0 {
+				// For zero expected fraction, check immediately.
+				require.Equal(t, tc.expectedFraction, fractionCompleted.FractionCompleted)
+			} else {
+				// For non-zero expected fraction, wait for async flush to complete.
+				testutils.SucceedsSoon(t, func() error {
+					progress = freshJob.Progress()
+					fractionCompleted, ok = progress.Progress.(*jobspb.Progress_FractionCompleted)
+					if !ok {
+						return errors.New("progress should be FractionCompleted type")
+					}
+					// Check if fraction is within epsilon (1% tolerance).
+					const epsilon = 0.01
+					if math.Abs(float64(fractionCompleted.FractionCompleted-tc.expectedFraction)) > epsilon {
+						return errors.Newf("fraction complete not within epsilon: expected %f ± %f, got %f",
+							tc.expectedFraction, epsilon, fractionCompleted.FractionCompleted)
+					}
+					return nil
+				})
+			}
+		})
+	}
+}