*: don't save table IDs in schema diff of FLASHBACK DATABASE

pkg/ddl/schema.go

@@ -319,18 +319,9 @@ func (w *worker) onRecoverSchema(d *ddlCtx, t *meta.Meta, job *model.Job) (ver i
 			}
 		}
 		schemaInfo.State = model.StatePublic
-		diffInfos := make([]schemaIDAndTableInfo, 0, len(recoverTbls))
-		for _, recoverInfo := range recoverTbls {
-			recoverInfo.TableInfo.State = model.StatePublic
-			recoverInfo.TableInfo.UpdateTS = t.StartTS
-			diffInfos = append(diffInfos, schemaIDAndTableInfo{
-				schemaID: schemaInfo.ID,
-				tblInfo:  recoverInfo.TableInfo,
-			})
-		}
 		// use to update InfoSchema
 		job.SchemaID = schemaInfo.ID
-		ver, err = updateSchemaVersion(d, t, job, diffInfos...)
+		ver, err = updateSchemaVersion(d, t, job)
 		if err != nil {
 			return ver, errors.Trace(err)
 		}

pkg/ddl/schema_version.go

@@ -276,6 +276,7 @@ func SetSchemaDiffForRecoverSchema(diff *model.SchemaDiff, job *model.Job) error
 			OldTableID:  recoverTabsInfo[i].TableInfo.ID,
 		}
 	}
+	diff.ReadTableFromMeta = true
 	return nil
 }
 

pkg/domain/domain.go

@@ -547,7 +547,7 @@ func (do *Domain) tryLoadSchemaDiffs(m *meta.Meta, usedVersion, newVersion int64
 	if err != nil {
 		return nil, nil, nil, errors.Trace(err)
 	}
-	builder.SetDeltaUpdateBundles()
+	builder.WithStore(do.store).SetDeltaUpdateBundles()
 	phyTblIDs := make([]int64, 0, len(diffs))
 	actions := make([]uint64, 0, len(diffs))
 	diffTypes := make([]string, 0, len(diffs))

pkg/executor/recover_test.go

@@ -24,6 +24,7 @@ import (
 	ddlutil "github.com/pingcap/tidb/pkg/ddl/util"
 	"github.com/pingcap/tidb/pkg/errno"
 	"github.com/pingcap/tidb/pkg/infoschema"
+	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/parser/auth"
 	"github.com/pingcap/tidb/pkg/sessionctx/variable"
 	"github.com/pingcap/tidb/pkg/store/mockstore"
@@ -633,6 +634,43 @@ func TestFlashbackSchema(t *testing.T) {
 	tk.MustExec("drop user 'testflashbackschema'@'localhost';")
 }
 
+func TestFlashbackSchemaWithManyTables(t *testing.T) {
+	testfailpoint.Enable(t, "github.com/pingcap/tidb/pkg/meta/autoid/mockAutoIDChange", `return(true)`)
+
+	backup := kv.TxnEntrySizeLimit.Load()
+	kv.TxnEntrySizeLimit.Store(50000)
+	t.Cleanup(func() {
+		kv.TxnEntrySizeLimit.Store(backup)
+	})
+
+	store := testkit.CreateMockStore(t, mockstore.WithStoreType(mockstore.EmbedUnistore))
+
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("set @@global.tidb_ddl_error_count_limit = 2")
+	tk.MustExec("set @@global.tidb_enable_fast_create_table=ON")
+	tk.MustExec("drop database if exists many_tables")
+	tk.MustExec("create database if not exists many_tables")
+	tk.MustExec("use many_tables")
+
+	timeBeforeDrop, _, safePointSQL, resetGC := MockGC(tk)
+	defer resetGC()
+
+	// Set GC safe point
+	tk.MustExec(fmt.Sprintf(safePointSQL, timeBeforeDrop))
+	// Set GC enable.
+	require.NoError(t, gcutil.EnableGC(tk.Session()))
+
+	for i := 0; i < 700; i++ {
+		tk.MustExec(fmt.Sprintf("create table t%d (a int)", i))
+	}
+
+	tk.MustExec("drop database many_tables")
+
+	tk.MustExec("flashback database many_tables")
+
+	tk.MustQuery("select count(*) from many_tables.t0").Check(testkit.Rows("0"))
+}
+
 // MockGC is used to make GC work in the test environment.
 func MockGC(tk *testkit.TestKit) (string, string, string, func()) {
 	originGC := ddlutil.IsEmulatorGCEnable()

pkg/infoschema/builder.go

@@ -55,6 +55,7 @@ type Builder struct {
 	factory func() (pools.Resource, error)
 	bundleInfoBuilder
 	infoData *Data
+	store    kv.Storage
 }
 
 // ApplyDiff applies SchemaDiff to the new InfoSchema.
@@ -1002,6 +1003,12 @@ func NewBuilder(r autoid.Requirement, factory func() (pools.Resource, error), in
 	return builder
 }
 
+// WithStore attaches the given store to builder.
+func (b *Builder) WithStore(s kv.Storage) *Builder {
+	b.store = s
+	return b
+}
+
 func tableBucketIdx(tableID int64) int {
 	intest.Assert(tableID > 0)
 	return int(tableID % bucketCount)

pkg/infoschema/infoschema_v2.go

@@ -871,6 +871,25 @@ func applyDropSchema(b *Builder, diff *model.SchemaDiff) []int64 {
 }
 
 func applyRecoverSchema(b *Builder, m *meta.Meta, diff *model.SchemaDiff) ([]int64, error) {
+	if diff.ReadTableFromMeta {
+		// recover tables under the database and set them to diff.AffectedOpts
+		s := b.store.GetSnapshot(kv.MaxVersion)
+		recoverMeta := meta.NewSnapshotMeta(s)
+		tables, err := recoverMeta.ListSimpleTables(diff.SchemaID)
+		if err != nil {
+			return nil, err
+		}
+		diff.AffectedOpts = make([]*model.AffectedOption, 0, len(tables))
+		for _, t := range tables {
+			diff.AffectedOpts = append(diff.AffectedOpts, &model.AffectedOption{
+				SchemaID:    diff.SchemaID,
+				OldSchemaID: diff.SchemaID,
+				TableID:     t.ID,
+				OldTableID:  t.ID,
+			})
+		}
+	}
+
 	if b.enableV2 {
 		return b.applyRecoverSchemaV2(m, diff)
 	}

pkg/parser/model/ddl.go

@@ -1199,6 +1199,9 @@ type SchemaDiff struct {
 	OldSchemaID int64 `json:"old_schema_id"`
 	// RegenerateSchemaMap means whether to rebuild the schema map when applying to the schema diff.
 	RegenerateSchemaMap bool `json:"regenerate_schema_map"`
+	// ReadTableFromMeta is set to avoid the diff is too large to be saved in SchemaDiff.
+	// infoschema should read latest meta directly.
+	ReadTableFromMeta bool `json:"read_table_from_meta,omitempty"`
 
 	AffectedOpts []*AffectedOption `json:"affected_options"`
 }