changefeedccl: deflake initial backfill checkpoint tests

This patch deflakes the initial backfill checkpoint tests, which filter
out resolved spans to force gaps during a backfill. The tests however
did not ensure that there would be at least one span that was not
filtered out, which would result in no span-level checkpoint being
created, leading the tests to time out.

Release note: None

Author: andyyang890

pkg/ccl/changefeedccl/alter_changefeed_test.go

@@ -1520,19 +1520,27 @@ func TestAlterChangefeedAddTargetsDuringBackfill(t *testing.T) {
 
 		// Emit resolved events for the majority of spans. Be extra paranoid and ensure that
 		// we have at least 1 span for which we don't emit resolvedFoo timestamp (to force checkpointing).
-		haveGaps := false
-		knobs.FilterSpanWithMutation = func(r *jobspb.ResolvedSpan) (bool, error) {
+		// We however also need to ensure there's at least one span that isn't filtered out.
+		var allowedOne, haveGaps bool
+		knobs.FilterSpanWithMutation = func(r *jobspb.ResolvedSpan) (filter bool, _ error) {
 			rndMu.Lock()
 			defer rndMu.Unlock()
+			defer func() {
+				t.Logf("resolved span: %s@%s, filter: %t", r.Span, r.Timestamp, filter)
+			}()
 
 			if r.Span.Equal(fooTableSpan) {
 				return true, nil
 			}
-			if haveGaps {
-				return rndMu.rnd.Intn(10) > 7, nil
+			if !allowedOne {
+				allowedOne = true
+				return false, nil
 			}
-			haveGaps = true
-			return true, nil
+			if !haveGaps {
+				haveGaps = true
+				return true, nil
+			}
+			return rndMu.rnd.Intn(10) > 7, nil
 		}
 
 		// Checkpoint progress frequently, and set the checkpoint size limit.
@@ -1582,8 +1590,10 @@ WITH resolved = '100ms', min_checkpoint_frequency='1ns'`)
 
 		// Collect spans we attempt to resolve after when we resume.
 		var resolvedFoo []roachpb.Span
-		knobs.FilterSpanWithMutation = func(r *jobspb.ResolvedSpan) (bool, error) {
-			t.Logf("resolved span: %#v", r)
+		knobs.FilterSpanWithMutation = func(r *jobspb.ResolvedSpan) (filter bool, _ error) {
+			defer func() {
+				t.Logf("resolved span: %s@%s, filter: %t", r.Span, r.Timestamp, filter)
+			}()
 			if !r.Span.Equal(fooTableSpan) {
 				resolvedFoo = append(resolvedFoo, r.Span)
 			}

pkg/ccl/changefeedccl/changefeed_test.go

@@ -9117,22 +9117,21 @@ func TestChangefeedBackfillCheckpoint(t *testing.T) {
 
 		// Emit resolved events for majority of spans.  Be extra paranoid and ensure that
 		// we have at least 1 span for which we don't emit resolved timestamp (to force checkpointing).
-		haveGaps := false
+		// We however also need to ensure there's at least one span that isn't filtered out.
+		var allowedOne, haveGaps bool
 		knobs.FilterSpanWithMutation = func(r *jobspb.ResolvedSpan) (bool, error) {
 			if r.Span.Equal(tableSpan) {
-				// Do not emit resolved events for the entire table span.
-				// We "simulate" large table by splitting single table span into many parts, so
-				// we want to resolve those sub-spans instead of the entire table span.
-				// However, we have to emit something -- otherwise the entire changefeed
-				// machine would not work.
-				r.Span.EndKey = tableSpan.Key.Next()
+				return true, nil
+			}
+			if !allowedOne {
+				allowedOne = true
 				return false, nil
 			}
-			if haveGaps {
-				return rnd.Intn(10) > 7, nil
+			if !haveGaps {
+				haveGaps = true
+				return true, nil
 			}
-			haveGaps = true
-			return true, nil
+			return rnd.Intn(10) > 7, nil
 		}
 
 		// Checkpoint progress frequently, and set the checkpoint size limit.