changefeedccl: fix job progress regressing

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -337,6 +337,10 @@ type changeFrontier struct {
 	// jobProgressedFn, if non-nil, is called to checkpoint the changefeed's
 	// progress in the corresponding system job entry.
 	jobProgressedFn func(context.Context, jobs.HighWaterProgressedFn) error
+	// highWaterAtStart is the greater of the job high-water and the timestamp the
+	// CHANGEFEED statement was run at. It's used in an assertion that we never
+	// regress the job high-water.
+	highWaterAtStart hlc.Timestamp
 	// passthroughBuf, in some but not all flows, contains changed row data to
 	// pass through unchanged to the gateway node.
 	passthroughBuf encDatumRowBuffer
@@ -441,13 +445,19 @@ func (cf *changeFrontier) Start(ctx context.Context) context.Context {
 	cf.sink = makeMetricsSink(cf.metrics, cf.sink)
 	cf.sink = &errorWrapperSink{wrapped: cf.sink}
 
+	cf.highWaterAtStart = cf.spec.Feed.StatementTime
 	if cf.spec.JobID != 0 {
 		job, err := cf.flowCtx.JobRegistry.LoadJob(ctx, cf.spec.JobID)
 		if err != nil {
 			cf.MoveToDraining(err)
 			return ctx
 		}
 		cf.jobProgressedFn = job.HighWaterProgressed
+
+		p := job.Progress()
+		if ts := p.GetHighWater(); ts != nil {
+			cf.highWaterAtStart.Forward(*ts)
+		}
 	}
 
 	cf.metrics.mu.Lock()
@@ -535,6 +545,20 @@ func (cf *changeFrontier) noteResolvedSpan(d sqlbase.EncDatum) error {
 			`unmarshalling resolved span: %x`, raw)
 	}
 
+	// Inserting a timestamp less than the one the changefeed flow started at
+	// could potentially regress the job progress. This is not expected, but it
+	// was a bug at one point, so assert to prevent regressions.
+	//
+	// TODO(dan): This is much more naturally expressed as an assert inside the
+	// job progress update closure, but it currently doesn't pass along the info
+	// we'd need to do it that way.
+	if !resolved.Timestamp.IsEmpty() && resolved.Timestamp.Less(cf.highWaterAtStart) {
+		log.ReportOrPanic(cf.Ctx, &cf.flowCtx.Settings.SV,
+			`got a span level timestamp %s for %s that is less than the initial high-water %s`,
+			log.Safe(resolved.Timestamp), resolved.Span, log.Safe(cf.highWaterAtStart))
+		return nil
+	}
+
 	frontierChanged := cf.sf.Forward(resolved.Span, resolved.Timestamp)
 	if frontierChanged {
 		newResolved := cf.sf.Frontier()

pkg/ccl/changefeedccl/poller.go

@@ -279,16 +279,12 @@ func (p *poller) rangefeedImpl(ctx context.Context) error {
 		// the faster-to-implement solution for now.
 		frontier := makeSpanFrontier(spans...)
 
+		rangeFeedStartTS := lastHighwater
 		for _, span := range p.spans {
-			req := &roachpb.RangeFeedRequest{
-				Header: roachpb.Header{
-					Timestamp: lastHighwater,
-				},
-				Span: span,
-			}
-			frontier.Forward(span, lastHighwater)
+			span := span
+			frontier.Forward(span, rangeFeedStartTS)
 			g.GoCtx(func(ctx context.Context) error {
-				return ds.RangeFeed(ctx, req, eventC)
+				return ds.RangeFeed(ctx, span, rangeFeedStartTS, eventC)
 			})
 		}
 		g.GoCtx(func(ctx context.Context) error {
@@ -302,6 +298,12 @@ func (p *poller) rangefeedImpl(ctx context.Context) error {
 							return err
 						}
 					case *roachpb.RangeFeedCheckpoint:
+						if !t.ResolvedTS.IsEmpty() && t.ResolvedTS.Less(rangeFeedStartTS) {
+							// RangeFeed happily forwards any closed timestamps it receives as
+							// soon as there are no outstanding intents under them.
+							// Changefeeds don't care about these at all, so throw them out.
+							continue
+						}
 						if err := memBuf.AddResolved(ctx, t.Span, t.ResolvedTS); err != nil {
 							return err
 						}

pkg/kv/dist_sender_rangefeed.go

@@ -39,18 +39,21 @@ type singleRangeInfo struct {
 // RangeFeed divides a RangeFeed request on range boundaries and establishes a
 // RangeFeed to each of the individual ranges. It streams back results on the
 // provided channel.
+//
+// Note that the timestamps in RangeFeedCheckpoint events that are streamed back
+// may be lower than the timestamp given here.
 func (ds *DistSender) RangeFeed(
-	ctx context.Context, args *roachpb.RangeFeedRequest, eventCh chan<- *roachpb.RangeFeedEvent,
+	ctx context.Context, span roachpb.Span, ts hlc.Timestamp, eventCh chan<- *roachpb.RangeFeedEvent,
 ) error {
 	ctx = ds.AnnotateCtx(ctx)
 	ctx, sp := tracing.EnsureChildSpan(ctx, ds.AmbientContext.Tracer, "dist sender")
 	defer sp.Finish()
 
-	startRKey, err := keys.Addr(args.Span.Key)
+	startRKey, err := keys.Addr(span.Key)
 	if err != nil {
 		return err
 	}
-	endRKey, err := keys.Addr(args.Span.EndKey)
+	endRKey, err := keys.Addr(span.EndKey)
 	if err != nil {
 		return err
 	}
@@ -76,7 +79,7 @@ func (ds *DistSender) RangeFeed(
 
 	// Kick off the initial set of ranges.
 	g.GoCtx(func(ctx context.Context) error {
-		return ds.divideAndSendRangeFeedToRanges(ctx, rs, args.Timestamp, rangeCh)
+		return ds.divideAndSendRangeFeedToRanges(ctx, rs, ts, rangeCh)
 	})
 
 	return g.Wait()