Feature add undo log manager delete

go.mod

@@ -8,8 +8,8 @@ require (
  github.com/agiledragon/gomonkey v2.0.2+incompatible
  github.com/apache/dubbo-getty v1.4.8
  github.com/dubbogo/gost v1.12.5
- github.com/golang/mock v1.6.0
  github.com/go-sql-driver/mysql v1.6.0
+ github.com/golang/mock v1.6.0
  github.com/google/uuid v1.3.0
  github.com/natefinch/lumberjack v2.0.0+incompatible
  github.com/pingcap/tidb v1.1.0-beta.0.20211124132551-4a1b2e9fe5b5

go.sum

@@ -1848,4 +1848,4 @@ sourcegraph.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67/go.m
 vimagination.zapto.org/byteio v0.0.0-20200222190125-d27cba0f0b10 h1:pxt6fVJP67Hxo1qk8JalUghLlk3abYByl+3e0JYfUlE=
 vimagination.zapto.org/byteio v0.0.0-20200222190125-d27cba0f0b10/go.mod h1:fl9OF22g6MTKgvHA1hqMXe/L7+ULWofVTwbC9loGu7A=
 vimagination.zapto.org/memio v0.0.0-20200222190306-588ebc67b97d h1:Mp6WiHHuiwHaknxTdxJ8pvC9/B4pOgW1PamKGexG7Fs=
-vimagination.zapto.org/memio v0.0.0-20200222190306-588ebc67b97d/go.mod h1:zHGDKp2tyvF4IAfLti4pKYqCJucXYmmKMb3UMrCHK/4=
+vimagination.zapto.org/memio v0.0.0-20200222190306-588ebc67b97d/go.mod h1:zHGDKp2tyvF4IAfLti4pKYqCJucXYmmKMb3UMrCHK/4=

pkg/common/util/slice_to_str.go

@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package util
+
+import (
+ "errors"
+ "strconv"
+ "strings"
+)
+
+// Int64Slice2Str
+func Int64Slice2Str(values interface{}, sep string) (string, error) {
+ v, ok := values.([]int64)
+ if !ok {
+ return "", errors.New("param type is fault")
+ }
+
+ var valuesText []string
+
+ for i := range v {
+ text := strconv.FormatInt(v[i], 10)
+ valuesText = append(valuesText, text)
+ }
+
+ return strings.Join(valuesText, sep), nil
+}

pkg/constant/client_table_columns_name.go

@@ -0,0 +1,27 @@
+package constant
+
+const (
+ // UndoLogId The constant undo_log column name xid, this field is not use in mysql
+ UndoLogId string = "id"
+
+ // UndoLogXid The constant undo_log column name xid
+ UndoLogXid = "xid"
+
+ // UndoLogBranchXid The constant undo_log column name branch_id
+ UndoLogBranchXid = "branch_id"
+
+ // UndoLogContext The constant undo_log column name context
+ UndoLogContext = "context"
+
+ // UndoLogRollBackInfo The constant undo_log column name rollback_info
+ UndoLogRollBackInfo = "rollback_info"
+
+ // UndoLogLogStatus The constant undo_log column name log_status
+ UndoLogLogStatus = "log_status"
+
+ // UndoLogLogCreated The constant undo_log column name log_created
+ UndoLogLogCreated = "log_created"
+
+ // UndoLogLogModified The constant undo_log column name log_modified
+ UndoLogLogModified = "log_modified"
+)

pkg/constant/undo.go

@@ -0,0 +1,8 @@
+package constant
+
+const (
+ DefaultTransactionUndoLogTable = "undo_log"
+ // UndoLogTableName Todo get from config
+ UndoLogTableName = DefaultTransactionUndoLogTable
+ DeleteUndoLogSql = "DELETE FROM" + UndoLogTableName + " WHERE " + UndoLogBranchXid + " = ? AND " + UndoLogXid + " = ?"
+)

pkg/datasource/sql/at.go

@@ -26,9 +26,10 @@ import (
  "sync"
  "time"
 
+ "github.com/seata/seata-go/pkg/datasource/sql/undo"
+
  "github.com/seata/seata-go/pkg/datasource/sql/datasource"
  "github.com/seata/seata-go/pkg/datasource/sql/types"
- "github.com/seata/seata-go/pkg/datasource/sql/undo"
  "github.com/seata/seata-go/pkg/protocol/branch"
  "github.com/seata/seata-go/pkg/protocol/message"
  "github.com/seata/seata-go/pkg/rm"

pkg/datasource/sql/db.go

@@ -21,9 +21,10 @@ import (
  "context"
  gosql "database/sql"
 
+ "github.com/seata/seata-go/pkg/datasource/sql/undo"
+
  "github.com/seata/seata-go/pkg/datasource/sql/datasource"
  "github.com/seata/seata-go/pkg/datasource/sql/types"
- "github.com/seata/seata-go/pkg/datasource/sql/undo"
  "github.com/seata/seata-go/pkg/protocol/branch"
 )
 

pkg/datasource/sql/tx.go

@@ -21,14 +21,15 @@ import (
  "context"
  "database/sql/driver"
 
+ "github.com/seata/seata-go/pkg/datasource/sql/undo"
+
  "github.com/seata/seata-go/pkg/common/log"
  "github.com/seata/seata-go/pkg/datasource/sql/datasource"
  "github.com/seata/seata-go/pkg/protocol/branch"
  "github.com/seata/seata-go/pkg/protocol/message"
 
  "github.com/pkg/errors"
  "github.com/seata/seata-go/pkg/datasource/sql/types"
- "github.com/seata/seata-go/pkg/datasource/sql/undo"
 )
 
 const REPORT_RETRY_COUNT = 5

pkg/datasource/sql/undo/base/undo.go

@@ -18,15 +18,24 @@
 package base
 
 import (
+ "context"
  "database/sql"
  "database/sql/driver"
+ "strings"
 
- "github.com/seata/seata-go/pkg/datasource/sql/types"
+ "github.com/pkg/errors"
+ "github.com/seata/seata-go/pkg/common/log"
+ "github.com/seata/seata-go/pkg/common/util"
+ "github.com/seata/seata-go/pkg/constant"
  "github.com/seata/seata-go/pkg/datasource/sql/undo"
+
+ "github.com/seata/seata-go/pkg/datasource/sql/types"
 )
 
 var _ undo.UndoLogManager = (*BaseUndoLogManager)(nil)
 
+var ErrorDeleteUndoLogParamsFault = errors.New("xid or branch_id can't nil")
+
 // BaseUndoLogManager
 type BaseUndoLogManager struct{}
 
@@ -39,8 +48,37 @@ func (m *BaseUndoLogManager) InsertUndoLog(l []undo.BranchUndoLog, tx driver.Tx)
  return nil
 }
 
-// DeleteUndoLog
+// DeleteUndoLogs exec delete undo log operate
 func (m *BaseUndoLogManager) DeleteUndoLogs(xid []string, branchID []int64, conn *sql.Conn) error {
+ // build delete undo log sql
+ batchDeleteSql, err := m.getBatchDeleteUndoLogSql(xid, branchID)
+ if err != nil {
+ log.Errorf("get undo sql log fail, err: %v", err)
+ return err
+ }
+
+ ctx := context.Background()
+
+ // prepare deal sql
+ stmt, err := conn.PrepareContext(ctx, batchDeleteSql)
+ if err != nil {
+ log.Errorf("prepare sql fail, err: %v", err)
+ return err
+ }
+
+ branchIDStr, err := util.Int64Slice2Str(branchID, ",")
+ if err != nil {
+ log.Errorf("slice to string transfer fail, err: %v", err)
+ return err
+ }
+
+ // exec sql stmt
+ _, err = stmt.ExecContext(ctx, branchIDStr, strings.Join(xid, ","))
+ if err != nil {
+ log.Errorf("exec delete undo log fail, err: %v", err)
+ return err
+ }
+
  return nil
 }
 
@@ -58,3 +96,41 @@ func (m *BaseUndoLogManager) RunUndo(xid string, branchID int64, conn *sql.Conn)
 func (m *BaseUndoLogManager) DBType() types.DBType {
  panic("implement me")
 }
+
+// getBatchDeleteUndoLogSql build batch delete undo log
+func (m *BaseUndoLogManager) getBatchDeleteUndoLogSql(xid []string, branchID []int64) (string, error) {
+ if len(xid) == 0 || len(branchID) == 0 {
+ return "", ErrorDeleteUndoLogParamsFault
+ }
+
+ var undoLogDeleteSql strings.Builder
+ undoLogDeleteSql.WriteString("DELETE FROM ")
+ undoLogDeleteSql.WriteString(constant.UndoLogTableName)
+ undoLogDeleteSql.WriteString(" WHERE ")
+ undoLogDeleteSql.WriteString(constant.UndoLogBranchXid)
+ undoLogDeleteSql.WriteString(" IN ")
+ m.appendInParam(len(branchID), &undoLogDeleteSql)
+ undoLogDeleteSql.WriteString(" AND ")
+ undoLogDeleteSql.WriteString(constant.UndoLogXid)
+ undoLogDeleteSql.WriteString(" IN ")
+ m.appendInParam(len(xid), &undoLogDeleteSql)
+
+ return undoLogDeleteSql.String(), nil
+}
+
+// appendInParam build in param
+func (m *BaseUndoLogManager) appendInParam(size int, str *strings.Builder) {
+ if size <= 0 {
+ return
+ }
+
+ str.WriteString(" (")
+ for i := 0; i < size; i++ {
+ str.WriteString("?")
+ if i < size-1 {
+ str.WriteString(",")
+ }
+ }
+
+ str.WriteString(") ")
+}

pkg/datasource/sql/undo/mysql/undo.go

@@ -21,8 +21,9 @@ import (
  "database/sql"
  "database/sql/driver"
 
- "github.com/seata/seata-go/pkg/datasource/sql/types"
  "github.com/seata/seata-go/pkg/datasource/sql/undo"
+
+ "github.com/seata/seata-go/pkg/datasource/sql/types"
  "github.com/seata/seata-go/pkg/datasource/sql/undo/base"
 )
 

pkg/datasource/sql/undo_test.go

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package sql
+
+import (
+ "context"
+ "database/sql"
+ "testing"
+
+ "github.com/seata/seata-go/pkg/datasource/sql/undo/base"
+ "github.com/stretchr/testify/assert"
+)
+
+// TestDeleteUndoLogs
+func TestDeleteUndoLogs(t *testing.T) {
+ // local test can annotation t.SkipNow()
+ t.SkipNow()
+
+ testDeleteUndoLogs := func() {
+ db, err := sql.Open("seata-mysql", "root:12345678@tcp(127.0.0.1:3306)/seata_order?multiStatements=true")
+ assert.Nil(t, err)
+
+ sqlConn, err := db.Conn(context.Background())
+ assert.Nil(t, err)
+
+ undoLogManager := new(base.BaseUndoLogManager)
+
+ err = undoLogManager.DeleteUndoLogs([]string{"1"}, []int64{1}, sqlConn)
+ assert.Nil(t, err)
+ }
+
+ t.Run("test_delete_undo_logs", func(t *testing.T) {
+ testDeleteUndoLogs()
+ })
+}

pkg/rm/rm_api_test.go

@@ -2,10 +2,11 @@ package rm
 
 import (
  "context"
- "github.com/golang/mock/gomock"
- "github.com/seata/seata-go/pkg/protocol/branch"
  "reflect"
  "sync"
+
+ "github.com/golang/mock/gomock"
+ "github.com/seata/seata-go/pkg/protocol/branch"
 )
 
 // MockResource is a mock of Resource interface.

testdata/mock_tcc.go

@@ -2,6 +2,7 @@ package testdata
 
 import (
  "context"
+
  "github.com/seata/seata-go/pkg/tm"
 )