Online check GC correctness

pkg/tnservice/cfg.go

@@ -124,6 +124,7 @@ type Config struct {
 		GCTTL          toml.Duration `toml:"gc-ttl"`
 		ScanGCInterval toml.Duration `toml:"scan-gc-interval"`
 		DisableGC      bool          `toml:"disable-gc"`
+		CheckGC        bool          `toml:"check-gc"`
 	}
 
 	Merge struct {

pkg/tnservice/factory.go

@@ -150,6 +150,7 @@ func (s *store) newTAEStorage(ctx context.Context, shard metadata.TNShard, facto
 		GCTTL:          s.cfg.GCCfg.GCTTL.Duration,
 		ScanGCInterval: s.cfg.GCCfg.ScanGCInterval.Duration,
 		DisableGC:      s.cfg.GCCfg.DisableGC,
+		CheckGC:        s.cfg.GCCfg.CheckGC,
 	}
 
 	mergeCfg := &options.MergeConfig{

pkg/vm/engine/tae/db/checkpoint/info.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/logutil"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/catalog"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/containers"
 	"time"
 )
@@ -32,6 +33,7 @@ type RunnerReader interface {
 	MaxGlobalCheckpoint() *CheckpointEntry
 	GetStage() types.TS
 	MaxLSN() uint64
+	GetCatalog() *catalog.Catalog
 }
 
 func (r *runner) collectCheckpointMetadata(start, end types.TS, ckpLSN, truncateLSN uint64) *containers.Batch {
@@ -102,6 +104,10 @@ func (r *runner) MaxCheckpoint() *CheckpointEntry {
 	return entry
 }
 
+func (r *runner) GetCatalog() *catalog.Catalog {
+	return r.catalog
+}
+
 func (r *runner) ICKPSeekLT(ts types.TS, cnt int) []*CheckpointEntry {
 	r.storage.Lock()
 	tree := r.storage.entries.Copy()

pkg/vm/engine/tae/db/gc/check.go

@@ -0,0 +1,214 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package gc
+
+import (
+	catalog2 "github.com/matrixorigin/matrixone/pkg/catalog"
+	"github.com/matrixorigin/matrixone/pkg/common/mpool"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/logutil"
+	"github.com/matrixorigin/matrixone/pkg/objectio"
+	"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/db/checkpoint"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/logtail"
+	"time"
+)
+
+const NotFoundLimit = 10
+
+type checker struct {
+	cleaner *checkpointCleaner
+}
+
+func (c *checker) getObjects() (map[string]struct{}, error) {
+	dirs, err := c.cleaner.fs.ListDir("")
+	if err != nil {
+		return nil, err
+	}
+	objects := make(map[string]struct{})
+	for _, entry := range dirs {
+		if entry.IsDir {
+			continue
+		}
+		objects[entry.Name] = struct{}{}
+	}
+	return objects, nil
+}
+
+func (c *checker) Check() error {
+	now := time.Now()
+	c.cleaner.inputs.RLock()
+	defer c.cleaner.inputs.RUnlock()
+	gcTables := c.cleaner.GetGCTables()
+
+	// Collect the objects and tombstones that the disk cleaner has already consumed
+	gcTable := NewGCTable()
+	for _, table := range gcTables {
+		gcTable.Merge(table)
+	}
+	objects := gcTable.objects
+	tombstones := gcTable.tombstones
+	entry := c.cleaner.GetMaxConsumed()
+	maxTs := entry.GetEnd()
+
+	// Collect the objects and tombstones that the disk cleaner has not consumed
+	checkpoints := c.cleaner.ckpClient.ICKPSeekLT(entry.GetEnd(), 40)
+	unconsumedTable := NewGCTable()
+	for _, ckp := range checkpoints {
+		_, data, err := logtail.LoadCheckpointEntriesFromKey(c.cleaner.ctx, c.cleaner.fs.Service,
+			ckp.GetLocation(), ckp.GetVersion(), nil, &types.TS{})
+		if err != nil {
+			logutil.Errorf("load checkpoint failed: %v", err)
+			continue
+		}
+		defer data.Close()
+		unconsumedTable.UpdateTable(data)
+		end := ckp.GetEnd()
+		if end.Greater(&maxTs) {
+			maxTs = ckp.GetEnd()
+		}
+	}
+	unconsumedObjects := unconsumedTable.objects
+	unconsumedTombstones := unconsumedTable.tombstones
+
+	// Collect all objects
+	allObjects, err := c.getObjects()
+	if err != nil {
+		return err
+	}
+
+	// Collect all checkpoint files
+	ckpfiles, _, err := checkpoint.ListSnapshotMeta(c.cleaner.ctx, c.cleaner.fs.Service, entry.GetStart(), nil)
+	if err != nil {
+		return err
+	}
+	// The number of checkpoint files is ckpObjectCount
+	ckpObjectCount := len(ckpfiles) * 2
+	allCount := len(allObjects)
+	for name := range allObjects {
+		isfound := false
+		if _, ok := objects[name]; ok {
+			isfound = true
+			delete(objects, name)
+		}
+		if _, ok := tombstones[name]; ok {
+			isfound = true
+			delete(tombstones, name)
+		}
+		if _, ok := unconsumedObjects[name]; ok {
+			isfound = true
+			delete(unconsumedObjects, name)
+		}
+		if _, ok := unconsumedTombstones[name]; ok {
+			isfound = true
+			delete(unconsumedTombstones, name)
+		}
+		if isfound {
+			delete(allObjects, name)
+		}
+	}
+
+	// Collect all objects in memory
+	catalog := c.cleaner.ckpClient.GetCatalog()
+	it := catalog.MakeDBIt(true)
+	bat := makeRespBatchFromSchema(logtail.BlkMetaSchema, common.DebugAllocator)
+	defer bat.Close()
+	end := types.BuildTS(time.Now().UnixNano(), 0)
+	for ; it.Valid(); it.Next() {
+		db := it.Get().GetPayload()
+		itTable := db.MakeTableIt(true)
+		for itTable.Valid() {
+			table := itTable.Get().GetPayload()
+			itObject := table.MakeObjectIt(true)
+			for itObject.Valid() {
+				objectEntry := itObject.Get().GetPayload()
+				stats := objectEntry.GetObjectStats()
+				delete(allObjects, stats.ObjectName().String())
+				itObject.Next()
+			}
+			it2 := table.GetDeleteList().Items()
+			for _, itt := range it2 {
+				_, _, _, err = itt.VisitDeletes(c.cleaner.ctx, maxTs, end, bat, nil, true)
+				if err != nil {
+					logutil.Errorf("visit deletes failed: %v", err)
+					continue
+				}
+			}
+			itTable.Next()
+		}
+	}
+	for i := 0; i < bat.Length(); i++ {
+		deltaLoc := objectio.Location(bat.GetVectorByName(catalog2.BlockMeta_DeltaLoc).Get(i).([]byte))
+		delete(allObjects, deltaLoc.Name().String())
+	}
+
+	if len(objects) != 0 || len(tombstones) != 0 || len(unconsumedObjects) != 0 || len(unconsumedTombstones) != 0 {
+		for name := range objects {
+			logutil.Errorf("[Check GC]lost object %s,", name)
+		}
+
+		for name := range tombstones {
+			logutil.Errorf("[Check GC]lost tombstone %s,", name)
+		}
+
+		for name := range unconsumedObjects {
+			logutil.Errorf("[Check GC]lost unconsumed object %s,", name)
+		}
+
+		for name := range unconsumedTombstones {
+			logutil.Errorf("[Check GC]lost unconsumed tombstone %s,", name)
+		}
+	}
+
+	if len(allObjects) > ckpObjectCount+NotFoundLimit {
+		for name := range allObjects {
+			logutil.Infof("[Check GC]not found object %s,", name)
+		}
+		logutil.Warnf("[Check GC]GC abnormal!!! const: %v, all objects: %d, not found: %d, checkpoint file: %d",
+			time.Since(now), allCount, len(allObjects)-ckpObjectCount, ckpObjectCount)
+	} else {
+		logutil.Infof("[Check GC]Check end!!! const: %v, all objects: %d, not found: %d",
+			time.Since(now), allCount, len(allObjects)-ckpObjectCount)
+	}
+	return nil
+}
+
+func makeRespBatchFromSchema(schema *catalog.Schema, mp *mpool.MPool) *containers.Batch {
+	bat := containers.NewBatch()
+
+	bat.AddVector(
+		catalog.AttrRowID,
+		containers.MakeVector(types.T_Rowid.ToType(), mp),
+	)
+	bat.AddVector(
+		catalog.AttrCommitTs,
+		containers.MakeVector(types.T_TS.ToType(), mp),
+	)
+	// Types() is not used, then empty schema can also be handled here
+	typs := schema.AllTypes()
+	attrs := schema.AllNames()
+	for i, attr := range attrs {
+		if attr == catalog.PhyAddrColumnName {
+			continue
+		}
+		bat.AddVector(
+			attr,
+			containers.MakeVector(typs[i], mp),
+		)
+	}
+	return bat
+}