Merge #144004

144004:  changefeedccl: fix premature shutdown due to schema change bug r=rharding6373 a=andyyang890

Fixes #144108

Test failures on master:
Fixes #143976
Fixes #144045
Fixes #144219

Test failures on release branches (won't be fixed until PR is backported):
Informs #144287
Informs #144291
Informs #144352

---

**changefeedccl: add more verbose logging around schema changes**

This patch adds more verbose logging to the change aggregator around
receiving and emitting resolved spans to help debug recurring changefeed
schema change test flakes.

Release note: None

---

**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.

---

**changefeedccl/resolvedspan: update assertion for forwarding boundary** 

This patch adds an assertion that the frontier will not forward the
boundary to a different type at the same time. This assertion is
important because if it's violated, the changefeed processors will not
shut down correctly during a schema change. One potential way this
could be violated in the future is a change to how boundary types are
determined on aggregators causing different boundaries to be sent from
aggregators in a mixed-version cluster for the same schema change.

Release note: None

Co-authored-by: Andy Yang <[email protected]>

Author: craig[bot]

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -868,14 +868,18 @@ func (ca *changeAggregator) flushBufferedEvents() error {
 // changeAggregator node to the changeFrontier node to allow the changeFrontier
 // to persist the overall changefeed's progress
 func (ca *changeAggregator) noteResolvedSpan(resolved jobspb.ResolvedSpan) (returnErr error) {
+	if log.V(2) {
+		log.Infof(ca.Ctx(), "resolved span from kv feed: %#v", resolved)
+	}
+
 	if resolved.Timestamp.IsEmpty() {
 		// @0.0 resolved timestamps could come in from rangefeed checkpoint.
 		// When rangefeed starts running, it emits @0.0 resolved timestamp.
 		// We don't care about those as far as checkpointing concerned.
 		return nil
 	}
 
-	advanced, err := ca.frontier.ForwardResolvedSpan(ca.Ctx(), resolved)
+	advanced, err := ca.frontier.ForwardResolvedSpan(resolved)
 	if err != nil {
 		return err
 	}
@@ -951,6 +955,9 @@ func (ca *changeAggregator) emitResolved(batch jobspb.ResolvedSpans) error {
 			RecentKvCount: ca.recentKVCount,
 		},
 	}
+	if log.V(2) {
+		log.Infof(ca.Ctx(), "progress update to be sent to change frontier: %#v", progressUpdate)
+	}
 	updateBytes, err := protoutil.Marshal(&progressUpdate)
 	if err != nil {
 		return err
@@ -1654,7 +1661,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/BUILD.bazel

@@ -15,7 +15,6 @@ go_library(
         "//pkg/roachpb",
         "//pkg/settings",
         "//pkg/util/hlc",
-        "//pkg/util/log",
         "//pkg/util/span",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_redact//:redact",

pkg/ccl/changefeedccl/resolvedspan/frontier.go

@@ -6,7 +6,6 @@
 package resolvedspan
 
 import (
-	"context"
 	"iter"
 	"slices"
 
@@ -16,7 +15,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
-	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/span"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/redact"
@@ -44,15 +42,15 @@ 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:
 		// Boundary resolved events should be ingested from the schema feed
 		// serially, where the changefeed won't ever observe a new schema change
 		// boundary until it has progressed past the current boundary.
-		if err := f.assertBoundaryNotEarlier(ctx, r); err != nil {
+		if err := f.assertBoundaryNotEarlierOrDifferent(r); err != nil {
 			return false, err
 		}
 	default:
@@ -89,8 +87,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:
@@ -103,21 +101,22 @@ func (f *CoordinatorFrontier) ForwardResolvedSpan(
 		// it is a BACKFILL we have already seen, then it is fine for it to be
 		// an earlier timestamp than the latest boundary.
 		boundaryTS := r.Timestamp
-		_, ok := slices.BinarySearchFunc(f.backfills, boundaryTS, func(elem hlc.Timestamp, ts hlc.Timestamp) int {
-			return elem.Compare(ts)
-		})
-		if ok {
+		if _, ok := slices.BinarySearchFunc(f.backfills, boundaryTS,
+			func(elem hlc.Timestamp, ts hlc.Timestamp) int {
+				return elem.Compare(ts)
+			},
+		); ok {
 			break
 		}
-		if err := f.assertBoundaryNotEarlier(ctx, r); err != nil {
+		if err := f.assertBoundaryNotEarlierOrDifferent(r); err != nil {
 			return false, err
 		}
 		f.backfills = append(f.backfills, boundaryTS)
 	case jobspb.ResolvedSpan_EXIT, jobspb.ResolvedSpan_RESTART:
 		// EXIT and RESTART are final boundaries that cause the changefeed
 		// processors to all move to draining and so should not be followed
 		// by any other boundaries.
-		if err := f.assertBoundaryNotEarlier(ctx, r); err != nil {
+		if err := f.assertBoundaryNotEarlierOrDifferent(r); err != nil {
 			return false, err
 		}
 	default:
@@ -130,9 +129,10 @@ func (f *CoordinatorFrontier) ForwardResolvedSpan(
 	// If the frontier changed, we check if the frontier has advanced past any known backfills.
 	if frontierChanged {
 		frontier := f.Frontier()
-		i, _ := slices.BinarySearchFunc(f.backfills, frontier, func(elem hlc.Timestamp, ts hlc.Timestamp) int {
-			return elem.Compare(ts)
-		})
+		i, _ := slices.BinarySearchFunc(f.backfills, frontier,
+			func(elem hlc.Timestamp, ts hlc.Timestamp) int {
+				return elem.Compare(ts)
+			})
 		f.backfills = f.backfills[i:]
 	}
 	return frontierChanged, nil
@@ -145,10 +145,11 @@ func (f *CoordinatorFrontier) ForwardResolvedSpan(
 // happening at different timestamps.
 func (f *CoordinatorFrontier) InBackfill(r jobspb.ResolvedSpan) bool {
 	boundaryTS := r.Timestamp
-	_, ok := slices.BinarySearchFunc(f.backfills, boundaryTS, func(elem hlc.Timestamp, ts hlc.Timestamp) int {
-		return elem.Next().Compare(ts)
-	})
-	if ok {
+	if _, ok := slices.BinarySearchFunc(f.backfills, boundaryTS,
+		func(elem hlc.Timestamp, ts hlc.Timestamp) int {
+			return elem.Next().Compare(ts)
+		},
+	); ok {
 		return true
 	}
 
@@ -161,11 +162,12 @@ func (f *CoordinatorFrontier) All() iter.Seq[jobspb.ResolvedSpan] {
 		for resolvedSpan := range f.resolvedSpanFrontier.All() {
 			// Check if it's at an earlier backfill boundary.
 			if resolvedSpan.BoundaryType == jobspb.ResolvedSpan_NONE {
-				for _, b := range f.backfills {
-					if b.Equal(resolvedSpan.Timestamp) {
-						resolvedSpan.BoundaryType = jobspb.ResolvedSpan_BACKFILL
-						break
-					}
+				if _, ok := slices.BinarySearchFunc(f.backfills, resolvedSpan.Timestamp,
+					func(elem hlc.Timestamp, ts hlc.Timestamp) int {
+						return elem.Compare(ts)
+					},
+				); ok {
+					resolvedSpan.BoundaryType = jobspb.ResolvedSpan_BACKFILL
 				}
 			}
 			if !yield(resolvedSpan) {
@@ -224,8 +226,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 +243,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
 }
@@ -284,23 +298,26 @@ func (f *resolvedSpanFrontier) InBackfill(r jobspb.ResolvedSpan) bool {
 	return false
 }
 
-// assertBoundaryNotEarlier is a helper method provided to assert that a
-// resolved span does not have an earlier boundary than the existing one.
-func (f *resolvedSpanFrontier) assertBoundaryNotEarlier(
-	ctx context.Context, r jobspb.ResolvedSpan,
-) error {
+// assertBoundaryNotEarlierOrDifferent is a helper method that asserts that a
+// resolved span does not have an earlier boundary than the existing one
+// nor is it at the same time as the existing one with a different type.
+func (f *resolvedSpanFrontier) assertBoundaryNotEarlierOrDifferent(r jobspb.ResolvedSpan) error {
 	boundaryType := r.BoundaryType
 	if boundaryType == jobspb.ResolvedSpan_NONE {
-		return errors.AssertionFailedf("assertBoundaryNotEarlier should not be called for NONE boundary")
+		return errors.AssertionFailedf(
+			"assertBoundaryNotEarlierOrDifferent should not be called for NONE boundary")
 	}
 	boundaryTS := r.Timestamp
+	newBoundary := newResolvedSpanBoundary(boundaryTS, boundaryType)
 	if f.boundary.After(boundaryTS) {
-		newBoundary := newResolvedSpanBoundary(boundaryTS, boundaryType)
-		err := errors.AssertionFailedf("received resolved span for %s "+
+		return errors.AssertionFailedf("received resolved span for %s "+
 			"with %v, which is earlier than previously received %v",
 			r.Span, newBoundary, f.boundary)
-		log.Errorf(ctx, "error while forwarding boundary resolved span: %v", err)
-		return err
+	}
+	if atBoundary, typ := f.boundary.At(boundaryTS); atBoundary && boundaryType != typ {
+		return errors.AssertionFailedf("received resolved span for %s "+
+			"with %v, which has a different type from previously received %v with same timestamp",
+			r.Span, newBoundary, f.boundary)
 	}
 	return nil
 }
@@ -375,9 +392,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.