changefeedccl: restore persisted span frontier from job_info

Changefeeds will now restore their persisted span frontiers on restart.

Release note: None

Author: andyyang890

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

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/sql/execinfrapb/processors_changefeeds.proto

@@ -64,6 +64,10 @@ message ChangeAggregatorSpec {
 
   // ProgressConfig is the configuration for the changefeed's progress tracking.
   optional ChangefeedProgressConfig progress_config = 10;
+
+  // ResolvedSpans contains the resolved spans that should be restored
+  // when the changefeed resumes.
+  repeated cockroach.sql.jobs.jobspb.ResolvedSpan resolved_spans = 11 [(gogoproto.nullable) = false];
 }
 
 // ChangeFrontierSpec is the specification for a processor that receives
@@ -98,6 +102,10 @@ message ChangeFrontierSpec {
 
   // ProgressConfig is the configuration for the changefeed's progress tracking.
   optional ChangefeedProgressConfig progress_config = 7;
+
+  // ResolvedSpans contains the resolved spans that should be restored
+  // when the changefeed resumes.
+  repeated cockroach.sql.jobs.jobspb.ResolvedSpan resolved_spans = 8 [(gogoproto.nullable) = false];
 }
 
 // ChangefeedProgressConfig is the configuration for the changefeed's progress tracking.