Merge #153621 #153652 #153659 #153680

153621: changefeedccl: restore persisted span frontier from job_info r=KeithCh a=andyyang890

Changefeeds will now restore their persisted span frontiers on restart.

Fixes #153201

Release note: None

153652: mmaprototype: thread ctx r=tbg a=tbg

This will make the traces in #153584 more informative.

Epic: CRDB-49117

153659: changefeedccl: delete per-table resolved timestamps code r=log-head a=andyyang890

Now that we periodically persist the entire span frontier, there is
no longer a need for per-table resolved timestamps and thus all the
relevant code can be deleted.

Fixes #153492

Release note: None

153680: roachtest: increase the verbosity of runMultiStoreRemove r=iskettaneh a=iskettaneh

Recent failures in runMultiStoreRemove seems to indicate a potential race between follower replica removal and leader fortification.

References: #153517

Release note: None

Co-authored-by: Andy Yang <[email protected]>
Co-authored-by: Tobias Grieger <[email protected]>
Co-authored-by: Ibrahim Kettaneh <[email protected]>

Author: 4 people

pkg/ccl/changefeedccl/cdcprogresspb/BUILD.bazel

@@ -7,11 +7,7 @@ proto_library(
     srcs = ["progress.proto"],
     strip_import_prefix = "/pkg",
     visibility = ["//visibility:public"],
-    deps = [
-        "//pkg/jobs/jobspb:jobspb_proto",
-        "//pkg/util/hlc:hlc_proto",
-        "@com_github_gogo_protobuf//gogoproto:gogo_proto",
-    ],
+    deps = ["@com_github_gogo_protobuf//gogoproto:gogo_proto"],
 )
 
 go_proto_library(
@@ -20,11 +16,7 @@ go_proto_library(
     importpath = "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcprogresspb",
     proto = ":cdcprogresspb_proto",
     visibility = ["//visibility:public"],
-    deps = [
-        "//pkg/jobs/jobspb",
-        "//pkg/util/hlc",
-        "@com_github_gogo_protobuf//gogoproto",
-    ],
+    deps = ["@com_github_gogo_protobuf//gogoproto"],
 )
 
 go_library(

pkg/ccl/changefeedccl/cdcprogresspb/progress.proto

@@ -8,19 +8,6 @@ package cockroach.ccl.changefeedccl.cdcprogresspb;
 option go_package = "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcprogresspb";
 
 import "gogoproto/gogo.proto";
-import "jobs/jobspb/jobs.proto";
-import "util/hlc/timestamp.proto";
-
-// ResolvedTables contains per-table resolved timestamp information.
-message ResolvedTables {
-  map<uint32, util.hlc.Timestamp> tables = 1 [
-    (gogoproto.castkey) = "github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb.ID",
-    (gogoproto.nullable) = false
-  ];
-
-  // TODO(#148124): Write to this field for aggregator-to-frontier messages.
-  cockroach.sql.jobs.jobspb.ResolvedSpan.BoundaryType boundary_type = 2;
-}
 
 // ProtectedTimestampRecords is a map from table descriptor IDs to protected timestamp record IDs.
 message ProtectedTimestampRecords {

pkg/ccl/changefeedccl/changefeed_dist.go

@@ -15,6 +15,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/ccl/kvccl/kvfollowerreadsccl"
 	"github.com/cockroachdb/cockroach/pkg/clusterversion"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobfrontier"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobsprofiler"
 	"github.com/cockroachdb/cockroach/pkg/keys"
@@ -470,6 +471,22 @@ func makePlan(
 			}
 		}
 
+		var resolvedSpans []jobspb.ResolvedSpan
+		if jobID != 0 {
+			if err := execCtx.ExecCfg().InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+				spans, ok, err := jobfrontier.GetAllResolvedSpans(ctx, txn, jobID)
+				if err != nil {
+					return err
+				}
+				if ok {
+					resolvedSpans = spans
+				}
+				return nil
+			}); err != nil {
+				return nil, nil, err
+			}
+		}
+
 		aggregatorSpecs := make([]*execinfrapb.ChangeAggregatorSpec, len(spanPartitions))
 		for i, sp := range spanPartitions {
 			if log.ExpensiveLogEnabled(ctx, 2) {
@@ -493,6 +510,7 @@ func makePlan(
 				Select:              execinfrapb.Expression{Expr: details.Select},
 				Description:         description,
 				ProgressConfig:      progressConfig,
+				ResolvedSpans:       resolvedSpans,
 			}
 		}
 
@@ -508,6 +526,7 @@ func makePlan(
 			UserProto:           execCtx.User().EncodeProto(),
 			Description:         description,
 			ProgressConfig:      progressConfig,
+			ResolvedSpans:       resolvedSpans,
 		}
 
 		if haveKnobs && maybeCfKnobs.OnDistflowSpec != nil {

pkg/ccl/changefeedccl/changefeed_job_info.go

@@ -23,8 +23,6 @@ const (
 )
 
 const (
-	// resolvedTablesFilename: ~changefeed/resolved-tables.binpb
-	resolvedTablesFilename = jobInfoFilenamePrefix + "resolved-tables" + jobInfoFilenameExtension
 	// perTableProtectedTimestampsFilename: ~changefeed/per-table-protected-timestamps.binpb
 	perTableProtectedTimestampsFilename = jobInfoFilenamePrefix + "per-table-protected-timestamps" + jobInfoFilenameExtension
 )
@@ -49,7 +47,6 @@ func writeChangefeedJobInfo(
 
 // readChangefeedJobInfo reads a changefeed job info protobuf from the
 // job_info table. A changefeed-specific filename is required.
-// TODO(#148119): Use this function to read.
 func readChangefeedJobInfo(
 	ctx context.Context, filename string, info protoutil.Message, txn isql.Txn, jobID jobspb.JobID,
 ) error {

pkg/ccl/changefeedccl/changefeed_job_info_test.go

@@ -9,55 +9,53 @@ import (
 	"context"
 	"testing"
 
+	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcprogresspb"
-	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdctest"
-	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
-	"github.com/cockroachdb/cockroach/pkg/jobs"
-	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
-	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/uuid"
 	"github.com/stretchr/testify/require"
 )
 
-func TestChangefeedJobInfoResolvedTables(t *testing.T) {
+func TestChangefeedJobInfoRoundTrip(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
-		ctx := context.Background()
-		sqlDB := sqlutils.MakeSQLRunner(s.DB)
-
-		// Make sure per-table tracking is enabled.
-		changefeedbase.TrackPerTableProgress.Override(ctx, &s.Server.ClusterSettings().SV, true)
-
-		sqlDB.Exec(t, `CREATE TABLE foo (a INT PRIMARY KEY, b STRING)`)
-		sqlDB.Exec(t, `CREATE TABLE bar (x INT PRIMARY KEY, y STRING)`)
-		feed := feed(t, f, `CREATE CHANGEFEED FOR foo, bar`)
-		defer closeFeed(t, feed)
-
-		sqlDB.Exec(t, `INSERT INTO foo VALUES (1, 'one')`)
-		sqlDB.Exec(t, `INSERT INTO bar VALUES (10, 'ten')`)
-		assertPayloads(t, feed, []string{
-			`foo: [1]->{"after": {"a": 1, "b": "one"}}`,
-			`bar: [10]->{"after": {"x": 10, "y": "ten"}}`,
-		})
-
-		// The ResolvedTables message should be persisted to the job_info table
-		// at the same time as the highwater being set.
-		enterpriseFeed := feed.(cdctest.EnterpriseTestFeed)
-		waitForHighwater(t, enterpriseFeed, s.Server.JobRegistry().(*jobs.Registry))
-
-		// Make sure the ResolvedTables message was persisted and can be decoded.
-		var resolvedTables cdcprogresspb.ResolvedTables
-		execCfg := s.Server.ExecutorConfig().(sql.ExecutorConfig)
-		err := execCfg.InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
-			return readChangefeedJobInfo(ctx, resolvedTablesFilename, &resolvedTables, txn, enterpriseFeed.JobID())
-		})
-		require.NoError(t, err)
-		require.Len(t, resolvedTables.Tables, 2)
+	ctx := context.Background()
+	srv := serverutils.StartServerOnly(t, base.TestServerArgs{})
+	defer srv.Stopper().Stop(ctx)
+
+	jobID := jobspb.JobID(123456)
+
+	// Create a basic progress record.
+	uuid1 := uuid.MakeV4()
+	uuid2 := uuid.MakeV4()
+	ptsRecords := cdcprogresspb.ProtectedTimestampRecords{
+		ProtectedTimestampRecords: map[descpb.ID]*uuid.UUID{
+			descpb.ID(100): &uuid1,
+			descpb.ID(200): &uuid2,
+		},
 	}
 
-	cdcTest(t, testFn, feedTestEnterpriseSinks)
+	// Write the progress record.
+	err := srv.InternalDB().(isql.DB).Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+		return writeChangefeedJobInfo(ctx,
+			perTableProtectedTimestampsFilename, &ptsRecords, txn, jobID)
+	})
+	require.NoError(t, err)
+
+	// Read the record back.
+	var readPTSRecords cdcprogresspb.ProtectedTimestampRecords
+	err = srv.InternalDB().(isql.DB).Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+		return readChangefeedJobInfo(ctx,
+			perTableProtectedTimestampsFilename, &readPTSRecords, txn, jobID)
+	})
+	require.NoError(t, err)
+
+	// Verify the read data matches the written data.
+	require.Equal(t, ptsRecords, readPTSRecords)
 }

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -656,6 +656,12 @@ func (ca *changeAggregator) setupSpansAndFrontier() (spans []roachpb.Span, err e
 		return nil, errors.Wrapf(err, "failed to restore span-level checkpoint")
 	}
 
+	for _, rs := range ca.spec.ResolvedSpans {
+		if _, err := ca.frontier.Forward(rs.Span, rs.Timestamp); err != nil {
+			return nil, errors.Wrapf(err, "failed to restore frontier")
+		}
+	}
+
 	return spans, nil
 }
 
@@ -1480,6 +1486,15 @@ func (cf *changeFrontier) Start(ctx context.Context) {
 		return
 	}
 
+	for _, rs := range cf.spec.ResolvedSpans {
+		if _, err := cf.frontier.Forward(rs.Span, rs.Timestamp); err != nil {
+			log.Changefeed.Warningf(cf.Ctx(),
+				"moving to draining due to error restoring frontier: %v", err)
+			cf.MoveToDraining(err)
+			return
+		}
+	}
+
 	if cf.knobs.AfterCoordinatorFrontierRestore != nil {
 		cf.knobs.AfterCoordinatorFrontierRestore(cf.frontier)
 	}
@@ -1877,20 +1892,6 @@ func (cf *changeFrontier) checkpointJobProgress(
 				progress.StatusMessage = fmt.Sprintf("running: resolved=%s", frontier)
 			}
 
-			// Write per-table progress if enabled.
-			if cf.spec.ProgressConfig != nil && cf.spec.ProgressConfig.PerTableTracking {
-				resolvedTables := &cdcprogresspb.ResolvedTables{
-					Tables: make(map[descpb.ID]hlc.Timestamp),
-				}
-				for tableID, tableFrontier := range cf.frontier.Frontiers() {
-					resolvedTables.Tables[tableID] = tableFrontier.Frontier()
-				}
-
-				if err := writeChangefeedJobInfo(ctx, resolvedTablesFilename, resolvedTables, txn, cf.spec.JobID); err != nil {
-					return errors.Wrap(err, "error writing resolved tables to job info")
-				}
-			}
-
 			ju.UpdateProgress(progress)
 
 			return nil

pkg/ccl/changefeedccl/changefeed_progress_test.go

@@ -11,14 +11,20 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdctest"
+	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobfrontier"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/desctestutils"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
+	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
+	"github.com/cockroachdb/cockroach/pkg/util/encoding"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/span"
 	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/require"
 )
@@ -54,9 +60,9 @@ func TestChangefeedFrontierPersistence(t *testing.T) {
 
 		// Make sure frontier gets persisted to job_info table.
 		jobID := foo.(cdctest.EnterpriseTestFeed).JobID()
+		var allSpans []jobspb.ResolvedSpan
 		testutils.SucceedsSoon(t, func() error {
 			var found bool
-			var allSpans []jobspb.ResolvedSpan
 			if err := s.Server.InternalDB().(isql.DB).Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
 				var err error
 				allSpans, found, err = jobfrontier.GetAllResolvedSpans(ctx, txn, jobID)
@@ -74,6 +80,17 @@ func TestChangefeedFrontierPersistence(t *testing.T) {
 			return nil
 		})
 
+		// Make sure the persisted spans cover the entire table.
+		fooTableSpan := desctestutils.
+			TestingGetPublicTableDescriptor(s.Server.DB(), s.Codec, "d", "foo").
+			PrimaryIndexSpan(s.Codec)
+		var spanGroup roachpb.SpanGroup
+		spanGroup.Add(fooTableSpan)
+		for _, rs := range allSpans {
+			spanGroup.Sub(rs.Span)
+		}
+		require.Zero(t, spanGroup.Len())
+
 		// Verify metric count and average latency have sensible values.
 		testutils.SucceedsSoon(t, func() error {
 			metricSnapshot := metric.CumulativeSnapshot()
@@ -91,3 +108,74 @@ func TestChangefeedFrontierPersistence(t *testing.T) {
 
 	cdcTest(t, testFn, feedTestEnterpriseSinks)
 }
+
+// TestChangefeedFrontierRestore verifies that changefeeds will correctly
+// restore progress from persisted span frontiers.
+func TestChangefeedFrontierRestore(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+		ctx := context.Background()
+
+		// Disable span-level checkpointing.
+		changefeedbase.SpanCheckpointInterval.Override(ctx, &s.Server.ClusterSettings().SV, 0)
+
+		// Create a table and a changefeed on it.
+		sqlDB.Exec(t, "CREATE TABLE foo (a INT PRIMARY KEY)")
+		foo := feed(t, f, "CREATE CHANGEFEED FOR foo WITH initial_scan='no'")
+		defer closeFeed(t, foo)
+		jobFeed := foo.(cdctest.EnterpriseTestFeed)
+
+		// Pause the changefeed.
+		require.NoError(t, jobFeed.Pause())
+
+		// Insert a few rows into the table and save the insert time.
+		var tsStr string
+		sqlDB.QueryRow(t, `INSERT INTO foo VALUES (1), (2), (3), (4), (5), (6)
+RETURNING cluster_logical_timestamp()`).Scan(&tsStr)
+		ts := parseTimeToHLC(t, tsStr)
+
+		// Make function to create spans for single rows in the table.
+		codec := s.Server.Codec()
+		fooDesc := desctestutils.TestingGetPublicTableDescriptor(s.Server.DB(), codec, "d", "foo")
+		rowSpan := func(key int64) roachpb.Span {
+			keyPrefix := func() []byte {
+				return rowenc.MakeIndexKeyPrefix(codec, fooDesc.GetID(), fooDesc.GetPrimaryIndexID())
+			}
+			return roachpb.Span{
+				Key:    encoding.EncodeVarintAscending(keyPrefix(), key),
+				EndKey: encoding.EncodeVarintAscending(keyPrefix(), key+1),
+			}
+		}
+
+		// Manually persist a span frontier that manually marks some of the
+		// inserted rows as resolved already.
+		hw, err := jobFeed.HighWaterMark()
+		require.NoError(t, err)
+		frontier, err := span.MakeFrontierAt(hw, fooDesc.PrimaryIndexSpan(codec))
+		require.NoError(t, err)
+		for _, id := range []int64{2, 5} {
+			_, err := frontier.Forward(rowSpan(id), ts)
+			require.NoError(t, err)
+		}
+		err = s.Server.InternalDB().(isql.DB).Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+			return jobfrontier.Store(ctx, txn, jobFeed.JobID(), "test frontier", frontier)
+		})
+		require.NoError(t, err)
+
+		// Resume the changefeed.
+		require.NoError(t, jobFeed.Resume())
+
+		// We should receive rows 1, 3, 4, 6 (rows 2 and 5 were marked as resolved).
+		assertPayloads(t, foo, []string{
+			`foo: [1]->{"after": {"a": 1}}`,
+			`foo: [3]->{"after": {"a": 3}}`,
+			`foo: [4]->{"after": {"a": 4}}`,
+			`foo: [6]->{"after": {"a": 6}}`,
+		})
+	}
+
+	cdcTest(t, testFn, feedTestEnterpriseSinks)
+}

pkg/cmd/roachtest/tests/multi_store_remove.go

@@ -60,6 +60,12 @@ func runMultiStoreRemove(ctx context.Context, t test.Test, c cluster.Cluster) {
 	startSettings := install.MakeClusterSettings()
 	// Speed up the replicate queue.
 	startSettings.Env = append(startSettings.Env, "COCKROACH_SCAN_INTERVAL=30s")
+
+	// Increase the verbosity of this test to help debug future failures.
+	startOpts.RoachprodOpts.ExtraArgs = append(startOpts.RoachprodOpts.ExtraArgs,
+		"--vmodule=*=1,raft=3",
+	)
+
 	c.Start(ctx, t.L(), startOpts, startSettings, c.Range(1, 3))
 
 	// Confirm that there are 6 stores live.

pkg/kv/kvserver/allocator/mmaprototype/allocator.go

@@ -29,7 +29,7 @@ type ChangeOptions struct {
 //   - changes to this interface to make the integration for the new allocator
 //     be less different than integration with the old allocator.
 type Allocator interface {
-	LoadSummaryForAllStores() string
+	LoadSummaryForAllStores(context.Context) string
 	Metrics() *MMAMetrics
 	// Methods to update the state of the external world. The allocator starts
 	// with no knowledge.