Repository: matrixorigin/matrixone
Pull request #13903 (2023-12-28 09:39:13)

Keep old usage and fix account type decode err
## What type of PR is this?

- [ ] API-change
- [x] BUG
- [x] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #12995, https://github.com/matrixorigin/MO-Cloud/issues/2146

## What this PR does / why we need it:

1. the storage usage data could be lost when the upgrade is because there is no object info in the catalog when first replaying the old ckp.
2. correct the account type in the storage usage bat to uint64.
3. adding more tests to test the behavior of the usage cache.
4. one way to provide support is to transfer the usage to the right account.


Patch:
diff --git a/pkg/frontend/show_account.go b/pkg/frontend/show_account.go
index 12242ff1c..a23733c89 100644
--- a/pkg/frontend/show_account.go
+++ b/pkg/frontend/show_account.go
@@ -182,7 +182,7 @@ func handleStorageUsageResponse(
 	result := make(map[int32]uint64, 0)
 
 	for x := range usage.AccIds {
-		result[usage.AccIds[x]] = usage.Sizes[x]
+		result[usage.AccIds[x]] += usage.Sizes[x]
 	}
 
 	return result, nil
@@ -196,7 +196,7 @@ func checkStorageUsageCache(accIds [][]int32) (result map[int32]uint64, succeed
 	result = make(map[int32]uint64)
 	for x := range accIds {
 		for y := range accIds[x] {
-			size, exist := cnUsageCache.GatherAccountSize(uint32(accIds[x][y]))
+			size, exist := cnUsageCache.GatherAccountSize(uint64(accIds[x][y]))
 			if !exist {
 				// one missed, update all
 				return nil, false
@@ -220,10 +220,12 @@ func updateStorageUsageCache(accIds []int32, sizes []uint64) {
 
 	// step 2: update
 	for x := range accIds {
-		cnUsageCache.Update(logtail.UsageData{
-			AccId: uint32(accIds[x]),
-			Size:  sizes[x],
-		})
+		usage := logtail.UsageData{AccId: uint64(accIds[x]), Size: sizes[x]}
+		if old, exist := cnUsageCache.Get(usage); exist {
+			usage.Size += old.Size
+		}
+
+		cnUsageCache.Update(usage)
 	}
 }
 
diff --git a/pkg/vm/engine/tae/catalog/database.go b/pkg/vm/engine/tae/catalog/database.go
index 0c7a7481d..3ce5b87ee 100644
--- a/pkg/vm/engine/tae/catalog/database.go
+++ b/pkg/vm/engine/tae/catalog/database.go
@@ -637,13 +637,13 @@ func (e *DBEntry) IsActive() bool {
 }
 
 // only for test
-func MockDBEntryWithAccInfo(accId uint32, dbId uint64) *DBEntry {
+func MockDBEntryWithAccInfo(accId uint64, dbId uint64) *DBEntry {
 	entry := &DBEntry{
 		ID: dbId,
 	}
 
 	entry.DBNode = &DBNode{}
-	entry.DBNode.acInfo.TenantID = accId
+	entry.DBNode.acInfo.TenantID = uint32(accId)
 
 	return entry
 }
diff --git a/pkg/vm/engine/tae/logtail/storage_usage.go b/pkg/vm/engine/tae/logtail/storage_usage.go
index fd532c287..ba909e47c 100644
--- a/pkg/vm/engine/tae/logtail/storage_usage.go
+++ b/pkg/vm/engine/tae/logtail/storage_usage.go
@@ -21,9 +21,11 @@ import (
 	"math"
 	"math/rand"
 	"sync"
+	"sync/atomic"
 	"time"
 	"unsafe"
 
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/fileservice"
@@ -90,7 +92,7 @@ const (
 const StorageUsageMagic uint64 = 0x1A2B3C4D5E6F
 
 type UsageData struct {
-	AccId uint32
+	AccId uint64
 	DbId  uint64
 	TblId uint64
 	Size  uint64
@@ -99,20 +101,21 @@ type UsageData struct {
 var zeroUsageData UsageData = UsageData{math.MaxUint32, math.MaxUint64, math.MaxUint64, math.MaxInt64}
 
 // MockUsageData generates accCnt * dbCnt * tblCnt UsageDatas.
-// the accIds, dbIds and tblIds are random produced
-func MockUsageData(accCnt, dbCnt, tblCnt int) (result []UsageData) {
+// the accIds, dbIds and tblIds are random produced.
+// this func ensure that all ids are different.
+func MockUsageData(accCnt, dbCnt, tblCnt int, allocator *atomic.Uint64) (result []UsageData) {
 	for x := 0; x < accCnt; x++ {
-		accId := rand.Uint32()
+		accId := allocator.Add(1)
 
 		for y := 0; y < dbCnt; y++ {
-			dbId := rand.Uint64()
+			dbId := allocator.Add(1)
 
 			for z := 0; z < tblCnt; z++ {
 				result = append(result, UsageData{
 					AccId: accId,
 					DbId:  dbId,
-					TblId: rand.Uint64(),
-					Size:  uint64(rand.Int63() % 0x3ffff),
+					TblId: allocator.Add(1),
+					Size:  uint64(rand.Int63() % 0x3fff),
 				})
 			}
 		}
@@ -235,8 +238,8 @@ func (c *StorageUsageCache) ClearForUpdate() {
 	c.data.Clear()
 }
 
-func (c *StorageUsageCache) GatherAllAccSize() (usages map[uint32]uint64) {
-	usages = make(map[uint32]uint64)
+func (c *StorageUsageCache) GatherAllAccSize() (usages map[uint64]uint64) {
+	usages = make(map[uint64]uint64)
 	c.data.Scan(func(item UsageData) bool {
 		usages[item.AccId] += item.Size
 		return true
@@ -245,7 +248,7 @@ func (c *StorageUsageCache) GatherAllAccSize() (usages map[uint32]uint64) {
 	return
 }
 
-func (c *StorageUsageCache) GatherAccountSize(id uint32) (size uint64, exist bool) {
+func (c *StorageUsageCache) GatherAccountSize(id uint64) (size uint64, exist bool) {
 	iter := c.data.Iter()
 	defer iter.Release()
 
@@ -279,20 +282,22 @@ func (c *StorageUsageCache) Delete(usage UsageData) {
 
 type TNUsageMemo struct {
 	sync.Mutex
-	cache    *StorageUsageCache
+	cache *StorageUsageCache
+	// has update
 	pending  bool
 	reqTrace []struct {
 		hint      string
 		timeStamp time.Time
-		accountId uint32
+		accountId uint64
 		totalSize uint64
 	}
 
 	newAccCache *StorageUsageCache
 
 	pendingReplay struct {
-		datas []*CheckpointData
-		vers  []uint32
+		datas   []*CheckpointData
+		vers    []uint32
+		delayed map[uint64]UsageData
 	}
 }
 
@@ -308,6 +313,10 @@ func (m *TNUsageMemo) PrepareReplay(datas []*CheckpointData, vers []uint32) {
 	m.pendingReplay.vers = vers
 }
 
+func (m *TNUsageMemo) GetDelayed() map[uint64]UsageData {
+	return m.pendingReplay.delayed
+}
+
 func (m *TNUsageMemo) GetNewAccCacheLatestUpdate() types.TS {
 	return types.BuildTS(m.newAccCache.lastUpdate.UnixNano(), 0)
 }
@@ -320,12 +329,12 @@ func (m *TNUsageMemo) ClearNewAccCache() {
 	m.newAccCache.ClearForUpdate()
 }
 
-func (m *TNUsageMemo) AddReqTrace(accountId uint32, tSize uint64, t time.Time, hint string) {
+func (m *TNUsageMemo) AddReqTrace(accountId uint64, tSize uint64, t time.Time, hint string) {
 	m.reqTrace = append(m.reqTrace,
 		struct {
 			hint      string
 			timeStamp time.Time
-			accountId uint32
+			accountId uint64
 			totalSize uint64
 		}{
 			hint:      hint,
@@ -335,7 +344,7 @@ func (m *TNUsageMemo) AddReqTrace(accountId uint32, tSize uint64, t time.Time, h
 		})
 }
 
-func (m *TNUsageMemo) GetAllReqTrace() (accountIds []uint32, timestamps []time.Time, sizes []uint64, hints []string) {
+func (m *TNUsageMemo) GetAllReqTrace() (accountIds []uint64, timestamps []time.Time, sizes []uint64, hints []string) {
 	m.EnterProcessing()
 	defer m.LeaveProcessing()
 
@@ -389,19 +398,19 @@ func (m *TNUsageMemo) HasUpdate() bool {
 	return m.pending
 }
 
-func (m *TNUsageMemo) gatherAccountSizeHelper(cache *StorageUsageCache, id uint32) (size uint64, exist bool) {
+func (m *TNUsageMemo) gatherAccountSizeHelper(cache *StorageUsageCache, id uint64) (size uint64, exist bool) {
 	return cache.GatherAccountSize(id)
 }
 
-func (m *TNUsageMemo) GatherAccountSize(id uint32) (size uint64, exist bool) {
+func (m *TNUsageMemo) GatherAccountSize(id uint64) (size uint64, exist bool) {
 	return m.gatherAccountSizeHelper(m.cache, id)
 }
 
-func (m *TNUsageMemo) GatherNewAccountSize(id uint32) (size uint64, exist bool) {
+func (m *TNUsageMemo) GatherNewAccountSize(id uint64) (size uint64, exist bool) {
 	return m.gatherAccountSizeHelper(m.newAccCache, id)
 }
 
-func (m *TNUsageMemo) GatherAllAccSize() (usages map[uint32]uint64) {
+func (m *TNUsageMemo) GatherAllAccSize() (usages map[uint64]uint64) {
 	return m.cache.GatherAllAccSize()
 }
 
@@ -452,31 +461,36 @@ func (m *TNUsageMemo) applyDeletes(
 			dbs = append(dbs, e)
 		case *catalog.TableEntry:
 			piovt := UsageData{
-				e.GetDB().GetTenantID(),
+				uint64(e.GetDB().GetTenantID()),
 				e.GetDB().GetID(), e.GetID(), 0}
 			if usage, exist := m.cache.Get(piovt); exist {
 				appendToStorageUsageBat(ckpData, usage, true, mp)
 				m.Delete(usage)
-				buf.WriteString(fmt.Sprintf("[tbl]%s_%d_%d_%d_%d; ",
+				buf.WriteString(fmt.Sprintf("[d-tbl]%s_%d_%d_%d_%d; ",
 					e.GetFullName(), usage.AccId, usage.DbId, usage.TblId, usage.Size))
 			}
 		}
 	}
 
+	isSameDBFunc := func(a UsageData, db *catalog.DBEntry) bool {
+		return a.AccId == uint64(db.GetTenantID()) && a.DbId == db.ID
+	}
+
 	usages := make([]UsageData, 0)
 	for _, db := range dbs {
 		iter := m.cache.Iter()
-		iter.Seek(UsageData{db.GetTenantID(), db.ID, 0, 0})
+		iter.Seek(UsageData{uint64(db.GetTenantID()), db.ID, 0, 0})
 
-		if iter.Item().DbId != db.ID || iter.Item().AccId != db.GetTenantID() {
+		if !isSameDBFunc(iter.Item(), db) {
 			iter.Release()
-			buf.WriteString(fmt.Sprintf("[db]%s_%d_%d_%d; ",
+			// db not found in cache
+			buf.WriteString(fmt.Sprintf("[n-db]%s_%d_%d_%d; ",
 				db.GetFullName(), db.GetTenantID(), db.GetID(), 0))
 			continue
 		}
 
 		usages = append(usages, iter.Item())
-		for iter.Next() && iter.Item().DbId == db.ID && iter.Item().AccId == db.GetTenantID() {
+		for iter.Next() && isSameDBFunc(iter.Item(), db) {
 			usages = append(usages, iter.Item())
 		}
 
@@ -489,7 +503,7 @@ func (m *TNUsageMemo) applyDeletes(
 			totalSize += usages[idx].Size
 		}
 
-		buf.WriteString(fmt.Sprintf("[db]%s_%d_%d_%d; ",
+		buf.WriteString(fmt.Sprintf("[d-db]%s_%d_%d_%d; ",
 			db.GetFullName(), db.GetTenantID(), db.GetID(), totalSize))
 
 		usages = usages[:0]
@@ -524,9 +538,9 @@ func (m *TNUsageMemo) replayIntoGCKP(collector *GlobalCollector) {
 	iter.Release()
 }
 
-func try2RemoveStaleData(usage UsageData, c *catalog.Catalog) (UsageData, bool) {
+func try2RemoveStaleData(usage UsageData, c *catalog.Catalog) (UsageData, string, bool) {
 	if c == nil {
-		return usage, false
+		return usage, "", false
 	}
 
 	var err error
@@ -536,39 +550,29 @@ func try2RemoveStaleData(usage UsageData, c *catalog.Catalog) (UsageData, bool)
 	dbEntry, err = c.GetDatabaseByID(usage.DbId)
 	if err != nil || dbEntry.HasDropCommitted() {
 		// the db has been deleted
-		return usage, true
+		name := "deleted"
+		if dbEntry != nil {
+			name = dbEntry.GetName()
+		}
+		log := fmt.Sprintf("[d-db]%s_%d_%d_%d; ", name, usage.AccId, usage.DbId, usage.Size)
+		return usage, log, true
 	}
 
 	tblEntry, err = dbEntry.GetTableEntryByID(usage.TblId)
 	if err != nil || tblEntry.HasDropCommitted() {
 		// the tbl has been deleted
-		return usage, true
-	}
-
-	size := int64(0)
-	processor := new(catalog.LoopProcessor)
-	processor.ObjectFn = func(entry *catalog.ObjectEntry) error {
-		if !entry.IsAppendable() {
-			if entry.HasDropCommitted() {
-				size -= int64(entry.GetCompSize())
-			} else if entry.IsCommitted() {
-				size += int64(entry.GetCompSize())
-			}
+		name := "deleted"
+		if tblEntry != nil {
+			name = tblEntry.GetFullName()
 		}
-		return nil
-	}
-
-	tblEntry.RecurLoop(processor)
-
-	if size < 0 {
-		size = 0
+		log := fmt.Sprintf("[d-tbl]%s_%d_%d_%d_%d; ", name, usage.AccId, usage.DbId, usage.TblId, usage.Size)
+		return usage, log, true
 	}
 
-	usage.Size = uint64(size)
-	return usage, false
+	return usage, "", false
 }
 
-func (m *TNUsageMemo) deleteAccount(accId uint32) (cleaned int) {
+func (m *TNUsageMemo) deleteAccount(accId uint64) {
 	trash := make([]UsageData, 0)
 	povit := UsageData{accId, 0, 0, 0}
 
@@ -578,7 +582,7 @@ func (m *TNUsageMemo) deleteAccount(accId uint32) (cleaned int) {
 
 	if iter.Item().AccId != accId {
 		iter.Release()
-		return 0
+		return
 	}
 
 	trash = append(trash, iter.Item())
@@ -595,19 +599,24 @@ func (m *TNUsageMemo) deleteAccount(accId uint32) (cleaned int) {
 	for idx := range trash {
 		m.Delete(trash[idx])
 	}
-
-	return len(trash)
 }
 
-func (m *TNUsageMemo) ClearDroppedAccounts(reserved map[uint32]struct{}) (cleaned int) {
+func (m *TNUsageMemo) ClearDroppedAccounts(reserved map[uint64]struct{}) string {
+	if reserved == nil {
+		return ""
+	}
+
+	var buf bytes.Buffer
+
 	usages := m.GatherAllAccSize()
 	for accId := range usages {
 		if _, ok := reserved[accId]; !ok {
 			// this account has been deleted
-			cleaned += m.deleteAccount(accId)
+			m.deleteAccount(accId)
+			buf.WriteString(fmt.Sprintf("%d; ", accId))
 		}
 	}
-	return
+	return buf.String()
 }
 
 // EstablishFromCKPs replays usage info which stored in ckps into the tn cache
@@ -615,12 +624,18 @@ func (m *TNUsageMemo) EstablishFromCKPs(c *catalog.Catalog) {
 	m.EnterProcessing()
 	defer m.LeaveProcessing()
 
+	var buf bytes.Buffer
+
 	defer func() {
-		for _, data := range m.pendingReplay.datas {
-			if data != nil {
-				data.Close()
+		for idx := range m.pendingReplay.datas {
+			if m.pendingReplay.datas[idx] != nil {
+				m.pendingReplay.datas[idx].Close()
 			}
+			m.pendingReplay.datas[idx] = nil
 		}
+		logutil.Info("[storage usage] replay:",
+			zap.String("remove old deleted db/tbl", buf.String()),
+			zap.Int("delayed %d tbl", len(m.pendingReplay.delayed)))
 	}()
 
 	for x := range m.pendingReplay.datas {
@@ -634,7 +649,8 @@ func (m *TNUsageMemo) EstablishFromCKPs(c *catalog.Catalog) {
 		accCol, dbCol, tblCol, sizeCol := getStorageUsageVectorCols(insVecs)
 
 		var skip bool
-		for y := 0; y < insVecs[UsageAccID].Length(); y++ {
+		var log string
+		for y := 0; y < len(accCol); y++ {
 			usage := UsageData{accCol[y], dbCol[y], tblCol[y], sizeCol[y]}
 
 			// these ckps, older than version 11, haven't del bat, we need clear the
@@ -643,10 +659,17 @@ func (m *TNUsageMemo) EstablishFromCKPs(c *catalog.Catalog) {
 			// (if a table or db recreate, it's id will change)
 			//
 			if m.pendingReplay.vers[x] < CheckpointVersion11 {
-				usage, skip = try2RemoveStaleData(usage, c)
+				// here only remove the deleted db and table.
+				// if table has deletes, we update it in gckp
+				usage, log, skip = try2RemoveStaleData(usage, c)
 				if skip {
+					buf.WriteString(log)
 					continue
 				}
+				if m.pendingReplay.delayed == nil {
+					m.pendingReplay.delayed = make(map[uint64]UsageData)
+				}
+				m.pendingReplay.delayed[usage.TblId] = usage
 			}
 
 			m.Update(usage, false)
@@ -660,7 +683,7 @@ func (m *TNUsageMemo) EstablishFromCKPs(c *catalog.Catalog) {
 		delVecs := getStorageUsageBatVectors(m.pendingReplay.datas[x].bats[StorageUsageDelIDX])
 		accCol, dbCol, tblCol, sizeCol = getStorageUsageVectorCols(delVecs)
 
-		for y := 0; y < delVecs[UsageAccID].Length(); y++ {
+		for y := 0; y < len(accCol); y++ {
 			usage := UsageData{accCol[y], dbCol[y], tblCol[y], sizeCol[y]}
 			m.Update(usage, true)
 		}
@@ -682,10 +705,10 @@ func getStorageUsageBatVectors(bat *containers.Batch) []*vector.Vector {
 }
 
 func getStorageUsageVectorCols(vecs []*vector.Vector) (
-	accCol []uint32, dbCol []uint64, tblCol []uint64, sizeCol []uint64) {
+	accCol []uint64, dbCol []uint64, tblCol []uint64, sizeCol []uint64) {
 
 	dbCol = vector.MustFixedCol[uint64](vecs[UsageDBID])
-	accCol = vector.MustFixedCol[uint32](vecs[UsageAccID])
+	accCol = vector.MustFixedCol[uint64](vecs[UsageAccID])
 	tblCol = vector.MustFixedCol[uint64](vecs[UsageTblID])
 	sizeCol = vector.MustFixedCol[uint64](vecs[UsageSize])
 
@@ -704,7 +727,7 @@ var lastDelUsage UsageData = zeroUsageData
 func appendToStorageUsageBat(data *CheckpointData, usage UsageData, del bool, mp *mpool.MPool) {
 	appendFunc := func(vecs []*vector.Vector) {
 		vector.AppendFixed[uint64](vecs[UsageSize], usage.Size, false, mp)
-		vector.AppendFixed[uint32](vecs[UsageAccID], usage.AccId, false, mp)
+		vector.AppendFixed[uint64](vecs[UsageAccID], usage.AccId, false, mp)
 		vector.AppendFixed[uint64](vecs[UsageDBID], usage.DbId, false, mp)
 		vector.AppendFixed[uint64](vecs[UsageTblID], usage.TblId, false, mp)
 	}
@@ -749,7 +772,7 @@ func objects2Usages(objs []*catalog.ObjectEntry) (usages []UsageData) {
 			DbId:  obj.GetTable().GetDB().GetID(),
 			Size:  uint64(obj.GetCompSize()),
 			TblId: obj.GetTable().GetID(),
-			AccId: obj.GetTable().GetDB().GetTenantID(),
+			AccId: uint64(obj.GetTable().GetDB().GetTenantID()),
 		}
 	}
 
@@ -769,6 +792,43 @@ func updateStorageUsageMeta(data *CheckpointData, tid uint64, start, end int32,
 	}
 }
 
+func tryUpdateDataFromOldVersion(collector *GlobalCollector) string {
+	memo := collector.UsageMemo
+	if len(memo.pendingReplay.delayed) == 0 {
+		return ""
+	}
+
+	var buf bytes.Buffer
+
+	tblChanges := make(map[uint64]int64)
+
+	usages := objects2Usages(collector.Usage.SegDeletes)
+	for idx := range usages {
+		tblChanges[usages[idx].TblId] -= int64(usages[idx].Size)
+	}
+
+	usages = objects2Usages(collector.Usage.SegInserts)
+	for idx := range usages {
+		tblChanges[usages[idx].TblId] += int64(usages[idx].Size)
+	}
+
+	for id, size := range tblChanges {
+		if size <= 0 {
+			size = 0
+		}
+		if usage, ok := memo.pendingReplay.delayed[id]; ok {
+			memo.Update(usage, true)
+			usage.Size = uint64(size)
+			memo.Update(usage, false)
+			delete(memo.pendingReplay.delayed, id)
+
+			buf.WriteString(fmt.Sprintf("[u-tbl]%d_%d_%d_%d; ",
+				usage.AccId, usage.DbId, usage.TblId, usage.Size))
+		}
+	}
+	return buf.String()
+}
+
 func applyChanges(collector *BaseCollector, tnUsageMemo *TNUsageMemo) string {
 	tnUsageMemo.newAccCache.ClearForUpdate()
 
@@ -787,6 +847,26 @@ func applyChanges(collector *BaseCollector, tnUsageMemo *TNUsageMemo) string {
 	return log
 }
 
+//func applyTransfer(collector *BaseCollector, memo *TNUsageMemo) string {
+//	var buf bytes.Buffer
+//	if len(memo.pendingTransfer.deletes) != 0 {
+//		memo.applySegDeletes(memo.pendingTransfer.deletes, collector.data, collector.Allocator())
+//		for _, usage := range memo.pendingTransfer.deletes {
+//			buf.WriteString(fmt.Sprintf("[td-tbl]%d_%d_%d_%d; ",
+//				usage.AccId, usage.DbId, usage.TblId, usage.Size))
+//		}
+//	}
+//	if len(memo.pendingTransfer.inserts) != 0 {
+//		memo.applySegInserts(memo.pendingTransfer.inserts, collector.data, collector.Allocator())
+//		for _, usage := range memo.pendingTransfer.inserts {
+//			buf.WriteString(fmt.Sprintf("[ti-tbl]%d_%d_%d_%d; ",
+//				usage.AccId, usage.DbId, usage.TblId, usage.Size))
+//		}
+//	}
+//
+//	return buf.String()
+//}
+
 func FillUsageBatOfGlobal(collector *GlobalCollector) {
 	start := time.Now()
 
@@ -796,10 +876,13 @@ func FillUsageBatOfGlobal(collector *GlobalCollector) {
 		v2.TaskGCkpCollectUsageDurationHistogram.Observe(time.Since(start).Seconds())
 	}()
 
-	cleaned := collector.UsageMemo.ClearDroppedAccounts(collector.Usage.ReservedAccIds)
+	log1 := tryUpdateDataFromOldVersion(collector)
+	log2 := collector.UsageMemo.ClearDroppedAccounts(collector.Usage.ReservedAccIds)
 	collector.UsageMemo.replayIntoGCKP(collector)
 
-	logutil.Infof("[storage usage] CKP[G]: cleaned %d accounts", cleaned)
+	logutil.Info("[storage usage] CKP[G]",
+		zap.String("update old data", log1),
+		zap.String("accounts cleaned", log2))
 }
 
 func FillUsageBatOfIncremental(collector *IncrementalCollector) {
@@ -812,11 +895,12 @@ func FillUsageBatOfIncremental(collector *IncrementalCollector) {
 		v2.TaskICkpCollectUsageDurationHistogram.Observe(time.Since(start).Seconds())
 	}()
 
-	log := applyChanges(collector.BaseCollector, collector.UsageMemo)
+	log1 := applyChanges(collector.BaseCollector, collector.UsageMemo)
+	//log2 := applyTransfer(collector.BaseCollector, collector.UsageMemo)
 
 	logutil.Info("[storage usage] CKP[I]",
 		zap.Float64("cache mem used", collector.UsageMemo.MemoryUsed()),
-		zap.String("applied deletes", log))
+		zap.String("applied deletes", log1))
 }
 
 // GetStorageUsageHistory is for debug to show these storage usage changes.
@@ -896,7 +980,7 @@ func GetStorageUsageHistory(
 }
 
 func cnBatchToUsageDatas(bat *batch.Batch) []UsageData {
-	accCol := vector.MustFixedCol[uint32](bat.GetVector(2))
+	accCol := vector.MustFixedCol[uint64](bat.GetVector(2))
 	dbCol := vector.MustFixedCol[uint64](bat.GetVector(3))
 	tblCol := vector.MustFixedCol[uint64](bat.GetVector(4))
 	sizeCol := vector.MustFixedCol[uint64](bat.GetVector(6))
@@ -983,3 +1067,58 @@ func loadStorageUsageBatch(
 	}
 	return bats, nil
 }
+
+func PairAccountVsDB(c *catalog.Catalog) map[uint64]uint64 {
+	pairs := make(map[uint64]uint64)
+
+	processor := new(catalog.LoopProcessor)
+	processor.DatabaseFn = func(entry *catalog.DBEntry) error {
+		pairs[entry.GetID()] = uint64(entry.GetTenantID())
+		return nil
+	}
+
+	c.RecurLoop(processor)
+	return pairs
+}
+
+func CorrectUsageWrongPlacement(c *catalog.Catalog) (int, float64, error) {
+	memo := c.GetUsageMemo().(*TNUsageMemo)
+	if memo == nil {
+		return 0, 0, moerr.NewInternalErrorNoCtx("tn usage cache is nil")
+	}
+
+	var buf bytes.Buffer
+
+	pairs := PairAccountVsDB(c)
+
+	memo.EnterProcessing()
+	defer memo.LeaveProcessing()
+
+	usages := memo.cache.data.Items()
+
+	anyTransferred := int(0)
+	transferredSize := float64(0)
+	for idx := range usages {
+		if pairs[usages[idx].DbId] != usages[idx].AccId {
+			anyTransferred++
+			transferredSize += float64(usages[idx].Size)
+
+			memo.Update(usages[idx], true)
+			buf.WriteString(fmt.Sprintf("[td-tbl]%d_%d_%d_%d; ",
+				usages[idx].AccId, usages[idx].DbId, usages[idx].TblId, usages[idx].Size))
+			//memo.pendingTransfer.deletes = append(memo.pendingTransfer.deletes, usages[idx])
+
+			usages[idx].AccId = pairs[usages[idx].DbId]
+			memo.Update(usages[idx], false)
+			buf.WriteString(fmt.Sprintf("[ti-tbl]%d_%d_%d_%d; ",
+				usages[idx].AccId, usages[idx].DbId, usages[idx].TblId, usages[idx].Size))
+			//memo.pendingTransfer.inserts = append(memo.pendingTransfer.inserts, usages[idx])
+		}
+	}
+
+	transferredSize /= 1024 * 1024
+	logutil.Info("[storage usage] apply transfer: ",
+		zap.String(fmt.Sprintf("transferred %d tbl, %f mb", anyTransferred, transferredSize), buf.String()))
+
+	return anyTransferred, transferredSize, nil
+}
diff --git a/pkg/vm/engine/tae/logtail/utils.go b/pkg/vm/engine/tae/logtail/utils.go
index 6cc03b6d3..b611edaf6 100644
--- a/pkg/vm/engine/tae/logtail/utils.go
+++ b/pkg/vm/engine/tae/logtail/utils.go
@@ -784,7 +784,7 @@ type BaseCollector struct {
 		Deletes        []interface{}
 		SegInserts     []*catalog.ObjectEntry
 		SegDeletes     []*catalog.ObjectEntry
-		ReservedAccIds map[uint32]struct{}
+		ReservedAccIds map[uint64]struct{}
 	}
 
 	UsageMemo *TNUsageMemo
@@ -845,7 +845,7 @@ func NewGlobalCollector(end types.TS, versionInterval time.Duration) *GlobalColl
 	collector.ObjectFn = collector.VisitObj
 	collector.BlockFn = collector.VisitBlk
 
-	collector.Usage.ReservedAccIds = make(map[uint32]struct{})
+	collector.Usage.ReservedAccIds = make(map[uint64]struct{})
 
 	return collector
 }
@@ -2790,7 +2790,7 @@ func (collector *GlobalCollector) VisitDB(entry *catalog.DBEntry) error {
 	}
 
 	currAccId := uint32(entry.GetTenantID())
-	collector.Usage.ReservedAccIds[currAccId] = struct{}{}
+	collector.Usage.ReservedAccIds[uint64(currAccId)] = struct{}{}
 
 	return collector.BaseCollector.VisitDB(entry)
 }
@@ -3083,6 +3083,7 @@ func (collector *BaseCollector) VisitObj(entry *catalog.ObjectEntry) (err error)
 
 func (collector *GlobalCollector) VisitObj(entry *catalog.ObjectEntry) error {
 	if collector.isEntryDeletedBeforeThreshold(entry.BaseEntryImpl) {
+		collector.Usage.SegDeletes = append(collector.Usage.SegDeletes, entry)
 		return nil
 	}
 	if collector.isEntryDeletedBeforeThreshold(entry.GetTable().BaseEntryImpl) {
diff --git a/pkg/vm/engine/tae/rpc/handle.go b/pkg/vm/engine/tae/rpc/handle.go
index 0a699080d..4044e0a3b 100644
--- a/pkg/vm/engine/tae/rpc/handle.go
+++ b/pkg/vm/engine/tae/rpc/handle.go
@@ -1139,7 +1139,8 @@ func traverseCatalogForNewAccounts(c *catalog2.Catalog, memo *logtail.TNUsageMem
 
 		tblIt := entry.MakeTableIt(true)
 		for tblIt.Valid() {
-			insUsage := logtail.UsageData{AccId: accId, DbId: entry.ID, TblId: tblIt.Get().GetPayload().ID}
+			insUsage := logtail.UsageData{
+				AccId: uint64(accId), DbId: entry.ID, TblId: tblIt.Get().GetPayload().ID}
 
 			tblEntry := tblIt.Get().GetPayload()
 			if tblEntry.HasDropCommitted() {
@@ -1191,11 +1192,11 @@ func (h *Handle) HandleStorageUsage(ctx context.Context, meta txn.TxnMeta,
 	newIds := make([]uint32, 0)
 	for _, id := range req.AccIds {
 		if usages != nil {
-			if size, exist := usages[uint32(id)]; exist {
-				memo.AddReqTrace(uint32(id), size, start, "req")
+			if size, exist := usages[uint64(id)]; exist {
+				memo.AddReqTrace(uint64(id), size, start, "req")
 				resp.AccIds = append(resp.AccIds, int32(id))
 				resp.Sizes = append(resp.Sizes, size)
-				delete(usages, uint32(id))
+				delete(usages, uint64(id))
 				continue
 			}
 		}
@@ -1204,7 +1205,7 @@ func (h *Handle) HandleStorageUsage(ctx context.Context, meta txn.TxnMeta,
 	}
 
 	for accId, size := range usages {
-		memo.AddReqTrace(accId, size, start, "oth")
+		memo.AddReqTrace(uint64(accId), size, start, "oth")
 		resp.AccIds = append(resp.AccIds, int32(accId))
 		resp.Sizes = append(resp.Sizes, size)
 	}
@@ -1213,10 +1214,10 @@ func (h *Handle) HandleStorageUsage(ctx context.Context, meta txn.TxnMeta,
 	traverseCatalogForNewAccounts(h.db.Catalog, memo, newIds)
 
 	for idx := range newIds {
-		if size, exist := memo.GatherNewAccountSize(newIds[idx]); exist {
+		if size, exist := memo.GatherNewAccountSize(uint64(newIds[idx])); exist {
 			resp.AccIds = append(resp.AccIds, int32(newIds[idx]))
 			resp.Sizes = append(resp.Sizes, size)
-			memo.AddReqTrace(newIds[idx], size, start, "new")
+			memo.AddReqTrace(uint64(newIds[idx]), size, start, "new")
 		}
 	}
 
diff --git a/pkg/vm/engine/tae/rpc/inspect.go b/pkg/vm/engine/tae/rpc/inspect.go
index 1c78899a1..fa9e0621a 100644
--- a/pkg/vm/engine/tae/rpc/inspect.go
+++ b/pkg/vm/engine/tae/rpc/inspect.go
@@ -194,15 +194,17 @@ func (c *objStatArg) Run() error {
 type storageUsageHistoryArg struct {
 	ctx    *inspectContext
 	detail *struct {
-		accId uint32
+		accId uint64
 		dbI   uint64
 		tblId uint64
 	}
 
 	trace *struct {
 		tStart, tEnd time.Time
-		accounts     map[uint32]struct{}
+		accounts     map[uint64]struct{}
 	}
+
+	transfer bool
 }
 
 func (c *storageUsageHistoryArg) FromCommand(cmd *cobra.Command) (err error) {
@@ -215,7 +217,7 @@ func (c *storageUsageHistoryArg) FromCommand(cmd *cobra.Command) (err error) {
 			return err
 		}
 		c.detail = &struct {
-			accId uint32
+			accId uint64
 			dbI   uint64
 			tblId uint64
 		}{accId: accId, dbI: dbId, tblId: tblId}
@@ -230,10 +232,19 @@ func (c *storageUsageHistoryArg) FromCommand(cmd *cobra.Command) (err error) {
 
 		c.trace = &struct {
 			tStart, tEnd time.Time
-			accounts     map[uint32]struct{}
+			accounts     map[uint64]struct{}
 		}{tStart: start, tEnd: end, accounts: accs}
 	}
 
+	expr, _ = cmd.Flags().GetString("transfer")
+	if expr != "" {
+		if expr == "*" {
+			c.transfer = true
+		} else {
+			return moerr.NewInvalidArgNoCtx(expr, "`storage_usage -f *` expected")
+		}
+	}
+
 	return nil
 }
 
@@ -242,6 +253,8 @@ func (c *storageUsageHistoryArg) Run() (err error) {
 		return storageUsageDetails(c)
 	} else if c.trace != nil {
 		return storageTrace(c)
+	} else if c.transfer {
+		return storageTransfer(c)
 	}
 	return moerr.NewInvalidArgNoCtx("", c.ctx.args)
 }
@@ -639,9 +652,11 @@ func initCommand(ctx context.Context, inspectCtx *inspectContext) *cobra.Command
 		Run:   RunFactory(&storageUsageHistoryArg{}),
 	}
 
+	// storage usage request history
 	storageUsageCmd.Flags().StringP("trace", "t", "", "format: -time time range or -acc account id list")
 	// storage usage details in ckp
 	storageUsageCmd.Flags().StringP("detail", "d", "", "format: accId{.dbName{.tableName}}")
+	storageUsageCmd.Flags().StringP("transfer", "f", "", "format: *")
 	rootCmd.AddCommand(storageUsageCmd)
 
 	return rootCmd
@@ -776,7 +791,7 @@ func (o *objectVisitor) OnTable(table *catalog.TableEntry) error {
 // the history of all
 // mo_ctl("dn", "inspect", "storage_usage -t *");
 func parseStorageUsageDetail(expr string, ac *db.AccessInfo, db *db.DB) (
-	accId uint32, dbId uint64, tblId uint64, err error) {
+	accId uint64, dbId uint64, tblId uint64, err error) {
 	strs := strings.Split(expr, ".")
 
 	if len(strs) == 0 || len(strs) > 3 {
@@ -800,7 +815,7 @@ func parseStorageUsageDetail(expr string, ac *db.AccessInfo, db *db.DB) (
 		return 0, 0, 0, err
 	}
 
-	accId = uint32(id)
+	accId = uint64(id)
 	dbId, tblId = math.MaxUint64, math.MaxUint64
 
 	var dbHdl handle.Database
@@ -849,7 +864,7 @@ func subString(src string, pos1, pos2 int) (string, error) {
 // no limit, show all request trace info
 // select mo_ctl("dn", "inspect", "-t ");
 func parseStorageUsageTrace(expr string, ac *db.AccessInfo, db *db.DB) (
-	tStart, tEnd time.Time, accounts map[uint32]struct{}, err error) {
+	tStart, tEnd time.Time, accounts map[uint64]struct{}, err error) {
 
 	var str string
 	tIdx := strings.Index(expr, "-time")
@@ -891,7 +906,7 @@ func parseStorageUsageTrace(expr string, ac *db.AccessInfo, db *db.DB) (
 		}
 		accs := strings.Split(str, " ")
 
-		accounts = make(map[uint32]struct{})
+		accounts = make(map[uint64]struct{})
 
 		var id int
 		for i := range accs {
@@ -899,7 +914,7 @@ func parseStorageUsageTrace(expr string, ac *db.AccessInfo, db *db.DB) (
 			if err != nil {
 				return
 			}
-			accounts[uint32(id)] = struct{}{}
+			accounts[uint64(id)] = struct{}{}
 		}
 	}
 
@@ -1049,7 +1064,7 @@ func storageUsageDetails(c *storageUsageHistoryArg) (err error) {
 
 func storageTrace(c *storageUsageHistoryArg) (err error) {
 
-	filter := func(accId uint32, stamp time.Time) bool {
+	filter := func(accId uint64, stamp time.Time) bool {
 		if !c.trace.tStart.IsZero() {
 			if stamp.UTC().Add(time.Hour*8).Before(c.trace.tStart) ||
 				stamp.UTC().Add(time.Hour*8).After(c.trace.tEnd) {
@@ -1091,3 +1106,13 @@ func storageTrace(c *storageUsageHistoryArg) (err error) {
 
 	return nil
 }
+
+func storageTransfer(c *storageUsageHistoryArg) (err error) {
+	cnt, size, err := logtail.CorrectUsageWrongPlacement(c.ctx.db.Catalog)
+	if err != nil {
+		return err
+	}
+
+	c.ctx.out.Write([]byte(fmt.Sprintf("transferred %d tbl, %f mb; ", cnt, size)))
+	return
+}


Test patch:
diff --git a/pkg/vm/engine/tae/db/test/storage_usage_test.go b/pkg/vm/engine/tae/db/test/storage_usage_test.go
index 2e7887d4f..a788fb762 100644
--- a/pkg/vm/engine/tae/db/test/storage_usage_test.go
+++ b/pkg/vm/engine/tae/db/test/storage_usage_test.go
@@ -15,21 +15,26 @@
 package test
 
 import (
+	"context"
 	"fmt"
+	"math"
+	"math/rand"
+	"sort"
+	"sync/atomic"
+	"testing"
+	"time"
+	"unsafe"
+
 	pkgcatalog "github.com/matrixorigin/matrixone/pkg/catalog"
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/catalog"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/containers"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/logtail"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/txn/txnbase"
 	"github.com/stretchr/testify/require"
-	"math"
-	"math/rand"
-	"sort"
-	"testing"
-	"time"
-	"unsafe"
 )
 
 func Test_StorageUsageCache(t *testing.T) {
@@ -37,8 +42,10 @@ func Test_StorageUsageCache(t *testing.T) {
 	cache := logtail.NewStorageUsageCache(logtail.WithLazyThreshold(1))
 	require.True(t, cache.IsExpired())
 
+	allocator := atomic.Uint64{}
+
 	accCnt, dbCnt, tblCnt := 10, 10, 10
-	usages := logtail.MockUsageData(accCnt, dbCnt, tblCnt)
+	usages := logtail.MockUsageData(accCnt, dbCnt, tblCnt, &allocator)
 	for idx := range usages {
 		cache.Update(usages[idx])
 	}
@@ -191,8 +198,11 @@ func mockDeletesAndInserts(
 }
 
 func Test_FillUsageBatOfIncremental(t *testing.T) {
+	allocator := atomic.Uint64{}
+	allocator.Store(pkgcatalog.MO_RESERVED_MAX + 1)
+
 	accCnt, dbCnt, tblCnt := 10, 10, 10
-	usages := logtail.MockUsageData(accCnt, dbCnt, tblCnt)
+	usages := logtail.MockUsageData(accCnt, dbCnt, tblCnt, &allocator)
 
 	memo := logtail.NewTNUsageMemo()
 	memo.Clear()
@@ -306,7 +316,7 @@ func Test_FillUsageBatOfIncremental(t *testing.T) {
 		delBat := ckpData.GetBatches()[logtail.StorageUsageDelIDX]
 		//insBat := ckpData.GetBatches()[logtail.StorageUsageInsIDX]
 
-		accCol := vector.MustFixedCol[uint32](delBat.GetVectorByName(pkgcatalog.SystemColAttr_AccID).GetDownstreamVector())
+		accCol := vector.MustFixedCol[uint64](delBat.GetVectorByName(pkgcatalog.SystemColAttr_AccID).GetDownstreamVector())
 		dbCol := vector.MustFixedCol[uint64](delBat.GetVectorByName(catalog.SnapshotAttr_DBID).GetDownstreamVector())
 		tblCol := vector.MustFixedCol[uint64](delBat.GetVectorByName(catalog.SnapshotAttr_TID).GetDownstreamVector())
 		sizeCol := vector.MustFixedCol[uint64](delBat.GetVectorByName(logtail.CheckpointMetaAttr_ObjectSize).GetDownstreamVector())
@@ -323,8 +333,11 @@ func Test_FillUsageBatOfIncremental(t *testing.T) {
 }
 
 func Test_FillUsageBatOfGlobal(t *testing.T) {
+	allocator := atomic.Uint64{}
+	allocator.Store(pkgcatalog.MO_RESERVED_MAX + 1)
+
 	accCnt, dbCnt, tblCnt := 10, 10, 10
-	usages := logtail.MockUsageData(accCnt, dbCnt, tblCnt)
+	usages := logtail.MockUsageData(accCnt, dbCnt, tblCnt, &allocator)
 
 	memo := logtail.NewTNUsageMemo()
 	memo.Clear()
@@ -333,8 +346,6 @@ func Test_FillUsageBatOfGlobal(t *testing.T) {
 	gCollector.UsageMemo = memo
 	defer gCollector.Close()
 
-	gCollector.Usage.ReservedAccIds = make(map[uint32]struct{})
-
 	for idx := range usages {
 		memo.Update(usages[idx], false)
 		gCollector.Usage.ReservedAccIds[usages[idx].AccId] = struct{}{}
@@ -359,7 +370,7 @@ func Test_FillUsageBatOfGlobal(t *testing.T) {
 			return memo.GetCache().LessFunc()(usages[i], usages[j])
 		})
 
-		accCol := vector.MustFixedCol[uint32](insBat.GetVectorByName(pkgcatalog.SystemColAttr_AccID).GetDownstreamVector())
+		accCol := vector.MustFixedCol[uint64](insBat.GetVectorByName(pkgcatalog.SystemColAttr_AccID).GetDownstreamVector())
 		dbCol := vector.MustFixedCol[uint64](insBat.GetVectorByName(catalog.SnapshotAttr_DBID).GetDownstreamVector())
 		tblCol := vector.MustFixedCol[uint64](insBat.GetVectorByName(catalog.SnapshotAttr_TID).GetDownstreamVector())
 		sizeCol := vector.MustFixedCol[uint64](insBat.GetVectorByName(logtail.CheckpointMetaAttr_ObjectSize).GetDownstreamVector())
@@ -373,11 +384,26 @@ func Test_FillUsageBatOfGlobal(t *testing.T) {
 	}
 }
 
+func appendUsageToBatch(bat *containers.Batch, usage logtail.UsageData) {
+	accVec := bat.GetVectorByName(pkgcatalog.SystemColAttr_AccID).GetDownstreamVector()
+	dbVec := bat.GetVectorByName(catalog.SnapshotAttr_DBID).GetDownstreamVector()
+	tblVec := bat.GetVectorByName(catalog.SnapshotAttr_TID).GetDownstreamVector()
+	sizeVec := bat.GetVectorByName(logtail.CheckpointMetaAttr_ObjectSize).GetDownstreamVector()
+
+	vector.AppendFixed(accVec, usage.AccId, false, common.DebugAllocator)
+	vector.AppendFixed(dbVec, usage.DbId, false, common.DebugAllocator)
+	vector.AppendFixed(tblVec, usage.TblId, false, common.DebugAllocator)
+	vector.AppendFixed(sizeVec, usage.Size, false, common.DebugAllocator)
+
+}
+
 func Test_EstablishFromCheckpoints(t *testing.T) {
-	version8Cnt, version9Cnt, version10Cnt := 3, 4, 5
+	version8Cnt, version9Cnt, version11Cnt := 3, 4, 5
+	allocator := atomic.Uint64{}
+	allocator.Store(pkgcatalog.MO_RESERVED_MAX + 1)
 
-	ckps := make([]*logtail.CheckpointData, version8Cnt+version9Cnt+version10Cnt)
-	vers := make([]uint32, version8Cnt+version9Cnt+version10Cnt)
+	ckps := make([]*logtail.CheckpointData, version8Cnt+version9Cnt+version11Cnt)
+	vers := make([]uint32, version8Cnt+version9Cnt+version11Cnt)
 
 	for idx := 0; idx < version8Cnt; idx++ {
 		data := logtail.NewCheckpointDataWithVersion(logtail.CheckpointVersion8, common.DebugAllocator)
@@ -385,51 +411,38 @@ func Test_EstablishFromCheckpoints(t *testing.T) {
 		vers = append(vers, logtail.CheckpointVersion8)
 	}
 
-	append2BatFunc := func(bat *containers.Batch, usage logtail.UsageData) {
-		accVec := bat.GetVectorByName(pkgcatalog.SystemColAttr_AccID).GetDownstreamVector()
-		dbVec := bat.GetVectorByName(catalog.SnapshotAttr_DBID).GetDownstreamVector()
-		tblVec := bat.GetVectorByName(catalog.SnapshotAttr_TID).GetDownstreamVector()
-		sizeVec := bat.GetVectorByName(logtail.CheckpointMetaAttr_ObjectSize).GetDownstreamVector()
-
-		vector.AppendFixed(accVec, usage.AccId, false, common.DebugAllocator)
-		vector.AppendFixed(dbVec, usage.DbId, false, common.DebugAllocator)
-		vector.AppendFixed(tblVec, usage.TblId, false, common.DebugAllocator)
-		vector.AppendFixed(sizeVec, usage.Size, false, common.DebugAllocator)
-
-	}
-
 	var usageIns, usageDel []logtail.UsageData
 
 	for idx := 0; idx < version9Cnt; idx++ {
 		data := logtail.NewCheckpointDataWithVersion(logtail.CheckpointVersion9, common.DebugAllocator)
 		insBat := data.GetBatches()[logtail.StorageUsageInsIDX]
 
-		usages := logtail.MockUsageData(10, 10, 10)
+		usages := logtail.MockUsageData(10, 10, 10, &allocator)
 		usageIns = append(usageIns, usages...)
 
 		for xx := range usages {
-			append2BatFunc(insBat, usages[xx])
+			appendUsageToBatch(insBat, usages[xx])
 		}
 
 		ckps = append(ckps, data)
 		vers = append(vers, logtail.CheckpointVersion9)
 	}
 
-	for idx := 0; idx < version10Cnt; idx++ {
+	for idx := 0; idx < version11Cnt; idx++ {
 		data := logtail.NewCheckpointDataWithVersion(logtail.CheckpointVersion11, common.DebugAllocator)
 		insBat := data.GetBatches()[logtail.StorageUsageInsIDX]
 		delBat := data.GetBatches()[logtail.StorageUsageDelIDX]
 
-		usages := logtail.MockUsageData(10, 10, 10)
+		usages := logtail.MockUsageData(10, 10, 10, &allocator)
 		usageIns = append(usageIns, usages...)
 		for xx := range usages {
-			append2BatFunc(insBat, usages[xx])
+			appendUsageToBatch(insBat, usages[xx])
 		}
 
-		usages = logtail.MockUsageData(10, 10, 10)
+		usages = logtail.MockUsageData(10, 10, 10, &allocator)
 		usageDel = append(usageDel, usages...)
 		for xx := range usages {
-			append2BatFunc(delBat, usages[xx])
+			appendUsageToBatch(delBat, usages[xx])
 		}
 
 		ckps = append(ckps, data)
@@ -457,7 +470,191 @@ func Test_EstablishFromCheckpoints(t *testing.T) {
 	for iter.Next() {
 		usage, exist := memo.Get(iter.Item())
 		require.True(t, exist)
+		//fmt.Println(usage)
+		//fmt.Println(iter.Item())
+		//fmt.Println()
 		require.Equal(t, usage, iter.Item())
 	}
 	iter.Release()
 }
+
+func Test_RemoveStaleAccounts(t *testing.T) {
+	// clear stale accounts happens in global ckp
+	allocator := atomic.Uint64{}
+	allocator.Store(pkgcatalog.MO_RESERVED_MAX + 1)
+
+	accCnt, dbCnt, tblCnt := 10000, 2, 2
+	usages := logtail.MockUsageData(accCnt, dbCnt, tblCnt, &allocator)
+
+	gCollector := logtail.NewGlobalCollector(types.TS{}, time.Second)
+	gCollector.UsageMemo = logtail.NewTNUsageMemo()
+	defer gCollector.Close()
+
+	for idx := range usages {
+		gCollector.UsageMemo.Update(usages[idx], false)
+		if rand.Int()%3 == 0 {
+			// mock the accounts deletion
+			continue
+		}
+		gCollector.Usage.ReservedAccIds[usages[idx].AccId] = struct{}{}
+	}
+
+	logtail.FillUsageBatOfGlobal(gCollector)
+
+	sizes := gCollector.UsageMemo.GatherAllAccSize()
+
+	require.Equal(t, len(gCollector.Usage.ReservedAccIds), len(sizes))
+
+	for accId := range sizes {
+		_, ok := gCollector.Usage.ReservedAccIds[accId]
+		require.True(t, ok)
+	}
+}
+
+func mockCkpDataWithVersion(version uint32, cnt int) (ckpDats []*logtail.CheckpointData, usages [][]logtail.UsageData) {
+	allocator := atomic.Uint64{}
+	allocator.Store(pkgcatalog.MO_RESERVED_MAX + 1)
+
+	for i := 0; i < cnt; i++ {
+		data := logtail.NewCheckpointDataWithVersion(version, common.DebugAllocator)
+
+		usage := logtail.MockUsageData(10, 10, 10, &allocator)
+		for xx := range usage {
+			appendUsageToBatch(data.GetBatches()[logtail.StorageUsageInsIDX], usage[xx])
+		}
+
+		ckpDats = append(ckpDats, data)
+		usages = append(usages, usage)
+	}
+
+	return
+}
+
+func Test_UpdateDataFromOldVersion(t *testing.T) {
+
+	memo := logtail.NewTNUsageMemo()
+	ctlog := catalog.MockCatalog()
+	defer ctlog.Close()
+
+	ctlog.SetUsageMemo(memo)
+
+	ckpDatas, _ := mockCkpDataWithVersion(logtail.CheckpointVersion9, 1)
+
+	// phase 1: all db/tbl have been deleted
+	{
+		memo.PrepareReplay(ckpDatas, []uint32{logtail.CheckpointVersion9})
+		memo.EstablishFromCKPs(ctlog)
+
+		require.Equal(t, 0, len(memo.GetDelayed()))
+		require.Equal(t, 0, memo.CacheLen())
+
+		for idx := range ckpDatas {
+			require.Nil(t, ckpDatas[idx])
+		}
+	}
+
+	createdTbl := make([]logtail.UsageData, 0)
+
+	// phase 2: part of them have been deleted
+	{
+
+		txnMgr := txnbase.NewTxnManager(
+			catalog.MockTxnStoreFactory(ctlog),
+			catalog.MockTxnFactory(ctlog),
+			types.NewMockHLCClock(1))
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+		defer cancel()
+
+		txnMgr.Start(ctx)
+		defer txnMgr.Stop()
+
+		txn, _ := txnMgr.StartTxn(nil)
+
+		ckpDatas, usages := mockCkpDataWithVersion(logtail.CheckpointVersion9, 1)
+
+		for xx := range usages {
+			for yy := range usages[xx] {
+				db, err := ctlog.GetDatabaseByID(usages[xx][yy].DbId)
+				if moerr.IsMoErrCode(err, moerr.OkExpectedEOB) || db == nil {
+					db, err = ctlog.CreateDBEntryWithID(usages[xx][yy].String(), "", "", usages[xx][yy].DbId, txn)
+				}
+
+				require.Nil(t, err)
+				require.NotNil(t, db)
+
+				if rand.Int()%3 == 0 {
+					continue
+				}
+
+				tbl, err := db.CreateTableEntryWithTableId(
+					catalog.MockSchema(1, 1), txn, nil, usages[xx][yy].TblId)
+				require.Nil(t, err)
+				require.NotNil(t, tbl)
+
+				createdTbl = append(createdTbl, usages[xx][yy])
+			}
+		}
+
+		require.Nil(t, txn.Commit(ctx))
+
+		memo.PrepareReplay(ckpDatas, []uint32{logtail.CheckpointVersion9})
+		memo.EstablishFromCKPs(ctlog)
+
+		for idx := range ckpDatas {
+			require.Nil(t, ckpDatas[idx])
+		}
+
+		require.Equal(t, len(createdTbl), len(memo.GetDelayed()))
+
+		sizes := memo.GatherAllAccSize()
+		for idx := range createdTbl {
+			_, ok := sizes[createdTbl[idx].AccId]
+			require.True(t, ok)
+
+			sizes[createdTbl[idx].AccId] -= createdTbl[idx].Size
+		}
+
+		for _, size := range sizes {
+			require.Equal(t, uint64(0), size)
+		}
+	}
+
+	{
+		// test update old data when global ckp
+		gCollector := logtail.NewGlobalCollector(types.TS{}, time.Second)
+		gCollector.UsageMemo = memo
+		defer gCollector.Close()
+
+		for _, usage := range createdTbl {
+			gCollector.Usage.ReservedAccIds[usage.AccId] = struct{}{}
+
+			db, err := ctlog.GetDatabaseByID(usage.DbId)
+			require.Nil(t, err)
+			require.NotNil(t, db)
+
+			tbl, err := db.GetTableEntryByID(usage.TblId)
+			require.Nil(t, err)
+			require.NotNil(t, tbl)
+
+			// double the size
+			obj := catalog.MockObjEntryWithTbl(tbl, usage.Size*2)
+			gCollector.Usage.SegInserts = append(gCollector.Usage.SegInserts, obj)
+		}
+
+		logtail.FillUsageBatOfGlobal(gCollector)
+		sizes := memo.GatherAllAccSize()
+
+		for idx := range createdTbl {
+			_, ok := sizes[createdTbl[idx].AccId]
+			require.True(t, ok)
+
+			sizes[createdTbl[idx].AccId] -= createdTbl[idx].Size * 2
+		}
+
+		for _, size := range sizes {
+			require.Equal(t, uint64(0), size)
+		}
+	}
+
+}