Merge branch 'main' into memorycache

Author: mergify[bot]

pkg/backup/backup_test.go

@@ -127,7 +127,8 @@ func TestBackupData(t *testing.T) {
 	locations = append(locations, backupTime.Format(time.DateTime))
 	location, err := db.ForceCheckpointForBackup(ctx, currTs, 20*time.Second)
 	assert.Nil(t, err)
-	db.BGCheckpointRunner.DisableCheckpoint()
+	err = db.BGCheckpointRunner.DisableCheckpoint(ctx)
+	assert.NoError(t, err)
 	locations = append(locations, location)
 	checkpoints := db.BGCheckpointRunner.GetAllCheckpoints()
 	files := make(map[string]string, 0)
@@ -232,7 +233,8 @@ func TestBackupData2(t *testing.T) {
 	locations = append(locations, backupTime.Format(time.DateTime))
 	location, err := db.ForceCheckpointForBackup(ctx, currTs, 20*time.Second)
 	assert.Nil(t, err)
-	db.BGCheckpointRunner.DisableCheckpoint()
+	err = db.BGCheckpointRunner.DisableCheckpoint(ctx)
+	assert.NoError(t, err)
 	locations = append(locations, location)
 	compacted := db.BGCheckpointRunner.GetCompacted()
 	checkpoints := db.BGCheckpointRunner.GetAllCheckpointsForBackup(compacted)
@@ -302,7 +304,8 @@ func TestBackupData3(t *testing.T) {
 	locations = append(locations, backupTime.Format(time.DateTime))
 	location, err := db.ForceCheckpointForBackup(ctx, currTs, 20*time.Second)
 	assert.Nil(t, err)
-	db.BGCheckpointRunner.DisableCheckpoint()
+	err = db.BGCheckpointRunner.DisableCheckpoint(ctx)
+	assert.NoError(t, err)
 	locations = append(locations, location)
 	compacted := db.BGCheckpointRunner.GetCompacted()
 	checkpoints := db.BGCheckpointRunner.GetAllCheckpointsForBackup(compacted)
@@ -382,7 +385,8 @@ func TestBackupData4(t *testing.T) {
 	locations = append(locations, backupTime.Format(time.DateTime))
 	location, err := db.ForceCheckpointForBackup(ctx, currTs, 20*time.Second)
 	assert.Nil(t, err)
-	db.BGCheckpointRunner.DisableCheckpoint()
+	err = db.BGCheckpointRunner.DisableCheckpoint(ctx)
+	assert.NoError(t, err)
 	locations = append(locations, location)
 	compacted := db.BGCheckpointRunner.GetCompacted()
 	checkpoints := db.BGCheckpointRunner.GetAllCheckpointsForBackup(compacted)

pkg/backup/tae.go

@@ -19,7 +19,6 @@ import (
 	"crypto/sha256"
 	"encoding/json"
 	"fmt"
-	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/db/gc/v3"
 	"io"
 	"os"
 	"path"
@@ -30,6 +29,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/matrixorigin/matrixone/pkg/objectio/ioutil"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/db/gc/v3"
+
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/runtime"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
@@ -39,7 +41,6 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/objectio"
 	"github.com/matrixorigin/matrixone/pkg/sql/plan/function/ctl"
 	"github.com/matrixorigin/matrixone/pkg/util/executor"
-	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/blockio"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/db/checkpoint"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/logtail"
@@ -305,7 +306,7 @@ func execBackup(
 		if err != nil {
 			return err
 		}
-		key, err := blockio.EncodeLocationFromString(metaLoc)
+		key, err := objectio.StringToLocation(metaLoc)
 		if err != nil {
 			return err
 		}
@@ -372,7 +373,7 @@ func execBackup(
 	taeFileList = append(taeFileList, sizeList...)
 	now = time.Now()
 	if trimInfo != "" {
-		cnLocation, err := blockio.EncodeLocationFromString(cnLoc)
+		cnLocation, err := objectio.StringToLocation(cnLoc)
 		if err != nil {
 			return err
 		}
@@ -426,32 +427,34 @@ func execBackup(
 
 // CopyCheckpointDir copy checkpoint dir from srcFs to dstFs
 // return taeFile list
-// copy: if copy is true,it means not to check the suffix name and copy all files.
+// doCopy: if doCopy is true,it means not to check the suffix name and doCopy all files.
 func copyFileAndGetMetaFiles(
 	ctx context.Context,
 	srcFs, dstFs fileservice.FileService,
 	dir string,
 	backup types.TS,
-	decodeFunc func(string) (types.TS, types.TS, string),
-	copy bool,
-) ([]*taeFile, []*checkpoint.MetaFile, []fileservice.DirEntry, error) {
+	decoder func(string) ioutil.TSRangeFile,
+	doCopy bool,
+) ([]*taeFile, []ioutil.TSRangeFile, []fileservice.DirEntry, error) {
 	files, err := fileservice.SortedList(srcFs.List(ctx, dir))
 	if err != nil {
 		return nil, nil, nil, err
 	}
 	taeFileList := make([]*taeFile, 0, len(files))
-	metaFiles := make([]*checkpoint.MetaFile, 0)
+	metaFiles := make([]ioutil.TSRangeFile, 0)
 	var checksum []byte
 	for i, file := range files {
 		if file.IsDir {
 			panic("not support dir")
 		}
-		start, end, ext := decodeFunc(file.Name)
-		if !backup.IsEmpty() && start.GE(&backup) {
+		meta := decoder(file.Name)
+		meta.SetIdx(i)
+
+		if !backup.IsEmpty() && meta.GetStart().GE(&backup) {
 			logutil.Infof("[Backup] skip file %v", file.Name)
 			continue
 		}
-		if copy || ext == blockio.AcctExt || ext == blockio.SnapshotExt {
+		if doCopy || meta.IsAcctExt() || meta.IsSnapshotExt() {
 			checksum, err = CopyFileWithRetry(ctx, srcFs, dstFs, file.Name, dir)
 			if err != nil {
 				return nil, nil, nil, err
@@ -465,19 +468,16 @@ func copyFileAndGetMetaFiles(
 			})
 		}
 
-		if copy || ext == blockio.CheckpointExt || ext == blockio.GCFullExt {
-			metaFile := checkpoint.NewMetaFile(i, start, end, file.Name)
-			metaFiles = append(metaFiles, metaFile)
+		if doCopy || meta.IsCKPFile() || meta.IsFullGCExt() {
+			metaFiles = append(metaFiles, meta)
 		}
 	}
 
 	if len(metaFiles) == 0 {
 		return taeFileList, metaFiles, files, nil
 	}
 	sort.Slice(metaFiles, func(i, j int) bool {
-		end1 := metaFiles[i].GetEnd()
-		end2 := metaFiles[j].GetEnd()
-		return end1.LT(&end2)
+		return metaFiles[i].GetEnd().LT(metaFiles[j].GetEnd())
 	})
 
 	return taeFileList, metaFiles, files, nil
@@ -492,17 +492,17 @@ func CopyGCDir(
 	var checksum []byte
 
 	taeFileList, metaFiles, files, err := copyFileAndGetMetaFiles(
-		ctx, srcFs, dstFs, dir, backup, blockio.DecodeGCMetadataFileName, false)
+		ctx, srcFs, dstFs, dir, backup, ioutil.DecodeGCMetadataName, false,
+	)
 	if err != nil {
 		return nil, err
 	}
 
-	copyFiles := make([]*checkpoint.MetaFile, 0)
+	copyFiles := make([]ioutil.TSRangeFile, 0)
 
 	for _, metaFile := range metaFiles {
-		name := metaFile.GetName()
 		window := gc.NewGCWindow(common.DebugAllocator, srcFs)
-		err = window.ReadTable(ctx, gc.GCMetaDir+name, srcFs)
+		err = window.ReadTable(ctx, metaFile.GetGCFullName(), srcFs)
 		if err != nil {
 			return nil, err
 		}
@@ -519,7 +519,7 @@ func CopyGCDir(
 			}
 			filesList = append(filesList, &taeFile{
 				path:     object.ObjectName().String(),
-				size:     files[metaFile.GetIndex()].Size,
+				size:     files[metaFile.GetIdx()].Size,
 				checksum: checksum,
 				needCopy: true,
 				ts:       backup,
@@ -534,13 +534,12 @@ func CopyGCDir(
 	for i, metaFile := range copyFiles {
 		name := metaFile.GetName()
 		if i == len(metaFiles)-1 {
-			end := metaFile.GetEnd()
-			if !min.IsEmpty() && end.LT(&min) {
+			if !min.IsEmpty() && metaFile.GetEnd().LT(&min) {
 				// It means that the gc consumption is too slow, and the gc water level needs to be raised.
 				// Otherwise, the gc will not work after the cluster is restored because it cannot find the checkpoint.
 				// The gc water level is determined by the name of the meta,
 				// so the name of the last gc meta needs to be modified.
-				name = blockio.UpdateGCMetadataFileName(name, end, min)
+				name = ioutil.InheritGCMetadataName(name, metaFile.GetEnd(), &min)
 			}
 		}
 		checksum, err = CopyFileWithRetry(ctx, srcFs, dstFs, metaFile.GetName(), dir, name)
@@ -549,7 +548,7 @@ func CopyGCDir(
 		}
 		taeFileList = append(taeFileList, &taeFile{
 			path:     dir + string(os.PathSeparator) + name,
-			size:     files[metaFile.GetIndex()].Size,
+			size:     files[metaFile.GetIdx()].Size,
 			checksum: checksum,
 			needCopy: true,
 			ts:       backup,
@@ -563,11 +562,14 @@ func CopyCheckpointDir(
 	srcFs, dstFs fileservice.FileService,
 	dir string, backup types.TS,
 ) ([]*taeFile, types.TS, error) {
-	decodeFunc := func(name string) (types.TS, types.TS, string) {
-		start, end, _ := blockio.DecodeCheckpointMetadataFileName(name)
-		return start, end, ""
+	decoder := func(name string) ioutil.TSRangeFile {
+		meta := ioutil.DecodeCKPMetaName(name)
+		meta.SetExt("")
+		return meta
 	}
-	taeFileList, metaFiles, _, err := copyFileAndGetMetaFiles(ctx, srcFs, dstFs, dir, backup, decodeFunc, true)
+	taeFileList, metaFiles, _, err := copyFileAndGetMetaFiles(
+		ctx, srcFs, dstFs, dir, backup, decoder, true,
+	)
 	if err != nil {
 		return nil, types.TS{}, err
 	}
@@ -577,7 +579,7 @@ func CopyCheckpointDir(
 	for i := len(metaFiles) - 1; i >= 0; i-- {
 		ckpStart := metaFiles[i].GetStart()
 		if ckpStart.IsEmpty() {
-			minTs = metaFiles[i].GetEnd()
+			minTs = *metaFiles[i].GetEnd()
 			break
 		}
 	}

pkg/bootstrap/custom_upgrade.go

@@ -29,6 +29,7 @@ import (
 )
 
 func (s *service) UpgradeTenant(ctx context.Context, tenantName string, retryCount uint32, isALLAccount bool) (bool, error) {
+	s.logger.Info("start manual bootstrap upgrade")
 	// Before manually upgrade, check if there are unready upgrade tasks in system upgrade environment
 	err := s.UpgradePreCheck(ctx)
 	if err != nil {