Repository: matrixorigin/matrixone
Pull request #6084 (2022-10-29 06:51:28)

vm/process: add Process.GetClusterDetails and initialize it in cnservice
## 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 #

## What this PR does / why we need it:



Patch:
diff --git a/pkg/cnservice/distributed_tae.go b/pkg/cnservice/distributed_tae.go
index f43f910fc..b7ffd6f36 100644
--- a/pkg/cnservice/distributed_tae.go
+++ b/pkg/cnservice/distributed_tae.go
@@ -20,7 +20,6 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/config"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/disttae"
-	"github.com/matrixorigin/matrixone/pkg/vm/engine/memoryengine"
 )
 
 func (s *service) initDistributedTAE(
@@ -54,10 +53,7 @@ func (s *service) initDistributedTAE(
 		s.fileService,
 		client,
 		hakeeper,
-		memoryengine.GetClusterDetailsFromHAKeeper(
-			ctx,
-			hakeeper,
-		),
+		pu.GetClusterDetails,
 	)
 
 	return nil
diff --git a/pkg/cnservice/memory_engine.go b/pkg/cnservice/memory_engine.go
index 75886f363..76204feb8 100644
--- a/pkg/cnservice/memory_engine.go
+++ b/pkg/cnservice/memory_engine.go
@@ -54,7 +54,7 @@ func (s *service) initMemoryEngine(
 	pu.StorageEngine = memoryengine.New(
 		ctx,
 		memoryengine.NewDefaultShardPolicy(mp),
-		memoryengine.GetClusterDetailsFromHAKeeper(ctx, hakeeper),
+		pu.GetClusterDetails,
 		memoryengine.NewHakeeperIDGenerator(hakeeper),
 	)
 
diff --git a/pkg/cnservice/options.go b/pkg/cnservice/options.go
index ddede5d65..49860905d 100644
--- a/pkg/cnservice/options.go
+++ b/pkg/cnservice/options.go
@@ -49,7 +49,8 @@ func WithMessageHandle(f func(ctx context.Context,
 	engine engine.Engine,
 	fs fileservice.FileService,
 	cli client.TxnClient,
-	mAcquirer func() morpc.Message) error) Option {
+	mAcquirer func() morpc.Message,
+	getClusterDetails engine.GetClusterDetailsFunc) error) Option {
 	return func(s *service) {
 		s.requestHandler = f
 	}
diff --git a/pkg/cnservice/server.go b/pkg/cnservice/server.go
index cb1197be8..0146b71f3 100644
--- a/pkg/cnservice/server.go
+++ b/pkg/cnservice/server.go
@@ -35,6 +35,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/txn/clock"
 	"github.com/matrixorigin/matrixone/pkg/txn/rpc"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/memoryengine"
 )
 
 func NewService(
@@ -78,12 +79,24 @@ func NewService(
 		},
 	}
 
-	pu := config.NewParameterUnit(&cfg.Frontend, nil, nil, nil)
+	hakeeper, err := srv.getHAKeeperClient()
+	if err != nil {
+		return nil, err
+	}
+
+	pu := config.NewParameterUnit(
+		&cfg.Frontend,
+		nil,
+		nil,
+		nil,
+		memoryengine.GetClusterDetailsFromHAKeeper(ctx, hakeeper),
+	)
 	cfg.Frontend.SetDefaultValues()
 	frontend.InitServerVersion(pu.SV.MoVersion)
 	if err = srv.initMOServer(ctx, pu); err != nil {
 		return nil, err
 	}
+	srv.pu = pu
 
 	server, err := morpc.NewRPCServer("cn-server", cfg.ListenAddress,
 		morpc.NewMessageCodec(srv.acquireMessage),
@@ -103,7 +116,7 @@ func NewService(
 	srv.storeEngine = pu.StorageEngine
 	srv._txnClient = pu.TxnClient
 
-	srv.requestHandler = func(ctx context.Context, message morpc.Message, cs morpc.ClientSession, engine engine.Engine, fService fileservice.FileService, cli client.TxnClient, messageAcquirer func() morpc.Message) error {
+	srv.requestHandler = func(ctx context.Context, message morpc.Message, cs morpc.ClientSession, engine engine.Engine, fService fileservice.FileService, cli client.TxnClient, messageAcquirer func() morpc.Message, getClusterDetails engine.GetClusterDetailsFunc) error {
 		return nil
 	}
 	for _, opt := range options {
@@ -161,7 +174,7 @@ func (s *service) releaseMessage(m *pipeline.Message) {
 }
 
 func (s *service) handleRequest(ctx context.Context, req morpc.Message, _ uint64, cs morpc.ClientSession) error {
-	go s.requestHandler(ctx, req, cs, s.storeEngine, s.fileService, s._txnClient, s.acquireMessage)
+	go s.requestHandler(ctx, req, cs, s.storeEngine, s.fileService, s._txnClient, s.acquireMessage, s.pu.GetClusterDetails)
 	return nil
 }
 
diff --git a/pkg/cnservice/server_task.go b/pkg/cnservice/server_task.go
index 342d31e28..525496c73 100644
--- a/pkg/cnservice/server_task.go
+++ b/pkg/cnservice/server_task.go
@@ -183,7 +183,13 @@ func (s *service) registerExecutors() {
 		return
 	}
 
-	pu := config.NewParameterUnit(&s.cfg.Frontend, nil, nil, nil)
+	pu := config.NewParameterUnit(
+		&s.cfg.Frontend,
+		nil,
+		nil,
+		nil,
+		s.pu.GetClusterDetails,
+	)
 	pu.StorageEngine = s.storeEngine
 	pu.TxnClient = s._txnClient
 	s.cfg.Frontend.SetDefaultValues()
diff --git a/pkg/cnservice/types.go b/pkg/cnservice/types.go
index 1b151312a..294d9d198 100644
--- a/pkg/cnservice/types.go
+++ b/pkg/cnservice/types.go
@@ -183,7 +183,7 @@ type service struct {
 	logger         *zap.Logger
 	server         morpc.RPCServer
 	requestHandler func(ctx context.Context, message morpc.Message, cs morpc.ClientSession, engine engine.Engine, fService fileservice.FileService, cli client.TxnClient,
-		messageAcquirer func() morpc.Message) error
+		messageAcquirer func() morpc.Message, getClusterDetails engine.GetClusterDetailsFunc) error
 	cancelMoServerFunc     context.CancelFunc
 	mo                     *frontend.MOServer
 	initHakeeperClientOnce sync.Once
@@ -195,6 +195,7 @@ type service struct {
 	storeEngine            engine.Engine
 	metadataFS             fileservice.ReplaceableFileService
 	fileService            fileservice.FileService
+	pu                     *config.ParameterUnit
 
 	stopper *stopper.Stopper
 
diff --git a/pkg/config/configuration.go b/pkg/config/configuration.go
index 871aa4f15..3c490b565 100644
--- a/pkg/config/configuration.go
+++ b/pkg/config/configuration.go
@@ -454,6 +454,9 @@ type ParameterUnit struct {
 
 	// FileService
 	FileService fileservice.FileService
+
+	// GetClusterDetails
+	GetClusterDetails engine.GetClusterDetailsFunc
 }
 
 func NewParameterUnit(
@@ -461,12 +464,14 @@ func NewParameterUnit(
 	storageEngine engine.Engine,
 	txnClient client.TxnClient,
 	clusterNodes engine.Nodes,
+	getClusterDetails engine.GetClusterDetailsFunc,
 ) *ParameterUnit {
 	return &ParameterUnit{
-		SV:            sv,
-		StorageEngine: storageEngine,
-		TxnClient:     txnClient,
-		ClusterNodes:  clusterNodes,
+		SV:                sv,
+		StorageEngine:     storageEngine,
+		TxnClient:         txnClient,
+		ClusterNodes:      clusterNodes,
+		GetClusterDetails: getClusterDetails,
 	}
 }
 
diff --git a/pkg/frontend/mysql_cmd_executor.go b/pkg/frontend/mysql_cmd_executor.go
index 8130db87d..a4e8e45ec 100644
--- a/pkg/frontend/mysql_cmd_executor.go
+++ b/pkg/frontend/mysql_cmd_executor.go
@@ -2372,6 +2372,7 @@ func (mce *MysqlCmdExecutor) doComQuery(requestCtx context.Context, sql string)
 		pu.TxnClient,
 		ses.GetTxnHandler().GetTxnOperator(),
 		pu.FileService,
+		pu.GetClusterDetails,
 	)
 	proc.Id = mce.getNextProcessId()
 	proc.Lim.Size = pu.SV.ProcessLimitationSize
diff --git a/pkg/frontend/util.go b/pkg/frontend/util.go
index 52ad9d476..ff3a77857 100644
--- a/pkg/frontend/util.go
+++ b/pkg/frontend/util.go
@@ -316,7 +316,7 @@ func getParameterUnit(configFile string, eng engine.Engine, txnClient TxnClient)
 		return nil, err
 	}
 	logutil.Info("Using Dump Storage Engine and Cluster Nodes.")
-	pu := mo_config.NewParameterUnit(sv, eng, txnClient, engine.Nodes{})
+	pu := mo_config.NewParameterUnit(sv, eng, txnClient, engine.Nodes{}, nil)
 
 	return pu, nil
 }
diff --git a/pkg/sql/compile/cnserviceDriver.go b/pkg/sql/compile/cnserviceDriver.go
index 1d6cd0e97..01cf9294c 100644
--- a/pkg/sql/compile/cnserviceDriver.go
+++ b/pkg/sql/compile/cnserviceDriver.go
@@ -76,9 +76,10 @@ import (
 
 // messageHandleHelper a structure records some elements to help handle messages.
 type messageHandleHelper struct {
-	storeEngine engine.Engine
-	fileService fileservice.FileService
-	acquirer    func() morpc.Message
+	storeEngine       engine.Engine
+	fileService       fileservice.FileService
+	getClusterDetails engine.GetClusterDetailsFunc
+	acquirer          func() morpc.Message
 }
 
 // processHelper a structure records information about source process. and to help
@@ -98,13 +99,15 @@ type processHelper struct {
 // write back Analysis Information and error info if error occurs to client.
 func CnServerMessageHandler(ctx context.Context, message morpc.Message,
 	cs morpc.ClientSession,
-	storeEngine engine.Engine, fileService fileservice.FileService, cli client.TxnClient, messageAcquirer func() morpc.Message) error {
+	storeEngine engine.Engine, fileService fileservice.FileService, cli client.TxnClient, messageAcquirer func() morpc.Message,
+	getClusterDetails engine.GetClusterDetailsFunc) error {
 	var errData []byte
 	// structure to help handle the message.
 	helper := &messageHandleHelper{
-		storeEngine: storeEngine,
-		fileService: fileService,
-		acquirer:    messageAcquirer,
+		storeEngine:       storeEngine,
+		fileService:       fileService,
+		acquirer:          messageAcquirer,
+		getClusterDetails: getClusterDetails,
 	}
 	// decode message and run it, get final analysis information and err info.
 	analysis, err := pipelineMessageHandle(ctx, message, cs, helper, cli)
@@ -1043,6 +1046,7 @@ func newCompile(ctx context.Context, message morpc.Message, pHelper *processHelp
 		pHelper.txnClient,
 		pHelper.txnOperator,
 		mHelper.fileService,
+		mHelper.getClusterDetails,
 	)
 	proc.UnixTime = pHelper.unixTime
 	proc.Id = pHelper.id
@@ -1060,7 +1064,7 @@ func newCompile(ctx context.Context, message morpc.Message, pHelper *processHelp
 		anal: &anaylze{},
 	}
 
-	c.fill = func(a any, b *batch.Batch) error {
+	c.fill = func(_ any, b *batch.Batch) error {
 		encodeData, errEncode := types.Encode(b)
 		if errEncode != nil {
 			return errEncode
diff --git a/pkg/vm/engine/disttae/engine.go b/pkg/vm/engine/disttae/engine.go
index 7b8a64ffe..07869da02 100644
--- a/pkg/vm/engine/disttae/engine.go
+++ b/pkg/vm/engine/disttae/engine.go
@@ -26,21 +26,18 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/fileservice"
-	"github.com/matrixorigin/matrixone/pkg/pb/logservice"
 	"github.com/matrixorigin/matrixone/pkg/txn/client"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine"
 	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
-type GetClusterDetailsFunc = func() (logservice.ClusterDetails, error)
-
 func New(
 	ctx context.Context,
 	mp *mpool.MPool,
 	fs fileservice.FileService,
 	cli client.TxnClient,
 	idGen IDGenerator,
-	getClusterDetails GetClusterDetailsFunc,
+	getClusterDetails engine.GetClusterDetailsFunc,
 ) *Engine {
 	cluster, err := getClusterDetails()
 	if err != nil {
@@ -223,6 +220,7 @@ func (e *Engine) New(ctx context.Context, op client.TxnOperator) error {
 		e.cli,
 		op,
 		e.fs,
+		e.getClusterDetails,
 	)
 	txn := &Transaction{
 		op:             op,
diff --git a/pkg/vm/engine/disttae/types.go b/pkg/vm/engine/disttae/types.go
index 515efe01e..ecb86c97a 100644
--- a/pkg/vm/engine/disttae/types.go
+++ b/pkg/vm/engine/disttae/types.go
@@ -96,7 +96,7 @@ type Engine struct {
 	db                *DB
 	cli               client.TxnClient
 	idGen             IDGenerator
-	getClusterDetails GetClusterDetailsFunc
+	getClusterDetails engine.GetClusterDetailsFunc
 	txns              map[string]*Transaction
 	// minimum heap of currently active transactions
 	txnHeap *transactionHeap
diff --git a/pkg/vm/engine/memoryengine/engine.go b/pkg/vm/engine/memoryengine/engine.go
index b01e96777..d12de6734 100644
--- a/pkg/vm/engine/memoryengine/engine.go
+++ b/pkg/vm/engine/memoryengine/engine.go
@@ -18,7 +18,6 @@ import (
 	"context"
 	"time"
 
-	logservicepb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 	"github.com/matrixorigin/matrixone/pkg/txn/client"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine"
 )
@@ -26,16 +25,14 @@ import (
 // Engine is an engine.Engine impl
 type Engine struct {
 	shardPolicy       ShardPolicy
-	getClusterDetails GetClusterDetailsFunc
+	getClusterDetails engine.GetClusterDetailsFunc
 	idGenerator       IDGenerator
 }
 
-type GetClusterDetailsFunc = func() (logservicepb.ClusterDetails, error)
-
 func New(
 	ctx context.Context,
 	shardPolicy ShardPolicy,
-	getClusterDetails GetClusterDetailsFunc,
+	getClusterDetails engine.GetClusterDetailsFunc,
 	idGenerator IDGenerator,
 ) *Engine {
 	_ = ctx
diff --git a/pkg/vm/engine/memoryengine/hakeeper.go b/pkg/vm/engine/memoryengine/hakeeper.go
index d0847752a..79889c442 100644
--- a/pkg/vm/engine/memoryengine/hakeeper.go
+++ b/pkg/vm/engine/memoryengine/hakeeper.go
@@ -22,13 +22,14 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/logservice"
 	logservicepb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine"
 )
 
 func GetClusterDetailsFromHAKeeper(
 	ctx context.Context,
 	client logservice.CNHAKeeperClient,
 ) (
-	get GetClusterDetailsFunc,
+	get engine.GetClusterDetailsFunc,
 ) {
 
 	var lock sync.Mutex
diff --git a/pkg/vm/engine/types.go b/pkg/vm/engine/types.go
index bc5f9df38..ade5c5f2d 100644
--- a/pkg/vm/engine/types.go
+++ b/pkg/vm/engine/types.go
@@ -22,6 +22,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/compress"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
+	logservicepb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/txn/client"
 )
@@ -211,3 +212,5 @@ type Engine interface {
 type Hints struct {
 	CommitOrRollbackTimeout time.Duration
 }
+
+type GetClusterDetailsFunc = func() (logservicepb.ClusterDetails, error)
diff --git a/pkg/vm/process/process.go b/pkg/vm/process/process.go
index 608ecd76d..db0287ad6 100644
--- a/pkg/vm/process/process.go
+++ b/pkg/vm/process/process.go
@@ -24,6 +24,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/fileservice"
 	"github.com/matrixorigin/matrixone/pkg/txn/client"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine"
 )
 
 // New creates a new Process.
@@ -34,13 +35,15 @@ func New(
 	txnClient client.TxnClient,
 	txnOperator client.TxnOperator,
 	fileService fileservice.FileService,
+	getClusterDetails engine.GetClusterDetailsFunc,
 ) *Process {
 	return &Process{
-		mp:          m,
-		Ctx:         ctx,
-		TxnClient:   txnClient,
-		TxnOperator: txnOperator,
-		FileService: fileService,
+		mp:                m,
+		Ctx:               ctx,
+		TxnClient:         txnClient,
+		TxnOperator:       txnOperator,
+		FileService:       fileService,
+		GetClusterDetails: getClusterDetails,
 	}
 }
 
diff --git a/pkg/vm/process/types.go b/pkg/vm/process/types.go
index 7fa019411..937255bd0 100644
--- a/pkg/vm/process/types.go
+++ b/pkg/vm/process/types.go
@@ -22,6 +22,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/fileservice"
 	"github.com/matrixorigin/matrixone/pkg/txn/client"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine"
 )
 
 // Analyze analyzes information for operator
@@ -122,6 +123,8 @@ type Process struct {
 
 	FileService fileservice.FileService
 
+	GetClusterDetails engine.GetClusterDetailsFunc
+
 	LoadTag bool
 }
 


Test patch:
diff --git a/pkg/frontend/authenticate_test.go b/pkg/frontend/authenticate_test.go
index ebac744e1..b7709fefd 100644
--- a/pkg/frontend/authenticate_test.go
+++ b/pkg/frontend/authenticate_test.go
@@ -142,7 +142,7 @@ func Test_checkSysExistsOrNot(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		defer ctrl.Finish()
 
-		pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil)
+		pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil, nil)
 		pu.SV.SetDefaultValues()
 
 		ctx := context.WithValue(context.TODO(), config.ParameterUnitKey, pu)
@@ -205,7 +205,7 @@ func Test_createTablesInMoCatalog(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		defer ctrl.Finish()
 
-		pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil)
+		pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil, nil)
 		pu.SV.SetDefaultValues()
 
 		ctx := context.WithValue(context.TODO(), config.ParameterUnitKey, pu)
@@ -239,7 +239,7 @@ func Test_checkTenantExistsOrNot(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		defer ctrl.Finish()
 
-		pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil)
+		pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil, nil)
 		pu.SV.SetDefaultValues()
 
 		ctx := context.WithValue(context.TODO(), config.ParameterUnitKey, pu)
@@ -289,7 +289,7 @@ func Test_createTablesInMoCatalogOfGeneralTenant(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		defer ctrl.Finish()
 
-		pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil)
+		pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil, nil)
 		pu.SV.SetDefaultValues()
 
 		ctx := context.WithValue(context.TODO(), config.ParameterUnitKey, pu)
@@ -337,7 +337,7 @@ func Test_checkUserExistsOrNot(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		defer ctrl.Finish()
 
-		pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil)
+		pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil, nil)
 		pu.SV.SetDefaultValues()
 
 		ctx := context.WithValue(context.TODO(), config.ParameterUnitKey, pu)
@@ -378,7 +378,7 @@ func Test_initUser(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		defer ctrl.Finish()
 
-		pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil)
+		pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil, nil)
 		pu.SV.SetDefaultValues()
 
 		ctx := context.WithValue(context.TODO(), config.ParameterUnitKey, pu)
@@ -441,7 +441,7 @@ func Test_initRole(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		defer ctrl.Finish()
 
-		pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil)
+		pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil, nil)
 		pu.SV.SetDefaultValues()
 
 		ctx := context.WithValue(context.TODO(), config.ParameterUnitKey, pu)
@@ -6030,7 +6030,7 @@ func Test_genRevokeCases(t *testing.T) {
 }
 
 func newSes(priv *privilege) *Session {
-	pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil)
+	pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil, nil)
 	pu.SV.SetDefaultValues()
 
 	ctx := context.WithValue(context.TODO(), config.ParameterUnitKey, pu)
diff --git a/pkg/frontend/internal_executor_test.go b/pkg/frontend/internal_executor_test.go
index 7b8039119..484c6201a 100644
--- a/pkg/frontend/internal_executor_test.go
+++ b/pkg/frontend/internal_executor_test.go
@@ -52,7 +52,7 @@ func (e *miniExec) PrepareSessionBeforeExecRequest(sess *Session) {
 
 func TestIe(t *testing.T) {
 	ctx := context.TODO()
-	pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil)
+	pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil, nil)
 	executor := newIe(pu, &miniExec{})
 	executor.ApplySessionOverride(ie.NewOptsBuilder().Username("dump").Finish())
 	sess := executor.newCmdSession(ctx, ie.NewOptsBuilder().Database("mo_catalog").Internal(true).Finish())
@@ -66,7 +66,7 @@ func TestIe(t *testing.T) {
 }
 
 func TestIeProto(t *testing.T) {
-	pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil)
+	pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil, nil)
 	executor := NewInternalExecutor(pu)
 	p := executor.proto
 	assert.True(t, p.IsEstablished())
diff --git a/pkg/frontend/load_test.go b/pkg/frontend/load_test.go
index 9e2dc1324..dc15260cc 100644
--- a/pkg/frontend/load_test.go
+++ b/pkg/frontend/load_test.go
@@ -27,7 +27,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	mock_frontend "github.com/matrixorigin/matrixone/pkg/frontend/test"
-	"github.com/matrixorigin/matrixone/pkg/vm/process"
+	"github.com/matrixorigin/matrixone/pkg/testutil"
 	"github.com/smartystreets/goconvey/convey"
 	"github.com/stretchr/testify/require"
 )
@@ -584,13 +584,7 @@ func Test_rowToColumnAndSaveToStorage(t *testing.T) {
 		if err != nil {
 			panic(err)
 		}
-		proc := process.New(
-			ctx,
-			mp,
-			nil,
-			nil,
-			nil,
-		)
+		proc := testutil.NewProcessWithMPool(mp)
 		field := [][]string{{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "8", "9"}}
 		Oid := []types.T{types.T_int8, types.T_int16, types.T_int32, types.T_int64, types.T_uint8, types.T_uint16,
 			types.T_uint32, types.T_uint64, types.T_float32, types.T_float64, types.T_char, types.T_date, types.T_datetime}
diff --git a/pkg/frontend/mysql_protocol_test.go b/pkg/frontend/mysql_protocol_test.go
index e1be3bded..fb173fcc8 100644
--- a/pkg/frontend/mysql_protocol_test.go
+++ b/pkg/frontend/mysql_protocol_test.go
@@ -91,7 +91,7 @@ func TestMysqlClientProtocol_Handshake(t *testing.T) {
 	//ion method: mysql -h 127.0.0.1 -P 6001 -udump -p
 
 	//before anything using the configuration
-	pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil)
+	pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil, nil)
 	_, err := toml.DecodeFile("test/system_vars_config.toml", pu.SV)
 	if err != nil {
 		panic(err)
@@ -1270,7 +1270,7 @@ func TestMysqlResultSet(t *testing.T) {
 	//	mysql-8.0.23 success
 	//./mysql-test-run 1st --extern user=root --extern port=6001 --extern host=127.0.0.1
 	//	matrixone failed: mysql-test-run: *** ERROR: Could not connect to extern server using command: '/Users/pengzhen/Documents/mysql-server-mysql-8.0.23/bld/runtime_output_directory//mysql --no-defaults --user=root --user=root --port=6001 --host=127.0.0.1 --silent --database=mysql --execute="SHOW GLOBAL VARIABLES"'
-	pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil)
+	pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil, nil)
 	_, err := toml.DecodeFile("test/system_vars_config.toml", pu.SV)
 	if err != nil {
 		panic(err)
diff --git a/pkg/frontend/routine_manager_test.go b/pkg/frontend/routine_manager_test.go
index 3d24ccd63..19e9bc48d 100644
--- a/pkg/frontend/routine_manager_test.go
+++ b/pkg/frontend/routine_manager_test.go
@@ -28,7 +28,7 @@ import (
 
 func create_test_server() *MOServer {
 	//before anything using the configuration
-	pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil)
+	pu := config.NewParameterUnit(&config.FrontendParameters{}, nil, nil, nil, nil)
 	_, err := toml.DecodeFile("test/system_vars_config.toml", pu.SV)
 	if err != nil {
 		panic(err)
diff --git a/pkg/frontend/session_test.go b/pkg/frontend/session_test.go
index c7c1a825f..d87b4b2a9 100644
--- a/pkg/frontend/session_test.go
+++ b/pkg/frontend/session_test.go
@@ -183,7 +183,7 @@ func TestSession_TxnBegin(t *testing.T) {
 		eng.EXPECT().Hints().Return(hints).AnyTimes()
 		eng.EXPECT().New(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 		eng.EXPECT().Commit(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
-		session := NewSession(proto, nil, config.NewParameterUnit(&config.FrontendParameters{}, eng, txnClient, nil), gSysVars)
+		session := NewSession(proto, nil, config.NewParameterUnit(&config.FrontendParameters{}, eng, txnClient, nil, nil), gSysVars)
 		session.SetRequestContext(context.Background())
 		return session
 	}
@@ -230,7 +230,7 @@ func TestVariables(t *testing.T) {
 		proto := NewMysqlClientProtocol(0, ioses, 1024, sv)
 		txnClient := mock_frontend.NewMockTxnClient(ctrl)
 		txnClient.EXPECT().New().AnyTimes()
-		session := NewSession(proto, nil, config.NewParameterUnit(&config.FrontendParameters{}, nil, txnClient, nil), gSysVars)
+		session := NewSession(proto, nil, config.NewParameterUnit(&config.FrontendParameters{}, nil, txnClient, nil, nil), gSysVars)
 		session.SetRequestContext(context.Background())
 		return session
 	}
@@ -539,7 +539,7 @@ func TestSession_TxnCompilerContext(t *testing.T) {
 		db.EXPECT().Relation(gomock.Any(), gomock.Any()).Return(table, nil).AnyTimes()
 		eng.EXPECT().Database(gomock.Any(), gomock.Any(), gomock.Any()).Return(db, nil).AnyTimes()
 
-		pu := config.NewParameterUnit(&config.FrontendParameters{}, eng, txnClient, nil)
+		pu := config.NewParameterUnit(&config.FrontendParameters{}, eng, txnClient, nil, nil)
 
 		gSysVars := &GlobalSystemVariables{}
 		InitGlobalSystemVariables(gSysVars)
diff --git a/pkg/frontend/test/engine_mock.go b/pkg/frontend/test/engine_mock.go
index 28e5c395c..c5d87c894 100644
--- a/pkg/frontend/test/engine_mock.go
+++ b/pkg/frontend/test/engine_mock.go
@@ -1,7 +1,7 @@
 // Code generated by MockGen. DO NOT EDIT.
-// Source: pkg/vm/engine/types.go
+// Source: ../../../pkg/vm/engine/types.go
 
-// Package mock_engine is a generated GoMock package.
+// Package mock_frontend is a generated GoMock package.
 package mock_frontend
 
 import (
@@ -414,11 +414,6 @@ func (m *MockDatabase) Create(arg0 context.Context, arg1 string, arg2 []engine.T
 	return ret0
 }
 
-func (m *MockDatabase) Truncate(ctx context.Context, s string) error {
-	//TODO implement me
-	panic("implement me")
-}
-
 // Create indicates an expected call of Create.
 func (mr *MockDatabaseMockRecorder) Create(arg0, arg1, arg2 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
@@ -469,6 +464,20 @@ func (mr *MockDatabaseMockRecorder) Relations(arg0 interface{}) *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Relations", reflect.TypeOf((*MockDatabase)(nil).Relations), arg0)
 }
 
+// Truncate mocks base method.
+func (m *MockDatabase) Truncate(arg0 context.Context, arg1 string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Truncate", arg0, arg1)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Truncate indicates an expected call of Truncate.
+func (mr *MockDatabaseMockRecorder) Truncate(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Truncate", reflect.TypeOf((*MockDatabase)(nil).Truncate), arg0, arg1)
+}
+
 // MockEngine is a mock of Engine interface.
 type MockEngine struct {
 	ctrl     *gomock.Controller
diff --git a/pkg/frontend/test/txn_mock.go b/pkg/frontend/test/txn_mock.go
index 97994374f..6abf519db 100644
--- a/pkg/frontend/test/txn_mock.go
+++ b/pkg/frontend/test/txn_mock.go
@@ -210,6 +210,160 @@ func (mr *MockTxnOperatorMockRecorder) WriteAndCommit(ctx, ops interface{}) *gom
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "WriteAndCommit", reflect.TypeOf((*MockTxnOperator)(nil).WriteAndCommit), ctx, ops)
 }
 
+// MockDebugableTxnOperator is a mock of DebugableTxnOperator interface.
+type MockDebugableTxnOperator struct {
+	ctrl     *gomock.Controller
+	recorder *MockDebugableTxnOperatorMockRecorder
+}
+
+// MockDebugableTxnOperatorMockRecorder is the mock recorder for MockDebugableTxnOperator.
+type MockDebugableTxnOperatorMockRecorder struct {
+	mock *MockDebugableTxnOperator
+}
+
+// NewMockDebugableTxnOperator creates a new mock instance.
+func NewMockDebugableTxnOperator(ctrl *gomock.Controller) *MockDebugableTxnOperator {
+	mock := &MockDebugableTxnOperator{ctrl: ctrl}
+	mock.recorder = &MockDebugableTxnOperatorMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockDebugableTxnOperator) EXPECT() *MockDebugableTxnOperatorMockRecorder {
+	return m.recorder
+}
+
+// ApplySnapshot mocks base method.
+func (m *MockDebugableTxnOperator) ApplySnapshot(data []byte) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ApplySnapshot", data)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// ApplySnapshot indicates an expected call of ApplySnapshot.
+func (mr *MockDebugableTxnOperatorMockRecorder) ApplySnapshot(data interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ApplySnapshot", reflect.TypeOf((*MockDebugableTxnOperator)(nil).ApplySnapshot), data)
+}
+
+// Commit mocks base method.
+func (m *MockDebugableTxnOperator) Commit(ctx context.Context) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Commit", ctx)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Commit indicates an expected call of Commit.
+func (mr *MockDebugableTxnOperatorMockRecorder) Commit(ctx interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Commit", reflect.TypeOf((*MockDebugableTxnOperator)(nil).Commit), ctx)
+}
+
+// Debug mocks base method.
+func (m *MockDebugableTxnOperator) Debug(ctx context.Context, ops []txn.TxnRequest) (*rpc.SendResult, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Debug", ctx, ops)
+	ret0, _ := ret[0].(*rpc.SendResult)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Debug indicates an expected call of Debug.
+func (mr *MockDebugableTxnOperatorMockRecorder) Debug(ctx, ops interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Debug", reflect.TypeOf((*MockDebugableTxnOperator)(nil).Debug), ctx, ops)
+}
+
+// Read mocks base method.
+func (m *MockDebugableTxnOperator) Read(ctx context.Context, ops []txn.TxnRequest) (*rpc.SendResult, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Read", ctx, ops)
+	ret0, _ := ret[0].(*rpc.SendResult)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Read indicates an expected call of Read.
+func (mr *MockDebugableTxnOperatorMockRecorder) Read(ctx, ops interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Read", reflect.TypeOf((*MockDebugableTxnOperator)(nil).Read), ctx, ops)
+}
+
+// Rollback mocks base method.
+func (m *MockDebugableTxnOperator) Rollback(ctx context.Context) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Rollback", ctx)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Rollback indicates an expected call of Rollback.
+func (mr *MockDebugableTxnOperatorMockRecorder) Rollback(ctx interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Rollback", reflect.TypeOf((*MockDebugableTxnOperator)(nil).Rollback), ctx)
+}
+
+// Snapshot mocks base method.
+func (m *MockDebugableTxnOperator) Snapshot() ([]byte, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Snapshot")
+	ret0, _ := ret[0].([]byte)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Snapshot indicates an expected call of Snapshot.
+func (mr *MockDebugableTxnOperatorMockRecorder) Snapshot() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Snapshot", reflect.TypeOf((*MockDebugableTxnOperator)(nil).Snapshot))
+}
+
+// Txn mocks base method.
+func (m *MockDebugableTxnOperator) Txn() txn.TxnMeta {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Txn")
+	ret0, _ := ret[0].(txn.TxnMeta)
+	return ret0
+}
+
+// Txn indicates an expected call of Txn.
+func (mr *MockDebugableTxnOperatorMockRecorder) Txn() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Txn", reflect.TypeOf((*MockDebugableTxnOperator)(nil).Txn))
+}
+
+// Write mocks base method.
+func (m *MockDebugableTxnOperator) Write(ctx context.Context, ops []txn.TxnRequest) (*rpc.SendResult, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Write", ctx, ops)
+	ret0, _ := ret[0].(*rpc.SendResult)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Write indicates an expected call of Write.
+func (mr *MockDebugableTxnOperatorMockRecorder) Write(ctx, ops interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Write", reflect.TypeOf((*MockDebugableTxnOperator)(nil).Write), ctx, ops)
+}
+
+// WriteAndCommit mocks base method.
+func (m *MockDebugableTxnOperator) WriteAndCommit(ctx context.Context, ops []txn.TxnRequest) (*rpc.SendResult, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "WriteAndCommit", ctx, ops)
+	ret0, _ := ret[0].(*rpc.SendResult)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// WriteAndCommit indicates an expected call of WriteAndCommit.
+func (mr *MockDebugableTxnOperatorMockRecorder) WriteAndCommit(ctx, ops interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "WriteAndCommit", reflect.TypeOf((*MockDebugableTxnOperator)(nil).WriteAndCommit), ctx, ops)
+}
+
 // MockTxnIDGenerator is a mock of TxnIDGenerator interface.
 type MockTxnIDGenerator struct {
 	ctrl     *gomock.Controller
diff --git a/pkg/frontend/test/types_mock.go b/pkg/frontend/test/types_mock.go
index 09b9fc0d9..c5122080c 100644
--- a/pkg/frontend/test/types_mock.go
+++ b/pkg/frontend/test/types_mock.go
@@ -1,5 +1,5 @@
 // Code generated by MockGen. DO NOT EDIT.
-// Source: pkg/frontend/types.go
+// Source: ../types.go
 
 // Package mock_frontend is a generated GoMock package.
 package mock_frontend
diff --git a/pkg/sql/plan/function/builtin/multi/from_unixtime_test.go b/pkg/sql/plan/function/builtin/multi/from_unixtime_test.go
index c8e581228..65a981125 100644
--- a/pkg/sql/plan/function/builtin/multi/from_unixtime_test.go
+++ b/pkg/sql/plan/function/builtin/multi/from_unixtime_test.go
@@ -15,7 +15,6 @@
 package multi
 
 import (
-	"context"
 	"testing"
 	"time"
 
@@ -503,17 +502,7 @@ func newTmpProcess() *process.Process {
 }
 
 func newProcessWithMPool(mp *mpool.MPool) *process.Process {
-	process := process.New(
-		context.Background(),
-		mp,
-		nil, // no txn client can be set
-		nil, // no txn operator can be set
-		testutil.NewFS(),
-	)
-	process.Lim.Size = 1 << 20
-	process.Lim.BatchRows = 1 << 20
-	process.Lim.BatchSize = 1 << 20
-	process.Lim.ReaderSize = 1 << 20
+	process := testutil.NewProcessWithMPool(mp)
 	process.SessionInfo.TimeZone = time.FixedZone("UTC0", 0)
 	return process
 }
diff --git a/pkg/testutil/util_new.go b/pkg/testutil/util_new.go
index b41c6339b..8a71f53f9 100644
--- a/pkg/testutil/util_new.go
+++ b/pkg/testutil/util_new.go
@@ -42,6 +42,7 @@ func NewProcessWithMPool(mp *mpool.MPool) *process.Process {
 		nil, // no txn client can be set
 		nil, // no txn operator can be set
 		NewFS(),
+		nil, // no get cluster details func can be set
 	)
 	proc.Lim.Size = 1 << 20
 	proc.Lim.BatchRows = 1 << 20