changefeedccl/resolvedspan: fix empty boundary handling bug

This patch fixes a bug where the empty boundary was being handled
incorrectly by the resolved span frontiers, which could lead to
a lot of unnecessary job progress saves during an initial scan.

Release note (bug fix): A bug where a changefeed could perform
many unnecessary job progress saves during an initial scan has
been fixed.

Author: andyyang890

pkg/ccl/changefeedccl/resolvedspan/frontier.go

@@ -255,18 +255,12 @@ func (f *resolvedSpanFrontier) ForwardResolvedSpan(
 		return false, err
 	}
 	f.latestTS.Forward(r.Timestamp)
-	if r.BoundaryType != jobspb.ResolvedSpan_NONE {
-		newBoundary := resolvedSpanBoundary{
-			ts:  r.Timestamp,
-			typ: r.BoundaryType,
-		}
-		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()
-		}
+	boundaryForwarded := f.boundary.Forward(r.Timestamp, r.BoundaryType)
+	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
 }
@@ -408,27 +402,35 @@ func newResolvedSpanBoundary(
 // At returns whether a timestamp is equal to the boundary timestamp
 // and if so, the boundary type as well.
 func (b *resolvedSpanBoundary) At(ts hlc.Timestamp) (bool, jobspb.ResolvedSpan_BoundaryType) {
-	if ts.Equal(b.ts) {
+	if b.IsSet() && ts.Equal(b.ts) {
 		return true, b.typ
 	}
 	return false, 0
 }
 
 // After returns whether the boundary is after a given timestamp.
 func (b *resolvedSpanBoundary) After(ts hlc.Timestamp) bool {
-	return b.ts.After(ts)
+	return b.IsSet() && b.ts.After(ts)
 }
 
 // Forward forwards the boundary to the new boundary if it is later.
 // It returns true if the boundary changed and false otherwise.
-func (b *resolvedSpanBoundary) Forward(newBoundary resolvedSpanBoundary) bool {
-	if newBoundary.After(b.ts) {
-		*b = newBoundary
+func (b *resolvedSpanBoundary) Forward(
+	ts hlc.Timestamp, typ jobspb.ResolvedSpan_BoundaryType,
+) bool {
+	if typ != jobspb.ResolvedSpan_NONE && ts.After(b.ts) {
+		b.ts = ts
+		b.typ = typ
 		return true
 	}
 	return false
 }
 
+// IsSet returns whether the boundary is set.
+func (b *resolvedSpanBoundary) IsSet() bool {
+	return b.typ != jobspb.ResolvedSpan_NONE
+}
+
 // SafeFormat implements the redact.SafeFormatter interface.
 func (b *resolvedSpanBoundary) SafeFormat(s redact.SafePrinter, _ rune) {
 	s.Printf("%v boundary (%v)", b.typ, b.ts)

pkg/ccl/changefeedccl/resolvedspan/frontier_test.go

@@ -179,15 +179,48 @@ func TestCoordinatorFrontier(t *testing.T) {
 }
 
 type frontier interface {
-	AddSpansAt(startAt hlc.Timestamp, spans ...roachpb.Span) error
+	AddSpansAt(hlc.Timestamp, ...roachpb.Span) error
 	Frontier() hlc.Timestamp
+	Forward(roachpb.Span, hlc.Timestamp) (bool, error)
 	ForwardResolvedSpan(jobspb.ResolvedSpan) (bool, error)
 	InBackfill(jobspb.ResolvedSpan) bool
 	AtBoundary() (bool, jobspb.ResolvedSpan_BoundaryType, hlc.Timestamp)
 	All() iter.Seq[jobspb.ResolvedSpan]
 	Frontiers() iter.Seq2[descpb.ID, span.ReadOnlyFrontier]
 }
 
+func newFrontier(
+	t testing.TB,
+	frontierType string,
+	statementTime hlc.Timestamp,
+	initialHighwater hlc.Timestamp,
+	codec resolvedspan.TableCodec,
+	perTableTracking bool,
+	spans ...roachpb.Span,
+) (frontier, error) {
+	switch frontierType {
+	case "aggregator":
+		return resolvedspan.NewAggregatorFrontier(
+			statementTime,
+			initialHighwater,
+			codec,
+			perTableTracking,
+			spans...,
+		)
+	case "coordinator":
+		return resolvedspan.NewCoordinatorFrontier(
+			statementTime,
+			initialHighwater,
+			codec,
+			perTableTracking,
+			spans...,
+		)
+	default:
+		t.Fatalf("unknown frontier type: %s", frontierType)
+	}
+	panic("unreachable")
+}
+
 func testBackfillSpan(
 	t *testing.T, f frontier, start, end string, ts hlc.Timestamp, frontierAfterSpan hlc.Timestamp,
 ) {
@@ -379,29 +412,15 @@ func TestFrontierPerTableResolvedTimestamps(t *testing.T) {
 			initialHighWater := hlc.Timestamp{}
 
 			// Create frontier with multiple table spans.
-			f, err := func() (frontier, error) {
-				switch frontierType {
-				case "aggregator":
-					return resolvedspan.NewAggregatorFrontier(
-						statementTime,
-						initialHighWater,
-						codec,
-						true, /* perTableTracking */
-						tableSpans...,
-					)
-				case "coordinator":
-					return resolvedspan.NewCoordinatorFrontier(
-						statementTime,
-						initialHighWater,
-						codec,
-						true, /* perTableTracking */
-						tableSpans...,
-					)
-				default:
-					t.Fatalf("unknown frontier type: %s", frontierType)
-				}
-				panic("unreachable")
-			}()
+			f, err := newFrontier(
+				t,
+				frontierType,
+				statementTime,
+				initialHighWater,
+				codec,
+				true, /* perTableTracking */
+				tableSpans...,
+			)
 			require.NoError(t, err)
 			require.Equal(t, initialHighWater, f.Frontier())
 
@@ -503,29 +522,15 @@ func TestFrontierForwardFullTableSpan(t *testing.T) {
 			statementTime := makeTS(5)
 			var initialHighWater hlc.Timestamp
 
-			f, err := func() (span.Frontier, error) {
-				switch frontierType {
-				case "aggregator":
-					return resolvedspan.NewAggregatorFrontier(
-						statementTime,
-						initialHighWater,
-						codec,
-						true, /* perTableTracking */
-						tableSpans...,
-					)
-				case "coordinator":
-					return resolvedspan.NewCoordinatorFrontier(
-						statementTime,
-						initialHighWater,
-						codec,
-						true, /* perTableTracking */
-						tableSpans...,
-					)
-				default:
-					t.Fatalf("unknown frontier type: %s", frontierType)
-				}
-				panic("unreachable")
-			}()
+			f, err := newFrontier(
+				t,
+				frontierType,
+				statementTime,
+				initialHighWater,
+				codec,
+				true, /* perTableTracking */
+				tableSpans...,
+			)
 			require.NoError(t, err)
 			require.Equal(t, initialHighWater, f.Frontier())
 
@@ -610,26 +615,14 @@ FROM [SHOW RANGES FROM TABLE foo WITH KEYS]`)
 						now := makeTS(timeutil.Now().Unix())
 
 						// Create the frontier and add all the spans.
-						f, err := func() (frontier, error) {
-							switch frontierType {
-							case "aggregator":
-								return resolvedspan.NewAggregatorFrontier(
-									now,
-									now,
-									codec,
-									perTableTracking,
-								)
-							case "coordinator":
-								return resolvedspan.NewCoordinatorFrontier(
-									now,
-									now,
-									codec,
-									perTableTracking,
-								)
-							default:
-								panic("unreachable")
-							}
-						}()
+						f, err := newFrontier(
+							b,
+							frontierType,
+							now,
+							now,
+							codec,
+							perTableTracking,
+						)
 						require.NoError(b, err)
 						require.NoError(b, f.AddSpansAt(now, spans...))
 
@@ -655,3 +648,63 @@ FROM [SHOW RANGES FROM TABLE foo WITH KEYS]`)
 		}
 	}
 }
+
+func TestFrontierAtBoundary(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	rnd, _ := randutil.NewTestRand()
+	perTableTracking := rnd.Intn(2) == 0
+	t.Logf("per-table tracking: %t", perTableTracking)
+
+	testutils.RunValues(t, "frontier type", []string{"aggregator", "coordinator"},
+		func(t *testing.T, frontierType string) {
+			statementTime := makeTS(timeutil.Now().Unix())
+			var initialHighWater hlc.Timestamp
+			f, err := newFrontier(
+				t,
+				frontierType,
+				statementTime,
+				initialHighWater,
+				mockCodec{},
+				perTableTracking,
+				makeSpan("a", "f"),
+			)
+			require.NoError(t, err)
+
+			// We can't be at boundary until a boundary is set.
+			atBoundary, _, _ := f.AtBoundary()
+			require.False(t, atBoundary)
+			_, err = f.ForwardResolvedSpan(jobspb.ResolvedSpan{
+				Span:      makeSpan("a", "f"),
+				Timestamp: statementTime,
+			})
+			require.NoError(t, err)
+			atBoundary, _, _ = f.AtBoundary()
+			require.False(t, atBoundary)
+
+			// Set a boundary by forwarding part of the span space.
+			ts := statementTime.AddDuration(3 * time.Second)
+			_, err = f.ForwardResolvedSpan(jobspb.ResolvedSpan{
+				Span:         makeSpan("a", "c"),
+				Timestamp:    ts,
+				BoundaryType: jobspb.ResolvedSpan_BACKFILL,
+			})
+			require.NoError(t, err)
+			atBoundary, _, _ = f.AtBoundary()
+			require.False(t, atBoundary)
+
+			// Verify the boundary is reached after forwarding
+			// the rest of the span space.
+			_, err = f.ForwardResolvedSpan(jobspb.ResolvedSpan{
+				Span:         makeSpan("c", "f"),
+				Timestamp:    ts,
+				BoundaryType: jobspb.ResolvedSpan_BACKFILL,
+			})
+			require.NoError(t, err)
+			atBoundary, boundaryType, boundaryTS := f.AtBoundary()
+			require.True(t, atBoundary)
+			require.Equal(t, jobspb.ResolvedSpan_BACKFILL, boundaryType)
+			require.Equal(t, ts, boundaryTS)
+		})
+}