Merge #148245 #148747

148245: changefeedccl: checkpoint on initial scan for core changefeeds r=asg0451,andyyang890 a=aerfrei

Previously we would only checkpoint during backfills for enterprise changefeeds, even though we had a test asserting otherwise. Adding checkpointing prevents us from emitting further duplicates when core changefeeds encounter transient errors during an initial scan.

Epic: none
Fixes: #143153

Release note: None

148747: general: ignore vscode configuration r=bghal a=bghal

Epic: none


Co-authored-by: Aerin Freilich <[email protected]>
Co-authored-by: Brendan Gerrity <[email protected]>

Author: 3 people

.gitignore

@@ -65,3 +65,4 @@ pkg/testutils/serverutils/*_generated.go
 
 .DS_Store
 .idea/
+.vscode/

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -951,8 +951,7 @@ func (ca *changeAggregator) noteResolvedSpan(resolved jobspb.ResolvedSpan) (retu
 
 	// At a lower frequency, we checkpoint specific spans in the job progress
 	// either in backfills or if the highwater mark is excessively lagging behind.
-	checkpointSpans := ca.spec.JobID != 0 && /* enterprise changefeed */
-		(ca.frontier.InBackfill(resolved) || ca.frontier.HasLaggingSpans(sv)) &&
+	checkpointSpans := (ca.frontier.InBackfill(resolved) || ca.frontier.HasLaggingSpans(sv)) &&
 		canCheckpointSpans(sv, ca.lastSpanFlush)
 
 	if checkpointSpans {

pkg/ccl/changefeedccl/changefeed_test.go

@@ -8829,9 +8829,11 @@ func TestCoreChangefeedBackfillScanCheckpoint(t *testing.T) {
 			context.Background(), &s.Server.ClusterSettings().SV, 100<<20)
 
 		emittedCount := 0
+		errorCount := 0
 		knobs.RaiseRetryableError = func() error {
 			emittedCount++
 			if emittedCount%200 == 0 {
+				errorCount++
 				return errors.New("test transient error")
 			}
 			return nil
@@ -8845,6 +8847,7 @@ func TestCoreChangefeedBackfillScanCheckpoint(t *testing.T) {
 			payloads[i] = fmt.Sprintf(`foo: [%d]->{"after": {"a": %d}}`, i, i)
 		}
 		assertPayloads(t, foo, payloads)
+		require.GreaterOrEqual(t, errorCount, 1)
 	}
 
 	cdcTest(t, testFn, feedTestForceSink("sinkless"))