fix(tests): make replay and shard read setup deterministic (#26051)

## What type of PR is this?

- [x] BUG
- [x] Test and CI

## What this PR does / why we need it

This PR fixes two independent nondeterministic test setups without
relaxing production behavior.

### TAE checkpoint replay

TestAppendAndGC2 created its database and tables through low-level TAE
catalog APIs. Those APIs mutate the in-memory catalog but do not append
rows to mo_database, mo_tables, and mo_columns. After a forced
checkpoint truncates the DDL WAL, checkpoint replay can therefore lose
the still-live database.

The prepared-DML replay fence added by #26027 exposed this invalid setup
as ExpectedEOB. The previous version of this PR skipped missing catalog
entries, but diagnostics proved the missing IDs were the live test
database and table. That production-code relaxation was removed.

The test now uses the full DDL helpers and, after restart, opens both
relations and scans every column to verify that each still contains all
99 rows.

### Shard read compatibility test

TestReadValidatesAllRemoteShardsBeforeSending could create its table
before all three CNs reported to the shard server. In the failed Ubuntu
job, all shards were initially allocated to cn1 and balancing the second
shard took 60 seconds. The shared setup context expires after 10
seconds, so the first remote compatibility check returned context
deadline exceeded after one Adjust call instead of reaching the intended
incompatible CN.

The test now waits for every CN before creating shards, bounds every
replica wait, uses separate setup and read contexts, and asserts the
exact incompatibility error. This keeps the test fast and prevents
unrelated timeout errors from satisfying require.Error.

No production code is changed.

## Validation

- TAE focused replay test: count=20
- TAE focused replay test: race count=10
- pkg/vm/engine/tae/db/test full package
- pkg/vm/engine/tae/txn/txnimpl full package
- shard compatibility test: count=30
- shard compatibility test: race count=10
- pkg/shardservice full package
- go build ./pkg/shardservice
- go vet ./pkg/shardservice
- git diff --check

Author: XuPeng-SH

pkg/shardservice/service_read_test.go

@@ -208,12 +208,23 @@ func TestReadValidatesAllRemoteShardsBeforeSending(t *testing.T) {
 	runServicesTest(
 		t,
 		"cn1,cn2,cn3",
-		func(ctx context.Context, _ *server, services []*service) {
+		func(_ context.Context, server *server, services []*service) {
 			s := services[0]
 			table := uint64(1)
-			mustAddTestShards(t, ctx, s, table, 3, 1, services[1:]...)
+			require.Eventually(t, func() bool {
+				server.r.RLock()
+				defer server.r.RUnlock()
+				return len(server.r.cns) == len(services)
+			}, 10*time.Second, 10*time.Millisecond)
+			func() {
+				setupCtx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+				defer cancel()
+				mustAddTestShards(t, setupCtx, s, table, 3, 1, services[1:]...)
+			}()
 			for _, service := range services {
-				waitReplicaCount(table, service, 1)
+				require.Eventually(t, func() bool {
+					return service.TableReplicaCount(table) == 1
+				}, 10*time.Second, 10*time.Millisecond)
 			}
 
 			cache, err := s.getShards(table)
@@ -253,8 +264,10 @@ func TestReadValidatesAllRemoteShardsBeforeSending(t *testing.T) {
 			client := &countingMethodBasedClient{MethodBasedClient: s.remote.client}
 			s.remote.client = client
 			adjustCalls := make(map[uint64]int)
+			readCtx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+			defer cancel()
 			err = s.Read(
-				ctx,
+				readCtx,
 				ReadRequest{
 					TableID: table,
 					Param: shard.ReadParam{Process: pipeline.ProcessInfo{
@@ -266,6 +279,7 @@ func TestReadValidatesAllRemoteShardsBeforeSending(t *testing.T) {
 				}),
 			)
 			require.Error(t, err)
+			require.ErrorContains(t, err, "incompatible or unknown commit")
 			require.Zero(t, client.asyncCalls.Load())
 			require.Len(t, adjustCalls, len(remoteTargets))
 			for _, calls := range adjustCalls {

pkg/vm/engine/tae/db/test/db_test.go

@@ -7077,13 +7077,13 @@ func TestAppendAndGC2(t *testing.T) {
 	schema2.Extra.ObjectMaxBlocks = 2
 	{
 		txn, _ := db.StartTxn(nil)
-		database, err := txn.CreateDatabase("db", "", "")
-		assert.Nil(t, err)
-		_, err = database.CreateRelation(schema1)
-		assert.Nil(t, err)
-		_, err = database.CreateRelation(schema2)
-		assert.Nil(t, err)
-		assert.Nil(t, txn.Commit(context.Background()))
+		database, err := testutil.CreateDatabase2(ctx, txn, "db")
+		require.NoError(t, err)
+		_, err = testutil.CreateRelation2(ctx, txn, database, schema1)
+		require.NoError(t, err)
+		_, err = testutil.CreateRelation2(ctx, txn, database, schema2)
+		require.NoError(t, err)
+		require.NoError(t, txn.Commit(ctx))
 	}
 	bat := catalog.MockBatch(schema1, int(schema1.Extra.BlockMaxRows*10-1))
 	defer bat.Close()
@@ -7110,6 +7110,21 @@ func TestAppendAndGC2(t *testing.T) {
 	metaFile := db.BGCheckpointRunner.GetCheckpointMetaFiles()
 	tae.Restart(ctx)
 	db = tae.DB
+	func() {
+		replayTxn, err := db.StartTxn(nil)
+		require.NoError(t, err)
+		defer func() {
+			require.NoError(t, replayTxn.Rollback(ctx))
+		}()
+		replayDB, err := replayTxn.GetDatabase("db")
+		require.NoError(t, err)
+		replayRel1, err := replayDB.GetRelationByName(schema1.Name)
+		require.NoError(t, err)
+		replayRel2, err := replayDB.GetRelationByName(schema2.Name)
+		require.NoError(t, err)
+		testutil.CheckAllColRowsByScan(t, replayRel1, bat.Length(), false)
+		testutil.CheckAllColRowsByScan(t, replayRel2, bat.Length(), false)
+	}()
 	files := make(map[string]struct{}, 0)
 	loadFiles := func(group uint32, lsn uint64, payload []byte, typ uint16, info any) driver.ReplayEntryState {
 		if group != wal.GroupFiles {