Repository: matrixorigin/matrixone
Pull request #13191 (2023-12-04 08:52:41)

add a session variable to control txn log
## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [x] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #13155 

## What this PR does / why we need it:
add a session variable to control txn log
1. set transaction_operator_open_log = on;  will open the switch.
2. txnoperator IsOpenLog() can get the operator switch value.


Patch:
diff --git a/pkg/frontend/txn.go b/pkg/frontend/txn.go
index 4c9a85ab2..f0aeb1593 100644
--- a/pkg/frontend/txn.go
+++ b/pkg/frontend/txn.go
@@ -17,9 +17,10 @@ package frontend
 import (
 	"context"
 	"fmt"
-	"github.com/google/uuid"
 	"sync"
 
+	"github.com/google/uuid"
+
 	"github.com/matrixorigin/matrixone/pkg/logutil"
 	"go.uber.org/zap"
 
@@ -157,6 +158,22 @@ func (th *TxnHandler) NewTxnOperator() (context.Context, TxnOperator, error) {
 		opts = append(opts,
 			client.WithUserTxn())
 	}
+
+	if th.ses != nil {
+		varVal, err := th.ses.GetSessionVar("transaction_operator_open_log")
+		if err != nil {
+			return nil, nil, err
+		}
+		if gsv, ok := GSysVariables.GetDefinitionOfSysVar("transaction_operator_open_log"); ok {
+			if svbt, ok2 := gsv.GetType().(SystemVariableBoolType); ok2 {
+				if svbt.IsTrue(varVal) {
+					opts = append(opts, client.WithTxnOpenLog())
+				}
+			}
+		}
+
+	}
+
 	th.txnOperator, err = th.txnClient.New(
 		txnCtx,
 		th.ses.getLastCommitTS(),
diff --git a/pkg/frontend/variables.go b/pkg/frontend/variables.go
index 9184e64b9..d7f124409 100644
--- a/pkg/frontend/variables.go
+++ b/pkg/frontend/variables.go
@@ -3482,6 +3482,14 @@ var gSysVarsDefs = map[string]SystemVariable{
 		Type:              InitSystemVariableStringType("gtid_purged"),
 		Default:           "",
 	},
+	"transaction_operator_open_log": {
+		Name:              "transaction_operator_open_log",
+		Scope:             ScopeSession,
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              InitSystemVariableBoolType("transaction_operator_open_log"),
+		Default:           int64(0),
+	},
 }
 
 func updateTimeZone(sess *Session, vars map[string]interface{}, name string, val interface{}) error {
diff --git a/pkg/txn/client/operator.go b/pkg/txn/client/operator.go
index abc7f4e15..50bcdb062 100644
--- a/pkg/txn/client/operator.go
+++ b/pkg/txn/client/operator.go
@@ -100,6 +100,15 @@ func WithTxnCNCoordinator() TxnOption {
 	}
 }
 
+// WithTxnOpenLog set txn open log
+func WithTxnOpenLog() TxnOption {
+	return func(tc *txnOperator) {
+		tc.mu.Lock()
+		defer tc.mu.Unlock()
+		tc.mu.openlog = true
+	}
+}
+
 // WithTxnLockService set txn lock service
 func WithTxnLockService(lockService lockservice.LockService) TxnOption {
 	return func(tc *txnOperator) {
@@ -175,6 +184,7 @@ type txnOperator struct {
 		lockTables   []lock.LockTable
 		callbacks    map[EventType][]func(txn.TxnMeta)
 		retry        bool
+		openlog      bool
 
 		lockSeq   uint64
 		waitLocks map[uint64]Lock
@@ -548,6 +558,18 @@ func (tc *txnOperator) IsRetry() bool {
 	return tc.mu.retry
 }
 
+func (tc *txnOperator) SetOpenLog(openlog bool) {
+	tc.mu.Lock()
+	defer tc.mu.Unlock()
+	tc.mu.openlog = openlog
+}
+
+func (tc *txnOperator) IsOpenLog() bool {
+	tc.mu.RLock()
+	defer tc.mu.RUnlock()
+	return tc.mu.openlog
+}
+
 func (tc *txnOperator) doAddLockTableLocked(value lock.LockTable) error {
 	for _, l := range tc.mu.lockTables {
 		if l.Table == value.Table {
diff --git a/pkg/txn/client/types.go b/pkg/txn/client/types.go
index db171fca9..70ea72e6e 100644
--- a/pkg/txn/client/types.go
+++ b/pkg/txn/client/types.go
@@ -155,6 +155,9 @@ type TxnOperator interface {
 	ResetRetry(bool)
 	IsRetry() bool
 
+	SetOpenLog(bool)
+	IsOpenLog() bool
+
 	// AppendEventCallback append callback. All append callbacks will be called sequentially
 	// if event happen.
 	AppendEventCallback(event EventType, callbacks ...func(txn.TxnMeta))
diff --git a/pkg/txn/storage/memorystorage/storage_txn_client.go b/pkg/txn/storage/memorystorage/storage_txn_client.go
index 895317464..ea1d5bc81 100644
--- a/pkg/txn/storage/memorystorage/storage_txn_client.go
+++ b/pkg/txn/storage/memorystorage/storage_txn_client.go
@@ -126,6 +126,14 @@ func (s *StorageTxnOperator) IsRetry() bool {
 	panic("unimplemented")
 }
 
+func (s *StorageTxnOperator) SetOpenLog(retry bool) {
+	panic("unimplemented")
+}
+
+func (s *StorageTxnOperator) IsOpenLog() bool {
+	panic("unimplemented")
+}
+
 func (s *StorageTxnOperator) AppendEventCallback(event client.EventType, callbacks ...func(txn.TxnMeta)) {
 	panic("unimplemented")
 }


Test patch:
diff --git a/pkg/frontend/session_test.go b/pkg/frontend/session_test.go
index f6a1d9109..d82ed1e39 100644
--- a/pkg/frontend/session_test.go
+++ b/pkg/frontend/session_test.go
@@ -16,11 +16,12 @@ package frontend
 
 import (
 	"context"
-	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"math"
 	"testing"
 	"time"
 
+	"github.com/matrixorigin/matrixone/pkg/pb/plan"
+
 	"github.com/fagongzi/goetty/v2/buf"
 	"github.com/golang/mock/gomock"
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
@@ -196,12 +197,15 @@ func TestTxnHandler_RollbackTxn(t *testing.T) {
 
 		pu, err := getParameterUnit("test/system_vars_config.toml", eng, txnClient)
 		convey.So(err, convey.ShouldBeNil)
+		var gSys GlobalSystemVariables
+		InitGlobalSystemVariables(&gSys)
 
 		txn := InitTxnHandler(eng, txnClient, nil, nil)
 		txn.ses = &Session{
 			requestCtx: ctx,
 			pu:         pu,
 			connectCtx: ctx,
+			gSysVars:   &gSys,
 		}
 		_, _, err = txn.NewTxn()
 		convey.So(err, convey.ShouldBeNil)
@@ -797,3 +801,59 @@ func TestSession_updateTimeZone(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, ses.GetTimeZone().String(), "UTC")
 }
+
+func TestTxnHandler_TxnOpenLog(t *testing.T) {
+	convey.Convey("txn open log", t, func() {
+		ctrl := gomock.NewController(t)
+		defer ctrl.Finish()
+
+		ctx := context.TODO()
+		txnOperator := mock_frontend.NewMockTxnOperator(ctrl)
+		txnOperator.EXPECT().Txn().Return(txn.TxnMeta{}).AnyTimes()
+		txnOperator.EXPECT().Rollback(gomock.Any()).Return(nil).AnyTimes()
+		txnOperator.EXPECT().Commit(gomock.Any()).Return(nil).AnyTimes()
+		// txnOperator.EXPECT().SetOpenLog(gomock.Any())
+		txnOperator.EXPECT().IsOpenLog().Return(true).AnyTimes()
+		txnClient := mock_frontend.NewMockTxnClient(ctrl)
+		cnt := 0
+		txnClient.EXPECT().New(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+			func(_ context.Context, _ timestamp.Timestamp, ootions ...client.TxnOption) (client.TxnOperator, error) {
+				cnt++
+				if cnt%2 != 0 {
+					return txnOperator, nil
+				} else {
+					return nil, moerr.NewInternalError(ctx, "startTxn failed")
+				}
+			}).AnyTimes()
+		eng := mock_frontend.NewMockEngine(ctrl)
+		eng.EXPECT().New(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+		eng.EXPECT().New(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+		eng.EXPECT().Hints().Return(engine.Hints{
+			CommitOrRollbackTimeout: time.Second,
+		}).AnyTimes()
+
+		ioses := mock_frontend.NewMockIOSession(ctrl)
+		ioses.EXPECT().OutBuf().Return(buf.NewByteBuf(1024)).AnyTimes()
+		ioses.EXPECT().Write(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+		ioses.EXPECT().RemoteAddress().Return("").AnyTimes()
+		ioses.EXPECT().Ref().AnyTimes()
+
+		pu, err := getParameterUnit("test/system_vars_config.toml", eng, txnClient)
+		convey.So(err, convey.ShouldBeNil)
+
+		var gSys GlobalSystemVariables
+		InitGlobalSystemVariables(&gSys)
+		gSys.sysVars["transaction_operator_open_log"] = 1
+		txn := InitTxnHandler(eng, txnClient, nil, nil)
+		txn.ses = &Session{
+			requestCtx: ctx,
+			pu:         pu,
+			connectCtx: ctx,
+			gSysVars:   &gSys,
+		}
+
+		_, op, err := txn.NewTxnOperator()
+		convey.So(err, convey.ShouldBeNil)
+		convey.So(op.IsOpenLog(), convey.ShouldBeTrue)
+	})
+}
diff --git a/pkg/frontend/test/txn_mock.go b/pkg/frontend/test/txn_mock.go
index 78a31bdcf..1c85d062a 100644
--- a/pkg/frontend/test/txn_mock.go
+++ b/pkg/frontend/test/txn_mock.go
@@ -398,6 +398,20 @@ func (mr *MockTxnOperatorMockRecorder) GetWorkspace() *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetWorkspace", reflect.TypeOf((*MockTxnOperator)(nil).GetWorkspace))
 }
 
+// IsOpenLog mocks base method.
+func (m *MockTxnOperator) IsOpenLog() bool {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "IsOpenLog")
+	ret0, _ := ret[0].(bool)
+	return ret0
+}
+
+// IsOpenLog indicates an expected call of IsOpenLog.
+func (mr *MockTxnOperatorMockRecorder) IsOpenLog() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsOpenLog", reflect.TypeOf((*MockTxnOperator)(nil).IsOpenLog))
+}
+
 // IsRetry mocks base method.
 func (m *MockTxnOperator) IsRetry() bool {
 	m.ctrl.T.Helper()
@@ -465,6 +479,18 @@ func (mr *MockTxnOperatorMockRecorder) Rollback(ctx interface{}) *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Rollback", reflect.TypeOf((*MockTxnOperator)(nil).Rollback), ctx)
 }
 
+// SetOpenLog mocks base method.
+func (m *MockTxnOperator) SetOpenLog(arg0 bool) {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "SetOpenLog", arg0)
+}
+
+// SetOpenLog indicates an expected call of SetOpenLog.
+func (mr *MockTxnOperatorMockRecorder) SetOpenLog(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetOpenLog", reflect.TypeOf((*MockTxnOperator)(nil).SetOpenLog), arg0)
+}
+
 // Snapshot mocks base method.
 func (m *MockTxnOperator) Snapshot() ([]byte, error) {
 	m.ctrl.T.Helper()
diff --git a/pkg/vm/engine/entire_engine_test.go b/pkg/vm/engine/entire_engine_test.go
index 25ad3824a..afaf5a594 100644
--- a/pkg/vm/engine/entire_engine_test.go
+++ b/pkg/vm/engine/entire_engine_test.go
@@ -362,6 +362,14 @@ func (o *testOperator) IsRetry() bool {
 	panic("unimplemented")
 }
 
+func (o *testOperator) SetOpenLog(retry bool) {
+	panic("unimplemented")
+}
+
+func (o *testOperator) IsOpenLog() bool {
+	panic("unimplemented")
+}
+
 func (o *testOperator) AppendEventCallback(event client.EventType, callbacks ...func(txn.TxnMeta)) {
 	panic("unimplemented")
 }
diff --git a/test/distributed/cases/database/system_variables.result b/test/distributed/cases/database/system_variables.result
index 9aeb7dce4..ce2989aab 100644
--- a/test/distributed/cases/database/system_variables.result
+++ b/test/distributed/cases/database/system_variables.result
@@ -84,6 +84,7 @@ show variables like 'trans%';
 Variable_name    Value
 transaction_alloc_block_size    8192
 transaction_isolation    REPEATABLE-READ
+transaction_operator_open_log    off
 transaction_prealloc_size    4096
 transaction_read_only    0
 show variables like 'transaction_isolation';