br: pitr sort tables before creating

br/pkg/restore/snap_client/BUILD.bazel

@@ -89,7 +89,7 @@ go_test(
     ],
     embed = [":snap_client"],
     flaky = True,
-    shard_count = 24,
+    shard_count = 25,
     deps = [
         "//br/pkg/errors",
         "//br/pkg/glue",

br/pkg/restore/snap_client/client.go

@@ -21,6 +21,7 @@ import (
 	"crypto/tls"
 	"encoding/json"
 	"slices"
+	"sort"
 	"strings"
 	"sync"
 	"time"
@@ -925,23 +926,49 @@ func (rc *SnapClient) createTables(
 	return cts, nil
 }
 
+// SortTablesBySchemaID sorts tables by their schema ID to ensure tables in the same schema
+// are processed together. It returns a new slice with sorted tables.
+func SortTablesBySchemaID(tables []*metautil.Table) []*metautil.Table {
+	if len(tables) <= 1 {
+		return tables
+	}
+
+	orderedTables := make([]*metautil.Table, len(tables))
+	copy(orderedTables, tables)
+
+	sort.SliceStable(orderedTables, func(i, j int) bool {
+		// first sort by schema ID
+		if orderedTables[i].DB.ID != orderedTables[j].DB.ID {
+			return orderedTables[i].DB.ID < orderedTables[j].DB.ID
+		}
+		// if schema IDs are equal, sort by table ID
+		return orderedTables[i].Info.ID < orderedTables[j].Info.ID
+	})
+
+	return orderedTables
+}
+
 func (rc *SnapClient) createTablesBatch(ctx context.Context, tables []*metautil.Table, newTS uint64) ([]*CreatedTable, error) {
 	eg, ectx := errgroup.WithContext(ctx)
 	rater := logutil.TraceRateOver(metrics.RestoreTableCreatedCount)
 	workers := tidbutil.NewWorkerPool(uint(len(rc.dbPool)), "Create Tables Worker")
-	numOfTables := len(tables)
+
+	// sort tables by schema ID to ensure tables in the same schema are processed together
+	orderedTables := SortTablesBySchemaID(tables)
+
+	numOfTables := len(orderedTables)
 	createdTables := struct {
 		sync.Mutex
 		tables []*CreatedTable
 	}{
-		tables: make([]*CreatedTable, 0, len(tables)),
+		tables: make([]*CreatedTable, 0, numOfTables),
 	}
 
 	for lastSent := 0; lastSent < numOfTables; lastSent += int(rc.batchDdlSize) {
-		end := min(lastSent+int(rc.batchDdlSize), len(tables))
+		end := min(lastSent+int(rc.batchDdlSize), numOfTables)
 		log.Info("create tables", zap.Int("table start", lastSent), zap.Int("table end", end))
 
-		tableSlice := tables[lastSent:end]
+		tableSlice := orderedTables[lastSent:end]
 		workers.ApplyWithIDInErrorGroup(eg, func(id uint64) error {
 			db := rc.dbPool[id%uint64(len(rc.dbPool))]
 			cts, err := rc.createTables(ectx, db, tableSlice, newTS) // ddl job for [lastSent:i)

br/pkg/restore/snap_client/client_test.go

@@ -349,3 +349,48 @@ func TestSetSpeedLimit(t *testing.T) {
 		require.Equal(t, mockStores[i].Id, recordStores.get(i))
 	}
 }
+
+func TestSortTablesBySchemaID(t *testing.T) {
+	// Create test tables with different schema IDs in mixed order
+	tables := []*metautil.Table{
+		createTestTable(2, 3),
+		createTestTable(1, 2),
+		createTestTable(3, 5),
+		createTestTable(1, 1),
+		createTestTable(2, 4),
+		createTestTable(3, 6),
+		createTestTable(6, 7),
+	}
+
+	sorted := snapclient.SortTablesBySchemaID(tables)
+
+	require.Len(t, sorted, 7, "Should have 7 tables after sorting")
+
+	expectedSchemaIDs := []int64{1, 1, 2, 2, 3, 3, 6}
+	expectedTableIDs := []int64{1, 2, 3, 4, 5, 6, 7}
+	actualSchemaIDs := make([]int64, 7)
+	actualTableIDs := make([]int64, 7)
+	for i, table := range sorted {
+		actualSchemaIDs[i] = table.DB.ID
+		actualTableIDs[i] = table.Info.ID
+	}
+
+	require.Equal(t, expectedSchemaIDs, actualSchemaIDs, "Tables should be sorted by schema ID")
+	require.Equal(t, expectedTableIDs, actualTableIDs, "Tables should be sorted by table ID")
+}
+
+// Helper function to create a test table with given IDs
+func createTestTable(schemaID, tableID int64) *metautil.Table {
+	dbInfo := &model.DBInfo{
+		ID: schemaID,
+	}
+
+	tableInfo := &model.TableInfo{
+		ID: tableID,
+	}
+
+	return &metautil.Table{
+		DB:   dbInfo,
+		Info: tableInfo,
+	}
+}