Repository: matrixorigin/matrixone
Pull request #14409 (2024-01-27 12:41:32)

Add GC's config for 1.1
## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [x] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue https://github.com/matrixorigin/matrixone/issues/13342

## What this PR does / why we need it:
Add GC configuration and add the following configuration in tn.toml:
[tn.GCCfg]
// ‘true’ means disabling gc and retaining all data. The default is false.
disable-gc = true
gc-ttl = "1h"
scan-gc-interval = "30m"


Patch:
diff --git a/pkg/tnservice/cfg.go b/pkg/tnservice/cfg.go
index f59e5b15e..a24d4bcc4 100644
--- a/pkg/tnservice/cfg.go
+++ b/pkg/tnservice/cfg.go
@@ -116,6 +116,12 @@ type Config struct {
 		ReservedWALEntryCount uint64        `toml:"reserved-WAL-entry-count"`
 	}
 
+	GCCfg struct {
+		GCTTL          toml.Duration `toml:"gc-ttl"`
+		ScanGCInterval toml.Duration `toml:"scan-gc-interval"`
+		DisableGC      bool          `toml:"disable-gc"`
+	}
+
 	LogtailServer struct {
 		ListenAddress              string        `toml:"listen-address"`
 		ServiceAddress             string        `toml:"service-address"`
@@ -226,6 +232,7 @@ func (c *Config) Validate() error {
 	if c.Ckp.ReservedWALEntryCount == 0 {
 		c.Ckp.ReservedWALEntryCount = defaultReservedWALEntryCount
 	}
+
 	if c.LogtailServer.ListenAddress == "" {
 		c.LogtailServer.ListenAddress = defaultLogtailListenAddress
 	}
diff --git a/pkg/tnservice/factory.go b/pkg/tnservice/factory.go
index c3e407dc0..88b7125d6 100644
--- a/pkg/tnservice/factory.go
+++ b/pkg/tnservice/factory.go
@@ -135,6 +135,12 @@ func (s *store) newTAEStorage(ctx context.Context, shard metadata.TNShard, facto
 		GlobalMinCount:        s.cfg.Ckp.GlobalMinCount,
 		ReservedWALEntryCount: s.cfg.Ckp.ReservedWALEntryCount,
 	}
+
+	gcCfg := &options.GCCfg{
+		GCTTL:          s.cfg.GCCfg.GCTTL.Duration,
+		ScanGCInterval: s.cfg.GCCfg.ScanGCInterval.Duration,
+		DisableGC:      s.cfg.GCCfg.DisableGC,
+	}
 	logtailServerAddr := s.logtailServiceListenAddr()
 	logtailServerCfg := &options.LogtailServerCfg{
 		RpcMaxMessageSize:      int64(s.cfg.LogtailServer.RpcMaxMessageSize),
@@ -157,6 +163,7 @@ func (s *store) newTAEStorage(ctx context.Context, shard metadata.TNShard, facto
 		fs,
 		s.rt,
 		ckpcfg,
+		gcCfg,
 		logtailServerAddr,
 		logtailServerCfg,
 		s.cfg.Txn.IncrementalDedup == "true",
diff --git a/pkg/txn/storage/tae/storage.go b/pkg/txn/storage/tae/storage.go
index 4aaddb849..81440e647 100644
--- a/pkg/txn/storage/tae/storage.go
+++ b/pkg/txn/storage/tae/storage.go
@@ -50,6 +50,7 @@ func NewTAEStorage(
 	fs fileservice.FileService,
 	rt runtime.Runtime,
 	ckpCfg *options.CheckpointCfg,
+	gcCfg *options.GCCfg,
 	logtailServerAddr string,
 	logtailServerCfg *options.LogtailServerCfg,
 	incrementalDedup bool,
@@ -61,6 +62,7 @@ func NewTAEStorage(
 		Lc:               logservicedriver.LogServiceClientFactory(factory),
 		Shard:            shard,
 		CheckpointCfg:    ckpCfg,
+		GCCfg:            gcCfg,
 		LogStoreT:        options.LogstoreLogservice,
 		IncrementalDedup: incrementalDedup,
 		Ctx:              ctx,
diff --git a/pkg/vm/engine/tae/db/gc/delete.go b/pkg/vm/engine/tae/db/gc/delete.go
index 42aa497d3..0aff37fdc 100644
--- a/pkg/vm/engine/tae/db/gc/delete.go
+++ b/pkg/vm/engine/tae/db/gc/delete.go
@@ -16,9 +16,9 @@ package gc
 
 import (
 	"context"
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"sync"
 
-	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/logutil"
 	"github.com/matrixorigin/matrixone/pkg/objectio"
 )
@@ -57,7 +57,7 @@ func (g *GCWorker) resetObjects() {
 	g.objects = make([]string, 0)
 }
 
-func (g *GCWorker) ExecDelete(ctx context.Context, names []string) error {
+func (g *GCWorker) ExecDelete(ctx context.Context, names []string, disableGC bool) error {
 	g.Lock()
 	g.objects = append(g.objects, names...)
 	if len(g.objects) == 0 {
@@ -67,8 +67,11 @@ func (g *GCWorker) ExecDelete(ctx context.Context, names []string) error {
 	}
 	g.Unlock()
 
-	logutil.Infof("[DB GC] files to delete: %v", g.objects)
-	err := g.fs.DelFiles(ctx, g.objects)
+	logutil.Infof("[DB GC] disableGC: %v, files to delete: %v", disableGC, g.objects)
+	var err error
+	if !disableGC {
+		err = g.fs.DelFiles(ctx, g.objects)
+	}
 	g.Lock()
 	defer g.Unlock()
 	if err != nil && !moerr.IsMoErrCode(err, moerr.ErrFileNotFound) {
diff --git a/pkg/vm/engine/tae/db/gc/diskcleaner.go b/pkg/vm/engine/tae/db/gc/diskcleaner.go
index dfbcdb544..9e8fc5e6d 100644
--- a/pkg/vm/engine/tae/db/gc/diskcleaner.go
+++ b/pkg/vm/engine/tae/db/gc/diskcleaner.go
@@ -46,6 +46,9 @@ type DiskCleaner struct {
 	fs  *objectio.ObjectFS
 	ctx context.Context
 
+	// disableGC is used to control whether to delete objects
+	disableGC bool
+
 	// ckpClient is used to get the instance of the specified checkpoint
 	ckpClient checkpoint.RunnerReader
 
@@ -101,12 +104,14 @@ func NewDiskCleaner(
 	fs *objectio.ObjectFS,
 	ckpClient checkpoint.RunnerReader,
 	catalog *catalog.Catalog,
+	disableGC bool,
 ) *DiskCleaner {
 	cleaner := &DiskCleaner{
 		ctx:       ctx,
 		fs:        fs,
 		ckpClient: ckpClient,
 		catalog:   catalog,
+		disableGC: disableGC,
 	}
 	cleaner.delWorker = NewGCWorker(fs, cleaner)
 	cleaner.processQueue = sm.NewSafeQueue(10000, 1000, cleaner.process)
@@ -351,7 +356,7 @@ func (cleaner *DiskCleaner) tryGC() error {
 	gc := cleaner.softGC()
 	// Delete files after softGC
 	// TODO:Requires Physical Removal Policy
-	err := cleaner.delWorker.ExecDelete(cleaner.ctx, gc)
+	err := cleaner.delWorker.ExecDelete(cleaner.ctx, gc, cleaner.disableGC)
 	if err != nil {
 		return err
 	}
diff --git a/pkg/vm/engine/tae/db/open.go b/pkg/vm/engine/tae/db/open.go
index baf86ee54..2ab9e5b62 100644
--- a/pkg/vm/engine/tae/db/open.go
+++ b/pkg/vm/engine/tae/db/open.go
@@ -175,7 +175,7 @@ func Open(ctx context.Context, dirname string, opts *options.Options) (db *DB, e
 		scanner)
 	db.BGScanner.Start()
 	// TODO: WithGCInterval requires configuration parameters
-	db.DiskCleaner = gc2.NewDiskCleaner(opts.Ctx, fs, db.BGCheckpointRunner, db.Catalog)
+	db.DiskCleaner = gc2.NewDiskCleaner(opts.Ctx, fs, db.BGCheckpointRunner, db.Catalog, opts.GCCfg.DisableGC)
 	db.DiskCleaner.Start()
 	db.DiskCleaner.AddChecker(
 		func(item any) bool {
diff --git a/pkg/vm/engine/tae/options/cfg.go b/pkg/vm/engine/tae/options/cfg.go
index adfa9d310..13eadf0d2 100644
--- a/pkg/vm/engine/tae/options/cfg.go
+++ b/pkg/vm/engine/tae/options/cfg.go
@@ -51,8 +51,9 @@ type CheckpointCfg struct {
 }
 
 type GCCfg struct {
-	GCTTL          time.Duration
-	ScanGCInterval time.Duration
+	GCTTL          time.Duration `toml:"gc-ttl"`
+	ScanGCInterval time.Duration `toml:"scan-gc-interval"`
+	DisableGC      bool          `toml:"disable-gc"`
 }
 
 type CatalogCfg struct {
diff --git a/pkg/vm/engine/tae/options/types.go b/pkg/vm/engine/tae/options/types.go
index a15b915bd..01736eb33 100644
--- a/pkg/vm/engine/tae/options/types.go
+++ b/pkg/vm/engine/tae/options/types.go
@@ -65,7 +65,7 @@ type Options struct {
 	StorageCfg    *StorageCfg    `toml:"storage-cfg"`
 	CheckpointCfg *CheckpointCfg `toml:"checkpoint-cfg"`
 	SchedulerCfg  *SchedulerCfg  `toml:"scheduler-cfg"`
-	GCCfg         *GCCfg
+	GCCfg         *GCCfg         `toml:"gc-cfg"`
 	LogtailCfg    *LogtailCfg
 	CatalogCfg    *CatalogCfg
 


Test patch:
diff --git a/pkg/vm/engine/tae/db/test/db_test.go b/pkg/vm/engine/tae/db/test/db_test.go
index ad2223de3..d342274aa 100644
--- a/pkg/vm/engine/tae/db/test/db_test.go
+++ b/pkg/vm/engine/tae/db/test/db_test.go
@@ -6102,7 +6102,7 @@ func TestGCWithCheckpoint(t *testing.T) {
 	opts := config.WithQuickScanAndCKPAndGCOpts(nil)
 	tae := testutil.NewTestEngine(ctx, ModuleName, t, opts)
 	defer tae.Close()
-	manager := gc.NewDiskCleaner(context.Background(), tae.Runtime.Fs, tae.BGCheckpointRunner, tae.Catalog)
+	manager := gc.NewDiskCleaner(context.Background(), tae.Runtime.Fs, tae.BGCheckpointRunner, tae.Catalog, false)
 	manager.Start()
 	defer manager.Stop()
 
@@ -6134,7 +6134,7 @@ func TestGCWithCheckpoint(t *testing.T) {
 		return entries[num-1].GetEnd().Equal(manager.GetMaxConsumed().GetEnd())
 	})
 	assert.True(t, entries[num-1].GetEnd().Equal(manager.GetMaxConsumed().GetEnd()))
-	manager2 := gc.NewDiskCleaner(context.Background(), tae.Runtime.Fs, tae.BGCheckpointRunner, tae.Catalog)
+	manager2 := gc.NewDiskCleaner(context.Background(), tae.Runtime.Fs, tae.BGCheckpointRunner, tae.Catalog, false)
 	manager2.Start()
 	defer manager2.Stop()
 	testutils.WaitExpect(5000, func() bool {
@@ -6156,7 +6156,7 @@ func TestGCDropDB(t *testing.T) {
 	opts := config.WithQuickScanAndCKPAndGCOpts(nil)
 	tae := testutil.NewTestEngine(ctx, ModuleName, t, opts)
 	defer tae.Close()
-	manager := gc.NewDiskCleaner(context.Background(), tae.Runtime.Fs, tae.BGCheckpointRunner, tae.Catalog)
+	manager := gc.NewDiskCleaner(context.Background(), tae.Runtime.Fs, tae.BGCheckpointRunner, tae.Catalog, false)
 	manager.Start()
 	defer manager.Stop()
 	schema := catalog.MockSchemaAll(3, 1)
@@ -6191,7 +6191,7 @@ func TestGCDropDB(t *testing.T) {
 		return entries[num-1].GetEnd().Equal(manager.GetMaxConsumed().GetEnd())
 	})
 	assert.True(t, entries[num-1].GetEnd().Equal(manager.GetMaxConsumed().GetEnd()))
-	manager2 := gc.NewDiskCleaner(context.Background(), tae.Runtime.Fs, tae.BGCheckpointRunner, tae.Catalog)
+	manager2 := gc.NewDiskCleaner(context.Background(), tae.Runtime.Fs, tae.BGCheckpointRunner, tae.Catalog, false)
 	manager2.Start()
 	defer manager2.Stop()
 	testutils.WaitExpect(5000, func() bool {
@@ -6214,7 +6214,7 @@ func TestGCDropTable(t *testing.T) {
 	opts := config.WithQuickScanAndCKPAndGCOpts(nil)
 	tae := testutil.NewTestEngine(ctx, ModuleName, t, opts)
 	defer tae.Close()
-	manager := gc.NewDiskCleaner(context.Background(), tae.Runtime.Fs, tae.BGCheckpointRunner, tae.Catalog)
+	manager := gc.NewDiskCleaner(context.Background(), tae.Runtime.Fs, tae.BGCheckpointRunner, tae.Catalog, false)
 	manager.Start()
 	defer manager.Stop()
 	schema := catalog.MockSchemaAll(3, 1)
@@ -6264,7 +6264,7 @@ func TestGCDropTable(t *testing.T) {
 		return entries[num-1].GetEnd().Equal(manager.GetMaxConsumed().GetEnd())
 	})
 	assert.True(t, entries[num-1].GetEnd().Equal(manager.GetMaxConsumed().GetEnd()))
-	manager2 := gc.NewDiskCleaner(context.Background(), tae.Runtime.Fs, tae.BGCheckpointRunner, tae.Catalog)
+	manager2 := gc.NewDiskCleaner(context.Background(), tae.Runtime.Fs, tae.BGCheckpointRunner, tae.Catalog, false)
 	manager2.Start()
 	defer manager2.Stop()
 	testutils.WaitExpect(5000, func() bool {