do not merge: force parquet cloud storage tests

This change forces all tests, including tests for `diff` and `end_time`
to run with the `cloudstorage` sink and `format=parquet` where possible.

Informs: #103129
Informs: #99028
Epic: CRDB-27372
Release note: None

pkg/ccl/changefeedccl/cdctest/nemeses.go

@@ -155,7 +155,7 @@ func RunNemesis(f TestFeedFactory, db *gosql.DB, isSinkless bool) (Validator, er
 	}
 
 	withFormatParquet := ""
-	if rand.Intn(2) < 1 {
+	if rand.Intn(2) < 2 {
 		withFormatParquet = ", format=parquet"
 	}
 	foo, err := f.Feed(fmt.Sprintf(`CREATE CHANGEFEED FOR foo WITH updated, resolved, diff %s`, withFormatParquet))

pkg/ccl/changefeedccl/changefeed_test.go

@@ -314,7 +314,7 @@ func TestChangefeedBasicQueryWrapped(t *testing.T) {
 		// Currently, parquet format (which may be injected by feed() call),  doesn't
 		// know how to handle tuple types (cdc_prev); so, force JSON format.
 		foo := feed(t, f, `
-CREATE CHANGEFEED WITH envelope='wrapped', format='json', diff
+CREATE CHANGEFEED WITH envelope='wrapped', format='parquet', diff
 AS SELECT b||a AS ba, event_op() AS op  FROM foo`)
 		defer closeFeed(t, foo)
 
@@ -343,7 +343,7 @@ AS SELECT b||a AS ba, event_op() AS op  FROM foo`)
 		})
 	}
 
-	cdcTest(t, testFn, feedTestForceSink("webhook"))
+	cdcTest(t, testFn, feedTestForceSink("cloudstorage"))
 }
 
 // Same test as TestChangefeedBasicQueryWrapped, but this time using AVRO.
@@ -577,7 +577,7 @@ func TestChangefeedDiff(t *testing.T) {
 		sqlDB.Exec(t, `INSERT INTO foo VALUES (0, 'initial')`)
 		sqlDB.Exec(t, `UPSERT INTO foo VALUES (0, 'updated')`)
 
-		foo := feed(t, f, `CREATE CHANGEFEED FOR foo WITH diff`)
+		foo := feed(t, f, `CREATE CHANGEFEED FOR foo WITH diff, format=parquet`)
 		defer closeFeed(t, foo)
 
 		// 'initial' is skipped because only the latest value ('updated') is
@@ -609,7 +609,7 @@ func TestChangefeedDiff(t *testing.T) {
 		})
 	}
 
-	cdcTest(t, testFn)
+	cdcTest(t, testFn, feedTestForceSink("cloudstorage"))
 }
 
 func TestChangefeedTenants(t *testing.T) {
@@ -7208,7 +7208,7 @@ func TestChangefeedEndTime(t *testing.T) {
 		sqlDB.Exec(t, "INSERT INTO foo VALUES (1), (2), (3)")
 
 		fakeEndTime := s.Server.Clock().Now().Add(int64(time.Hour), 0).AsOfSystemTime()
-		feed := feed(t, f, "CREATE CHANGEFEED FOR foo WITH end_time = $1", fakeEndTime)
+		feed := feed(t, f, "CREATE CHANGEFEED FOR foo WITH end_time = $1, format=parquet", fakeEndTime)
 		defer closeFeed(t, feed)
 
 		assertPayloads(t, feed, []string{
@@ -7225,7 +7225,7 @@ func TestChangefeedEndTime(t *testing.T) {
 		}))
 	}
 
-	cdcTest(t, testFn, feedTestEnterpriseSinks)
+	cdcTest(t, testFn, feedTestForceSink("cloudstorage"))
 }
 
 func TestChangefeedEndTimeWithCursor(t *testing.T) {

pkg/ccl/changefeedccl/helpers_test.go

@@ -858,6 +858,9 @@ func randomSinkTypeWithOptions(options feedTestOptions) string {
 			sinkWeights[sinkType] = 0
 		}
 	}
+	if weight, ok := sinkWeights["cloudstorage"]; ok && weight != 0 {
+		sinkWeights = map[string]int{"cloudstorage": 1}
+	}
 	weightTotal := 0
 	for _, weight := range sinkWeights {
 		weightTotal += weight

pkg/ccl/changefeedccl/testfeed_test.go

@@ -1076,7 +1076,7 @@ func (f *cloudFeedFactory) Feed(
 			}
 		}
 		randNum := rand.Intn(5)
-		if randNum < 3 {
+		if randNum < 0 {
 			parquetPossible = false
 		}
 		if parquetPossible {