changefeedccl: fix flakey tests

Our test SQL connection doesn't always read its own writes:
it can use different connections for different queries.
So
```
INSERT INTO foo;
feed(CREATE CHANGEFEED FOR foo)
```
is a race condition. This just doesn't matter for most tests
because if the insert happens after the initial scan, it creates
the same payload. But tests with initial_scan='only' won't see it
at all.

Fixes #92211

Release note: None

pkg/ccl/changefeedccl/changefeed_test.go

@@ -6313,6 +6313,12 @@ func TestChangefeedOnlyInitialScan(t *testing.T) {
 				sqlDB.Exec(t, `CREATE TABLE foo (a INT PRIMARY KEY)`)
 				sqlDB.Exec(t, `INSERT INTO foo (a) SELECT * FROM generate_series(1, 5000);`)
 
+				// Most changefeed tests can afford to have a race condition between the initial
+				// inserts and starting the feed because the output looks the same for an initial
+				// scan and an insert. For tests with initial_scan=only, though, we can't start the feed
+				// until it's going to see all the initial inserts in the initial scan.
+				sqlDB.CheckQueryResultsRetry(t, `SELECT count(*) FROM foo`, [][]string{{`5000`}})
+
 				feed := feed(t, f, changefeedStmt)
 
 				sqlDB.Exec(t, "INSERT INTO foo VALUES (5005), (5007), (5009)")
@@ -6404,6 +6410,8 @@ func TestChangefeedOnlyInitialScanCSV(t *testing.T) {
 				sqlDB.Exec(t, "INSERT INTO foo VALUES (1, 'Alice'), (2, 'Bob'), (3, 'Carol')")
 				sqlDB.Exec(t, "INSERT INTO bar VALUES (1, 'a'), (2, 'b'), (3, 'c')")
 
+				sqlDB.CheckQueryResultsRetry(t, `SELECT count(*) FROM foo,bar`, [][]string{{`9`}})
+
 				feed := feed(t, f, testData.changefeedStmt)
 
 				sqlDB.Exec(t, "INSERT INTO foo VALUES (4, 'Doug'), (5, 'Elaine'), (6, 'Fred')")
@@ -6461,6 +6469,8 @@ func TestChangefeedOnlyInitialScanCSVSinkless(t *testing.T) {
 				sqlDB.Exec(t, "CREATE TABLE foo (id INT PRIMARY KEY, name STRING)")
 				sqlDB.Exec(t, "INSERT INTO foo VALUES (1, 'Alice'), (2, 'Bob'), (3, 'Carol')")
 
+				sqlDB.CheckQueryResultsRetry(t, `SELECT count(*) FROM foo`, [][]string{{`3`}})
+
 				feed := feed(t, f, changefeedStmt)
 
 				sqlDB.Exec(t, "INSERT INTO foo VALUES (4, 'Doug'), (5, 'Elaine'), (6, 'Fred')")