diff --git a/logservice/schemastore/disk_format.go b/logservice/schemastore/disk_format.go index 5b95d3d0b7..3fd59cfac8 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 ( @@ -306,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) @@ -355,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 { @@ -366,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, @@ -739,27 +807,19 @@ 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, + tableInfo *model.TableInfo, 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)) + marshalBuf []byte, +) (int64, string, []int64, []byte, error) { + tableInfoValue, err := json.Marshal(tableInfo) + if err != nil { + return 0, "", nil, marshalBuf, errors.WrapError(errors.ErrMarshalFailed, err) } // write table info to batch tableKey, err := tableInfoKey(ts, tableInfo.ID) @@ -771,10 +831,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 @@ -806,7 +867,7 @@ func addTableInfoToBatchWithEncryption( partitionIDs = append(partitionIDs, partition.ID) } } - return tableInfo.ID, tableInfo.Name.O, partitionIDs + return tableInfo.ID, tableInfo.Name.O, partitionIDs, marshalBuf, nil } // persistSchemaSnapshot write database/table/partition info to disks. @@ -850,58 +911,75 @@ 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 } - 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 { + tableID, tableName, partitionIDs, marshalBuf, err := addTableInfoToBatchWithEncryption( + batch, snapTs, dbInfo, tableInfo, encMgr, keyspaceID, + tableInfoEntryMarshalBuf) + tableInfoEntryMarshalBuf = marshalBuf + if err != nil { + callbackErr = err + return callbackErr } - for _, rawTable := range rawTables { - if !isTableRawKey(rawTable.Field) { - continue + 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() + } + // 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") + 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. @@ -972,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, @@ -986,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))) @@ -1013,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) @@ -1027,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, @@ -1042,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) @@ -1055,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{ diff --git a/logservice/schemastore/disk_format_test.go b/logservice/schemastore/disk_format_test.go index 9e54464dcb..b6a284b993 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, nil) require.NoError(t, err) - addTableInfoToBatch(batch, snapshotTs, dbInfo, value) } 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 bbde0c809e..ff8882cffc 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,11 +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, nil) if err != nil { - log.Panic("marshal table info fail", zap.Error(err)) + log.Panic("add table info to batch fail", zap.Error(err)) } - addTableInfoToBatch(batch, snapTs, dbInfo.dbInfo, tableInfoValue) } } if err := batch.Commit(pebble.NoSync); err != nil {