From bc793e9dc47e5894479137b886f720c07614656f Mon Sep 17 00:00:00 2001 From: wang1309 <1028960576@qq.com> Date: Fri, 2 Sep 2022 08:05:01 +0800 Subject: [PATCH] feature: add undo log manager delete (#240) feature:add DeleteUndoLogs func, issue:#217 --- go.mod | 2 +- go.sum | 2 +- pkg/common/util/slice_to_str.go | 41 ++++++++++ pkg/constant/client_table_columns_name.go | 44 ++++++++++ pkg/constant/undo.go | 26 ++++++ pkg/datasource/sql/at.go | 5 +- pkg/datasource/sql/db.go | 3 +- pkg/datasource/sql/tx.go | 3 +- pkg/datasource/sql/undo/base/undo.go | 97 ++++++++++++++++++++++- pkg/datasource/sql/undo/mysql/undo.go | 13 ++- pkg/datasource/sql/undo/undo.go | 5 +- pkg/datasource/sql/undo_test.go | 73 +++++++++++++++++ pkg/rm/rm_api_test.go | 5 +- sample/tcc/grpc/pb/tcc_grpc.pb.go | 5 +- testdata/mock_tcc.go | 1 + 15 files changed, 308 insertions(+), 17 deletions(-) create mode 100644 pkg/common/util/slice_to_str.go create mode 100644 pkg/constant/client_table_columns_name.go create mode 100644 pkg/constant/undo.go create mode 100644 pkg/datasource/sql/undo_test.go diff --git a/go.mod b/go.mod index da22814d3..f14164372 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index e60b96713..d28d015c4 100644 --- a/go.sum +++ b/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= \ No newline at end of file +vimagination.zapto.org/memio v0.0.0-20200222190306-588ebc67b97d/go.mod h1:zHGDKp2tyvF4IAfLti4pKYqCJucXYmmKMb3UMrCHK/4= diff --git a/pkg/common/util/slice_to_str.go b/pkg/common/util/slice_to_str.go new file mode 100644 index 000000000..03f2d82ac --- /dev/null +++ b/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 +} diff --git a/pkg/constant/client_table_columns_name.go b/pkg/constant/client_table_columns_name.go new file mode 100644 index 000000000..390c38ee2 --- /dev/null +++ b/pkg/constant/client_table_columns_name.go @@ -0,0 +1,44 @@ +/* + * 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 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" +) diff --git a/pkg/constant/undo.go b/pkg/constant/undo.go new file mode 100644 index 000000000..b92254755 --- /dev/null +++ b/pkg/constant/undo.go @@ -0,0 +1,26 @@ +/* + * 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 constant + +const ( + DeleteFrom = "DELETE FROM " + DefaultTransactionUndoLogTable = " undo_log " + // UndoLogTableName Todo get from config + UndoLogTableName = DefaultTransactionUndoLogTable + DeleteUndoLogSql = DeleteFrom + UndoLogTableName + " WHERE " + UndoLogBranchXid + " = ? AND " + UndoLogXid + " = ?" +) diff --git a/pkg/datasource/sql/at.go b/pkg/datasource/sql/at.go index e4bc6f6a8..c3d97bb1b 100644 --- a/pkg/datasource/sql/at.go +++ b/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" @@ -242,7 +243,7 @@ func (w *asyncATWorker) dealWithGroupedContexts(resID string, phaseCtxs []phaseT for i := range phaseCtxs { phaseCtx := phaseCtxs[i] - if err := undoMgr.DeleteUndoLogs([]string{phaseCtx.Xid}, []int64{phaseCtx.BranchID}, conn); err != nil { + if err := undoMgr.BatchDeleteUndoLog([]string{phaseCtx.Xid}, []int64{phaseCtx.BranchID}, conn); err != nil { w.commitQueue <- phaseCtx } } diff --git a/pkg/datasource/sql/db.go b/pkg/datasource/sql/db.go index a6c0c4b83..5957c4e35 100644 --- a/pkg/datasource/sql/db.go +++ b/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" ) diff --git a/pkg/datasource/sql/tx.go b/pkg/datasource/sql/tx.go index 2b8b6c201..7b8c10468 100644 --- a/pkg/datasource/sql/tx.go +++ b/pkg/datasource/sql/tx.go @@ -21,6 +21,8 @@ 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" @@ -28,7 +30,6 @@ import ( "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 diff --git a/pkg/datasource/sql/undo/base/undo.go b/pkg/datasource/sql/undo/base/undo.go index cd7f2d90e..38f2b2997 100644 --- a/pkg/datasource/sql/undo/base/undo.go +++ b/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,52 @@ func (m *BaseUndoLogManager) InsertUndoLog(l []undo.BranchUndoLog, tx driver.Tx) return nil } -// DeleteUndoLog -func (m *BaseUndoLogManager) DeleteUndoLogs(xid []string, branchID []int64, conn *sql.Conn) error { +// DeleteUndoLog exec delete single undo log operate +func (m *BaseUndoLogManager) DeleteUndoLog(ctx context.Context, xid string, branchID int64, conn *sql.Conn) error { + stmt, err := conn.PrepareContext(ctx, constant.DeleteUndoLogSql) + if err != nil { + log.Errorf("[DeleteUndoLog] prepare sql fail, err: %v", err) + return err + } + + if _, err = stmt.ExecContext(ctx, branchID, xid); err != nil { + log.Errorf("[DeleteUndoLog] exec delete undo log fail, err: %v", err) + return err + } + + return nil +} + +// BatchDeleteUndoLog exec delete undo log operate +func (m *BaseUndoLogManager) BatchDeleteUndoLog(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 + if _, err = stmt.ExecContext(ctx, branchIDStr, strings.Join(xid, ",")); err != nil { + log.Errorf("exec delete undo log fail, err: %v", err) + return err + } + return nil } @@ -58,3 +111,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(constant.DeleteFrom) + 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(") ") +} diff --git a/pkg/datasource/sql/undo/mysql/undo.go b/pkg/datasource/sql/undo/mysql/undo.go index eacbf263f..c11d50c7e 100644 --- a/pkg/datasource/sql/undo/mysql/undo.go +++ b/pkg/datasource/sql/undo/mysql/undo.go @@ -18,11 +18,13 @@ package mysql import ( + "context" "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" ) @@ -42,8 +44,13 @@ func (m *undoLogManager) InsertUndoLog(l []undo.BranchUndoLog, tx driver.Tx) err } // DeleteUndoLog -func (m *undoLogManager) DeleteUndoLogs(xid []string, branchID []int64, conn *sql.Conn) error { - return m.Base.DeleteUndoLogs(xid, branchID, conn) +func (m *undoLogManager) DeleteUndoLog(ctx context.Context, xid string, branchID int64, conn *sql.Conn) error { + return m.Base.DeleteUndoLog(ctx, xid, branchID, conn) +} + +// BatchDeleteUndoLog +func (m *undoLogManager) BatchDeleteUndoLog(xid []string, branchID []int64, conn *sql.Conn) error { + return m.Base.BatchDeleteUndoLog(xid, branchID, conn) } // FlushUndoLog diff --git a/pkg/datasource/sql/undo/undo.go b/pkg/datasource/sql/undo/undo.go index 7d291adf2..4ee1707b0 100644 --- a/pkg/datasource/sql/undo/undo.go +++ b/pkg/datasource/sql/undo/undo.go @@ -18,6 +18,7 @@ package undo import ( + "context" "database/sql" "database/sql/driver" "errors" @@ -51,7 +52,9 @@ type UndoLogManager interface { // InsertUndoLog InsertUndoLog(l []BranchUndoLog, tx driver.Tx) error // DeleteUndoLog - DeleteUndoLogs(xid []string, branchID []int64, conn *sql.Conn) error + DeleteUndoLog(ctx context.Context, xid string, branchID int64, conn *sql.Conn) error + // BatchDeleteUndoLog + BatchDeleteUndoLog(xid []string, branchID []int64, conn *sql.Conn) error // FlushUndoLog FlushUndoLog(txCtx *types.TransactionContext, tx driver.Tx) error // RunUndo diff --git a/pkg/datasource/sql/undo_test.go b/pkg/datasource/sql/undo_test.go new file mode 100644 index 000000000..c8baf4c26 --- /dev/null +++ b/pkg/datasource/sql/undo_test.go @@ -0,0 +1,73 @@ +/* + * 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" +) + +// TestBatchDeleteUndoLogs +func TestBatchDeleteUndoLogs(t *testing.T) { + // local test can annotation t.SkipNow() + t.SkipNow() + + testBatchDeleteUndoLogs := func() { + db, err := sql.Open(SeataMySQLDriver, "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.BatchDeleteUndoLog([]string{"1"}, []int64{1}, sqlConn) + assert.Nil(t, err) + } + + t.Run("test_batch_delete_undo_logs", func(t *testing.T) { + testBatchDeleteUndoLogs() + }) +} + +func TestDeleteUndoLogs(t *testing.T) { + // local test can annotation t.SkipNow() + t.SkipNow() + + testDeleteUndoLogs := func() { + db, err := sql.Open(SeataMySQLDriver, "root:12345678@tcp(127.0.0.1:3306)/seata_order?multiStatements=true") + assert.Nil(t, err) + + ctx := context.Background() + sqlConn, err := db.Conn(ctx) + assert.Nil(t, err) + + undoLogManager := new(base.BaseUndoLogManager) + + err = undoLogManager.DeleteUndoLog(ctx, "1", 1, sqlConn) + assert.Nil(t, err) + } + + t.Run("test_delete_undo_logs", func(t *testing.T) { + testDeleteUndoLogs() + }) +} diff --git a/pkg/rm/rm_api_test.go b/pkg/rm/rm_api_test.go index fa1d8419b..3b9931231 100644 --- a/pkg/rm/rm_api_test.go +++ b/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. diff --git a/sample/tcc/grpc/pb/tcc_grpc.pb.go b/sample/tcc/grpc/pb/tcc_grpc.pb.go index a4629e850..087b9cc3f 100644 --- a/sample/tcc/grpc/pb/tcc_grpc.pb.go +++ b/sample/tcc/grpc/pb/tcc_grpc.pb.go @@ -23,12 +23,13 @@ package pb import ( + reflect "reflect" + sync "sync" + protoreflect "google.golang.org/protobuf/reflect/protoreflect" protoimpl "google.golang.org/protobuf/runtime/protoimpl" anypb "google.golang.org/protobuf/types/known/anypb" wrapperspb "google.golang.org/protobuf/types/known/wrapperspb" - reflect "reflect" - sync "sync" ) const ( diff --git a/testdata/mock_tcc.go b/testdata/mock_tcc.go index 683d19dee..d912cff6d 100644 --- a/testdata/mock_tcc.go +++ b/testdata/mock_tcc.go @@ -2,6 +2,7 @@ package testdata import ( "context" + "github.com/seata/seata-go/pkg/tm" )