Repository: matrixorigin/matrixone
Pull request #4930 (2022-09-08 01:02:57)

add Process.TxnClient and use it in auto increment codes
## 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:
add Process.TxnClient and use it in auto increment codes


Patch:
diff --git a/pkg/frontend/mysql_cmd_executor.go b/pkg/frontend/mysql_cmd_executor.go
index 358a1ebb7..4f84e2a5f 100644
--- a/pkg/frontend/mysql_cmd_executor.go
+++ b/pkg/frontend/mysql_cmd_executor.go
@@ -1661,6 +1661,7 @@ func (mce *MysqlCmdExecutor) doComQuery(requestCtx context.Context, sql string)
 	proc := process.New(
 		requestCtx,
 		mheap.New(ses.GuestMmu),
+		ses.Pu.TxnClient,
 		ses.GetTxnHandler().txn,
 		ses.Pu.FileService,
 	)
diff --git a/pkg/sql/colexec/auto_incr.go b/pkg/sql/colexec/auto_incr.go
index d6be2eacb..e491e89ed 100644
--- a/pkg/sql/colexec/auto_incr.go
+++ b/pkg/sql/colexec/auto_incr.go
@@ -26,7 +26,6 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/sql/errors"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine"
-	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/moengine"
 	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
@@ -98,19 +97,19 @@ func getRangeFromAutoIncrTable(param *AutoIncrParam, bat *batch.Batch, tableID s
 }
 
 func getOneColRangeFromAutoIncrTable(param *AutoIncrParam, bat *batch.Batch, name string, pos int) (int64, int64, error) {
-	taeEngine, ok := param.eg.(moengine.TxnEngine)
-	if !ok {
-		return 0, 0, errors.New("", "the engine is not tae")
-	}
-
-	txnCtx, err := taeEngine.StartTxn(nil)
+	txnOperator, err := param.proc.TxnClient.New()
 	if err != nil {
 		return 0, 0, err
 	}
 
 	oriNum, step := getCurrentIndex(param, name)
 	if oriNum < 0 {
-		if err2 := txnCtx.Rollback(); err2 != nil {
+		ctx, cancel := context.WithTimeout(
+			param.ctx,
+			param.eg.Hints().CommitOrRollbackTimeout,
+		)
+		defer cancel()
+		if err2 := txnOperator.Rollback(ctx); err2 != nil {
 			return 0, 0, err2
 		}
 		return 0, 0, errors.New("", "GetIndex from auto_increment table fail")
@@ -147,12 +146,22 @@ func getOneColRangeFromAutoIncrTable(param *AutoIncrParam, bat *batch.Batch, nam
 		return 0, 0, errors.New("", "auto_incrment column constant value overflows bigint")
 	}
 	if err := updateAutoIncrTable(param, maxNum, name); err != nil {
-		if err2 := txnCtx.Rollback(); err2 != nil {
+		ctx, cancel := context.WithTimeout(
+			param.ctx,
+			param.eg.Hints().CommitOrRollbackTimeout,
+		)
+		defer cancel()
+		if err2 := txnOperator.Rollback(ctx); err2 != nil {
 			return 0, 0, err2
 		}
 		return 0, 0, err
 	}
-	err = txnCtx.Commit()
+	ctx, cancel := context.WithTimeout(
+		param.ctx,
+		param.eg.Hints().CommitOrRollbackTimeout,
+	)
+	defer cancel()
+	err = txnOperator.Commit(ctx)
 	if err != nil {
 		return 0, 0, err
 	}
diff --git a/pkg/sql/compile/cnserviceDriver.go b/pkg/sql/compile/cnserviceDriver.go
index b41c09c22..d59641d10 100644
--- a/pkg/sql/compile/cnserviceDriver.go
+++ b/pkg/sql/compile/cnserviceDriver.go
@@ -807,6 +807,7 @@ func newCompile(
 		proc: process.New(
 			ctx,
 			mheap.New(guest.New(1<<30, host.New(1<<20))),
+			nil, //TODO must set txn client
 			nil, //TODO must set txn operator
 			nil, //TODO must set file service
 		),
diff --git a/pkg/vm/process/process.go b/pkg/vm/process/process.go
index 74095f2bb..dd89ecefa 100644
--- a/pkg/vm/process/process.go
+++ b/pkg/vm/process/process.go
@@ -31,12 +31,14 @@ import (
 func New(
 	ctx context.Context,
 	m *mheap.Mheap,
+	txnClient client.TxnClient,
 	txnOperator client.TxnOperator,
 	fileService fileservice.FileService,
 ) *Process {
 	return &Process{
 		mp:          m,
 		Ctx:         ctx,
+		TxnClient:   txnClient,
 		TxnOperator: txnOperator,
 		FileService: fileService,
 	}
@@ -56,6 +58,7 @@ func NewFromProc(p *Process, ctx context.Context, regNumber int) *Process {
 	proc.Id = p.Id
 	proc.mp = p.Mp()
 	proc.Lim = p.Lim
+	proc.TxnClient = p.TxnClient
 	proc.TxnOperator = p.TxnOperator
 	proc.AnalInfos = p.AnalInfos
 	proc.SessionInfo = p.SessionInfo
diff --git a/pkg/vm/process/types.go b/pkg/vm/process/types.go
index daa9994f1..9c7fe5ccd 100644
--- a/pkg/vm/process/types.go
+++ b/pkg/vm/process/types.go
@@ -107,6 +107,8 @@ type Process struct {
 	// unix timestamp
 	UnixTime int64
 
+	TxnClient client.TxnClient
+
 	TxnOperator client.TxnOperator
 
 	AnalInfos []*AnalyzeInfo


Test patch:
diff --git a/pkg/sql/colexec/external/external_test.go b/pkg/sql/colexec/external/external_test.go
index a472b8f5f..220405bc5 100644
--- a/pkg/sql/colexec/external/external_test.go
+++ b/pkg/sql/colexec/external/external_test.go
@@ -18,7 +18,6 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
-	"math"
 	"os"
 	"testing"
 
@@ -26,7 +25,6 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/sql/parsers/tree"
 	"github.com/matrixorigin/matrixone/pkg/sql/plan"
 	"github.com/matrixorigin/matrixone/pkg/testutil"
-	"github.com/matrixorigin/matrixone/pkg/vm/mheap"
 	"github.com/matrixorigin/matrixone/pkg/vm/mmu/guest"
 	"github.com/matrixorigin/matrixone/pkg/vm/mmu/host"
 	"github.com/matrixorigin/matrixone/pkg/vm/process"
@@ -335,11 +333,7 @@ func Test_GetBatchData(t *testing.T) {
 			simdCsvLineArray: [][]string{line},
 		}
 
-		mmu := &guest.Mmu{
-			Limit: math.MaxInt64,
-			Mmu:   &host.Mmu{},
-		}
-		proc := process.New(context.TODO(), mheap.New(mmu), nil, nil)
+		proc := testutil.NewProc()
 		_, err := GetBatchData(param, plh, proc)
 		convey.So(err, convey.ShouldBeNil)
 
diff --git a/pkg/testutil/util_new.go b/pkg/testutil/util_new.go
index 1bc904334..6bf558033 100644
--- a/pkg/testutil/util_new.go
+++ b/pkg/testutil/util_new.go
@@ -46,7 +46,8 @@ func NewProcessWithMheap(heap *mheap.Mheap) *process.Process {
 	proc := process.New(
 		context.Background(),
 		heap,
-		nil,
+		nil, // no txn client can be set
+		nil, // no txn operator can be set
 		NewFS(),
 	)
 	proc.Lim.Size = 1 << 20
diff --git a/pkg/vectorize/momath/math_test.go b/pkg/vectorize/momath/math_test.go
index 2ff15f6e9..ed5bb6fd8 100644
--- a/pkg/vectorize/momath/math_test.go
+++ b/pkg/vectorize/momath/math_test.go
@@ -16,13 +16,13 @@ package momath
 
 import (
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
-	"github.com/stretchr/testify/require"
 	"github.com/matrixorigin/matrixone/pkg/testutil"
+	"github.com/stretchr/testify/require"
 	"math"
 	"testing"
 )
 
-func TestLn(t *testing.T){
+func TestLn(t *testing.T) {
 	as := []float64{1, math.Exp(0), math.Exp(1), math.Exp(10), math.Exp(100), math.Exp(99), math.Exp(-1)}
 	cs := make([]float64, 7)
 
@@ -30,14 +30,14 @@ func TestLn(t *testing.T){
 	cv := testutil.MakeFloat64Vector(cs, nil)
 
 	err := Ln(av, cv)
-	if err != nil{
+	if err != nil {
 		panic(err)
 	}
 	cols := vector.MustTCols[float64](cv)
 	require.Equal(t, []float64{0.0, 0.0, 1.0, 10.0, 100.0, 99.0, -1}, cols)
 }
 
-func TestExP(t *testing.T){
+func TestExP(t *testing.T) {
 	as := []float64{-1, 0, 1, 2, 10, 100}
 	cs := make([]float64, 6)
 
@@ -45,14 +45,14 @@ func TestExP(t *testing.T){
 	cv := testutil.MakeFloat64Vector(cs, nil)
 
 	err := Exp(av, cv)
-	if err != nil{
+	if err != nil {
 		panic(err)
 	}
 	cols := vector.MustTCols[float64](cv)
 	require.Equal(t, []float64{math.Exp(-1), math.Exp(0), math.Exp(1), math.Exp(2), math.Exp(10), math.Exp(100)}, cols)
 }
 
-func TestSin(t *testing.T){
+func TestSin(t *testing.T) {
 	as := []float64{-math.Pi / 2, 0, math.Pi / 2}
 	cs := make([]float64, 3)
 
@@ -60,14 +60,14 @@ func TestSin(t *testing.T){
 	cv := testutil.MakeFloat64Vector(cs, nil)
 
 	err := Sin(av, cv)
-	if err != nil{
+	if err != nil {
 		panic(err)
 	}
 	cols := vector.MustTCols[float64](cv)
 	require.Equal(t, []float64{-1, 0, 1}, cols)
 }
 
-func TestCos(t *testing.T){
+func TestCos(t *testing.T) {
 	as := []float64{-math.Pi, 0, math.Pi}
 	cs := make([]float64, 3)
 
@@ -75,14 +75,14 @@ func TestCos(t *testing.T){
 	cv := testutil.MakeFloat64Vector(cs, nil)
 
 	err := Cos(av, cv)
-	if err != nil{
+	if err != nil {
 		panic(err)
 	}
 	cols := vector.MustTCols[float64](cv)
 	require.Equal(t, []float64{-1, 1, -1}, cols)
 }
 
-func TestTan(t *testing.T){
+func TestTan(t *testing.T) {
 	as := []float64{0}
 	cs := make([]float64, 1)
 
@@ -90,14 +90,14 @@ func TestTan(t *testing.T){
 	cv := testutil.MakeFloat64Vector(cs, nil)
 
 	err := Tan(av, cv)
-	if err != nil{
+	if err != nil {
 		panic(err)
 	}
 	cols := vector.MustTCols[float64](cv)
 	require.Equal(t, []float64{0}, cols)
 }
 
-func TestSinh(t *testing.T){
+func TestSinh(t *testing.T) {
 	as := []float64{0}
 	cs := make([]float64, 1)
 
@@ -105,15 +105,14 @@ func TestSinh(t *testing.T){
 	cv := testutil.MakeFloat64Vector(cs, nil)
 
 	err := Sinh(av, cv)
-	if err != nil{
+	if err != nil {
 		panic(err)
 	}
 	cols := vector.MustTCols[float64](cv)
 	require.Equal(t, []float64{0}, cols)
 }
 
-
-func TestAcos(t *testing.T){
+func TestAcos(t *testing.T) {
 	as := []float64{1}
 	cs := make([]float64, 1)
 
@@ -121,14 +120,14 @@ func TestAcos(t *testing.T){
 	cv := testutil.MakeFloat64Vector(cs, nil)
 
 	err := Acos(av, cv)
-	if err != nil{
+	if err != nil {
 		panic(err)
 	}
 	cols := vector.MustTCols[float64](cv)
 	require.Equal(t, []float64{0}, cols)
 }
 
-func TestAtan(t *testing.T){
+func TestAtan(t *testing.T) {
 	as := []float64{0}
 	cs := make([]float64, 1)
 
@@ -136,10 +135,9 @@ func TestAtan(t *testing.T){
 	cv := testutil.MakeFloat64Vector(cs, nil)
 
 	err := Atan(av, cv)
-	if err != nil{
+	if err != nil {
 		panic(err)
 	}
 	cols := vector.MustTCols[float64](cv)
 	require.Equal(t, []float64{0}, cols)
 }
-
diff --git a/pkg/vm/engine/tae/common/mempool_test.go b/pkg/vm/engine/tae/common/mempool_test.go
index 91cabcd80..516ee11e4 100644
--- a/pkg/vm/engine/tae/common/mempool_test.go
+++ b/pkg/vm/engine/tae/common/mempool_test.go
@@ -55,9 +55,7 @@ func TestPool(t *testing.T) {
 	var wg sync.WaitGroup
 
 	now := time.Now()
-	// hm := host.New(1 << 48)
-	// gm := guest.New(1<<48, hm)
-	// proc := process.New(gm, mempool.New(1<<48, 8))
+	// proc := testutil.NewProc()
 
 	// for i := 0; i < 1024*512; i++ {
 	// 	wg.Add(1)