From fb3369968d548a653e2535c5feaf1e0d211325c3 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Fri, 14 Aug 2026 14:27:05 +0800 Subject: [PATCH 1/5] schemastore: reduce memory usage during initialization --- logservice/schemastore/disk_format.go | 98 +++++++++++++++++---------- 1 file changed, 63 insertions(+), 35 deletions(-) diff --git a/logservice/schemastore/disk_format.go b/logservice/schemastore/disk_format.go index 5b95d3d0b7..d3e0b9111e 100644 --- a/logservice/schemastore/disk_format.go +++ b/logservice/schemastore/disk_format.go @@ -761,6 +761,20 @@ func addTableInfoToBatchWithEncryption( if err := json.Unmarshal(tableInfoValue, &tableInfo); err != nil { log.Fatal("unmarshal table info failed", zap.Error(err)) } + return addTableInfoModelToBatchWithEncryption( + batch, ts, dbInfo, &tableInfo, tableInfoValue, encMgr, keyspaceID, + ) +} + +func addTableInfoModelToBatchWithEncryption( + batch *pebble.Batch, + ts uint64, + dbInfo *model.DBInfo, + tableInfo *model.TableInfo, + tableInfoValue []byte, + encMgr encryption.EncryptionManager, + keyspaceID uint32, +) (int64, string, []int64) { // write table info to batch tableKey, err := tableInfoKey(ts, tableInfo.ID) if err != nil { @@ -854,54 +868,68 @@ func persistSchemaSnapshotWithEncryption( if filter.IsSysSchema(dbInfo.Name.O) { continue } - batch := db.NewBatch() - addSchemaInfoToBatchWithEncryption(batch, snapTs, dbInfo, encMgr, keyspaceID) for { - rawTables, err := meta.GetMetasByDBID(dbInfo.ID) - if err == nil { - var tablesInDB map[int64]bool - if collectMetaInfo { - tablesInDB = make(map[int64]bool) + batch := db.NewBatch() + addSchemaInfoToBatchWithEncryption(batch, snapTs, dbInfo, encMgr, keyspaceID) + var tablesInDB map[int64]bool + if collectMetaInfo { + tablesInDB = make(map[int64]bool) + } + + var callbackErr error + err := meta.IterTables(dbInfo.ID, func(tableInfo *model.TableInfo) error { + tableInfoValue, err := json.Marshal(tableInfo) + if err != nil { + callbackErr = errors.WrapError(errors.ErrMarshalFailed, err) + return callbackErr } - for _, rawTable := range rawTables { - if !isTableRawKey(rawTable.Field) { - continue + tableID, tableName, partitionIDs := addTableInfoModelToBatchWithEncryption( + batch, snapTs, dbInfo, tableInfo, tableInfoValue, encMgr, keyspaceID) + if collectMetaInfo { + tableMap[tableID] = &BasicTableInfo{ + SchemaID: dbInfo.ID, + Name: tableName, } - tableID, tableName, partitionIDs := addTableInfoToBatchWithEncryption( - batch, snapTs, dbInfo, rawTable.Value, encMgr, keyspaceID) - if collectMetaInfo { - tableMap[tableID] = &BasicTableInfo{ - SchemaID: dbInfo.ID, - Name: tableName, - } - tablesInDB[tableID] = true - if len(partitionIDs) > 0 { - partitionMap[tableID] = make(BasicPartitionInfo) - for _, partitionID := range partitionIDs { - partitionMap[tableID].AddPartitionIDs(partitionID) - } - } + tablesInDB[tableID] = true + if len(partitionIDs) > 0 { + partitionMap[tableID] = make(BasicPartitionInfo) + partitionMap[tableID].AddPartitionIDs(partitionIDs...) } - // 8M is arbitrary, we can adjust it later - if batch.Len() >= 8*1024*1024 { - if err := batch.Commit(pebble.NoSync); err != nil { - return nil, nil, nil, err - } - batch = db.NewBatch() + } + // Bound the buffered snapshot data and reuse its allocation. + if batch.Len() >= 8*1024*1024 { + if err := batch.Commit(pebble.NoSync); err != nil { + callbackErr = errors.WrapError( + errors.ErrUnexpected, err, "commit schema snapshot batch") + return callbackErr } + batch.Reset() + } + return nil + }) + if callbackErr != nil { + _ = batch.Close() + return nil, nil, nil, callbackErr + } + if err == nil { + if err := batch.Commit(pebble.NoSync); err != nil { + _ = batch.Close() + return nil, nil, nil, errors.WrapError( + errors.ErrUnexpected, err, "commit schema snapshot batch") + } + if err := batch.Close(); err != nil { + return nil, nil, nil, errors.WrapError( + errors.ErrUnexpected, err, "close schema snapshot batch") } if collectMetaInfo { - databaseInfo := &BasicDatabaseInfo{ + databaseMap[dbInfo.ID] = &BasicDatabaseInfo{ Name: dbInfo.Name.O, Tables: tablesInDB, } - databaseMap[dbInfo.ID] = databaseInfo - } - if err := batch.Commit(pebble.NoSync); err != nil { - return nil, nil, nil, err } break } + _ = batch.Close() // If this error is caused by the GC life time is shorter than transaction duration, // it means the snapshot is lost forever, so we should return the error to the caller. From b3ec978120958c3cde9485be7b47abdc5fbd1af5 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Fri, 14 Aug 2026 14:41:04 +0800 Subject: [PATCH 2/5] remove unused code --- logservice/schemastore/disk_format.go | 28 +------------------ logservice/schemastore/disk_format_test.go | 2 +- .../schemastore/persist_storage_test_utils.go | 3 +- 3 files changed, 4 insertions(+), 29 deletions(-) diff --git a/logservice/schemastore/disk_format.go b/logservice/schemastore/disk_format.go index d3e0b9111e..56a8142760 100644 --- a/logservice/schemastore/disk_format.go +++ b/logservice/schemastore/disk_format.go @@ -739,34 +739,8 @@ func addSchemaInfoToBatchWithEncryption(batch *pebble.Batch, ts uint64, info *mo batch.Set(schemaKey, schemaValue, pebble.NoSync) } -func addTableInfoToBatch( - batch *pebble.Batch, - ts uint64, - dbInfo *model.DBInfo, - tableInfoValue []byte, -) (int64, string, []int64) { - return addTableInfoToBatchWithEncryption(batch, ts, dbInfo, tableInfoValue, nil, 0) -} - // addTableInfoToBatchWithEncryption encrypts and adds table info to batch if encryption is enabled func addTableInfoToBatchWithEncryption( - batch *pebble.Batch, - ts uint64, - dbInfo *model.DBInfo, - tableInfoValue []byte, - encMgr encryption.EncryptionManager, - keyspaceID uint32, -) (int64, string, []int64) { - tableInfo := model.TableInfo{} - if err := json.Unmarshal(tableInfoValue, &tableInfo); err != nil { - log.Fatal("unmarshal table info failed", zap.Error(err)) - } - return addTableInfoModelToBatchWithEncryption( - batch, ts, dbInfo, &tableInfo, tableInfoValue, encMgr, keyspaceID, - ) -} - -func addTableInfoModelToBatchWithEncryption( batch *pebble.Batch, ts uint64, dbInfo *model.DBInfo, @@ -883,7 +857,7 @@ func persistSchemaSnapshotWithEncryption( callbackErr = errors.WrapError(errors.ErrMarshalFailed, err) return callbackErr } - tableID, tableName, partitionIDs := addTableInfoModelToBatchWithEncryption( + tableID, tableName, partitionIDs := addTableInfoToBatchWithEncryption( batch, snapTs, dbInfo, tableInfo, tableInfoValue, encMgr, keyspaceID) if collectMetaInfo { tableMap[tableID] = &BasicTableInfo{ diff --git a/logservice/schemastore/disk_format_test.go b/logservice/schemastore/disk_format_test.go index 9e54464dcb..bf25ef7bbd 100644 --- a/logservice/schemastore/disk_format_test.go +++ b/logservice/schemastore/disk_format_test.go @@ -305,7 +305,7 @@ func TestGetAllPhysicalTablesSkipsViews(t *testing.T) { for _, info := range []*model.TableInfo{tableInfo, viewInfo} { value, err := json.Marshal(info) require.NoError(t, err) - addTableInfoToBatch(batch, snapshotTs, dbInfo, value) + addTableInfoToBatchWithEncryption(batch, snapshotTs, dbInfo, info, value, nil, 0) } require.NoError(t, batch.Commit(pebble.NoSync)) diff --git a/logservice/schemastore/persist_storage_test_utils.go b/logservice/schemastore/persist_storage_test_utils.go index bbde0c809e..17da9314df 100644 --- a/logservice/schemastore/persist_storage_test_utils.go +++ b/logservice/schemastore/persist_storage_test_utils.go @@ -137,7 +137,8 @@ func mockWriteKVSnapOnDisk(db *pebble.DB, snapTs uint64, dbInfos []mockDBInfo) { if err != nil { log.Panic("marshal table info fail", zap.Error(err)) } - addTableInfoToBatch(batch, snapTs, dbInfo.dbInfo, tableInfoValue) + addTableInfoToBatchWithEncryption( + batch, snapTs, dbInfo.dbInfo, tableInfo, tableInfoValue, nil, 0) } } if err := batch.Commit(pebble.NoSync); err != nil { From bf194c511b66e5545b264d6e479f768786a42764 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Fri, 14 Aug 2026 14:55:58 +0800 Subject: [PATCH 3/5] fix --- logservice/schemastore/disk_format.go | 16 +++++++++------- logservice/schemastore/disk_format_test.go | 4 ++-- .../schemastore/persist_storage_test_utils.go | 8 +++----- 3 files changed, 14 insertions(+), 14 deletions(-) diff --git a/logservice/schemastore/disk_format.go b/logservice/schemastore/disk_format.go index 56a8142760..9072687294 100644 --- a/logservice/schemastore/disk_format.go +++ b/logservice/schemastore/disk_format.go @@ -745,10 +745,13 @@ func addTableInfoToBatchWithEncryption( ts uint64, dbInfo *model.DBInfo, tableInfo *model.TableInfo, - tableInfoValue []byte, encMgr encryption.EncryptionManager, keyspaceID uint32, -) (int64, string, []int64) { +) (int64, string, []int64, error) { + tableInfoValue, err := json.Marshal(tableInfo) + if err != nil { + return 0, "", nil, errors.WrapError(errors.ErrMarshalFailed, err) + } // write table info to batch tableKey, err := tableInfoKey(ts, tableInfo.ID) if err != nil { @@ -794,7 +797,7 @@ func addTableInfoToBatchWithEncryption( partitionIDs = append(partitionIDs, partition.ID) } } - return tableInfo.ID, tableInfo.Name.O, partitionIDs + return tableInfo.ID, tableInfo.Name.O, partitionIDs, nil } // persistSchemaSnapshot write database/table/partition info to disks. @@ -852,13 +855,12 @@ func persistSchemaSnapshotWithEncryption( var callbackErr error err := meta.IterTables(dbInfo.ID, func(tableInfo *model.TableInfo) error { - tableInfoValue, err := json.Marshal(tableInfo) + tableID, tableName, partitionIDs, err := addTableInfoToBatchWithEncryption( + batch, snapTs, dbInfo, tableInfo, encMgr, keyspaceID) if err != nil { - callbackErr = errors.WrapError(errors.ErrMarshalFailed, err) + callbackErr = err return callbackErr } - tableID, tableName, partitionIDs := addTableInfoToBatchWithEncryption( - batch, snapTs, dbInfo, tableInfo, tableInfoValue, encMgr, keyspaceID) if collectMetaInfo { tableMap[tableID] = &BasicTableInfo{ SchemaID: dbInfo.ID, diff --git a/logservice/schemastore/disk_format_test.go b/logservice/schemastore/disk_format_test.go index bf25ef7bbd..d689baa69e 100644 --- a/logservice/schemastore/disk_format_test.go +++ b/logservice/schemastore/disk_format_test.go @@ -303,9 +303,9 @@ func TestGetAllPhysicalTablesSkipsViews(t *testing.T) { }() addSchemaInfoToBatch(batch, snapshotTs, dbInfo) for _, info := range []*model.TableInfo{tableInfo, viewInfo} { - value, err := json.Marshal(info) + _, _, _, err := addTableInfoToBatchWithEncryption( + batch, snapshotTs, dbInfo, info, nil, 0) require.NoError(t, err) - addTableInfoToBatchWithEncryption(batch, snapshotTs, dbInfo, info, value, nil, 0) } require.NoError(t, batch.Commit(pebble.NoSync)) diff --git a/logservice/schemastore/persist_storage_test_utils.go b/logservice/schemastore/persist_storage_test_utils.go index 17da9314df..0565a4ae95 100644 --- a/logservice/schemastore/persist_storage_test_utils.go +++ b/logservice/schemastore/persist_storage_test_utils.go @@ -15,7 +15,6 @@ package schemastore import ( "context" - "encoding/json" "fmt" "os" "strings" @@ -133,12 +132,11 @@ func mockWriteKVSnapOnDisk(db *pebble.DB, snapTs uint64, dbInfos []mockDBInfo) { for _, dbInfo := range dbInfos { addSchemaInfoToBatch(batch, snapTs, dbInfo.dbInfo) for _, tableInfo := range dbInfo.tables { - tableInfoValue, err := json.Marshal(tableInfo) + _, _, _, err := addTableInfoToBatchWithEncryption( + batch, snapTs, dbInfo.dbInfo, tableInfo, nil, 0) if err != nil { - log.Panic("marshal table info fail", zap.Error(err)) + log.Panic("add table info to batch fail", zap.Error(err)) } - addTableInfoToBatchWithEncryption( - batch, snapTs, dbInfo.dbInfo, tableInfo, tableInfoValue, nil, 0) } } if err := batch.Commit(pebble.NoSync); err != nil { From 3d18507244953f7d034115e1e280f7f18414c8c2 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Sun, 16 Aug 2026 18:09:14 +0800 Subject: [PATCH 4/5] improve --- logservice/schemastore/disk_format.go | 26 +++++++++------ logservice/schemastore/disk_format_test.go | 32 +++++++++++++++++-- .../schemastore/persist_storage_test_utils.go | 4 +-- 3 files changed, 49 insertions(+), 13 deletions(-) diff --git a/logservice/schemastore/disk_format.go b/logservice/schemastore/disk_format.go index 9072687294..8dc1d1bc78 100644 --- a/logservice/schemastore/disk_format.go +++ b/logservice/schemastore/disk_format.go @@ -59,7 +59,8 @@ const ( snapshotPartitionKeyPrefix = "sp_" ddlKeyPrefix = "ds_" - schemaStoreKeyMaskLen = 8 + schemaStoreKeyMaskLen = 8 + schemaSnapshotBatchFlushSize = 512 * 1024 ) const ( @@ -747,10 +748,11 @@ func addTableInfoToBatchWithEncryption( tableInfo *model.TableInfo, encMgr encryption.EncryptionManager, keyspaceID uint32, -) (int64, string, []int64, error) { + marshalBuf []byte, +) (int64, string, []int64, []byte, error) { tableInfoValue, err := json.Marshal(tableInfo) if err != nil { - return 0, "", nil, errors.WrapError(errors.ErrMarshalFailed, err) + return 0, "", nil, marshalBuf, errors.WrapError(errors.ErrMarshalFailed, err) } // write table info to batch tableKey, err := tableInfoKey(ts, tableInfo.ID) @@ -762,10 +764,11 @@ func addTableInfoToBatchWithEncryption( SchemaName: dbInfo.Name.O, TableInfoValue: tableInfoValue, } - tableInfoEntryValue, err := tableInfoEntry.MarshalMsg(nil) + tableInfoEntryValue, err := tableInfoEntry.MarshalMsg(marshalBuf[:0]) if err != nil { log.Fatal("marshal table info entry failed", zap.Error(err)) } + marshalBuf = tableInfoEntryValue keyMask := uint64(0) // Encrypt if encryption is enabled @@ -797,7 +800,7 @@ func addTableInfoToBatchWithEncryption( partitionIDs = append(partitionIDs, partition.ID) } } - return tableInfo.ID, tableInfo.Name.O, partitionIDs, nil + return tableInfo.ID, tableInfo.Name.O, partitionIDs, marshalBuf, nil } // persistSchemaSnapshot write database/table/partition info to disks. @@ -841,6 +844,7 @@ func persistSchemaSnapshotWithEncryption( tableMap = make(map[int64]*BasicTableInfo) partitionMap = make(map[int64]BasicPartitionInfo) } + var tableInfoEntryMarshalBuf []byte for _, dbInfo := range dbInfos { if filter.IsSysSchema(dbInfo.Name.O) { continue @@ -855,8 +859,11 @@ func persistSchemaSnapshotWithEncryption( var callbackErr error err := meta.IterTables(dbInfo.ID, func(tableInfo *model.TableInfo) error { - tableID, tableName, partitionIDs, err := addTableInfoToBatchWithEncryption( - batch, snapTs, dbInfo, tableInfo, encMgr, keyspaceID) + tableID, tableName, partitionIDs, marshalBuf, err := + addTableInfoToBatchWithEncryption( + batch, snapTs, dbInfo, tableInfo, encMgr, keyspaceID, + tableInfoEntryMarshalBuf) + tableInfoEntryMarshalBuf = marshalBuf if err != nil { callbackErr = err return callbackErr @@ -872,8 +879,9 @@ func persistSchemaSnapshotWithEncryption( partitionMap[tableID].AddPartitionIDs(partitionIDs...) } } - // Bound the buffered snapshot data and reuse its allocation. - if batch.Len() >= 8*1024*1024 { + // Keep the batch below Pebble's 1 MiB retention limit so Reset + // can reuse the backing buffer instead of allocating it again. + if batch.Len() >= schemaSnapshotBatchFlushSize { if err := batch.Commit(pebble.NoSync); err != nil { callbackErr = errors.WrapError( errors.ErrUnexpected, err, "commit schema snapshot batch") diff --git a/logservice/schemastore/disk_format_test.go b/logservice/schemastore/disk_format_test.go index d689baa69e..b6a284b993 100644 --- a/logservice/schemastore/disk_format_test.go +++ b/logservice/schemastore/disk_format_test.go @@ -303,8 +303,8 @@ func TestGetAllPhysicalTablesSkipsViews(t *testing.T) { }() addSchemaInfoToBatch(batch, snapshotTs, dbInfo) for _, info := range []*model.TableInfo{tableInfo, viewInfo} { - _, _, _, err := addTableInfoToBatchWithEncryption( - batch, snapshotTs, dbInfo, info, nil, 0) + _, _, _, _, err := addTableInfoToBatchWithEncryption( + batch, snapshotTs, dbInfo, info, nil, 0, nil) require.NoError(t, err) } require.NoError(t, batch.Commit(pebble.NoSync)) @@ -329,6 +329,34 @@ func TestGetAllPhysicalTablesSkipsViews(t *testing.T) { }, tables) } +func TestAddTableInfoToBatchReusesMarshalBuffer(t *testing.T) { + db, err := pebble.Open(t.TempDir(), &pebble.Options{}) + require.NoError(t, err) + defer func() { + require.NoError(t, db.Close()) + }() + + const snapshotTs = uint64(100) + dbInfo := &model.DBInfo{ID: 100, Name: ast.NewCIStr("test")} + tableInfo := newEligibleTableInfoForTest(200, "t1") + batch := db.NewBatch() + defer func() { + require.NoError(t, batch.Close()) + }() + + _, _, _, marshalBuf, err := addTableInfoToBatchWithEncryption( + batch, snapshotTs, dbInfo, tableInfo, nil, 0, nil) + require.NoError(t, err) + require.NotEmpty(t, marshalBuf) + firstByte := &marshalBuf[0] + + _, _, _, marshalBuf, err = addTableInfoToBatchWithEncryption( + batch, snapshotTs, dbInfo, tableInfo, nil, 0, marshalBuf) + require.NoError(t, err) + require.NotEmpty(t, marshalBuf) + require.Same(t, firstByte, &marshalBuf[0]) +} + type snapshotLostByGCError struct{} func (snapshotLostByGCError) Error() string { diff --git a/logservice/schemastore/persist_storage_test_utils.go b/logservice/schemastore/persist_storage_test_utils.go index 0565a4ae95..ff8882cffc 100644 --- a/logservice/schemastore/persist_storage_test_utils.go +++ b/logservice/schemastore/persist_storage_test_utils.go @@ -132,8 +132,8 @@ func mockWriteKVSnapOnDisk(db *pebble.DB, snapTs uint64, dbInfos []mockDBInfo) { for _, dbInfo := range dbInfos { addSchemaInfoToBatch(batch, snapTs, dbInfo.dbInfo) for _, tableInfo := range dbInfo.tables { - _, _, _, err := addTableInfoToBatchWithEncryption( - batch, snapTs, dbInfo.dbInfo, tableInfo, nil, 0) + _, _, _, _, err := addTableInfoToBatchWithEncryption( + batch, snapTs, dbInfo.dbInfo, tableInfo, nil, 0, nil) if err != nil { log.Panic("add table info to batch fail", zap.Error(err)) } From c210178c408f5f95299d9177be0b75fad1b153bb Mon Sep 17 00:00:00 2001 From: lidezhu Date: Mon, 17 Aug 2026 14:32:55 +0800 Subject: [PATCH 5/5] more fix --- logservice/schemastore/disk_format.go | 155 ++++++++++++++++++++++---- 1 file changed, 134 insertions(+), 21 deletions(-) diff --git a/logservice/schemastore/disk_format.go b/logservice/schemastore/disk_format.go index 8dc1d1bc78..3fd59cfac8 100644 --- a/logservice/schemastore/disk_format.go +++ b/logservice/schemastore/disk_format.go @@ -307,6 +307,18 @@ func loadTablesInKVSnapWithEncryption( databaseMap map[int64]*BasicDatabaseInfo, encMgr encryption.EncryptionManager, keyspaceID uint32, +) (map[int64]*BasicTableInfo, map[int64]BasicPartitionInfo, error) { + return loadTablesInKVSnapWithEncryptionAndCallback( + snap, gcTs, databaseMap, encMgr, keyspaceID, nil) +} + +func loadTablesInKVSnapWithEncryptionAndCallback( + snap *pebble.Snapshot, + gcTs uint64, + databaseMap map[int64]*BasicDatabaseInfo, + encMgr encryption.EncryptionManager, + keyspaceID uint32, + onTable func(schemaName string, tableInfo *model.TableInfo), ) (map[int64]*BasicTableInfo, map[int64]BasicPartitionInfo, error) { tablesInKVSnap := make(map[int64]*BasicTableInfo) partitionsInKVSnap := make(map[int64]BasicPartitionInfo) @@ -356,6 +368,9 @@ func loadTablesInKVSnapWithEncryption( SchemaID: table_info_entry.SchemaID, Name: tableInfo.Name.O, } + if onTable != nil { + onTable(table_info_entry.SchemaName, &tableInfo) + } if tableInfo.Partition != nil { partitionInfo := make(BasicPartitionInfo) for _, partition := range tableInfo.Partition.Definitions { @@ -367,6 +382,58 @@ func loadTablesInKVSnapWithEncryption( return tablesInKVSnap, partitionsInKVSnap, nil } +type physicalTableTraits struct { + isView bool + eligible bool + splitable bool +} + +func physicalTableTraitsFromModel( + tableInfo *model.TableInfo, + tableFilter filter.Filter, +) physicalTableTraits { + eligible := true + if tableFilter != nil { + // IsEligible only needs these structural fields. Avoid WrapTableInfo here, + // which would build and retain a full column schema for every table. + compactTableInfo := &common.TableInfo{ + HasPKOrNotNullUK: common.OriginalHasPKOrNotNullUK(tableInfo), + View: tableInfo.View, + Sequence: tableInfo.Sequence, + } + eligible = tableFilter.IsEligibleTable(compactTableInfo) + } + return physicalTableTraits{ + isView: tableInfo.View != nil, + eligible: eligible, + splitable: isSplitable(tableInfo), + } +} + +func physicalTableTraitsFromCommon( + tableInfo *common.TableInfo, + tableFilter filter.Filter, +) physicalTableTraits { + eligible := true + if tableFilter != nil { + eligible = tableFilter.IsEligibleTable(tableInfo) + } + splitable := tableInfo.GetPkColInfo() != nil + if splitable { + for _, index := range tableInfo.GetIndices() { + if !index.Primary && index.Unique { + splitable = false + break + } + } + } + return physicalTableTraits{ + isView: tableInfo.IsView(), + eligible: eligible, + splitable: splitable, + } +} + func loadFullTablesInKVSnap( snap *pebble.Snapshot, gcTs uint64, @@ -859,10 +926,9 @@ func persistSchemaSnapshotWithEncryption( var callbackErr error err := meta.IterTables(dbInfo.ID, func(tableInfo *model.TableInfo) error { - tableID, tableName, partitionIDs, marshalBuf, err := - addTableInfoToBatchWithEncryption( - batch, snapTs, dbInfo, tableInfo, encMgr, keyspaceID, - tableInfoEntryMarshalBuf) + tableID, tableName, partitionIDs, marshalBuf, err := addTableInfoToBatchWithEncryption( + batch, snapTs, dbInfo, tableInfo, encMgr, keyspaceID, + tableInfoEntryMarshalBuf) tableInfoEntryMarshalBuf = marshalBuf if err != nil { callbackErr = err @@ -984,6 +1050,32 @@ func cleanObsoleteData(db *pebble.DB, oldGcTs uint64, gcTs uint64) { } } +func loadPhysicalTableTraitsAtTs( + storageSnap *pebble.Snapshot, + gcTs uint64, + snapVersion uint64, + physicalTableID int64, + ddlHistory []uint64, + tableFilter filter.Filter, + encMgr encryption.EncryptionManager, + keyspaceID uint32, +) (physicalTableTraits, error) { + store := newEmptyVersionedTableInfoStore(physicalTableID) + if err := addTableInfoFromKVSnap(store, gcTs, storageSnap, encMgr, keyspaceID); err != nil { + return physicalTableTraits{}, err + } + for _, version := range ddlHistory { + ddlEvent := readPersistedDDLEventWithEncryption(storageSnap, version, encMgr, keyspaceID) + store.applyDDLFromPersistStorage(&ddlEvent) + } + store.setTableInfoInitialized() + tableInfo, err := store.getTableInfo(snapVersion) + if err != nil { + return physicalTableTraits{}, err + } + return physicalTableTraitsFromCommon(tableInfo, tableFilter), nil +} + func loadAllPhysicalTablesAtTs( storageSnap *pebble.Snapshot, gcTs uint64, @@ -998,13 +1090,17 @@ func loadAllPhysicalTablesAtTs( return nil, err } - tableInfoMap, tableMap, partitionMap, err := loadFullTablesInKVSnapWithEncryption( - storageSnap, gcTs, databaseMap, encMgr, keyspaceID) + tableTraits := make(map[int64]physicalTableTraits) + tableMap, partitionMap, err := loadTablesInKVSnapWithEncryptionAndCallback( + storageSnap, gcTs, databaseMap, encMgr, keyspaceID, + func(_ string, tableInfo *model.TableInfo) { + tableTraits[tableInfo.ID] = physicalTableTraitsFromModel(tableInfo, tableFilter) + }) if err != nil { return nil, err } log.Info("after load tables in kv snap", - zap.Int("tableInfoMapLen", len(tableInfoMap)), + zap.Int("tableTraitsLen", len(tableTraits)), zap.Int("tableMapLen", len(tableMap)), zap.Int("partitionMapLen", len(partitionMap))) @@ -1025,6 +1121,8 @@ func loadAllPhysicalTablesAtTs( log.Fatal("new iterator failed", zap.Error(err)) } defer snapIter.Close() + tablesDDLHistory := make(map[int64][]uint64) + tableTriggerDDLHistory := make([]uint64, 0) for snapIter.First(); snapIter.Valid(); snapIter.Next() { ddlValue := snapIter.Value() ddlValue, err = decryptValueIfNeeded(snapIter.Key(), ddlValue, encMgr, keyspaceID) @@ -1039,12 +1137,13 @@ func loadAllPhysicalTablesAtTs( if !ok { log.Panic("unknown ddl type", zap.Any("ddlType", ddlEvent.Type), zap.String("query", ddlEvent.Query)) } - // Note: updateFullTableInfoFunc must be called before updateSchemaMetadataFunc, - // because it depends on some info which may be updated by updateSchemaMetadataFunc. - handler.updateFullTableInfoFunc(updateFullTableInfoFuncArgs{ - event: &ddlEvent, - databaseMap: databaseMap, - tableInfoMap: tableInfoMap, + tableTriggerDDLHistory = handler.updateDDLHistoryFunc(updateDDLHistoryFuncArgs{ + ddlEvent: &ddlEvent, + databaseMap: databaseMap, + tableMap: tableMap, + partitionMap: partitionMap, + tablesDDLHistory: tablesDDLHistory, + tableTriggerDDLHistory: tableTriggerDDLHistory, }) handler.updateSchemaMetadataFunc(updateSchemaMetadataFuncArgs{ event: &ddlEvent, @@ -1054,7 +1153,7 @@ func loadAllPhysicalTablesAtTs( }) } log.Info("after load tables from ddl", - zap.Int("tableInfoMapLen", len(tableInfoMap)), + zap.Int("tableTraitsLen", len(tableTraits)), zap.Int("tableMapLen", len(tableMap)), zap.Int("partitionMapLen", len(partitionMap))) tables := make([]commonEvent.Table, 0) @@ -1067,27 +1166,41 @@ func loadAllPhysicalTablesAtTs( zap.Any("databaseMapLen", len(databaseMap))) } schemaName := databaseMap[tableInfo.SchemaID].Name - fullTableInfo, ok := tableInfoMap[tableID] - if !ok { - log.Panic("table info not found", zap.Int64("tableID", tableID)) + physicalTableID := tableID + if partitionInfo, ok := partitionMap[tableID]; ok { + for partitionID := range partitionInfo { + physicalTableID = partitionID + break + } + } + traits, ok := tableTraits[tableID] + ddlHistory := tablesDDLHistory[physicalTableID] + if !ok || len(ddlHistory) > 0 { + traits, err = loadPhysicalTableTraitsAtTs( + storageSnap, gcTs, snapVersion, physicalTableID, ddlHistory, + tableFilter, encMgr, keyspaceID) + if err != nil { + return nil, err + } } // Views have no physical KV events. Their DDLs are replicated by the // table-trigger dispatcher, so creating a table dispatcher for a view can // only leave an orphan after the view is dropped. - if fullTableInfo.View != nil { + if traits.isView { continue } if tableFilter != nil { if tableFilter.ShouldIgnoreTable(schemaName, tableInfo.Name) { continue } - if !tableFilter.IsEligibleTable(common.WrapTableInfo(schemaName, fullTableInfo)) { - log.Info("table is not eligible, should ignore this table", zap.String("schema", schemaName), zap.String("table", tableInfo.Name), zap.Any("tableInfo", fullTableInfo)) + if !traits.eligible { + log.Info("table is not eligible, should ignore this table", + zap.String("schema", schemaName), zap.String("table", tableInfo.Name)) continue } } - splitable := isSplitable(fullTableInfo) + splitable := traits.splitable if partitionInfo, ok := partitionMap[tableID]; ok { for partitionID := range partitionInfo { tables = append(tables, commonEvent.Table{