Merge #155055 #155057 #155067

155055: sql/inspect: optimize progress updates to avoid unnecessary writes r=spilchen a=spilchen

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

155057: tests: avoid use of loopback dialer in Benchmark{Parallel,}Sysbench r=tbg a=tbg

When follow-the-workload is disabled (#153865), leases no longer reliably stay on n1. Since the workload clients are always created on n1, they were able to use the loopback dialer (which avoids going through the TCP layer by using inter-progress communication instead) prior to that change.
We want these benchmarks to be independent of lease placement, so we disable the loopback dialer. That this was not done initially is likely an oversight.

This commit will cause a a benchmark regression, but it is due to changing what is being benchmarked. The regression does not result in any production change.

Informs #153866.

Epic: CRDB-55052
^-- because this is required for #153865.

155067: roachtest: only run liquibase test on GCE r=rafiss a=rafiss

The test needs custom ports, which only works in GCE.

fixes #155040
Release note: None

Co-authored-by: Matt Spilchen <[email protected]>
Co-authored-by: Tobias Grieger <[email protected]>
Co-authored-by: Rafi Shamim <[email protected]>

Author: 4 people

pkg/cmd/roachtest/tests/liquibase.go

@@ -156,11 +156,12 @@ sudo ln -sf /home/ubuntu/cockroach /cockroach/cockroach.sh`); err != nil {
 	}
 
 	r.Add(registry.TestSpec{
-		Name:             "liquibase",
-		Owner:            registry.OwnerSQLFoundations,
-		Cluster:          r.MakeClusterSpec(1),
-		Leases:           registry.MetamorphicLeases,
-		CompatibleClouds: registry.AllExceptAWS,
+		Name:    "liquibase",
+		Owner:   registry.OwnerSQLFoundations,
+		Cluster: r.MakeClusterSpec(1),
+		Leases:  registry.MetamorphicLeases,
+		// This test uses custom ports, which is Only supported on GCE.
+		CompatibleClouds: registry.OnlyGCE,
 		Suites:           registry.Suites(registry.Nightly, registry.Tool),
 		Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
 			runLiquibase(ctx, t, c)

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
+				})
+			}
+		})
+	}
+}

pkg/sql/tests/BUILD.bazel

@@ -89,6 +89,7 @@ go_test(
         "//pkg/kv/kvserver/concurrency",
         "//pkg/multitenant/tenantcapabilitiespb",
         "//pkg/roachpb",
+        "//pkg/rpc",
         "//pkg/rpc/nodedialer",
         "//pkg/security/securityassets",
         "//pkg/security/securitytest",

pkg/sql/tests/sysbench_test.go

@@ -27,7 +27,9 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/rpc"
 	"github.com/cockroachdb/cockroach/pkg/rpc/nodedialer"
+	"github.com/cockroachdb/cockroach/pkg/server"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
@@ -164,25 +166,27 @@ func newTestCluster(
 	// also moderately less realistic.
 	disableBackgroundWork(st)
 	const cacheSize = 2 * 1024 * 1024 * 1024 // 2GB
+	knobs := base.TestingKnobs{
+		DialerKnobs: nodedialer.DialerTestingKnobs{
+			TestingNoLocalClientOptimization: !localRPCFastPath,
+		},
+		Server: &server.TestingKnobs{
+			ContextTestingKnobs: rpc.ContextTestingKnobs{
+				NoLoopbackDialer: !localRPCFastPath,
+			},
+		},
+	}
 	return serverutils.StartCluster(b, nodes, base.TestClusterArgs{
 		ServerArgs: base.TestServerArgs{
 			Settings:  st,
 			CacheSize: cacheSize,
-			Knobs: base.TestingKnobs{
-				DialerKnobs: nodedialer.DialerTestingKnobs{
-					TestingNoLocalClientOptimization: !localRPCFastPath,
-				},
-			},
+			Knobs:     knobs,
 		}},
 	)
 }
 
 // sysbenchSQL is SQL-based implementation of sysbenchDriver. It runs SQL
 // statements against a single node cluster.
-//
-// TODO(nvanbenschoten): add a 3-node cluster variant of this driver.
-// TODO(nvanbenschoten): add a variant of this driver which bypasses the gRPC
-// local fast-path optimization.
 type sysbenchSQL struct {
 	ctx     context.Context
 	stopper *stop.Stopper
@@ -376,10 +380,6 @@ func (s *sysbenchSQLClient) prepConn() {
 // sysbenchKV is KV-based implementation of sysbenchDriver. It bypasses the SQL
 // layer and runs the workload directly against the KV layer, on a single node
 // cluster.
-//
-// TODO(nvanbenschoten): add a 3-node cluster variant of this driver.
-// TODO(nvanbenschoten): add a variant of this driver which bypasses the gRPC
-// local fast-path optimization.
 type sysbenchKV struct {
 	ctx         context.Context
 	db          *kv.DB