fix lock not found or changed in new txn

pkg/lockservice/lock_table_keeper.go

@@ -189,7 +189,7 @@ func (k *lockTableKeeper) doKeepLockTableBind(ctx context.Context) {
 	defer cancel()
 	resp, err := k.client.Send(ctx, req)
 	if err != nil {
-		logKeepBindFailed(err)
+		logKeepBindFailed(k.serviceID, err)
 		return
 	}
 	defer releaseResponse(resp)

pkg/lockservice/log.go

@@ -301,10 +301,13 @@ func logCheckDeadLockFailed(
 	}
 }
 
-func logKeepBindFailed(err error) {
+func logKeepBindFailed(
+	serviceID string,
+	err error) {
 	logger := getWithSkipLogger()
 	if logger.Enabled(zap.ErrorLevel) {
 		logger.Error("failed to keep lock table bind",
+			zap.String("serviceID", serviceID),
 			zap.Error(err))
 	}
 }

pkg/sql/colexec/lockop/lock_op.go

@@ -18,6 +18,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"github.com/matrixorigin/matrixone/pkg/txn/client"
 	"strings"
 
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
@@ -427,12 +428,19 @@ func doLock(
 	key := txnOp.AddWaitLock(tableID, rows, options)
 	defer txnOp.RemoveWaitLock(key)
 
-	result, err := lockService.Lock(
-		ctx,
-		tableID,
-		rows,
-		txn.ID,
-		options)
+	var err error
+	var result lock.Result
+	for i := 0; i < 10; i++ {
+		result, err = lockService.Lock(
+			ctx,
+			tableID,
+			rows,
+			txn.ID,
+			options)
+		if !canRetryLock(txnOp, err) {
+			break
+		}
+	}
 	if err != nil {
 		return false, false, timestamp.Timestamp{}, err
 	}
@@ -508,6 +516,17 @@ func doLock(
 	return true, result.TableDefChanged, snapshotTS, nil
 }
 
+func canRetryLock(txn client.TxnOperator, err error) bool {
+	if !moerr.IsMoErrCode(err, moerr.ErrLockTableBindChanged) ||
+		!moerr.IsMoErrCode(err, moerr.ErrLockTableNotFound) {
+		return false
+	}
+	if txn.LockTableCount() == 0 {
+		return true
+	}
+	return false
+}
+
 // DefaultLockOptions create a default lock operation. The parker is used to
 // encode primary key into lock row.
 func DefaultLockOptions(parker *types.Packer) LockOptions {

pkg/txn/client/operator.go

@@ -552,6 +552,15 @@ func (tc *txnOperator) AddLockTable(value lock.LockTable) error {
 	return tc.doAddLockTableLocked(value)
 }
 
+func (tc *txnOperator) LockTableCount() int32 {
+	tc.mu.RLock()
+	defer tc.mu.RUnlock()
+	if tc.mu.txn.Mode != txn.TxnMode_Pessimistic {
+		panic("lock in optimistic mode")
+	}
+	return int32(len(tc.mu.lockTables))
+}
+
 func (tc *txnOperator) ResetRetry(retry bool) {
 	tc.mu.Lock()
 	defer tc.mu.Unlock()

pkg/txn/client/types.go

@@ -146,6 +146,8 @@ type TxnOperator interface {
 	AddWaitLock(tableID uint64, rows [][]byte, opt lock.LockOptions) uint64
 	// RemoveWaitLock remove wait lock for current txn
 	RemoveWaitLock(key uint64)
+	// LockTableCount get quality of lock table
+	LockTableCount() int32
 
 	// AddWorkspace for the transaction
 	AddWorkspace(workspace Workspace)

pkg/txn/storage/memorystorage/storage_txn_client.go

@@ -291,6 +291,10 @@ func (s *StorageTxnOperator) AddLockTable(lock.LockTable) error {
 	panic("should not call")
 }
 
+func (s *StorageTxnOperator) LockTableCount() int32 {
+	panic("should not call")
+}
+
 func (s *StorageTxnOperator) UpdateSnapshot(ctx context.Context, ts timestamp.Timestamp) error {
 	panic("should not call")
 }

pkg/vm/engine/entire_engine_test.go

@@ -360,6 +360,10 @@ func (o *testOperator) AddLockTable(lock.LockTable) error {
 	return nil
 }
 
+func (o *testOperator) LockTableCount() int32 {
+	return 0
+}
+
 func (o *testOperator) UpdateSnapshot(ctx context.Context, ts timestamp.Timestamp) error {
 	panic("should not call")
 }