changefeedccl: deflake tests that expect frequent span-level checkpoints

In a recent commit, the interval at which change aggregators would
flush their frontiers to the change frontier was decoupled from
the span-level checkpoint interval (which is configurable via a
cluster setting). This caused some tests that only set the cluster
setting (and not `min_checkpoint_frequency`) to a low value to
sometimes flake. This patch updates all such tests to also configure
the `min_checkpoint_frequency` option.

Release note: None

Author: andyyang890

pkg/ccl/changefeedccl/alter_changefeed_test.go

@@ -1542,7 +1542,8 @@ func TestAlterChangefeedAddTargetsDuringBackfill(t *testing.T) {
 			context.Background(), &s.Server.ClusterSettings().SV, maxCheckpointSize)
 
 		registry := s.Server.JobRegistry().(*jobs.Registry)
-		testFeed := feed(t, f, `CREATE CHANGEFEED FOR foo WITH resolved = '100ms'`)
+		testFeed := feed(t, f, `CREATE CHANGEFEED FOR foo
+WITH resolved = '100ms', min_checkpoint_frequency='1ns'`)
 
 		g := ctxgroup.WithContext(context.Background())
 		g.Go(func() error {

pkg/ccl/changefeedccl/changefeed_test.go

@@ -2802,7 +2802,8 @@ func TestChangefeedLaggingSpanCheckpointing(t *testing.T) {
 
 	var jobID jobspb.JobID
 	sqlDB.QueryRow(t,
-		`CREATE CHANGEFEED FOR foo INTO 'null://' WITH resolved='50ms', no_initial_scan, cursor=$1`, tsStr,
+		`CREATE CHANGEFEED FOR foo INTO 'null://'
+WITH resolved='50ms', min_checkpoint_frequency='50ms', no_initial_scan, cursor=$1`, tsStr,
 	).Scan(&jobID)
 
 	// Helper to read job progress
@@ -2944,7 +2945,8 @@ func TestChangefeedSchemaChangeBackfillCheckpoint(t *testing.T) {
 		}
 
 		// Setup changefeed job details, avoid relying on initial scan functionality
-		baseFeed := feed(t, f, `CREATE CHANGEFEED FOR foo WITH resolved='100ms', min_checkpoint_frequency='100ms', no_initial_scan`)
+		baseFeed := feed(t, f, `CREATE CHANGEFEED FOR foo
+WITH resolved='100ms', min_checkpoint_frequency='1ns', no_initial_scan`)
 		jobFeed := baseFeed.(cdctest.EnterpriseTestFeed)
 		jobRegistry := s.Server.JobRegistry().(*jobs.Registry)
 
@@ -9140,7 +9142,8 @@ func TestChangefeedBackfillCheckpoint(t *testing.T) {
 			context.Background(), &s.Server.ClusterSettings().SV, maxCheckpointSize)
 
 		registry := s.Server.JobRegistry().(*jobs.Registry)
-		foo := feed(t, f, `CREATE CHANGEFEED FOR foo WITH resolved='100ms'`)
+		foo := feed(t, f, `CREATE CHANGEFEED FOR foo
+WITH resolved='100ms', min_checkpoint_frequency='1ns'`)
 		// Some test feeds (kafka) are not buffered, so we have to consume messages.
 		var shouldDrain int32 = 1
 		g := ctxgroup.WithContext(context.Background())
@@ -12000,7 +12003,8 @@ func TestChangefeedProtectedTimestampUpdate(t *testing.T) {
 		require.Equal(t, int64(0), managePTSCount)
 		require.Equal(t, int64(0), managePTSErrorCount)
 
-		createStmt := `CREATE CHANGEFEED FOR foo WITH resolved='10ms', no_initial_scan`
+		createStmt := `CREATE CHANGEFEED FOR foo
+WITH resolved='10ms', min_checkpoint_frequency='10ms', no_initial_scan`
 		testFeed := feed(t, f, createStmt)
 		defer closeFeed(t, testFeed)
 
@@ -12113,7 +12117,8 @@ func TestChangefeedProtectedTimestampUpdateError(t *testing.T) {
 			return errors.New("test error")
 		}
 
-		createStmt := `CREATE CHANGEFEED FOR foo WITH resolved='10ms', no_initial_scan`
+		createStmt := `CREATE CHANGEFEED FOR foo
+WITH resolved='10ms', min_checkpoint_frequency='10ms', no_initial_scan`
 		testFeed := feed(t, f, createStmt)
 		defer closeFeed(t, testFeed)
 

pkg/ccl/changefeedccl/protected_timestamps_test.go

@@ -970,7 +970,8 @@ func TestChangefeedProtectedTimestampUpdateForMultipleTables(t *testing.T) {
 		require.Equal(t, int64(0), managePTSCount)
 		require.Equal(t, int64(0), managePTSErrorCount)
 
-		createStmt := `CREATE CHANGEFEED FOR foo, bar WITH resolved='10ms', initial_scan='no'`
+		createStmt := `CREATE CHANGEFEED FOR foo, bar
+WITH resolved='10ms', min_checkpoint_frequency='100ms', initial_scan='no'`
 		testFeed := feed(t, f, createStmt)
 		defer closeFeed(t, testFeed)
 
@@ -1114,7 +1115,8 @@ func TestChangefeedPerTableProtectedTimestampProgression(t *testing.T) {
 			}
 		}
 
-		createStmt := `CREATE CHANGEFEED FOR table1, table2, table3 WITH resolved='100ms'`
+		createStmt := `CREATE CHANGEFEED FOR table1, table2, table3
+WITH resolved='100ms', min_checkpoint_frequency='100ms'`
 		testFeed := feed(t, f, createStmt)
 		defer closeFeed(t, testFeed)