changefeedccl: fix premature shutdown due to schema change bug

This patch fixes a bug that could potentially cause a changefeed
to erroneously complete when one of its watched tables encounters a
schema change has been fixed.

The root cause for the bug was that if we happened to get a rangefeed
checkpoint at precisely `ts.Prev()` for some schema change timestamp
`ts`, the kv feed would deliver a resolved span with `ts` and a NONE
boundary to the change aggregator, which would advance its frontier;
then when the resolved span with `ts` and a RESTART boundary was
sent to the change aggregator, the frontier would not be advanced and
so would not be flushed to the change frontier. The change frontier
would then read a nil row from the change aggregator and shut the
changefeed down as if it had completed successfully.

This bug has been fixed by modifying the resolved span frontier
forwarding logic to consider forwarding to the current timestamp
but with a non-NONE boundary type to be advancing the frontier.

Two alternative solutions that were ruled out were:
1. Unconditionally flushing the frontier when we get a non-NONE boundary
   instead of only flushing when the frontier is advanced.
     - Problem: We would flush the frontier O(spans) number of times.
2. Making the kv feed not emit resolved events for rangefeed checkpoints
   that are at a schema change boundary.
     - Problem: We wouldn't be able to save the per-span progress at the
       schema change boundary.

Release note (bug fix): A bug that could potentially cause a changefeed
to erroneously complete when one of its watched tables encounters a
schema change has been fixed.

Author: andyyang890

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -878,7 +878,7 @@ func (ca *changeAggregator) noteResolvedSpan(resolved jobspb.ResolvedSpan) (retu
 		return nil
 	}
 
-	advanced, err := ca.frontier.ForwardResolvedSpan(ca.Ctx(), resolved)
+	advanced, err := ca.frontier.ForwardResolvedSpan(resolved)
 	if err != nil {
 		return err
 	}
@@ -1660,7 +1660,7 @@ func (cf *changeFrontier) noteAggregatorProgress(d rowenc.EncDatum) error {
 }
 
 func (cf *changeFrontier) forwardFrontier(resolved jobspb.ResolvedSpan) error {
-	frontierChanged, err := cf.frontier.ForwardResolvedSpan(cf.Ctx(), resolved)
+	frontierChanged, err := cf.frontier.ForwardResolvedSpan(resolved)
 	if err != nil {
 		return err
 	}

pkg/ccl/changefeedccl/resolvedspan/frontier.go

@@ -6,7 +6,6 @@
 package resolvedspan
 
 import (
-	"context"
 	"iter"
 	"slices"
 
@@ -44,8 +43,8 @@ func NewAggregatorFrontier(
 // ForwardResolvedSpan forwards the progress of a resolved span and also does
 // some boundary validation.
 func (f *AggregatorFrontier) ForwardResolvedSpan(
-	ctx context.Context, r jobspb.ResolvedSpan,
-) (bool, error) {
+	r jobspb.ResolvedSpan,
+) (forwarded bool, err error) {
 	switch boundaryType := r.BoundaryType; boundaryType {
 	case jobspb.ResolvedSpan_NONE:
 	case jobspb.ResolvedSpan_BACKFILL, jobspb.ResolvedSpan_EXIT, jobspb.ResolvedSpan_RESTART:
@@ -89,8 +88,8 @@ func NewCoordinatorFrontier(
 // ForwardResolvedSpan forwards the progress of a resolved span and also does
 // some boundary validation.
 func (f *CoordinatorFrontier) ForwardResolvedSpan(
-	ctx context.Context, r jobspb.ResolvedSpan,
-) (bool, error) {
+	r jobspb.ResolvedSpan,
+) (forwarded bool, err error) {
 	switch boundaryType := r.BoundaryType; boundaryType {
 	case jobspb.ResolvedSpan_NONE:
 	case jobspb.ResolvedSpan_BACKFILL:
@@ -224,8 +223,14 @@ func newResolvedSpanFrontier(
 }
 
 // ForwardResolvedSpan forwards the progress of a resolved span.
-func (f *resolvedSpanFrontier) ForwardResolvedSpan(r jobspb.ResolvedSpan) (bool, error) {
-	forwarded, err := f.Forward(r.Span, r.Timestamp)
+// The frontier is considered forwarded if either the frontier
+// timestamp advances or the current frontier timestamp becomes
+// a boundary timestamp (for some non-NONE boundary type)
+// and all the spans are at the boundary timestamp already.
+func (f *resolvedSpanFrontier) ForwardResolvedSpan(
+	r jobspb.ResolvedSpan,
+) (forwarded bool, err error) {
+	forwarded, err = f.Forward(r.Span, r.Timestamp)
 	if err != nil {
 		return false, err
 	}
@@ -235,7 +240,13 @@ func (f *resolvedSpanFrontier) ForwardResolvedSpan(r jobspb.ResolvedSpan) (bool,
 			ts:  r.Timestamp,
 			typ: r.BoundaryType,
 		}
-		f.boundary.Forward(newBoundary)
+		boundaryForwarded := f.boundary.Forward(newBoundary)
+		if boundaryForwarded && !forwarded {
+			// The frontier is considered forwarded if the boundary type
+			// changes to non-NONE and all the spans are at the boundary
+			// timestamp already.
+			forwarded, _, _ = f.AtBoundary()
+		}
 	}
 	return forwarded, nil
 }
@@ -375,9 +386,9 @@ func (b *resolvedSpanBoundary) At(ts hlc.Timestamp) (bool, jobspb.ResolvedSpan_B
 	return false, 0
 }
 
-// After returns whether a timestamp is later than the boundary timestamp.
+// After returns whether the boundary is after a given timestamp.
 func (b *resolvedSpanBoundary) After(ts hlc.Timestamp) bool {
-	return ts.Less(b.ts)
+	return b.ts.After(ts)
 }
 
 // Forward forwards the boundary to the new boundary if it is later.

pkg/ccl/changefeedccl/resolvedspan/frontier_test.go

@@ -155,7 +155,7 @@ func TestCoordinatorFrontier(t *testing.T) {
 
 type frontier interface {
 	Frontier() hlc.Timestamp
-	ForwardResolvedSpan(context.Context, jobspb.ResolvedSpan) (bool, error)
+	ForwardResolvedSpan(jobspb.ResolvedSpan) (bool, error)
 	InBackfill(jobspb.ResolvedSpan) bool
 	AtBoundary() (bool, jobspb.ResolvedSpan_BoundaryType, hlc.Timestamp)
 	All() iter.Seq[jobspb.ResolvedSpan]
@@ -171,7 +171,7 @@ func testBackfillSpan(
 ) {
 	backfillSpan := makeResolvedSpan(start, end, ts, jobspb.ResolvedSpan_NONE)
 	require.True(t, f.InBackfill(backfillSpan))
-	_, err := f.ForwardResolvedSpan(ctx, backfillSpan)
+	_, err := f.ForwardResolvedSpan(backfillSpan)
 	require.NoError(t, err)
 	require.Equal(t, frontierAfterSpan, f.Frontier())
 }
@@ -186,7 +186,7 @@ func testBoundarySpan(
 	frontierAfterSpan hlc.Timestamp,
 ) {
 	boundarySpan := makeResolvedSpan(start, end, boundaryTS, boundaryType)
-	_, err := f.ForwardResolvedSpan(ctx, boundarySpan)
+	_, err := f.ForwardResolvedSpan(boundarySpan)
 	require.NoError(t, err)
 
 	if finalBoundarySpan := frontierAfterSpan.Equal(boundaryTS); finalBoundarySpan {
@@ -220,7 +220,7 @@ func testIllegalBoundarySpan(
 	boundaryType jobspb.ResolvedSpan_BoundaryType,
 ) {
 	boundarySpan := makeResolvedSpan(start, end, boundaryTS, boundaryType)
-	_, err := f.ForwardResolvedSpan(ctx, boundarySpan)
+	_, err := f.ForwardResolvedSpan(boundarySpan)
 	require.True(t, errors.HasAssertionFailure(err))
 }
 
@@ -244,3 +244,53 @@ func makeSpan(start, end string) roachpb.Span {
 		EndKey: roachpb.Key(end),
 	}
 }
+
+func TestAggregatorFrontier_ForwardResolvedSpan(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	// Create a fresh frontier with no progress.
+	f, err := resolvedspan.NewAggregatorFrontier(
+		hlc.Timestamp{},
+		hlc.Timestamp{},
+		makeSpan("a", "f"),
+	)
+	require.NoError(t, err)
+	require.Zero(t, f.Frontier())
+
+	t.Run("advance frontier with no boundary", func(t *testing.T) {
+		// Forwarding part of the span space to 10 should not advance the frontier.
+		forwarded, err := f.ForwardResolvedSpan(
+			makeResolvedSpan("a", "b", makeTS(10), jobspb.ResolvedSpan_NONE))
+		require.NoError(t, err)
+		require.False(t, forwarded)
+		require.Zero(t, f.Frontier())
+
+		// Forwarding the rest of the span space to 10 should advance the frontier.
+		forwarded, err = f.ForwardResolvedSpan(
+			makeResolvedSpan("b", "f", makeTS(10), jobspb.ResolvedSpan_NONE))
+		require.NoError(t, err)
+		require.True(t, forwarded)
+		require.Equal(t, makeTS(10), f.Frontier())
+	})
+
+	t.Run("advance frontier with same timestamp and new boundary", func(t *testing.T) {
+		// Forwarding part of the span space to 10 again with a non-NONE boundary
+		// should be considered forwarding the frontier because we're learning
+		// about a new boundary.
+		forwarded, err := f.ForwardResolvedSpan(
+			makeResolvedSpan("c", "f", makeTS(10), jobspb.ResolvedSpan_RESTART))
+		require.NoError(t, err)
+		require.True(t, forwarded)
+		require.Equal(t, makeTS(10), f.Frontier())
+
+		// Forwarding the rest of the span space to 10 again with a non-NONE boundary
+		// should not be considered forwarding the frontier because we already
+		// know about the new boundary.
+		forwarded, err = f.ForwardResolvedSpan(
+			makeResolvedSpan("a", "c", makeTS(10), jobspb.ResolvedSpan_RESTART))
+		require.NoError(t, err)
+		require.False(t, forwarded)
+		require.Equal(t, makeTS(10), f.Frontier())
+	})
+}