Repository: apache/incubator-seata-go

Issue:
Improve test coverage for pkg/datasource/sql/undo
- **Package**: `seata.apache.org/seata-go/pkg/datasource/sql/undo`
- **Current Coverage**: 0.0%
- **Target Coverage**: >= 70%
- **Description**: The package currently has low test coverage. We need to add more unit tests to improve code quality and maintainability.
- **Task**:
  1. Analyze the code in this package
  2. Identify code paths that are not covered by tests
  3. Write unit tests to cover those paths
  4. Ensure tests are comprehensive and meaningful

Discussion:
@LucienShen-Liu: plz assign to me.@slievrly: @LucienShen-Liu  I have assigned it to you and look forward to your PR submission.@LucienShen-Liu: > [@LucienShen-Liu](https://github.com/LucienShen-Liu) I have assigned it to you and look forward to your PR submission.

Merged done

Solution patch:
diff --git a/go.mod b/go.mod
index 91dc37d..3c55e2b 100644
--- a/go.mod
+++ b/go.mod
@@ -82,6 +82,7 @@ require (
 	github.com/prometheus/procfs v0.7.3 // indirect
 	github.com/satori/go.uuid v1.2.1-0.20181028125025-b2ce2384e17b // indirect
 	github.com/shirou/gopsutil/v3 v3.22.2 // indirect
+	github.com/stretchr/objx v0.5.0 // indirect
 	github.com/tklauser/go-sysconf v0.3.10 // indirect
 	github.com/tklauser/numcpus v0.4.0 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
diff --git a/go.sum b/go.sum
index abbd1cc..e76a9f0 100644
--- a/go.sum
+++ b/go.sum
@@ -715,6 +715,7 @@ github.com/streadway/handy v0.0.0-20190108123426-d5acb3125c2a/go.mod h1:qNTQ5P5J
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/objx v0.5.0 h1:1zr/of2m5FGMsad5YfcqgdqdWrIhu+EBEJRhR1U7z/c=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
diff --git a/pkg/datasource/sql/undo/base/undo.go b/pkg/datasource/sql/undo/base/undo.go
index 75cb4b2..f52087c 100644
--- a/pkg/datasource/sql/undo/base/undo.go
+++ b/pkg/datasource/sql/undo/base/undo.go
@@ -238,7 +238,7 @@ func (m *BaseUndoLogManager) FlushUndoLog(tranCtx *types.TransactionContext, con
 		XID:          tranCtx.XID,
 		Context:      undoLogContent,
 		RollbackInfo: rollbackInfo,
-		LogStatus:    undo.UndoLogStatueNormnal,
+		LogStatus:    undo.UndoLogStatusNormal,
 	}, conn)
 }
 
@@ -368,13 +368,13 @@ func (m *BaseUndoLogManager) Undo(ctx context.Context, dbType types.DBType, xid
 			log.Errorf("[Undo] delete undo fail, err: %v", err)
 			return err
 		}
-		log.Infof("xid %v branch %v, undo_log deleted with %v", xid, branchID, undo.UndoLogStatueGlobalFinished)
+		log.Infof("xid %v branch %v, undo_log deleted with %v", xid, branchID, undo.UndoLogStatusGlobalFinished)
 	} else {
 		if err = m.insertUndoLogWithGlobalFinished(ctx, xid, uint64(branchID), conn); err != nil {
 			log.Errorf("[Undo] insert undo with global finished fail, err: %v", err)
 			return err
 		}
-		log.Infof("xid %v branch %v, undo_log added with %v", xid, branchID, undo.UndoLogStatueGlobalFinished)
+		log.Infof("xid %v branch %v, undo_log added with %v", xid, branchID, undo.UndoLogStatusGlobalFinished)
 	}
 
 	if err = tx.Commit(); err != nil {
diff --git a/pkg/datasource/sql/undo/undo.go b/pkg/datasource/sql/undo/undo.go
index fac3eab..8e1e5a0 100644
--- a/pkg/datasource/sql/undo/undo.go
+++ b/pkg/datasource/sql/undo/undo.go
@@ -94,11 +94,11 @@ func GetUndoLogManager(d types.DBType) (UndoLogManager, error) {
 	return v.mgr, nil
 }
 
-type UndoLogStatue int8
+type UndoLogStatus int8
 
 const (
-	UndoLogStatueNormnal        UndoLogStatue = 0
-	UndoLogStatueGlobalFinished UndoLogStatue = 1
+	UndoLogStatusNormal         UndoLogStatus = 0
+	UndoLogStatusGlobalFinished UndoLogStatus = 1
 )
 
 type UndologRecord struct {
@@ -106,13 +106,13 @@ type UndologRecord struct {
 	XID          string        `json:"xid"`
 	Context      []byte        `json:"context"`
 	RollbackInfo []byte        `json:"rollbackInfo"`
-	LogStatus    UndoLogStatue `json:"logStatus"`
+	LogStatus    UndoLogStatus `json:"logStatus"`
 	LogCreated   []byte        `json:"logCreated"`
 	LogModified  []byte        `json:"logModified"`
 }
 
 func (u *UndologRecord) CanUndo() bool {
-	return u.LogStatus == UndoLogStatueNormnal
+	return u.LogStatus == UndoLogStatusNormal
 }
 
 // BranchUndoLog



Test patch:
diff --git a/pkg/datasource/sql/undo/base/undo_test.go b/pkg/datasource/sql/undo/base/undo_test.go
index 8370594..10b5115 100644
--- a/pkg/datasource/sql/undo/base/undo_test.go
+++ b/pkg/datasource/sql/undo/base/undo_test.go
@@ -22,42 +22,43 @@ import (
 	"database/sql/driver"
 	"encoding/json"
 	"errors"
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/stretchr/testify/require"
+	"seata.apache.org/seata-go/pkg/datasource/sql/types"
 	"strings"
 	"testing"
 
-	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 
-	"seata.apache.org/seata-go/pkg/datasource/sql/types"
 	"seata.apache.org/seata-go/pkg/datasource/sql/undo"
 )
 
 func TestGetUndoLogTableName(t *testing.T) {
 	tests := []struct {
-		name       string
-		configName string
-		expected   string
+		name     string
+		config   string
+		expected string
 	}{
 		{
-			name:       "default table name",
-			configName: "",
-			expected:   " undo_log ",
+			name:     "default table name",
+			config:   "",
+			expected: " undo_log ",
 		},
 		{
-			name:       "custom table name",
-			configName: "custom_undo_log",
-			expected:   "custom_undo_log",
+			name:     "custom table name",
+			config:   "custom_undo_log",
+			expected: "custom_undo_log",
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			// Save original config and restore after test
-			originalTable := undo.UndoConfig.LogTable
-			defer func() { undo.UndoConfig.LogTable = originalTable }()
+			originalConfig := undo.UndoConfig.LogTable
+			defer func() {
+				undo.UndoConfig.LogTable = originalConfig
+			}()
 
-			undo.UndoConfig.LogTable = tt.configName
+			undo.UndoConfig.LogTable = tt.config
 			result := getUndoLogTableName()
 			assert.Equal(t, tt.expected, result)
 		})
@@ -65,133 +66,95 @@ func TestGetUndoLogTableName(t *testing.T) {
 }
 
 func TestGetCheckUndoLogTableExistSql(t *testing.T) {
-	originalTable := undo.UndoConfig.LogTable
-	defer func() { undo.UndoConfig.LogTable = originalTable }()
-
-	undo.UndoConfig.LogTable = ""
-	sql := getCheckUndoLogTableExistSql()
-	assert.Contains(t, sql, "SELECT 1 FROM")
-	assert.Contains(t, sql, "undo_log")
-	assert.Contains(t, sql, "LIMIT 1")
+	originalConfig := undo.UndoConfig.LogTable
+	defer func() {
+		undo.UndoConfig.LogTable = originalConfig
+	}()
+
+	undo.UndoConfig.LogTable = "test_undo_log"
+	result := getCheckUndoLogTableExistSql()
+	expected := "SELECT 1 FROM test_undo_log LIMIT 1"
+	assert.Equal(t, expected, result)
 }
 
 func TestGetInsertUndoLogSql(t *testing.T) {
-	sql := getInsertUndoLogSql()
-	assert.Contains(t, sql, "INSERT INTO")
-	assert.Contains(t, sql, "branch_id")
-	assert.Contains(t, sql, "xid")
-	assert.Contains(t, sql, "context")
-	assert.Contains(t, sql, "rollback_info")
-	assert.Contains(t, sql, "log_status")
+	originalConfig := undo.UndoConfig.LogTable
+	defer func() {
+		undo.UndoConfig.LogTable = originalConfig
+	}()
+
+	undo.UndoConfig.LogTable = "test_undo_log"
+	result := getInsertUndoLogSql()
+	expected := "INSERT INTO test_undo_log(branch_id,xid,context,rollback_info,log_status,log_created,log_modified) VALUES (?, ?, ?, ?, ?, now(6), now(6))"
+	assert.Equal(t, expected, result)
 }
 
 func TestGetSelectUndoLogSql(t *testing.T) {
-	sql := getSelectUndoLogSql()
-	assert.Contains(t, sql, "SELECT")
-	assert.Contains(t, sql, "branch_id")
-	assert.Contains(t, sql, "xid")
-	assert.Contains(t, sql, "FOR UPDATE")
+	originalConfig := undo.UndoConfig.LogTable
+	defer func() {
+		undo.UndoConfig.LogTable = originalConfig
+	}()
+
+	undo.UndoConfig.LogTable = "test_undo_log"
+	result := getSelectUndoLogSql()
+	expected := "SELECT `branch_id`,`xid`,`context`,`rollback_info`,`log_status` FROM test_undo_log WHERE branch_id = ? AND xid = ? FOR UPDATE"
+	assert.Equal(t, expected, result)
 }
 
 func TestGetDeleteUndoLogSql(t *testing.T) {
-	sql := getDeleteUndoLogSql()
-	assert.Contains(t, sql, "DELETE FROM")
-	assert.Contains(t, sql, "WHERE branch_id = ? AND xid = ?")
+	originalConfig := undo.UndoConfig.LogTable
+	defer func() {
+		undo.UndoConfig.LogTable = originalConfig
+	}()
+
+	undo.UndoConfig.LogTable = "test_undo_log"
+	result := getDeleteUndoLogSql()
+	expected := "DELETE FROM test_undo_log WHERE branch_id = ? AND xid = ?"
+	assert.Equal(t, expected, result)
 }
 
 func TestNewBaseUndoLogManager(t *testing.T) {
 	manager := NewBaseUndoLogManager()
 	assert.NotNil(t, manager)
+	assert.IsType(t, &BaseUndoLogManager{}, manager)
 }
 
 func TestBaseUndoLogManager_Init(t *testing.T) {
 	manager := NewBaseUndoLogManager()
-	// Init should not panic
-	manager.Init()
-}
-
-func TestInt64Slice2Str(t *testing.T) {
-	tests := []struct {
-		name      string
-		values    interface{}
-		sep       string
-		expected  string
-		expectErr bool
-	}{
-		{
-			name:      "valid slice with comma separator",
-			values:    []int64{1, 2, 3},
-			sep:       ",",
-			expected:  "1,2,3",
-			expectErr: false,
-		},
-		{
-			name:      "valid slice with dash separator",
-			values:    []int64{100, 200},
-			sep:       "-",
-			expected:  "100-200",
-			expectErr: false,
-		},
-		{
-			name:      "empty slice",
-			values:    []int64{},
-			sep:       ",",
-			expected:  "",
-			expectErr: false,
-		},
-		{
-			name:      "single element",
-			values:    []int64{42},
-			sep:       ",",
-			expected:  "42",
-			expectErr: false,
-		},
-		{
-			name:      "invalid type",
-			values:    []string{"a", "b"},
-			sep:       ",",
-			expected:  "",
-			expectErr: true,
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			result, err := Int64Slice2Str(tt.values, tt.sep)
-			if tt.expectErr {
-				assert.Error(t, err)
-			} else {
-				assert.NoError(t, err)
-				assert.Equal(t, tt.expected, result)
-			}
-		})
-	}
+	assert.NotPanics(t, func() {
+		manager.Init()
+	})
 }
 
 func TestBaseUndoLogManager_canUndo(t *testing.T) {
 	manager := NewBaseUndoLogManager()
 
 	tests := []struct {
-		name     string
-		state    int32
-		expected bool
+		name  string
+		state int32
+		want  bool
 	}{
 		{
-			name:     "normal status can undo",
-			state:    UndoLogStatusNormal,
-			expected: true,
+			name:  "normal status - can undo",
+			state: UndoLogStatusNormal,
+			want:  true,
 		},
 		{
-			name:     "global finished status cannot undo",
-			state:    UndoLogStatusGlobalFinished,
-			expected: false,
+			name:  "global finished status - cannot undo",
+			state: UndoLogStatusGlobalFinished,
+			want:  false,
+		},
+		{
+			name:  "invalid status",
+			state: 999,
+			want:  false,
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			result := manager.canUndo(tt.state)
-			assert.Equal(t, tt.expected, result)
+			assert.Equal(t, tt.want, result)
 		})
 	}
 }
@@ -200,39 +163,40 @@ func TestBaseUndoLogManager_UnmarshalContext(t *testing.T) {
 	manager := NewBaseUndoLogManager()
 
 	tests := []struct {
-		name      string
-		input     []byte
-		expected  map[string]string
-		expectErr bool
+		name        string
+		input       []byte
+		expected    map[string]string
+		expectError bool
 	}{
 		{
-			name:  "valid json context",
+			name:  "valid json",
 			input: []byte(`{"key1":"value1","key2":"value2"}`),
 			expected: map[string]string{
 				"key1": "value1",
 				"key2": "value2",
 			},
-			expectErr: false,
+			expectError: false,
 		},
 		{
-			name:      "empty json",
-			input:     []byte(`{}`),
-			expected:  map[string]string{},
-			expectErr: false,
+			name:        "invalid json",
+			input:       []byte(`{"key1":value1`),
+			expected:    nil,
+			expectError: true,
 		},
 		{
-			name:      "invalid json",
-			input:     []byte(`{invalid json`),
-			expected:  nil,
-			expectErr: true,
+			name:        "empty json",
+			input:       []byte(`{}`),
+			expected:    map[string]string{},
+			expectError: false,
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			result, err := manager.UnmarshalContext(tt.input)
-			if tt.expectErr {
+			if tt.expectError {
 				assert.Error(t, err)
+				assert.Nil(t, result)
 			} else {
 				assert.NoError(t, err)
 				assert.Equal(t, tt.expected, result)
@@ -249,599 +213,261 @@ func TestBaseUndoLogManager_getSerializer(t *testing.T) {
 		context  map[string]string
 		expected string
 	}{
-		{
-			name: "context with serializer key",
-			context: map[string]string{
-				"serializerKey": "json",
-			},
-			expected: "json",
-		},
 		{
 			name:     "nil context",
 			context:  nil,
 			expected: "",
 		},
-		{
-			name:     "context without serializer key",
-			context:  map[string]string{"other": "value"},
-			expected: "",
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			result := manager.getSerializer(tt.context)
-			assert.Equal(t, tt.expected, result)
-		})
-	}
-}
-
-func TestBaseUndoLogManager_appendInParam(t *testing.T) {
-	manager := NewBaseUndoLogManager()
-
-	tests := []struct {
-		name     string
-		size     int
-		expected string
-	}{
-		{
-			name:     "size 3",
-			size:     3,
-			expected: " (?,?,?) ",
-		},
-		{
-			name:     "size 1",
-			size:     1,
-			expected: " (?) ",
-		},
-		{
-			name:     "size 0",
-			size:     0,
-			expected: "",
-		},
-		{
-			name:     "negative size",
-			size:     -1,
-			expected: "",
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			var sb strings.Builder
-			manager.appendInParam(tt.size, &sb)
-			assert.Equal(t, tt.expected, sb.String())
-		})
-	}
-}
-
-func TestBaseUndoLogManager_getBatchDeleteUndoLogSql(t *testing.T) {
-	manager := NewBaseUndoLogManager()
-
-	tests := []struct {
-		name      string
-		xid       []string
-		branchID  []int64
-		expectErr bool
-	}{
-		{
-			name:      "valid input",
-			xid:       []string{"xid1", "xid2"},
-			branchID:  []int64{1, 2},
-			expectErr: false,
-		},
-		{
-			name:      "empty xid",
-			xid:       []string{},
-			branchID:  []int64{1},
-			expectErr: true,
-		},
-		{
-			name:      "empty branchID",
-			xid:       []string{"xid1"},
-			branchID:  []int64{},
-			expectErr: true,
-		},
-		{
-			name:      "both empty",
-			xid:       []string{},
-			branchID:  []int64{},
-			expectErr: true,
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			result, err := manager.getBatchDeleteUndoLogSql(tt.xid, tt.branchID)
-			if tt.expectErr {
-				assert.Error(t, err)
-			} else {
-				assert.NoError(t, err)
-				assert.Contains(t, result, "DELETE FROM")
-				assert.Contains(t, result, "branch_id IN")
-				assert.Contains(t, result, "xid IN")
-			}
-		})
-	}
-}
-
-func TestBaseUndoLogManager_encodeDecodeUndoLogCtx(t *testing.T) {
-	manager := NewBaseUndoLogManager()
-
-	tests := []struct {
-		name  string
-		input map[string]string
-	}{
-		{
-			name: "simple context",
-			input: map[string]string{
-				"key1": "value1",
-				"key2": "value2",
-			},
-		},
-		{
-			name:  "empty context",
-			input: map[string]string{},
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			encoded := manager.encodeUndoLogCtx(tt.input)
-			decoded := manager.decodeUndoLogCtx(encoded)
-			assert.Equal(t, tt.input, decoded)
-		})
-	}
-}
-
-func TestBaseUndoLogManager_DBType(t *testing.T) {
-	manager := NewBaseUndoLogManager()
-
-	defer func() {
-		if r := recover(); r == nil {
-			t.Errorf("DBType should panic")
-		}
-	}()
-
-	manager.DBType()
-}
-
-func TestBaseUndoLogManager_InsertUndoLogWithSqlConn(t *testing.T) {
-	db, mock, err := sqlmock.New()
-	require.NoError(t, err)
-	defer db.Close()
-
-	ctx := context.Background()
-	conn, err := db.Conn(ctx)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	manager := NewBaseUndoLogManager()
-
-	record := undo.UndologRecord{
-		BranchID:     123,
-		XID:          "test-xid",
-		Context:      []byte("test-context"),
-		RollbackInfo: []byte("test-rollback"),
-		LogStatus:    undo.UndoLogStatueNormnal,
-	}
-
-	t.Run("successful insert", func(t *testing.T) {
-		mock.ExpectPrepare("INSERT INTO").
-			ExpectExec().
-			WithArgs(record.BranchID, record.XID, record.Context, record.RollbackInfo, int64(record.LogStatus)).
-			WillReturnResult(sqlmock.NewResult(1, 1))
-
-		err := manager.InsertUndoLogWithSqlConn(ctx, record, conn)
-		assert.NoError(t, err)
-		assert.NoError(t, mock.ExpectationsWereMet())
-	})
-}
-
-func TestBaseUndoLogManager_DeleteUndoLog(t *testing.T) {
-	db, mock, err := sqlmock.New()
-	require.NoError(t, err)
-	defer db.Close()
-
-	ctx := context.Background()
-	conn, err := db.Conn(ctx)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	manager := NewBaseUndoLogManager()
-
-	t.Run("successful delete", func(t *testing.T) {
-		mock.ExpectPrepare("DELETE FROM").
-			ExpectExec().
-			WithArgs(int64(123), "test-xid").
-			WillReturnResult(sqlmock.NewResult(0, 1))
-
-		err := manager.DeleteUndoLog(ctx, "test-xid", 123, conn)
-		assert.NoError(t, err)
-		assert.NoError(t, mock.ExpectationsWereMet())
-	})
-
-	t.Run("delete with prepare error", func(t *testing.T) {
-		mock.ExpectPrepare("DELETE FROM").
-			WillReturnError(errors.New("prepare error"))
-
-		err := manager.DeleteUndoLog(ctx, "test-xid", 123, conn)
-		assert.Error(t, err)
-	})
-}
-
-func TestBaseUndoLogManager_FlushUndoLog(t *testing.T) {
-	manager := NewBaseUndoLogManager()
-
-	t.Run("empty round images", func(t *testing.T) {
-		tranCtx := &types.TransactionContext{
-			RoundImages: &types.RoundRecordImage{},
-		}
-
-		mockConn := &mockDriverConn{}
-		err := manager.FlushUndoLog(tranCtx, mockConn)
-		assert.NoError(t, err)
-	})
-
-	t.Run("empty before and after images", func(t *testing.T) {
-		roundImages := &types.RoundRecordImage{}
-		tranCtx := &types.TransactionContext{
-			XID:         "test-xid",
-			BranchID:    123,
-			RoundImages: roundImages,
-		}
-
-		mockConn := &mockDriverConn{}
-		err := manager.FlushUndoLog(tranCtx, mockConn)
-		assert.NoError(t, err)
-	})
-}
-
-func TestBaseUndoLogManager_RunUndo(t *testing.T) {
-	manager := NewBaseUndoLogManager()
-	db, _, err := sqlmock.New()
-	require.NoError(t, err)
-	defer db.Close()
-
-	err = manager.RunUndo(context.Background(), "test-xid", 123, db, "test_db")
-	assert.NoError(t, err)
-}
-
-func TestBaseUndoLogManager_getRollbackInfo(t *testing.T) {
-	manager := NewBaseUndoLogManager()
-
-	t.Run("without compression", func(t *testing.T) {
-		input := []byte("test data")
-		context := map[string]string{}
-
-		result, err := manager.getRollbackInfo(input, context)
-		assert.NoError(t, err)
-		assert.Equal(t, input, result)
-	})
-}
-
-// Mock implementations for testing
-
-type mockDriverConn struct {
-	driver.Conn
-	prepareFunc func(query string) (driver.Stmt, error)
-}
-
-func (m *mockDriverConn) Prepare(query string) (driver.Stmt, error) {
-	if m.prepareFunc != nil {
-		return m.prepareFunc(query)
-	}
-	return &mockDriverStmt{}, nil
-}
-
-func (m *mockDriverConn) Begin() (driver.Tx, error) {
-	return &mockDriverTx{}, nil
-}
-
-func (m *mockDriverConn) Close() error {
-	return nil
-}
-
-type mockDriverStmt struct {
-	driver.Stmt
-	execFunc func(args []driver.Value) (driver.Result, error)
-}
-
-func (m *mockDriverStmt) Close() error {
-	return nil
-}
-
-func (m *mockDriverStmt) NumInput() int {
-	return -1
-}
-
-func (m *mockDriverStmt) Exec(args []driver.Value) (driver.Result, error) {
-	if m.execFunc != nil {
-		return m.execFunc(args)
-	}
-	return &mockDriverResult{}, nil
-}
-
-func (m *mockDriverStmt) Query(args []driver.Value) (driver.Rows, error) {
-	return nil, nil
-}
-
-type mockDriverResult struct {
-	driver.Result
-}
-
-func (m *mockDriverResult) LastInsertId() (int64, error) {
-	return 0, nil
-}
-
-func (m *mockDriverResult) RowsAffected() (int64, error) {
-	return 1, nil
-}
-
-type mockDriverTx struct {
-	driver.Tx
-}
-
-func (m *mockDriverTx) Commit() error {
-	return nil
-}
-
-func (m *mockDriverTx) Rollback() error {
-	return nil
-}
-
-func TestUndoLogStatusConstants(t *testing.T) {
-	assert.Equal(t, int(0), UndoLogStatusNormal)
-	assert.Equal(t, int(1), UndoLogStatusGlobalFinished)
-}
-
-func TestBaseUndoLogManager_serializeBranchUndoLog(t *testing.T) {
-	manager := NewBaseUndoLogManager()
-
-	branchLog := &undo.BranchUndoLog{
-		Xid:      "test-xid",
-		BranchID: 123,
-		Logs:     []undo.SQLUndoLog{},
-	}
-
-	t.Run("with json serializer", func(t *testing.T) {
-		result, err := manager.serializeBranchUndoLog(branchLog, "json")
-		if err == nil {
-			assert.NotNil(t, result)
-		}
-	})
-}
-
-func TestBaseUndoLogManager_deserializeBranchUndoLog(t *testing.T) {
-	manager := NewBaseUndoLogManager()
-
-	branchLog := &undo.BranchUndoLog{
-		Xid:      "test-xid",
-		BranchID: 123,
-		Logs:     []undo.SQLUndoLog{},
-	}
-
-	// Serialize first
-	data, err := json.Marshal(branchLog)
-	require.NoError(t, err)
-
-	t.Run("deserialize with context", func(t *testing.T) {
-		logCtx := map[string]string{
-			"serializerKey": "json",
-		}
-
-		result, err := manager.deserializeBranchUndoLog(data, logCtx)
-		if err == nil {
-			assert.NotNil(t, result)
-		}
-	})
-}
-
-func TestBaseUndoLogManager_InsertUndoLog(t *testing.T) {
-	manager := NewBaseUndoLogManager()
-
-	record := undo.UndologRecord{
-		BranchID:     123,
-		XID:          "test-xid",
-		Context:      []byte("test-context"),
-		RollbackInfo: []byte("test-rollback"),
-		LogStatus:    undo.UndoLogStatueNormnal,
-	}
-
-	t.Run("with mock conn - prepare error", func(t *testing.T) {
-		mockConn := &mockDriverConn{
-			prepareFunc: func(query string) (driver.Stmt, error) {
-				return nil, errors.New("prepare error")
-			},
-		}
-
-		err := manager.InsertUndoLog(record, mockConn)
-		assert.Error(t, err)
-	})
-
-	t.Run("with mock conn - exec error", func(t *testing.T) {
-		mockConn := &mockDriverConn{
-			prepareFunc: func(query string) (driver.Stmt, error) {
-				return &mockDriverStmt{
-					execFunc: func(args []driver.Value) (driver.Result, error) {
-						return nil, errors.New("exec error")
-					},
-				}, nil
-			},
-		}
-
-		err := manager.InsertUndoLog(record, mockConn)
-		assert.Error(t, err)
-	})
-
-	t.Run("with mock conn - success", func(t *testing.T) {
-		mockConn := &mockDriverConn{
-			prepareFunc: func(query string) (driver.Stmt, error) {
-				return &mockDriverStmt{
-					execFunc: func(args []driver.Value) (driver.Result, error) {
-						return &mockDriverResult{}, nil
-					},
-				}, nil
+		{
+			name:     "empty context",
+			context:  map[string]string{},
+			expected: "",
+		},
+		{
+			name: "context with serializer",
+			context: map[string]string{
+				"serializerKey": "json",
 			},
-		}
+			expected: "json",
+		},
+		{
+			name: "context without serializer",
+			context: map[string]string{
+				"otherKey": "value",
+			},
+			expected: "",
+		},
+	}
 
-		err := manager.InsertUndoLog(record, mockConn)
-		assert.NoError(t, err)
-	})
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := manager.getSerializer(tt.context)
+			assert.Equal(t, tt.expected, result)
+		})
+	}
 }
 
-func TestBaseUndoLogManager_FlushUndoLog_WithImages(t *testing.T) {
+func TestBaseUndoLogManager_getBatchDeleteUndoLogSql(t *testing.T) {
 	manager := NewBaseUndoLogManager()
 
-	t.Run("with before and after images", func(t *testing.T) {
-		// Create round images with data
-		roundImages := &types.RoundRecordImage{}
-
-		// Add a before image
-		beforeImage := &types.RecordImage{
-			TableName: "test_table",
-			SQLType:   types.SQLTypeInsert,
-			Rows:      []types.RowImage{},
-		}
-		roundImages.AppendBeofreImage(beforeImage)
-
-		// Add an after image
-		afterImage := &types.RecordImage{
-			TableName: "test_table",
-			SQLType:   types.SQLTypeInsert,
-			Rows:      []types.RowImage{},
-		}
-		roundImages.AppendAfterImage(afterImage)
-
-		tranCtx := &types.TransactionContext{
-			XID:         "test-xid",
-			BranchID:    123,
-			RoundImages: roundImages,
-		}
-
-		mockConn := &mockDriverConn{
-			prepareFunc: func(query string) (driver.Stmt, error) {
-				return &mockDriverStmt{
-					execFunc: func(args []driver.Value) (driver.Result, error) {
-						return &mockDriverResult{}, nil
-					},
-				}, nil
-			},
-		}
-
-		// Save original config
-		originalSerialization := undo.UndoConfig.LogSerialization
-		originalCompressType := undo.UndoConfig.CompressConfig.Type
-		defer func() {
-			undo.UndoConfig.LogSerialization = originalSerialization
-			undo.UndoConfig.CompressConfig.Type = originalCompressType
-		}()
+	originalConfig := undo.UndoConfig.LogTable
+	defer func() {
+		undo.UndoConfig.LogTable = originalConfig
+	}()
+	undo.UndoConfig.LogTable = "test_undo_log"
 
-		undo.UndoConfig.LogSerialization = "json"
-		undo.UndoConfig.CompressConfig.Type = "none"
+	tests := []struct {
+		name        string
+		xid         []string
+		branchID    []int64
+		expectError bool
+		expected    string
+	}{
+		{
+			name:        "empty xid",
+			xid:         []string{},
+			branchID:    []int64{1, 2},
+			expectError: true,
+		},
+		{
+			name:        "empty branchID",
+			xid:         []string{"xid1", "xid2"},
+			branchID:    []int64{},
+			expectError: true,
+		},
+		{
+			name:        "both empty",
+			xid:         []string{},
+			branchID:    []int64{},
+			expectError: true,
+		},
+		{
+			name:     "valid inputs",
+			xid:      []string{"xid1", "xid2"},
+			branchID: []int64{1, 2},
+			expected: " DELETE FROM test_undo_log WHERE branch_id IN  (?,?)  AND xid IN  (?,?) ",
+		},
+		{
+			name:     "single values",
+			xid:      []string{"xid1"},
+			branchID: []int64{1},
+			expected: " DELETE FROM test_undo_log WHERE branch_id IN  (?)  AND xid IN  (?) ",
+		},
+	}
 
-		err := manager.FlushUndoLog(tranCtx, mockConn)
-		// Error is expected due to serialization issues, but we're testing the code path
-		if err != nil {
-			assert.Error(t, err)
-		}
-	})
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, err := manager.getBatchDeleteUndoLogSql(tt.xid, tt.branchID)
+			if tt.expectError {
+				assert.Error(t, err)
+				assert.Contains(t, err.Error(), "xid or branch_id can't nil")
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.expected, result)
+			}
+		})
+	}
 }
 
-func TestBaseUndoLogManager_getRollbackInfo_WithCompression(t *testing.T) {
+func TestBaseUndoLogManager_appendInParam(t *testing.T) {
 	manager := NewBaseUndoLogManager()
 
-	t.Run("with none compression type", func(t *testing.T) {
-		input := []byte("test data")
-		context := map[string]string{
-			"compressorTypeKey": "none",
-		}
+	tests := []struct {
+		name     string
+		size     int
+		expected string
+	}{
+		{
+			name:     "zero size",
+			size:     0,
+			expected: "",
+		},
+		{
+			name:     "negative size",
+			size:     -1,
+			expected: "",
+		},
+		{
+			name:     "single parameter",
+			size:     1,
+			expected: " (?) ",
+		},
+		{
+			name:     "multiple parameters",
+			size:     3,
+			expected: " (?,?,?) ",
+		},
+		{
+			name:     "five parameters",
+			size:     5,
+			expected: " (?,?,?,?,?) ",
+		},
+	}
 
-		result, err := manager.getRollbackInfo(input, context)
-		assert.NoError(t, err)
-		assert.Equal(t, input, result)
-	})
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var builder strings.Builder
+			manager.appendInParam(tt.size, &builder)
+			result := builder.String()
+			assert.Equal(t, tt.expected, result)
+		})
+	}
 }
 
-func TestBaseUndoLogManager_BatchDeleteUndoLog(t *testing.T) {
-	db, mock, err := sqlmock.New()
-	require.NoError(t, err)
-	defer db.Close()
+func TestInt64Slice2Str(t *testing.T) {
+	tests := []struct {
+		name        string
+		values      interface{}
+		sep         string
+		expected    string
+		expectError bool
+	}{
+		{
+			name:     "valid int64 slice",
+			values:   []int64{1, 2, 3, 4, 5},
+			sep:      ",",
+			expected: "1,2,3,4,5",
+		},
+		{
+			name:     "single value",
+			values:   []int64{42},
+			sep:      ",",
+			expected: "42",
+		},
+		{
+			name:     "empty slice",
+			values:   []int64{},
+			sep:      ",",
+			expected: "",
+		},
+		{
+			name:     "different separator",
+			values:   []int64{1, 2, 3},
+			sep:      "|",
+			expected: "1|2|3",
+		},
+		{
+			name:        "invalid type",
+			values:      []string{"1", "2", "3"},
+			sep:         ",",
+			expectError: true,
+		},
+		{
+			name:        "nil value",
+			values:      nil,
+			sep:         ",",
+			expectError: true,
+		},
+	}
 
-	ctx := context.Background()
-	conn, err := db.Conn(ctx)
-	require.NoError(t, err)
-	defer conn.Close()
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, err := Int64Slice2Str(tt.values, tt.sep)
+			if tt.expectError {
+				assert.Error(t, err)
+				assert.Contains(t, err.Error(), "param type is fault")
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.expected, result)
+			}
+		})
+	}
+}
 
+func TestBaseUndoLogManager_encodeDecodeUndoLogCtx(t *testing.T) {
 	manager := NewBaseUndoLogManager()
 
-	t.Run("successful batch delete", func(t *testing.T) {
-		xids := []string{"xid1", "xid2"}
-		branchIDs := []int64{100, 200}
-
-		mock.ExpectPrepare("DELETE FROM").
-			ExpectExec().
-			WithArgs("100,200", "xid1,xid2").
-			WillReturnResult(sqlmock.NewResult(0, 2))
-
-		err := manager.BatchDeleteUndoLog(xids, branchIDs, conn)
-		assert.NoError(t, err)
-	})
-
-	t.Run("prepare error", func(t *testing.T) {
-		xids := []string{"xid1"}
-		branchIDs := []int64{100}
-
-		mock.ExpectPrepare("DELETE FROM").
-			WillReturnError(errors.New("prepare failed"))
-
-		err := manager.BatchDeleteUndoLog(xids, branchIDs, conn)
-		assert.Error(t, err)
-	})
-
-	t.Run("exec error", func(t *testing.T) {
-		xids := []string{"xid1"}
-		branchIDs := []int64{100}
+	testContext := map[string]string{
+		"key1": "value1",
+		"key2": "value2",
+		"key3": "value3",
+	}
 
-		mock.ExpectPrepare("DELETE FROM").
-			ExpectExec().
-			WillReturnError(errors.New("exec failed"))
+	encoded := manager.encodeUndoLogCtx(testContext)
+	assert.NotNil(t, encoded)
 
-		err := manager.BatchDeleteUndoLog(xids, branchIDs, conn)
-		assert.Error(t, err)
-	})
+	decoded := manager.decodeUndoLogCtx(encoded)
+	assert.Equal(t, testContext, decoded)
 }
 
-func TestCanUndoLogRecord(t *testing.T) {
+func TestBaseUndoLogManager_getRollbackInfo(t *testing.T) {
+	manager := NewBaseUndoLogManager()
+
 	tests := []struct {
-		name     string
-		record   undo.UndologRecord
-		expected bool
+		name           string
+		rollbackInfo   []byte
+		undoContext    map[string]string
+		expectedOutput []byte
+		expectError    bool
 	}{
 		{
-			name: "can undo - normal status",
-			record: undo.UndologRecord{
-				LogStatus: undo.UndoLogStatueNormnal,
-			},
-			expected: true,
+			name:           "no compression",
+			rollbackInfo:   []byte("test data"),
+			undoContext:    map[string]string{},
+			expectedOutput: []byte("test data"),
 		},
 		{
-			name: "cannot undo - global finished",
-			record: undo.UndologRecord{
-				LogStatus: undo.UndoLogStatueGlobalFinished,
+			name:         "context without compressor",
+			rollbackInfo: []byte("test data"),
+			undoContext: map[string]string{
+				"otherKey": "value",
 			},
-			expected: false,
+			expectedOutput: []byte("test data"),
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			result := tt.record.CanUndo()
-			assert.Equal(t, tt.expected, result)
+			result, err := manager.getRollbackInfo(tt.rollbackInfo, tt.undoContext)
+			if tt.expectError {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.expectedOutput, result)
+			}
 		})
 	}
 }
-
 func TestBaseUndoLogManager_InsertUndoLogWithSqlConn_Errors(t *testing.T) {
 	db, mock, err := sqlmock.New()
 	require.NoError(t, err)
@@ -859,7 +485,7 @@ func TestBaseUndoLogManager_InsertUndoLogWithSqlConn_Errors(t *testing.T) {
 		XID:          "test-xid",
 		Context:      []byte("test-context"),
 		RollbackInfo: []byte("test-rollback"),
-		LogStatus:    undo.UndoLogStatueNormnal,
+		LogStatus:    undo.UndoLogStatusNormal,
 	}
 
 	t.Run("prepare error", func(t *testing.T) {
@@ -1699,3 +1325,85 @@ func TestBaseUndoLogManager_HasUndoLogTable(t *testing.T) {
 		assert.NoError(t, mock.ExpectationsWereMet())
 	})
 }
+
+// Mock implementations for driver interfaces
+type mockDriverConn struct {
+	prepareFunc func(query string) (driver.Stmt, error)
+	beginFunc   func() (driver.Tx, error)
+	closeFunc   func() error
+}
+
+func (m *mockDriverConn) Prepare(query string) (driver.Stmt, error) {
+	if m.prepareFunc != nil {
+		return m.prepareFunc(query)
+	}
+	return nil, errors.New("prepare not implemented")
+}
+
+func (m *mockDriverConn) Close() error {
+	if m.closeFunc != nil {
+		return m.closeFunc()
+	}
+	return nil
+}
+
+func (m *mockDriverConn) Begin() (driver.Tx, error) {
+	if m.beginFunc != nil {
+		return m.beginFunc()
+	}
+	return nil, errors.New("begin not implemented")
+}
+
+type mockDriverStmt struct {
+	closeFunc    func() error
+	numInputFunc func() int
+	execFunc     func(args []driver.Value) (driver.Result, error)
+	queryFunc    func(args []driver.Value) (driver.Rows, error)
+}
+
+func (m *mockDriverStmt) Close() error {
+	if m.closeFunc != nil {
+		return m.closeFunc()
+	}
+	return nil
+}
+
+func (m *mockDriverStmt) NumInput() int {
+	if m.numInputFunc != nil {
+		return m.numInputFunc()
+	}
+	return -1
+}
+
+func (m *mockDriverStmt) Exec(args []driver.Value) (driver.Result, error) {
+	if m.execFunc != nil {
+		return m.execFunc(args)
+	}
+	return nil, errors.New("exec not implemented")
+}
+
+func (m *mockDriverStmt) Query(args []driver.Value) (driver.Rows, error) {
+	if m.queryFunc != nil {
+		return m.queryFunc(args)
+	}
+	return nil, errors.New("query not implemented")
+}
+
+type mockDriverResult struct {
+	lastInsertIdFunc func() (int64, error)
+	rowsAffectedFunc func() (int64, error)
+}
+
+func (m *mockDriverResult) LastInsertId() (int64, error) {
+	if m.lastInsertIdFunc != nil {
+		return m.lastInsertIdFunc()
+	}
+	return 0, nil
+}
+
+func (m *mockDriverResult) RowsAffected() (int64, error) {
+	if m.rowsAffectedFunc != nil {
+		return m.rowsAffectedFunc()
+	}
+	return 1, nil
+}
diff --git a/pkg/datasource/sql/undo/builder/mysql_delete_undo_log_builder_test.go b/pkg/datasource/sql/undo/builder/mysql_delete_undo_log_builder_test.go
index 1dee724..e70c370 100644
--- a/pkg/datasource/sql/undo/builder/mysql_delete_undo_log_builder_test.go
+++ b/pkg/datasource/sql/undo/builder/mysql_delete_undo_log_builder_test.go
@@ -18,12 +18,16 @@
 package builder
 
 import (
+	"context"
 	"database/sql/driver"
+	"fmt"
+	"strings"
 	"testing"
 
-	"seata.apache.org/seata-go/pkg/util/log"
-
 	"github.com/stretchr/testify/assert"
+
+	"seata.apache.org/seata-go/pkg/datasource/sql/types"
+	"seata.apache.org/seata-go/pkg/util/log"
 )
 
 func TestBuildDeleteBeforeImageSQL(t *testing.T) {
@@ -72,3 +76,60 @@ func TestBuildDeleteBeforeImageSQL(t *testing.T) {
 		})
 	}
 }
+
+func TestGetMySQLDeleteUndoLogBuilder(t *testing.T) {
+	builder := GetMySQLDeleteUndoLogBuilder()
+	assert.NotNil(t, builder)
+	assert.IsType(t, &MySQLDeleteUndoLogBuilder{}, builder)
+}
+
+func TestMySQLDeleteUndoLogBuilder_GetExecutorType(t *testing.T) {
+	builder := &MySQLDeleteUndoLogBuilder{}
+	executorType := builder.GetExecutorType()
+	assert.Equal(t, types.DeleteExecutor, executorType)
+}
+
+func TestMySQLDeleteUndoLogBuilder_BeforeImage(t *testing.T) {
+	log.Init()
+	builder := &MySQLDeleteUndoLogBuilder{}
+
+	// Test with nil values - should handle NamedValues
+	execCtx := &types.ExecContext{
+		Values: nil,
+		NamedValues: []driver.NamedValue{
+			{Name: "0", Value: 100},
+		},
+		Query: "DELETE FROM t_user WHERE id = ?",
+		Conn:  nil,
+	}
+
+	// Since conn is nil, this should panic or return error, let's catch the panic
+	defer func() {
+		if r := recover(); r != nil {
+			// Expected panic due to nil conn or index error
+			panicStr := fmt.Sprintf("%v", r)
+			assert.True(t,
+				strings.Contains(panicStr, "nil pointer") ||
+					strings.Contains(panicStr, "index out of range"),
+				"Expected panic related to nil pointer or index error, got: %s", panicStr)
+		}
+	}()
+
+	_, err := builder.BeforeImage(context.Background(), execCtx)
+	// If we reach here, there should be an error
+	if err == nil {
+		t.Error("Expected error or panic when conn is nil")
+	}
+}
+
+func TestMySQLDeleteUndoLogBuilder_AfterImage(t *testing.T) {
+	builder := &MySQLDeleteUndoLogBuilder{}
+
+	execCtx := &types.ExecContext{}
+	beforeImages := []*types.RecordImage{}
+
+	images, err := builder.AfterImage(context.Background(), execCtx, beforeImages)
+	// AfterImage for DELETE should return nil
+	assert.NoError(t, err)
+	assert.Nil(t, images)
+}
diff --git a/pkg/datasource/sql/undo/builder/mysql_insert_undo_log_builder_test.go b/pkg/datasource/sql/undo/builder/mysql_insert_undo_log_builder_test.go
index 0eac4d4..3fe0b59 100644
--- a/pkg/datasource/sql/undo/builder/mysql_insert_undo_log_builder_test.go
+++ b/pkg/datasource/sql/undo/builder/mysql_insert_undo_log_builder_test.go
@@ -1039,3 +1039,42 @@ func TestMySQLInsertUndoLogBuilder_autoGeneratePks(t *testing.T) {
 		})
 	}
 }
+
+func TestGetMySQLInsertUndoLogBuilder(t *testing.T) {
+	builder := GetMySQLInsertUndoLogBuilder()
+	assert.NotNil(t, builder)
+	assert.IsType(t, &MySQLInsertUndoLogBuilder{}, builder)
+}
+
+func TestMySQLInsertUndoLogBuilder_GetExecutorType(t *testing.T) {
+	builder := &MySQLInsertUndoLogBuilder{}
+	executorType := builder.GetExecutorType()
+	assert.Equal(t, types.InsertExecutor, executorType)
+}
+
+func TestMySQLInsertUndoLogBuilder_BeforeImage(t *testing.T) {
+	builder := &MySQLInsertUndoLogBuilder{}
+
+	execCtx := &types.ExecContext{}
+
+	images, err := builder.BeforeImage(context.Background(), execCtx)
+	// BeforeImage for INSERT should return empty images
+	assert.NoError(t, err)
+	assert.Len(t, images, 0)
+}
+
+func TestMySQLInsertUndoLogBuilder_AfterImage(t *testing.T) {
+	builder := &MySQLInsertUndoLogBuilder{}
+
+	// Test with nil conn - should return empty result
+	execCtx := &types.ExecContext{
+		Values: []driver.Value{100, "test"},
+		Query:  "INSERT INTO t_user (id, name) VALUES (?, ?)",
+		Conn:   nil,
+	}
+	beforeImages := []*types.RecordImage{}
+
+	images, err := builder.AfterImage(context.Background(), execCtx, beforeImages)
+	assert.NoError(t, err)
+	assert.Len(t, images, 0)
+}
diff --git a/pkg/datasource/sql/undo/builder/mysql_multi_undo_log_builder_test.go b/pkg/datasource/sql/undo/builder/mysql_multi_undo_log_builder_test.go
index c2f653d..156f1f7 100644
--- a/pkg/datasource/sql/undo/builder/mysql_multi_undo_log_builder_test.go
+++ b/pkg/datasource/sql/undo/builder/mysql_multi_undo_log_builder_test.go
@@ -16,3 +16,211 @@
  */
 
 package builder
+
+import (
+	"context"
+	"database/sql/driver"
+	"fmt"
+	"testing"
+
+	"github.com/arana-db/parser/ast"
+	"github.com/arana-db/parser/model"
+	"github.com/stretchr/testify/assert"
+
+	"seata.apache.org/seata-go/pkg/datasource/sql/types"
+)
+
+func TestGetMySQLMultiDeleteUndoLogBuilder(t *testing.T) {
+	builder := GetMySQLMultiDeleteUndoLogBuilder()
+	assert.NotNil(t, builder)
+	assert.IsType(t, &MySQLMultiDeleteUndoLogBuilder{}, builder)
+}
+
+func TestMySQLMultiDeleteUndoLogBuilder_GetExecutorType(t *testing.T) {
+	builder := &MySQLMultiDeleteUndoLogBuilder{}
+	executorType := builder.GetExecutorType()
+	assert.Equal(t, types.MultiDeleteExecutor, executorType)
+}
+
+func TestMySQLMultiDeleteUndoLogBuilder_AfterImage(t *testing.T) {
+	builder := &MySQLMultiDeleteUndoLogBuilder{}
+
+	execCtx := &types.ExecContext{}
+	beforeImages := []*types.RecordImage{}
+
+	images, err := builder.AfterImage(context.Background(), execCtx, beforeImages)
+	// AfterImage for multi DELETE should return nil
+	assert.NoError(t, err)
+	assert.Nil(t, images)
+}
+
+func TestMySQLMultiDeleteUndoLogBuilder_BeforeImage_SingleDelete(t *testing.T) {
+	builder := &MySQLMultiDeleteUndoLogBuilder{}
+
+	// Test with single delete - should delegate to regular delete builder
+	// We expect this to fail due to nil conn, but we're testing the delegation logic
+	execCtx := &types.ExecContext{
+		Query:  "DELETE FROM t_user WHERE id = ?",
+		Values: []driver.Value{100},
+	}
+
+	// This should delegate to single delete builder (based on query splitting logic)
+	defer func() {
+		if r := recover(); r != nil {
+			// Expected panic due to nil structures - this confirms delegation happened
+			assert.Contains(t, fmt.Sprintf("%v", r), "nil pointer")
+		}
+	}()
+
+	_, err := builder.BeforeImage(context.Background(), execCtx)
+	// If we reach here without panic, there should be an error
+	if err == nil {
+		t.Error("Expected error or panic due to incomplete context")
+	}
+}
+
+func TestGetMySQLMultiUpdateUndoLogBuilder(t *testing.T) {
+	builder := GetMySQLMultiUpdateUndoLogBuilder()
+	assert.NotNil(t, builder)
+	assert.IsType(t, &MySQLMultiUpdateUndoLogBuilder{}, builder)
+}
+
+func TestMySQLMultiUpdateUndoLogBuilder_GetExecutorType(t *testing.T) {
+	builder := &MySQLMultiUpdateUndoLogBuilder{}
+	executorType := builder.GetExecutorType()
+	assert.Equal(t, types.UpdateExecutor, executorType)
+}
+
+func TestMySQLMultiUpdateUndoLogBuilder_AfterImage(t *testing.T) {
+	builder := &MySQLMultiUpdateUndoLogBuilder{}
+
+	// Test with nil conn - expect panic that we can catch
+	execCtx := &types.ExecContext{
+		Conn: nil,
+		ParseContext: &types.ParseContext{
+			UpdateStmt: &ast.UpdateStmt{
+				TableRefs: &ast.TableRefsClause{
+					TableRefs: &ast.Join{
+						Left: &ast.TableSource{
+							Source: &ast.TableName{
+								Name: model.CIStr{O: "t_user"},
+							},
+						},
+					},
+				},
+			},
+		},
+		MetaDataMap: map[string]types.TableMeta{
+			"t_user": {
+				TableName: "t_user",
+				Indexs: map[string]types.IndexMeta{
+					"id": {
+						IType: types.IndexTypePrimaryKey,
+						Columns: []types.ColumnMeta{
+							{ColumnName: "id"},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	beforeImages := []*types.RecordImage{
+		{
+			TableName: "t_user",
+			Rows: []types.RowImage{
+				{
+					Columns: []types.ColumnImage{
+						{
+							ColumnName: "id",
+							Value:      100,
+						},
+					},
+				},
+			},
+		},
+	}
+
+	// Expect panic due to nil conn
+	defer func() {
+		if r := recover(); r != nil {
+			assert.Contains(t, fmt.Sprintf("%v", r), "nil pointer")
+		}
+	}()
+
+	_, err := builder.AfterImage(context.Background(), execCtx, beforeImages)
+	// If we reach here without panic, there should be an error
+	if err == nil {
+		t.Error("Expected error or panic due to nil conn")
+	}
+}
+
+func TestMySQLMultiUpdateUndoLogBuilder_buildAfterImageSQL(t *testing.T) {
+	builder := &MySQLMultiUpdateUndoLogBuilder{}
+
+	beforeImage := &types.RecordImage{
+		TableName: "t_user",
+		Rows: []types.RowImage{
+			{
+				Columns: []types.ColumnImage{
+					{
+						ColumnName: "id",
+						Value:      100,
+					},
+				},
+			},
+		},
+	}
+
+	meta := types.TableMeta{
+		TableName: "t_user",
+		Indexs: map[string]types.IndexMeta{
+			"id": {
+				IType: types.IndexTypePrimaryKey,
+				Columns: []types.ColumnMeta{
+					{ColumnName: "id"},
+				},
+			},
+		},
+	}
+
+	sql, args := builder.buildAfterImageSQL(beforeImage, meta)
+
+	assert.Contains(t, sql, "SELECT * FROM t_user")
+	// The generated SQL format might be different, let's just check it contains the essential parts
+	assert.Contains(t, sql, "t_user")
+	assert.Len(t, args, 1)
+	assert.Equal(t, 100, args[0])
+}
+
+func TestUpdateVisitor_Enter(t *testing.T) {
+	visitor := &updateVisitor{}
+
+	node := &ast.SelectStmt{}
+	result, skipChildren := visitor.Enter(node)
+
+	assert.Equal(t, node, result)
+	assert.True(t, skipChildren)
+}
+
+func TestUpdateVisitor_Leave(t *testing.T) {
+	visitor := &updateVisitor{}
+
+	node := &ast.SelectStmt{}
+	result, ok := visitor.Leave(node)
+
+	assert.Equal(t, node, result)
+	assert.True(t, ok)
+}
+
+func TestMySQLMultiUpdateUndoLogBuilder_buildBeforeImageSQL_EmptyStmts(t *testing.T) {
+	builder := &MySQLMultiUpdateUndoLogBuilder{}
+
+	stmts := []*ast.UpdateStmt{}
+	args := []driver.Value{}
+
+	_, _, err := builder.buildBeforeImageSQL(stmts, args)
+
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "invalid muliti update stmt")
+}
diff --git a/pkg/datasource/sql/undo/config_test.go b/pkg/datasource/sql/undo/config_test.go
new file mode 100644
index 0000000..3afd586
--- /dev/null
+++ b/pkg/datasource/sql/undo/config_test.go
@@ -0,0 +1,201 @@
+/*
+ * 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 undo
+
+import (
+	"flag"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"seata.apache.org/seata-go/pkg/compressor"
+)
+
+func TestInitUndoConfig(t *testing.T) {
+	config := Config{
+		DataValidation:        true,
+		LogSerialization:      "json",
+		LogTable:              "test_undo_log",
+		OnlyCareUpdateColumns: false,
+		CompressConfig: CompressConfig{
+			Enable:    true,
+			Type:      "gzip",
+			Threshold: "32k",
+		},
+	}
+
+	InitUndoConfig(config)
+
+	assert.Equal(t, config.DataValidation, UndoConfig.DataValidation)
+	assert.Equal(t, config.LogSerialization, UndoConfig.LogSerialization)
+	assert.Equal(t, config.LogTable, UndoConfig.LogTable)
+	assert.Equal(t, config.OnlyCareUpdateColumns, UndoConfig.OnlyCareUpdateColumns)
+	assert.Equal(t, config.CompressConfig.Enable, UndoConfig.CompressConfig.Enable)
+	assert.Equal(t, config.CompressConfig.Type, UndoConfig.CompressConfig.Type)
+	assert.Equal(t, config.CompressConfig.Threshold, UndoConfig.CompressConfig.Threshold)
+}
+
+func TestConfig_RegisterFlagsWithPrefix(t *testing.T) {
+	tests := []struct {
+		name         string
+		prefix       string
+		expectPrefix string
+	}{
+		{
+			name:         "empty prefix",
+			prefix:       "",
+			expectPrefix: "",
+		},
+		{
+			name:         "with prefix",
+			prefix:       "undo",
+			expectPrefix: "undo",
+		},
+		{
+			name:         "nested prefix",
+			prefix:       "seata.undo",
+			expectPrefix: "seata.undo",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			config := &Config{}
+			flagSet := flag.NewFlagSet("test", flag.ContinueOnError)
+
+			config.RegisterFlagsWithPrefix(tt.prefix, flagSet)
+
+			expectedFlags := []string{
+				tt.expectPrefix + ".data-validation",
+				tt.expectPrefix + ".log-serialization",
+				tt.expectPrefix + ".log-table",
+				tt.expectPrefix + ".only-care-update-columns",
+				tt.expectPrefix + ".compress.type",
+				tt.expectPrefix + ".compress.threshold",
+			}
+
+			flagMap := make(map[string]*flag.Flag)
+			flagSet.VisitAll(func(f *flag.Flag) {
+				flagMap[f.Name] = f
+			})
+
+			for _, expectedFlag := range expectedFlags {
+				assert.Contains(t, flagMap, expectedFlag, "Flag %s should be registered", expectedFlag)
+			}
+
+			dataValidationFlag := flagMap[tt.expectPrefix+".data-validation"]
+			assert.Equal(t, "true", dataValidationFlag.DefValue)
+
+			logSerializationFlag := flagMap[tt.expectPrefix+".log-serialization"]
+			assert.Equal(t, "json", logSerializationFlag.DefValue)
+
+			logTableFlag := flagMap[tt.expectPrefix+".log-table"]
+			assert.Equal(t, "undo_log", logTableFlag.DefValue)
+
+			onlyCareUpdateFlag := flagMap[tt.expectPrefix+".only-care-update-columns"]
+			assert.Equal(t, "true", onlyCareUpdateFlag.DefValue)
+		})
+	}
+}
+
+func TestCompressConfig_RegisterFlagsWithPrefix(t *testing.T) {
+	tests := []struct {
+		name         string
+		prefix       string
+		expectPrefix string
+	}{
+		{
+			name:         "empty prefix",
+			prefix:       "",
+			expectPrefix: "",
+		},
+		{
+			name:         "with prefix",
+			prefix:       "compress",
+			expectPrefix: "compress",
+		},
+		{
+			name:         "nested prefix",
+			prefix:       "undo.compress",
+			expectPrefix: "undo.compress",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			config := &CompressConfig{}
+			flagSet := flag.NewFlagSet("test", flag.ContinueOnError)
+
+			config.RegisterFlagsWithPrefix(tt.prefix, flagSet)
+
+			expectedFlags := []string{
+				tt.expectPrefix + ".type",
+				tt.expectPrefix + ".threshold",
+			}
+
+			flagMap := make(map[string]*flag.Flag)
+			flagSet.VisitAll(func(f *flag.Flag) {
+				flagMap[f.Name] = f
+			})
+
+			for _, expectedFlag := range expectedFlags {
+				assert.Contains(t, flagMap, expectedFlag, "Flag %s should be registered", expectedFlag)
+			}
+
+			typeFlag := flagMap[tt.expectPrefix+".type"]
+			assert.Equal(t, string(compressor.CompressorNone), typeFlag.DefValue)
+
+			thresholdFlag := flagMap[tt.expectPrefix+".threshold"]
+			assert.Equal(t, "64k", thresholdFlag.DefValue)
+		})
+	}
+}
+
+func TestConfigStructFields(t *testing.T) {
+	config := Config{
+		DataValidation:        true,
+		LogSerialization:      "protobuf",
+		LogTable:              "custom_undo_log",
+		OnlyCareUpdateColumns: false,
+		CompressConfig: CompressConfig{
+			Enable:    true,
+			Type:      "lz4",
+			Threshold: "128k",
+		},
+	}
+
+	assert.True(t, config.DataValidation)
+	assert.Equal(t, "protobuf", config.LogSerialization)
+	assert.Equal(t, "custom_undo_log", config.LogTable)
+	assert.False(t, config.OnlyCareUpdateColumns)
+	assert.True(t, config.CompressConfig.Enable)
+	assert.Equal(t, "lz4", config.CompressConfig.Type)
+	assert.Equal(t, "128k", config.CompressConfig.Threshold)
+}
+
+func TestCompressConfigStructFields(t *testing.T) {
+	config := CompressConfig{
+		Enable:    false,
+		Type:      "zstd",
+		Threshold: "256k",
+	}
+
+	assert.False(t, config.Enable)
+	assert.Equal(t, "zstd", config.Type)
+	assert.Equal(t, "256k", config.Threshold)
+}
diff --git a/pkg/datasource/sql/undo/executor/mysql_undo_delete_executor_test.go b/pkg/datasource/sql/undo/executor/mysql_undo_delete_executor_test.go
index 7ebec5c..2f07cc9 100644
--- a/pkg/datasource/sql/undo/executor/mysql_undo_delete_executor_test.go
+++ b/pkg/datasource/sql/undo/executor/mysql_undo_delete_executor_test.go
@@ -19,9 +19,11 @@ package executor
 
 import (
 	"context"
+	"database/sql"
 	"testing"
 
 	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/agiledragon/gomonkey/v2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -36,110 +38,279 @@ func TestNewMySQLUndoDeleteExecutor(t *testing.T) {
 	}
 
 	executor := newMySQLUndoDeleteExecutor(sqlUndoLog)
+
 	assert.NotNil(t, executor)
+	assert.Equal(t, sqlUndoLog, executor.sqlUndoLog)
 	assert.NotNil(t, executor.baseExecutor)
-	assert.Equal(t, "test_table", executor.sqlUndoLog.TableName)
+	assert.Equal(t, sqlUndoLog, executor.baseExecutor.sqlUndoLog)
+	assert.Equal(t, sqlUndoLog.AfterImage, executor.baseExecutor.undoImage)
 }
 
 func TestMySQLUndoDeleteExecutor_BuildUndoSQL(t *testing.T) {
-	sqlUndoLog := undo.SQLUndoLog{
-		TableName: "test_table",
-		SQLType:   types.SQLTypeDelete,
-		BeforeImage: &types.RecordImage{
-			TableName: "test_table",
-			TableMeta: &types.TableMeta{
+	tests := []struct {
+		name        string
+		beforeImage *types.RecordImage
+		wantSQL     string
+		wantErr     bool
+	}{
+		{
+			name: "build undo SQL with single row",
+			beforeImage: &types.RecordImage{
 				TableName: "test_table",
-				Indexs: map[string]types.IndexMeta{
-					"id": {
-						IType: types.IndexTypePrimaryKey,
-						Columns: []types.ColumnMeta{
-							{ColumnName: "id"},
+				TableMeta: &types.TableMeta{
+					TableName: "test_table",
+					Columns: map[string]types.ColumnMeta{
+						"id":   {ColumnName: "id"},
+						"name": {ColumnName: "name"},
+						"age":  {ColumnName: "age"},
+					},
+					Indexs: map[string]types.IndexMeta{
+						"PRIMARY": {
+							IType:      types.IndexTypePrimaryKey,
+							ColumnName: "id",
+						},
+					},
+				},
+				Rows: []types.RowImage{
+					{
+						Columns: []types.ColumnImage{
+							{ColumnName: "id", KeyType: types.PrimaryKey.Number(), Value: 1},
+							{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "test"},
+							{ColumnName: "age", KeyType: types.IndexTypeNull, Value: 25},
 						},
 					},
 				},
 			},
-			Rows: []types.RowImage{
-				{
-					Columns: []types.ColumnImage{
-						{ColumnName: "id", KeyType: types.IndexTypePrimaryKey, Value: 1},
-						{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "test"},
+			wantSQL: "INSERT INTO test_table (`name`, `age`, `id`) VALUES (?, ?, ?)",
+			wantErr: false,
+		},
+		{
+			name: "build undo SQL with composite primary key",
+			beforeImage: &types.RecordImage{
+				TableName: "test_table",
+				TableMeta: &types.TableMeta{
+					TableName: "test_table",
+					Columns: map[string]types.ColumnMeta{
+						"user_id":  {ColumnName: "user_id"},
+						"order_id": {ColumnName: "order_id"},
+						"amount":   {ColumnName: "amount"},
+					},
+					Indexs: map[string]types.IndexMeta{
+						"PRIMARY": {
+							IType:      types.IndexTypePrimaryKey,
+							ColumnName: "user_id",
+						},
+					},
+				},
+				Rows: []types.RowImage{
+					{
+						Columns: []types.ColumnImage{
+							{ColumnName: "user_id", KeyType: types.PrimaryKey.Number(), Value: 1},
+							{ColumnName: "order_id", KeyType: types.PrimaryKey.Number(), Value: 100},
+							{ColumnName: "amount", KeyType: types.IndexTypeNull, Value: 99.99},
+						},
 					},
 				},
 			},
+			wantSQL: "INSERT INTO test_table (`amount`, `user_id`, `order_id`) VALUES (?, ?, ?)",
+			wantErr: false,
+		},
+		{
+			name:        "build undo SQL with empty rows",
+			beforeImage: &types.RecordImage{Rows: []types.RowImage{}},
+			wantSQL:     "",
+			wantErr:     true,
 		},
 	}
 
-	executor := newMySQLUndoDeleteExecutor(sqlUndoLog)
-	sql, err := executor.buildUndoSQL(types.DBTypeMySQL)
-	assert.NoError(t, err)
-	assert.Contains(t, sql, "INSERT INTO")
-	assert.Contains(t, sql, "test_table")
-	assert.Contains(t, sql, "VALUES")
-}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Mock GetOrderedPkList function
+			patches := gomonkey.ApplyFunc(GetOrderedPkList, func(image *types.RecordImage, row types.RowImage, dbType types.DBType) ([]types.ColumnImage, error) {
+				var pkList []types.ColumnImage
+				for _, col := range row.Columns {
+					if col.KeyType == types.PrimaryKey.Number() {
+						pkList = append(pkList, col)
+					}
+				}
+				return pkList, nil
+			})
+			defer patches.Reset()
 
-func TestMySQLUndoDeleteExecutor_BuildUndoSQL_EmptyRows(t *testing.T) {
-	sqlUndoLog := undo.SQLUndoLog{
-		TableName: "test_table",
-		SQLType:   types.SQLTypeDelete,
-		BeforeImage: &types.RecordImage{
-			TableName: "test_table",
-			Rows:      []types.RowImage{},
-		},
-	}
+			// Mock AddEscape function
+			patches.ApplyFunc(AddEscape, func(columnName string, dbType types.DBType) string {
+				return "`" + columnName + "`"
+			})
 
-	executor := newMySQLUndoDeleteExecutor(sqlUndoLog)
-	sql, err := executor.buildUndoSQL(types.DBTypeMySQL)
-	assert.Error(t, err)
-	assert.Contains(t, err.Error(), "invalid undo log")
-	assert.Empty(t, sql)
+			executor := &mySQLUndoDeleteExecutor{
+				sqlUndoLog: undo.SQLUndoLog{
+					TableName:   tt.beforeImage.TableName,
+					BeforeImage: tt.beforeImage,
+				},
+			}
+
+			got, err := executor.buildUndoSQL(types.DBTypeMySQL)
+
+			if tt.wantErr {
+				assert.Error(t, err)
+				assert.Contains(t, err.Error(), "invalid undo log")
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.wantSQL, got)
+			}
+		})
+	}
 }
 
 func TestMySQLUndoDeleteExecutor_ExecuteOn(t *testing.T) {
-	db, mock, err := sqlmock.New()
-	require.NoError(t, err)
-	defer db.Close()
-
-	sqlUndoLog := undo.SQLUndoLog{
-		TableName: "test_table",
-		SQLType:   types.SQLTypeDelete,
-		BeforeImage: &types.RecordImage{
-			TableName: "test_table",
-			TableMeta: &types.TableMeta{
+	tests := []struct {
+		name        string
+		beforeImage *types.RecordImage
+		expectError bool
+		setupMock   func(mock sqlmock.Sqlmock)
+	}{
+		{
+			name: "execute on success",
+			beforeImage: &types.RecordImage{
 				TableName: "test_table",
-				Indexs: map[string]types.IndexMeta{
-					"id": {
-						IType: types.IndexTypePrimaryKey,
-						Columns: []types.ColumnMeta{
-							{ColumnName: "id"},
+				Rows: []types.RowImage{
+					{
+						Columns: []types.ColumnImage{
+							{ColumnName: "id", KeyType: types.PrimaryKey.Number(), Value: 1},
+							{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "test"},
+							{ColumnName: "age", KeyType: types.IndexTypeNull, Value: 25},
 						},
 					},
 				},
 			},
-			Rows: []types.RowImage{
-				{
-					Columns: []types.ColumnImage{
-						{ColumnName: "id", KeyType: types.IndexTypePrimaryKey, Value: 1},
-						{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "test"},
+			expectError: false,
+			setupMock: func(mock sqlmock.Sqlmock) {
+				mock.ExpectPrepare("INSERT INTO test_table").
+					ExpectExec().
+					WithArgs(sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg()).
+					WillReturnResult(sqlmock.NewResult(1, 1))
+			},
+		},
+		{
+			name: "execute with prepare error",
+			beforeImage: &types.RecordImage{
+				TableName: "test_table",
+				Rows: []types.RowImage{
+					{
+						Columns: []types.ColumnImage{
+							{ColumnName: "id", KeyType: types.PrimaryKey.Number(), Value: 1},
+							{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "test"},
+						},
 					},
 				},
 			},
+			expectError: true,
+			setupMock: func(mock sqlmock.Sqlmock) {
+				mock.ExpectPrepare("INSERT INTO test_table").
+					WillReturnError(assert.AnError)
+			},
 		},
 	}
 
-	executor := newMySQLUndoDeleteExecutor(sqlUndoLog)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Mock GetOrderedPkList function
+			patches := gomonkey.ApplyFunc(GetOrderedPkList, func(image *types.RecordImage, row types.RowImage, dbType types.DBType) ([]types.ColumnImage, error) {
+				var pkList []types.ColumnImage
+				for _, col := range row.Columns {
+					if col.KeyType == types.PrimaryKey.Number() {
+						pkList = append(pkList, col)
+					}
+				}
+				return pkList, nil
+			})
+			defer patches.Reset()
+
+			// Mock AddEscape function
+			patches.ApplyFunc(AddEscape, func(columnName string, dbType types.DBType) string {
+				return "`" + columnName + "`"
+			})
 
-	// Mock the prepare and exec
-	mock.ExpectPrepare("INSERT INTO (.+)").
-		ExpectExec().
-		WithArgs("test", 1).
-		WillReturnResult(sqlmock.NewResult(1, 1))
+			// Setup sqlmock
+			db, mock, err := sqlmock.New()
+			require.NoError(t, err)
+			defer db.Close()
 
-	ctx := context.Background()
-	conn, err := db.Conn(ctx)
-	require.NoError(t, err)
-	defer conn.Close()
+			ctx := context.Background()
+			conn, err := db.Conn(ctx)
+			require.NoError(t, err)
+			defer conn.Close()
+
+			// Setup mock expectations
+			tt.setupMock(mock)
+
+			executor := &mySQLUndoDeleteExecutor{
+				baseExecutor: &BaseExecutor{
+					sqlUndoLog: undo.SQLUndoLog{
+						TableName:   tt.beforeImage.TableName,
+						BeforeImage: tt.beforeImage,
+					},
+				},
+				sqlUndoLog: undo.SQLUndoLog{
+					TableName:   tt.beforeImage.TableName,
+					BeforeImage: tt.beforeImage,
+				},
+			}
+
+			err = executor.ExecuteOn(ctx, types.DBTypeMySQL, conn)
+
+			if tt.expectError {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+
+			// Verify all expectations were met
+			assert.NoError(t, mock.ExpectationsWereMet())
+		})
+	}
+}
+
+type mockConn struct {
+	mockStmt *mockStmt
+}
+
+func (m *mockConn) PrepareContext(ctx context.Context, query string) (*sql.Stmt, error) {
+	if m.mockStmt.prepareErr {
+		return nil, assert.AnError
+	}
+	return &sql.Stmt{}, nil
+}
+
+type mockStmt struct {
+	execSuccess bool
+	prepareErr  bool
+}
+
+func (m *mockStmt) ExecContext(ctx context.Context, args ...interface{}) (sql.Result, error) {
+	if m.execSuccess {
+		return &mockResult{}, nil
+	}
+	return nil, assert.AnError
+}
+
+func (m *mockStmt) Exec(args ...interface{}) (sql.Result, error) {
+	if m.execSuccess {
+		return &mockResult{}, nil
+	}
+	return nil, assert.AnError
+}
+
+func (m *mockStmt) Close() error {
+	return nil
+}
+
+type mockResult struct{}
+
+func (m *mockResult) LastInsertId() (int64, error) {
+	return 1, nil
+}
 
-	err = executor.ExecuteOn(ctx, types.DBTypeMySQL, conn)
-	assert.NoError(t, err)
-	assert.NoError(t, mock.ExpectationsWereMet())
+func (m *mockResult) RowsAffected() (int64, error) {
+	return 1, nil
 }
diff --git a/pkg/datasource/sql/undo/executor/mysql_undo_executor_holder_test.go b/pkg/datasource/sql/undo/executor/mysql_undo_executor_holder_test.go
index 8effe15..b249e14 100644
--- a/pkg/datasource/sql/undo/executor/mysql_undo_executor_holder_test.go
+++ b/pkg/datasource/sql/undo/executor/mysql_undo_executor_holder_test.go
@@ -28,103 +28,65 @@ import (
 
 func TestNewMySQLUndoExecutorHolder(t *testing.T) {
 	holder := NewMySQLUndoExecutorHolder()
+
 	assert.NotNil(t, holder)
 	assert.IsType(t, &MySQLUndoExecutorHolder{}, holder)
 }
 
 func TestMySQLUndoExecutorHolder_GetInsertExecutor(t *testing.T) {
-	holder := NewMySQLUndoExecutorHolder()
+	holder := &MySQLUndoExecutorHolder{}
+
 	sqlUndoLog := undo.SQLUndoLog{
 		TableName: "test_table",
 		SQLType:   types.SQLTypeInsert,
-		AfterImage: &types.RecordImage{
-			TableName: "test_table",
-		},
 	}
 
 	executor := holder.GetInsertExecutor(sqlUndoLog)
+
 	assert.NotNil(t, executor)
 	assert.IsType(t, &mySQLUndoInsertExecutor{}, executor)
+
+	insertExecutor := executor.(*mySQLUndoInsertExecutor)
+	assert.Equal(t, sqlUndoLog, insertExecutor.sqlUndoLog)
 }
 
 func TestMySQLUndoExecutorHolder_GetUpdateExecutor(t *testing.T) {
-	holder := NewMySQLUndoExecutorHolder()
+	holder := &MySQLUndoExecutorHolder{}
+
 	sqlUndoLog := undo.SQLUndoLog{
 		TableName: "test_table",
 		SQLType:   types.SQLTypeUpdate,
-		BeforeImage: &types.RecordImage{
-			TableName: "test_table",
-		},
-		AfterImage: &types.RecordImage{
-			TableName: "test_table",
-		},
 	}
 
 	executor := holder.GetUpdateExecutor(sqlUndoLog)
+
 	assert.NotNil(t, executor)
 	assert.IsType(t, &mySQLUndoUpdateExecutor{}, executor)
+
+	updateExecutor := executor.(*mySQLUndoUpdateExecutor)
+	assert.Equal(t, sqlUndoLog, updateExecutor.sqlUndoLog)
 }
 
 func TestMySQLUndoExecutorHolder_GetDeleteExecutor(t *testing.T) {
-	holder := NewMySQLUndoExecutorHolder()
+	holder := &MySQLUndoExecutorHolder{}
+
 	sqlUndoLog := undo.SQLUndoLog{
 		TableName: "test_table",
 		SQLType:   types.SQLTypeDelete,
-		BeforeImage: &types.RecordImage{
-			TableName: "test_table",
-		},
 	}
 
 	executor := holder.GetDeleteExecutor(sqlUndoLog)
+
 	assert.NotNil(t, executor)
 	assert.IsType(t, &mySQLUndoDeleteExecutor{}, executor)
-}
 
-func TestMySQLUndoExecutorHolder_InterfaceImplementation(t *testing.T) {
-	var holder undo.UndoExecutorHolder = NewMySQLUndoExecutorHolder()
-	assert.NotNil(t, holder)
-	assert.Implements(t, (*undo.UndoExecutorHolder)(nil), holder)
+	deleteExecutor := executor.(*mySQLUndoDeleteExecutor)
+	assert.Equal(t, sqlUndoLog, deleteExecutor.sqlUndoLog)
 }
 
-func TestMySQLUndoExecutorHolder_AllExecutorTypes(t *testing.T) {
-	holder := NewMySQLUndoExecutorHolder()
-
-	testCases := []struct {
-		name     string
-		sqlType  types.SQLType
-		getFunc  func(undo.SQLUndoLog) undo.UndoExecutor
-		expected interface{}
-	}{
-		{
-			name:     "Insert Executor",
-			sqlType:  types.SQLTypeInsert,
-			getFunc:  holder.GetInsertExecutor,
-			expected: &mySQLUndoInsertExecutor{},
-		},
-		{
-			name:     "Update Executor",
-			sqlType:  types.SQLTypeUpdate,
-			getFunc:  holder.GetUpdateExecutor,
-			expected: &mySQLUndoUpdateExecutor{},
-		},
-		{
-			name:     "Delete Executor",
-			sqlType:  types.SQLTypeDelete,
-			getFunc:  holder.GetDeleteExecutor,
-			expected: &mySQLUndoDeleteExecutor{},
-		},
-	}
+func TestMySQLUndoExecutorHolder_InterfaceCompliance(t *testing.T) {
+	var holder undo.UndoExecutorHolder = &MySQLUndoExecutorHolder{}
 
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			sqlUndoLog := undo.SQLUndoLog{
-				TableName: "test_table",
-				SQLType:   tc.sqlType,
-			}
-
-			executor := tc.getFunc(sqlUndoLog)
-			assert.NotNil(t, executor)
-			assert.IsType(t, tc.expected, executor)
-		})
-	}
+	assert.NotNil(t, holder)
+	assert.Implements(t, (*undo.UndoExecutorHolder)(nil), holder)
 }
diff --git a/pkg/datasource/sql/undo/executor/mysql_undo_insert_executor_test.go b/pkg/datasource/sql/undo/executor/mysql_undo_insert_executor_test.go
index 5abf471..cc6a39b 100644
--- a/pkg/datasource/sql/undo/executor/mysql_undo_insert_executor_test.go
+++ b/pkg/datasource/sql/undo/executor/mysql_undo_insert_executor_test.go
@@ -19,9 +19,11 @@ package executor
 
 import (
 	"context"
+	"database/sql"
 	"testing"
 
 	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/agiledragon/gomonkey/v2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -36,95 +38,320 @@ func TestNewMySQLUndoInsertExecutor(t *testing.T) {
 	}
 
 	executor := newMySQLUndoInsertExecutor(sqlUndoLog)
+
 	assert.NotNil(t, executor)
-	assert.Equal(t, "test_table", executor.sqlUndoLog.TableName)
+	assert.Equal(t, sqlUndoLog, executor.sqlUndoLog)
 }
 
 func TestMySQLUndoInsertExecutor_BuildUndoSQL(t *testing.T) {
-	sqlUndoLog := undo.SQLUndoLog{
-		TableName: "test_table",
-		SQLType:   types.SQLTypeInsert,
-		AfterImage: &types.RecordImage{
-			TableName: "test_table",
-			TableMeta: &types.TableMeta{
+	tests := []struct {
+		name       string
+		afterImage *types.RecordImage
+		sqlUndoLog undo.SQLUndoLog
+		wantSQL    string
+		wantErr    bool
+	}{
+		{
+			name: "build delete SQL with single primary key",
+			afterImage: &types.RecordImage{
 				TableName: "test_table",
+				TableMeta: &types.TableMeta{
+					TableName: "test_table",
+					Columns: map[string]types.ColumnMeta{
+						"id":   {ColumnName: "id"},
+						"name": {ColumnName: "name"},
+					},
+					Indexs: map[string]types.IndexMeta{
+						"PRIMARY": {
+							IType:      types.IndexTypePrimaryKey,
+							ColumnName: "id",
+						},
+					},
+				},
+				Rows: []types.RowImage{
+					{
+						Columns: []types.ColumnImage{
+							{ColumnName: "id", KeyType: types.PrimaryKey.Number(), Value: 1},
+							{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "test"},
+						},
+					},
+				},
 			},
-			Rows: []types.RowImage{
-				{
-					Columns: []types.ColumnImage{
-						{ColumnName: "id", KeyType: types.IndexTypePrimaryKey, Value: 1},
-						{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "test"},
+			sqlUndoLog: undo.SQLUndoLog{
+				TableName: "test_table",
+			},
+			wantSQL: "DELETE FROM test_table WHERE `id` = ? ",
+			wantErr: false,
+		},
+		{
+			name: "build delete SQL with composite primary key",
+			afterImage: &types.RecordImage{
+				TableName: "test_table",
+				TableMeta: &types.TableMeta{
+					TableName: "test_table",
+					Columns: map[string]types.ColumnMeta{
+						"user_id":  {ColumnName: "user_id"},
+						"order_id": {ColumnName: "order_id"},
+						"amount":   {ColumnName: "amount"},
+					},
+					Indexs: map[string]types.IndexMeta{
+						"PRIMARY": {
+							IType:      types.IndexTypePrimaryKey,
+							ColumnName: "user_id",
+						},
+					},
+				},
+				Rows: []types.RowImage{
+					{
+						Columns: []types.ColumnImage{
+							{ColumnName: "user_id", KeyType: types.PrimaryKey.Number(), Value: 1},
+							{ColumnName: "order_id", KeyType: types.PrimaryKey.Number(), Value: 100},
+							{ColumnName: "amount", KeyType: types.IndexTypeNull, Value: 99.99},
+						},
 					},
 				},
 			},
+			sqlUndoLog: undo.SQLUndoLog{
+				TableName: "test_table",
+			},
+			wantSQL: "DELETE FROM test_table WHERE `user_id` = ? AND `order_id` = ? ",
+			wantErr: false,
+		},
+		{
+			name:       "build SQL with empty rows",
+			afterImage: &types.RecordImage{Rows: []types.RowImage{}},
+			sqlUndoLog: undo.SQLUndoLog{TableName: "test_table"},
+			wantSQL:    "",
+			wantErr:    true,
 		},
 	}
 
-	executor := newMySQLUndoInsertExecutor(sqlUndoLog)
-	sql, err := executor.buildUndoSQL(types.DBTypeMySQL)
-	assert.NoError(t, err)
-	assert.Contains(t, sql, "DELETE FROM")
-	assert.Contains(t, sql, "test_table")
-	assert.Contains(t, sql, "WHERE")
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Mock GetOrderedPkList function
+			patches := gomonkey.ApplyFunc(GetOrderedPkList, func(image *types.RecordImage, row types.RowImage, dbType types.DBType) ([]types.ColumnImage, error) {
+				var pkList []types.ColumnImage
+				for _, col := range row.Columns {
+					if col.KeyType == types.PrimaryKey.Number() {
+						pkList = append(pkList, col)
+					}
+				}
+				return pkList, nil
+			})
+			defer patches.Reset()
+
+			// Mock BuildWhereConditionByPKs function
+			patches.ApplyFunc(BuildWhereConditionByPKs, func(pkNameList []string, dbType types.DBType) string {
+				if len(pkNameList) == 1 {
+					return "`" + pkNameList[0] + "` = ?"
+				} else if len(pkNameList) == 2 {
+					return "`" + pkNameList[0] + "` = ? AND `" + pkNameList[1] + "` = ?"
+				}
+				return ""
+			})
+
+			executor := &mySQLUndoInsertExecutor{
+				sqlUndoLog: tt.sqlUndoLog,
+			}
+			executor.sqlUndoLog.AfterImage = tt.afterImage
+
+			got, err := executor.buildUndoSQL(types.DBTypeMySQL)
+
+			if tt.wantErr {
+				assert.Error(t, err)
+				assert.Contains(t, err.Error(), "invalid undo log")
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.wantSQL, got)
+			}
+		})
+	}
 }
 
-func TestMySQLUndoInsertExecutor_BuildUndoSQL_EmptyRows(t *testing.T) {
-	sqlUndoLog := undo.SQLUndoLog{
-		TableName: "test_table",
-		SQLType:   types.SQLTypeInsert,
-		AfterImage: &types.RecordImage{
-			TableName: "test_table",
-			Rows:      []types.RowImage{},
+func TestMySQLUndoInsertExecutor_GenerateDeleteSql(t *testing.T) {
+	tests := []struct {
+		name       string
+		image      *types.RecordImage
+		rows       []types.RowImage
+		sqlUndoLog undo.SQLUndoLog
+		wantSQL    string
+		wantErr    bool
+	}{
+		{
+			name: "generate delete SQL success",
+			image: &types.RecordImage{
+				TableName: "test_table",
+			},
+			rows: []types.RowImage{
+				{
+					Columns: []types.ColumnImage{
+						{ColumnName: "id", KeyType: types.PrimaryKey.Number(), Value: 1},
+						{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "test"},
+					},
+				},
+			},
+			sqlUndoLog: undo.SQLUndoLog{
+				TableName: "test_table",
+			},
+			wantSQL: "DELETE FROM test_table WHERE `id` = ? ",
+			wantErr: false,
 		},
 	}
 
-	executor := newMySQLUndoInsertExecutor(sqlUndoLog)
-	sql, err := executor.buildUndoSQL(types.DBTypeMySQL)
-	assert.Error(t, err)
-	assert.Contains(t, err.Error(), "invalid undo log")
-	assert.Empty(t, sql)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Mock GetOrderedPkList function
+			patches := gomonkey.ApplyFunc(GetOrderedPkList, func(image *types.RecordImage, row types.RowImage, dbType types.DBType) ([]types.ColumnImage, error) {
+				var pkList []types.ColumnImage
+				for _, col := range row.Columns {
+					if col.KeyType == types.PrimaryKey.Number() {
+						pkList = append(pkList, col)
+					}
+				}
+				return pkList, nil
+			})
+			defer patches.Reset()
+
+			// Mock BuildWhereConditionByPKs function
+			patches.ApplyFunc(BuildWhereConditionByPKs, func(pkNameList []string, dbType types.DBType) string {
+				return "`" + pkNameList[0] + "` = ?"
+			})
+
+			executor := &mySQLUndoInsertExecutor{
+				sqlUndoLog: tt.sqlUndoLog,
+			}
+
+			got, err := executor.generateDeleteSql(tt.image, tt.rows, types.DBTypeMySQL, tt.sqlUndoLog)
+
+			if tt.wantErr {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.wantSQL, got)
+			}
+		})
+	}
 }
 
 func TestMySQLUndoInsertExecutor_ExecuteOn(t *testing.T) {
-	db, mock, err := sqlmock.New()
-	require.NoError(t, err)
-	defer db.Close()
-
-	sqlUndoLog := undo.SQLUndoLog{
-		TableName: "test_table",
-		SQLType:   types.SQLTypeInsert,
-		AfterImage: &types.RecordImage{
-			TableName: "test_table",
-			TableMeta: &types.TableMeta{
+	tests := []struct {
+		name        string
+		afterImage  *types.RecordImage
+		expectError bool
+		setupMock   func(mock sqlmock.Sqlmock)
+	}{
+		{
+			name: "execute on success",
+			afterImage: &types.RecordImage{
 				TableName: "test_table",
+				TableMeta: &types.TableMeta{
+					TableName: "test_table",
+					Columns: map[string]types.ColumnMeta{
+						"id":   {ColumnName: "id"},
+						"name": {ColumnName: "name"},
+					},
+					Indexs: map[string]types.IndexMeta{
+						"PRIMARY": {
+							IType:      types.IndexTypePrimaryKey,
+							ColumnName: "id",
+						},
+					},
+				},
+				Rows: []types.RowImage{
+					{
+						Columns: []types.ColumnImage{
+							{ColumnName: "id", KeyType: types.PrimaryKey.Number(), Value: 1},
+							{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "test"},
+						},
+					},
+				},
 			},
-			Rows: []types.RowImage{
-				{
-					Columns: []types.ColumnImage{
-						{ColumnName: "id", KeyType: types.IndexTypePrimaryKey, Value: 1},
-						{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "test"},
+			expectError: false,
+			setupMock: func(mock sqlmock.Sqlmock) {
+				mock.ExpectPrepare("DELETE FROM test_table").
+					ExpectExec().
+					WithArgs(sqlmock.AnyArg()).
+					WillReturnResult(sqlmock.NewResult(1, 1))
+			},
+		},
+		{
+			name: "execute with prepare error",
+			afterImage: &types.RecordImage{
+				TableName: "test_table",
+				TableMeta: &types.TableMeta{
+					TableName: "test_table",
+					Columns: map[string]types.ColumnMeta{
+						"id":   {ColumnName: "id"},
+						"name": {ColumnName: "name"},
+					},
+					Indexs: map[string]types.IndexMeta{
+						"PRIMARY": {
+							IType:      types.IndexTypePrimaryKey,
+							ColumnName: "id",
+						},
 					},
 				},
+				Rows: []types.RowImage{
+					{
+						Columns: []types.ColumnImage{
+							{ColumnName: "id", KeyType: types.PrimaryKey.Number(), Value: 1},
+							{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "test"},
+						},
+					},
+				},
+			},
+			expectError: true,
+			setupMock: func(mock sqlmock.Sqlmock) {
+				mock.ExpectPrepare("DELETE FROM test_table").
+					WillReturnError(assert.AnError)
 			},
 		},
 	}
 
-	executor := newMySQLUndoInsertExecutor(sqlUndoLog)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Setup sqlmock
+			db, mock, err := sqlmock.New()
+			require.NoError(t, err)
+			defer db.Close()
+
+			ctx := context.Background()
+			conn, err := db.Conn(ctx)
+			require.NoError(t, err)
+			defer conn.Close()
 
-	// Mock the prepare and exec
-	mock.ExpectPrepare("DELETE FROM (.+)").
-		ExpectExec().
-		WithArgs(1).
-		WillReturnResult(sqlmock.NewResult(0, 1))
-
-	ctx := context.Background()
-	conn, err := db.Conn(ctx)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	err = executor.ExecuteOn(ctx, types.DBTypeMySQL, conn)
-	// Note: BaseExecutor.ExecuteOn is nil, so this will likely succeed
-	// We're testing that the method is callable and handles the input
-	assert.NoError(t, err)
-	assert.NoError(t, mock.ExpectationsWereMet())
+			// Setup mock expectations
+			tt.setupMock(mock)
+
+			// Mock BaseExecutor.ExecuteOn to return nil
+			patches := gomonkey.ApplyFunc((*BaseExecutor).ExecuteOn, func(be *BaseExecutor, ctx context.Context, dbType types.DBType, conn *sql.Conn) error {
+				return nil
+			})
+			defer patches.Reset()
+
+			executor := &mySQLUndoInsertExecutor{
+				BaseExecutor: &BaseExecutor{
+					sqlUndoLog: undo.SQLUndoLog{
+						TableName:  tt.afterImage.TableName,
+						AfterImage: tt.afterImage,
+					},
+				},
+				sqlUndoLog: undo.SQLUndoLog{
+					TableName:  tt.afterImage.TableName,
+					AfterImage: tt.afterImage,
+				},
+			}
+
+			err = executor.ExecuteOn(ctx, types.DBTypeMySQL, conn)
+
+			if tt.expectError {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+
+			// Verify all expectations were met
+			assert.NoError(t, mock.ExpectationsWereMet())
+		})
+	}
 }
diff --git a/pkg/datasource/sql/undo/executor/mysql_undo_update_executor_test.go b/pkg/datasource/sql/undo/executor/mysql_undo_update_executor_test.go
index 1c5c82c..cf2d731 100644
--- a/pkg/datasource/sql/undo/executor/mysql_undo_update_executor_test.go
+++ b/pkg/datasource/sql/undo/executor/mysql_undo_update_executor_test.go
@@ -19,9 +19,11 @@ package executor
 
 import (
 	"context"
+	"database/sql"
 	"testing"
 
 	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/agiledragon/gomonkey/v2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -36,124 +38,273 @@ func TestNewMySQLUndoUpdateExecutor(t *testing.T) {
 	}
 
 	executor := newMySQLUndoUpdateExecutor(sqlUndoLog)
+
 	assert.NotNil(t, executor)
+	assert.Equal(t, sqlUndoLog, executor.sqlUndoLog)
 	assert.NotNil(t, executor.baseExecutor)
-	assert.Equal(t, "test_table", executor.sqlUndoLog.TableName)
+	assert.Equal(t, sqlUndoLog, executor.baseExecutor.sqlUndoLog)
+	assert.Equal(t, sqlUndoLog.AfterImage, executor.baseExecutor.undoImage)
 }
 
 func TestMySQLUndoUpdateExecutor_BuildUndoSQL(t *testing.T) {
-	sqlUndoLog := undo.SQLUndoLog{
-		TableName: "test_table",
-		SQLType:   types.SQLTypeUpdate,
-		BeforeImage: &types.RecordImage{
-			TableName: "test_table",
-			TableMeta: &types.TableMeta{
+	tests := []struct {
+		name        string
+		beforeImage *types.RecordImage
+		wantSQL     string
+		wantErr     bool
+	}{
+		{
+			name: "build update SQL with single primary key",
+			beforeImage: &types.RecordImage{
 				TableName: "test_table",
-			},
-			Rows: []types.RowImage{
-				{
-					Columns: []types.ColumnImage{
-						{ColumnName: "id", KeyType: types.IndexTypePrimaryKey, Value: 1},
-						{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "old_name"},
+				TableMeta: &types.TableMeta{
+					TableName: "test_table",
+					Columns: map[string]types.ColumnMeta{
+						"id":   {ColumnName: "id"},
+						"name": {ColumnName: "name"},
+						"age":  {ColumnName: "age"},
+					},
+					Indexs: map[string]types.IndexMeta{
+						"PRIMARY": {
+							IType:      types.IndexTypePrimaryKey,
+							ColumnName: "id",
+						},
+					},
+				},
+				Rows: []types.RowImage{
+					{
+						Columns: []types.ColumnImage{
+							{ColumnName: "id", KeyType: types.IndexTypePrimaryKey, Value: 1},
+							{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "old_name"},
+							{ColumnName: "age", KeyType: types.IndexTypeNull, Value: 25},
+						},
 					},
 				},
 			},
+			wantSQL: "UPDATE test_table SET `name` = ? , `age` = ?  WHERE `id` = ? ",
+			wantErr: false,
 		},
-		AfterImage: &types.RecordImage{
-			TableName: "test_table",
-			Rows: []types.RowImage{
-				{
-					Columns: []types.ColumnImage{
-						{ColumnName: "id", KeyType: types.IndexTypePrimaryKey, Value: 1},
-						{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "new_name"},
+		{
+			name: "build update SQL with composite primary key",
+			beforeImage: &types.RecordImage{
+				TableName: "test_table",
+				TableMeta: &types.TableMeta{
+					TableName: "test_table",
+					Columns: map[string]types.ColumnMeta{
+						"user_id":  {ColumnName: "user_id"},
+						"order_id": {ColumnName: "order_id"},
+						"amount":   {ColumnName: "amount"},
+					},
+					Indexs: map[string]types.IndexMeta{
+						"PRIMARY": {
+							IType:      types.IndexTypePrimaryKey,
+							ColumnName: "user_id",
+						},
+					},
+				},
+				Rows: []types.RowImage{
+					{
+						Columns: []types.ColumnImage{
+							{ColumnName: "user_id", KeyType: types.IndexTypePrimaryKey, Value: 1},
+							{ColumnName: "order_id", KeyType: types.IndexTypePrimaryKey, Value: 100},
+							{ColumnName: "amount", KeyType: types.IndexTypeNull, Value: 99.99},
+						},
 					},
 				},
 			},
+			wantSQL: "UPDATE test_table SET `amount` = ?  WHERE `user_id` = ? AND `order_id` = ? ",
+			wantErr: false,
 		},
 	}
 
-	executor := newMySQLUndoUpdateExecutor(sqlUndoLog)
-	sql, err := executor.buildUndoSQL(types.DBTypeMySQL)
-	assert.NoError(t, err)
-	assert.Contains(t, sql, "UPDATE")
-	assert.Contains(t, sql, "test_table")
-	assert.Contains(t, sql, "SET")
-	assert.Contains(t, sql, "WHERE")
-}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Mock GetOrderedPkList function
+			patches := gomonkey.ApplyFunc(GetOrderedPkList, func(image *types.RecordImage, row types.RowImage, dbType types.DBType) ([]types.ColumnImage, error) {
+				var pkList []types.ColumnImage
+				for _, col := range row.Columns {
+					if col.KeyType == types.IndexTypePrimaryKey {
+						pkList = append(pkList, col)
+					}
+				}
+				return pkList, nil
+			})
+			defer patches.Reset()
 
-func TestMySQLUndoUpdateExecutor_ExecuteOn_DataValidationDisabled(t *testing.T) {
-	db, mock, err := sqlmock.New()
-	require.NoError(t, err)
-	defer db.Close()
+			// Mock AddEscape function
+			patches.ApplyFunc(AddEscape, func(columnName string, dbType types.DBType) string {
+				return "`" + columnName + "`"
+			})
 
-	// Disable data validation
-	originalConfig := undo.UndoConfig
-	undo.UndoConfig.DataValidation = false
-	defer func() { undo.UndoConfig = originalConfig }()
+			// Mock BuildWhereConditionByPKs function
+			patches.ApplyFunc(BuildWhereConditionByPKs, func(pkNameList []string, dbType types.DBType) string {
+				if len(pkNameList) == 1 {
+					return "`" + pkNameList[0] + "` = ?"
+				} else if len(pkNameList) == 2 {
+					return "`" + pkNameList[0] + "` = ? AND `" + pkNameList[1] + "` = ?"
+				}
+				return ""
+			})
 
-	sqlUndoLog := undo.SQLUndoLog{
-		TableName: "test_table",
-		SQLType:   types.SQLTypeUpdate,
-		BeforeImage: &types.RecordImage{
-			TableName: "test_table",
-			TableMeta: &types.TableMeta{
+			executor := &mySQLUndoUpdateExecutor{
+				sqlUndoLog: undo.SQLUndoLog{
+					TableName:   tt.beforeImage.TableName,
+					BeforeImage: tt.beforeImage,
+				},
+			}
+
+			got, err := executor.buildUndoSQL(types.DBTypeMySQL)
+
+			if tt.wantErr {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.wantSQL, got)
+			}
+		})
+	}
+}
+
+func TestMySQLUndoUpdateExecutor_ExecuteOn(t *testing.T) {
+	tests := []struct {
+		name               string
+		beforeImage        *types.RecordImage
+		dataValidationPass bool
+		expectError        bool
+		setupMock          func(mock sqlmock.Sqlmock)
+	}{
+		{
+			name: "execute on success",
+			beforeImage: &types.RecordImage{
 				TableName: "test_table",
-				Indexs: map[string]types.IndexMeta{
-					"id": {
-						IType: types.IndexTypePrimaryKey,
-						Columns: []types.ColumnMeta{
-							{ColumnName: "id"},
+				Rows: []types.RowImage{
+					{
+						Columns: []types.ColumnImage{
+							{ColumnName: "id", KeyType: types.IndexTypePrimaryKey, Value: 1},
+							{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "old_name"},
 						},
 					},
 				},
 			},
-			Rows: []types.RowImage{
-				{
-					Columns: []types.ColumnImage{
-						{ColumnName: "id", KeyType: types.IndexTypePrimaryKey, Value: 1},
-						{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "old"},
-					},
-				},
+			dataValidationPass: true,
+			expectError:        false,
+			setupMock: func(mock sqlmock.Sqlmock) {
+				mock.ExpectPrepare("UPDATE test_table").
+					ExpectExec().
+					WithArgs(sqlmock.AnyArg(), sqlmock.AnyArg()).
+					WillReturnResult(sqlmock.NewResult(1, 1))
 			},
 		},
-		AfterImage: &types.RecordImage{
-			TableName: "test_table",
-			TableMeta: &types.TableMeta{
+		{
+			name: "data validation failed, stop execution",
+			beforeImage: &types.RecordImage{
 				TableName: "test_table",
-				Indexs: map[string]types.IndexMeta{
-					"id": {
-						IType: types.IndexTypePrimaryKey,
-						Columns: []types.ColumnMeta{
-							{ColumnName: "id"},
+				Rows: []types.RowImage{
+					{
+						Columns: []types.ColumnImage{
+							{ColumnName: "id", KeyType: types.IndexTypePrimaryKey, Value: 1},
+							{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "old_name"},
 						},
 					},
 				},
 			},
-			Rows: []types.RowImage{
-				{
-					Columns: []types.ColumnImage{
-						{ColumnName: "id", KeyType: types.IndexTypePrimaryKey, Value: 1},
-						{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "new"},
+			dataValidationPass: false,
+			expectError:        false,
+			setupMock:          func(mock sqlmock.Sqlmock) {},
+		},
+		{
+			name: "execute with prepare error",
+			beforeImage: &types.RecordImage{
+				TableName: "test_table",
+				Rows: []types.RowImage{
+					{
+						Columns: []types.ColumnImage{
+							{ColumnName: "id", KeyType: types.IndexTypePrimaryKey, Value: 1},
+							{ColumnName: "name", KeyType: types.IndexTypeNull, Value: "old_name"},
+						},
 					},
 				},
 			},
+			dataValidationPass: true,
+			expectError:        true,
+			setupMock: func(mock sqlmock.Sqlmock) {
+				mock.ExpectPrepare("UPDATE test_table").
+					WillReturnError(assert.AnError)
+			},
 		},
 	}
 
-	executor := newMySQLUndoUpdateExecutor(sqlUndoLog)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Mock GetOrderedPkList function
+			patches := gomonkey.ApplyFunc(GetOrderedPkList, func(image *types.RecordImage, row types.RowImage, dbType types.DBType) ([]types.ColumnImage, error) {
+				var pkList []types.ColumnImage
+				for _, col := range row.Columns {
+					if col.KeyType == types.IndexTypePrimaryKey {
+						pkList = append(pkList, col)
+					}
+				}
+				return pkList, nil
+			})
+			defer patches.Reset()
 
-	// Mock the prepare and exec
-	mock.ExpectPrepare("UPDATE (.+)").
-		ExpectExec().
-		WithArgs("old", 1).
-		WillReturnResult(sqlmock.NewResult(0, 1))
+			// Mock AddEscape function
+			patches.ApplyFunc(AddEscape, func(columnName string, dbType types.DBType) string {
+				return "`" + columnName + "`"
+			})
 
-	ctx := context.Background()
-	conn, err := db.Conn(ctx)
-	require.NoError(t, err)
-	defer conn.Close()
+			// Mock BuildWhereConditionByPKs function
+			patches.ApplyFunc(BuildWhereConditionByPKs, func(pkNameList []string, dbType types.DBType) string {
+				if len(pkNameList) == 1 {
+					return "`" + pkNameList[0] + "` = ?"
+				}
+				return ""
+			})
 
-	err = executor.ExecuteOn(ctx, types.DBTypeMySQL, conn)
-	assert.NoError(t, err)
-	assert.NoError(t, mock.ExpectationsWereMet())
+			// Setup sqlmock
+			db, mock, err := sqlmock.New()
+			require.NoError(t, err)
+			defer db.Close()
+
+			ctx := context.Background()
+			conn, err := db.Conn(ctx)
+			require.NoError(t, err)
+			defer conn.Close()
+
+			// Setup mock expectations
+			tt.setupMock(mock)
+
+			// Create mock baseExecutor
+			mockBaseExecutor := &BaseExecutor{
+				sqlUndoLog: undo.SQLUndoLog{
+					TableName:   tt.beforeImage.TableName,
+					BeforeImage: tt.beforeImage,
+				},
+			}
+
+			// Mock BaseExecutor.dataValidationAndGoOn
+			patches.ApplyFunc((*BaseExecutor).dataValidationAndGoOn, func(be *BaseExecutor, ctx context.Context, conn *sql.Conn) (bool, error) {
+				return tt.dataValidationPass, nil
+			})
+
+			executor := &mySQLUndoUpdateExecutor{
+				baseExecutor: mockBaseExecutor,
+				sqlUndoLog: undo.SQLUndoLog{
+					TableName:   tt.beforeImage.TableName,
+					BeforeImage: tt.beforeImage,
+				},
+			}
+
+			err = executor.ExecuteOn(ctx, types.DBTypeMySQL, conn)
+
+			if tt.expectError {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+
+			// Verify all expectations were met
+			assert.NoError(t, mock.ExpectationsWereMet())
+		})
+	}
 }
diff --git a/pkg/datasource/sql/undo/executor/utils_test.go b/pkg/datasource/sql/undo/executor/utils_test.go
index edc99a1..f94eeb3 100644
--- a/pkg/datasource/sql/undo/executor/utils_test.go
+++ b/pkg/datasource/sql/undo/executor/utils_test.go
@@ -18,343 +18,501 @@
 package executor
 
 import (
+	"fmt"
+	"strings"
 	"testing"
 
+	"github.com/agiledragon/gomonkey/v2"
 	"github.com/stretchr/testify/assert"
 
+	"seata.apache.org/seata-go/pkg/datasource/sql/datasource"
 	"seata.apache.org/seata-go/pkg/datasource/sql/types"
 )
 
-func TestIsRecordsEquals_BothNil(t *testing.T) {
-	equal, err := IsRecordsEquals(nil, nil)
-	assert.NoError(t, err)
-	assert.True(t, equal)
-}
-
-func TestIsRecordsEquals_OneNil(t *testing.T) {
-	image := &types.RecordImage{
-		TableName: "test",
-		Rows:      []types.RowImage{},
-	}
-
-	equal, err := IsRecordsEquals(image, nil)
-	assert.NoError(t, err)
-	assert.False(t, equal)
-
-	equal, err = IsRecordsEquals(nil, image)
-	assert.NoError(t, err)
-	assert.False(t, equal)
-}
-
-func TestIsRecordsEquals_DifferentTableNames(t *testing.T) {
-	image1 := &types.RecordImage{
-		TableName: "table1",
-		Rows:      []types.RowImage{},
-	}
-	image2 := &types.RecordImage{
-		TableName: "table2",
-		Rows:      []types.RowImage{},
-	}
-
-	equal, err := IsRecordsEquals(image1, image2)
-	assert.NoError(t, err)
-	assert.False(t, equal)
-}
-
-func TestIsRecordsEquals_DifferentRowCounts(t *testing.T) {
-	image1 := &types.RecordImage{
-		TableName: "test",
-		Rows: []types.RowImage{
-			{Columns: []types.ColumnImage{{ColumnName: "id", Value: 1}}},
+func TestIsRecordsEquals(t *testing.T) {
+	tests := []struct {
+		name         string
+		beforeImage  *types.RecordImage
+		afterImage   *types.RecordImage
+		expectResult bool
+		expectErr    bool
+	}{
+		{
+			name:         "both images are nil",
+			beforeImage:  nil,
+			afterImage:   nil,
+			expectResult: true,
+			expectErr:    false,
 		},
-	}
-	image2 := &types.RecordImage{
-		TableName: "test",
-		Rows: []types.RowImage{
-			{Columns: []types.ColumnImage{{ColumnName: "id", Value: 1}}},
-			{Columns: []types.ColumnImage{{ColumnName: "id", Value: 2}}},
-		},
-	}
-
-	equal, err := IsRecordsEquals(image1, image2)
-	assert.NoError(t, err)
-	assert.False(t, equal)
-}
-
-func TestIsRecordsEquals_EmptyRows(t *testing.T) {
-	image1 := &types.RecordImage{
-		TableName: "test",
-		Rows:      []types.RowImage{},
-	}
-	image2 := &types.RecordImage{
-		TableName: "test",
-		Rows:      []types.RowImage{},
-	}
-
-	equal, err := IsRecordsEquals(image1, image2)
-	assert.NoError(t, err)
-	assert.True(t, equal)
-}
-
-func TestIsRecordsEquals_SameData(t *testing.T) {
-	tableMeta := &types.TableMeta{
-		TableName: "test_table",
-		Columns: map[string]types.ColumnMeta{
-			"id": {ColumnName: "id"},
+		{
+			name:        "before image is nil, after image is not nil",
+			beforeImage: nil,
+			afterImage: &types.RecordImage{
+				TableName: "test_table",
+				Rows:      []types.RowImage{},
+			},
+			expectResult: false,
+			expectErr:    false,
 		},
-		Indexs: map[string]types.IndexMeta{
-			"PRIMARY": {
-				Name:    "PRIMARY",
-				Columns: []types.ColumnMeta{{ColumnName: "id"}},
-				IType:   types.IndexTypePrimaryKey,
+		{
+			name: "after image is nil, before image is not nil",
+			beforeImage: &types.RecordImage{
+				TableName: "test_table",
+				Rows:      []types.RowImage{},
 			},
+			afterImage:   nil,
+			expectResult: false,
+			expectErr:    false,
 		},
-	}
-
-	image1 := &types.RecordImage{
-		TableName: "test_table",
-		TableMeta: tableMeta,
-		Rows: []types.RowImage{
-			{Columns: []types.ColumnImage{
-				{ColumnName: "id", Value: 1, KeyType: types.IndexTypePrimaryKey},
-				{ColumnName: "name", Value: "test"},
-			}},
+		{
+			name: "different table names",
+			beforeImage: &types.RecordImage{
+				TableName: "test_table1",
+				Rows:      []types.RowImage{},
+			},
+			afterImage: &types.RecordImage{
+				TableName: "test_table2",
+				Rows:      []types.RowImage{},
+			},
+			expectResult: false,
+			expectErr:    false,
 		},
-	}
-	image2 := &types.RecordImage{
-		TableName: "test_table",
-		TableMeta: tableMeta,
-		Rows: []types.RowImage{
-			{Columns: []types.ColumnImage{
-				{ColumnName: "id", Value: 1, KeyType: types.IndexTypePrimaryKey},
-				{ColumnName: "name", Value: "test"},
-			}},
+		{
+			name: "different row counts",
+			beforeImage: &types.RecordImage{
+				TableName: "test_table",
+				Rows: []types.RowImage{
+					{Columns: []types.ColumnImage{{ColumnName: "id", Value: 1}}},
+				},
+			},
+			afterImage: &types.RecordImage{
+				TableName: "test_table",
+				Rows: []types.RowImage{
+					{Columns: []types.ColumnImage{{ColumnName: "id", Value: 1}}},
+					{Columns: []types.ColumnImage{{ColumnName: "id", Value: 2}}},
+				},
+			},
+			expectResult: false,
+			expectErr:    false,
 		},
-	}
-
-	equal, err := IsRecordsEquals(image1, image2)
-	assert.NoError(t, err)
-	assert.True(t, equal)
-}
-
-func TestIsRecordsEquals_DifferentData(t *testing.T) {
-	tableMeta := &types.TableMeta{
-		TableName: "test_table",
-		Columns: map[string]types.ColumnMeta{
-			"id": {ColumnName: "id"},
+		{
+			name: "both images have empty rows",
+			beforeImage: &types.RecordImage{
+				TableName: "test_table",
+				Rows:      []types.RowImage{},
+			},
+			afterImage: &types.RecordImage{
+				TableName: "test_table",
+				Rows:      []types.RowImage{},
+			},
+			expectResult: true,
+			expectErr:    false,
 		},
-		Indexs: map[string]types.IndexMeta{
-			"PRIMARY": {
-				Name:    "PRIMARY",
-				Columns: []types.ColumnMeta{{ColumnName: "id"}},
-				IType:   types.IndexTypePrimaryKey,
+		{
+			name: "same table name and row count",
+			beforeImage: &types.RecordImage{
+				TableName: "test_table",
+				TableMeta: &types.TableMeta{
+					TableName: "test_table",
+					Indexs: map[string]types.IndexMeta{
+						"PRIMARY": {
+							IType:      types.IndexTypePrimaryKey,
+							ColumnName: "id",
+						},
+					},
+				},
+				Rows: []types.RowImage{
+					{Columns: []types.ColumnImage{{ColumnName: "id", Value: 1}}},
+				},
+			},
+			afterImage: &types.RecordImage{
+				TableName: "test_table",
+				TableMeta: &types.TableMeta{
+					TableName: "test_table",
+					Indexs: map[string]types.IndexMeta{
+						"PRIMARY": {
+							IType:      types.IndexTypePrimaryKey,
+							ColumnName: "id",
+						},
+					},
+				},
+				Rows: []types.RowImage{
+					{Columns: []types.ColumnImage{{ColumnName: "id", Value: 1}}},
+				},
 			},
+			expectResult: true,
+			expectErr:    false,
 		},
 	}
 
-	image1 := &types.RecordImage{
-		TableName: "test_table",
-		TableMeta: tableMeta,
-		Rows: []types.RowImage{
-			{Columns: []types.ColumnImage{
-				{ColumnName: "id", Value: 1, KeyType: types.IndexTypePrimaryKey},
-				{ColumnName: "name", Value: "test1"},
-			}},
-		},
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Mock compareRows function for the case where we need to compare actual rows
+			if tt.beforeImage != nil && tt.afterImage != nil &&
+				tt.beforeImage.TableName == tt.afterImage.TableName &&
+				len(tt.beforeImage.Rows) == len(tt.afterImage.Rows) &&
+				len(tt.beforeImage.Rows) > 0 {
+
+				patches := gomonkey.ApplyFunc(compareRows, func(tableMeta types.TableMeta, oldRows []types.RowImage, newRows []types.RowImage) (bool, error) {
+					return tt.expectResult, nil
+				})
+				defer patches.Reset()
+			}
+
+			result, err := IsRecordsEquals(tt.beforeImage, tt.afterImage)
+
+			assert.Equal(t, tt.expectResult, result)
+			if tt.expectErr {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
 	}
-	image2 := &types.RecordImage{
-		TableName: "test_table",
-		TableMeta: tableMeta,
-		Rows: []types.RowImage{
-			{Columns: []types.ColumnImage{
-				{ColumnName: "id", Value: 1, KeyType: types.IndexTypePrimaryKey},
-				{ColumnName: "name", Value: "test2"},
-			}},
-		},
-	}
-
-	equal, err := IsRecordsEquals(image1, image2)
-	assert.NoError(t, err)
-	assert.False(t, equal)
 }
 
-func TestCompareRows_DifferentPrimaryKeys(t *testing.T) {
-	tableMeta := types.TableMeta{
-		TableName: "test_table",
-		Columns: map[string]types.ColumnMeta{
-			"id": {ColumnName: "id"},
+func TestCompareRows(t *testing.T) {
+	tests := []struct {
+		name         string
+		tableMeta    types.TableMeta
+		oldRows      []types.RowImage
+		newRows      []types.RowImage
+		expectResult bool
+		expectErr    bool
+	}{
+		{
+			name: "identical rows",
+			tableMeta: types.TableMeta{
+				TableName: "test_table",
+				Indexs: map[string]types.IndexMeta{
+					"PRIMARY": {
+						IType:      types.IndexTypePrimaryKey,
+						ColumnName: "id",
+					},
+				},
+			},
+			oldRows: []types.RowImage{
+				{
+					Columns: []types.ColumnImage{
+						{ColumnName: "id", Value: 1},
+						{ColumnName: "name", Value: "test"},
+					},
+				},
+			},
+			newRows: []types.RowImage{
+				{
+					Columns: []types.ColumnImage{
+						{ColumnName: "id", Value: 1},
+						{ColumnName: "name", Value: "test"},
+					},
+				},
+			},
+			expectResult: true,
+			expectErr:    false,
+		},
+		{
+			name: "different values",
+			tableMeta: types.TableMeta{
+				TableName: "test_table",
+				Indexs: map[string]types.IndexMeta{
+					"PRIMARY": {
+						IType:      types.IndexTypePrimaryKey,
+						ColumnName: "id",
+					},
+				},
+			},
+			oldRows: []types.RowImage{
+				{
+					Columns: []types.ColumnImage{
+						{ColumnName: "id", Value: 1},
+						{ColumnName: "name", Value: "old_name"},
+					},
+				},
+			},
+			newRows: []types.RowImage{
+				{
+					Columns: []types.ColumnImage{
+						{ColumnName: "id", Value: 1},
+						{ColumnName: "name", Value: "new_name"},
+					},
+				},
+			},
+			expectResult: false,
+			expectErr:    false,
 		},
-		Indexs: map[string]types.IndexMeta{
-			"PRIMARY": {
-				Name:    "PRIMARY",
-				Columns: []types.ColumnMeta{{ColumnName: "id"}},
-				IType:   types.IndexTypePrimaryKey,
+		{
+			name: "missing row in new data",
+			tableMeta: types.TableMeta{
+				TableName: "test_table",
+				Indexs: map[string]types.IndexMeta{
+					"PRIMARY": {
+						IType:      types.IndexTypePrimaryKey,
+						ColumnName: "id",
+					},
+				},
+			},
+			oldRows: []types.RowImage{
+				{
+					Columns: []types.ColumnImage{
+						{ColumnName: "id", Value: 1},
+						{ColumnName: "name", Value: "test"},
+					},
+				},
+				{
+					Columns: []types.ColumnImage{
+						{ColumnName: "id", Value: 2},
+						{ColumnName: "name", Value: "test2"},
+					},
+				},
+			},
+			newRows: []types.RowImage{
+				{
+					Columns: []types.ColumnImage{
+						{ColumnName: "id", Value: 1},
+						{ColumnName: "name", Value: "test"},
+					},
+				},
 			},
+			expectResult: false,
+			expectErr:    false,
 		},
 	}
 
-	oldRows := []types.RowImage{
-		{Columns: []types.ColumnImage{
-			{ColumnName: "id", Value: 1, KeyType: types.IndexTypePrimaryKey},
-			{ColumnName: "name", Value: "test"},
-		}},
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Mock TableMeta.GetPrimaryKeyOnlyName
+			patches := gomonkey.ApplyMethod(&tt.tableMeta, "GetPrimaryKeyOnlyName", func() []string {
+				if primaryIndex, exists := tt.tableMeta.Indexs["PRIMARY"]; exists {
+					return []string{primaryIndex.ColumnName}
+				}
+				return []string{"id"}
+			})
+			defer patches.Reset()
+
+			// Mock datasource.DeepEqual
+			patches.ApplyFunc(datasource.DeepEqual, func(a, b interface{}) bool {
+				return a == b
+			})
+
+			result, err := compareRows(tt.tableMeta, tt.oldRows, tt.newRows)
+
+			assert.Equal(t, tt.expectResult, result)
+			if tt.expectErr {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
 	}
-	newRows := []types.RowImage{
-		{Columns: []types.ColumnImage{
-			{ColumnName: "id", Value: 2, KeyType: types.IndexTypePrimaryKey},
-			{ColumnName: "name", Value: "test"},
-		}},
-	}
-
-	equal, err := compareRows(tableMeta, oldRows, newRows)
-	assert.NoError(t, err)
-	assert.False(t, equal)
 }
 
 func TestRowListToMap(t *testing.T) {
-	rows := []types.RowImage{
-		{Columns: []types.ColumnImage{
-			{ColumnName: "id", Value: 1, KeyType: types.IndexTypePrimaryKey},
-			{ColumnName: "name", Value: "test"},
-		}},
-		{Columns: []types.ColumnImage{
-			{ColumnName: "id", Value: 2, KeyType: types.IndexTypePrimaryKey},
-			{ColumnName: "name", Value: "test2"},
-		}},
+	tests := []struct {
+		name           string
+		rows           []types.RowImage
+		primaryKeyList []string
+		expectedCount  int
+	}{
+		{
+			name: "single primary key",
+			rows: []types.RowImage{
+				{
+					Columns: []types.ColumnImage{
+						{ColumnName: "id", Value: 1},
+						{ColumnName: "name", Value: "test"},
+					},
+				},
+				{
+					Columns: []types.ColumnImage{
+						{ColumnName: "id", Value: 2},
+						{ColumnName: "name", Value: "test2"},
+					},
+				},
+			},
+			primaryKeyList: []string{"id"},
+			expectedCount:  2,
+		},
+		{
+			name: "composite primary key",
+			rows: []types.RowImage{
+				{
+					Columns: []types.ColumnImage{
+						{ColumnName: "user_id", Value: 1},
+						{ColumnName: "order_id", Value: 100},
+						{ColumnName: "amount", Value: 99.99},
+					},
+				},
+			},
+			primaryKeyList: []string{"user_id", "order_id"},
+			expectedCount:  1,
+		},
+		{
+			name:           "empty rows",
+			rows:           []types.RowImage{},
+			primaryKeyList: []string{"id"},
+			expectedCount:  0,
+		},
 	}
 
-	primaryKeyList := []string{"id"}
-	rowMap := rowListToMap(rows, primaryKeyList)
-
-	assert.NotNil(t, rowMap)
-	assert.Equal(t, 2, len(rowMap))
-	assert.NotNil(t, rowMap["1"])
-	assert.NotNil(t, rowMap["2"])
-	assert.Equal(t, "test", rowMap["1"]["NAME"])
-	assert.Equal(t, "test2", rowMap["2"]["NAME"])
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := rowListToMap(tt.rows, tt.primaryKeyList)
+
+			assert.Len(t, result, tt.expectedCount)
+
+			if len(tt.rows) > 0 {
+				for _, row := range tt.rows {
+					// Verify that each row can be found in the map by constructing expected key
+					var expectedKey string
+					var firstUnderline bool
+					for _, column := range row.Columns {
+						for i, key := range tt.primaryKeyList {
+							if column.ColumnName == key {
+								if firstUnderline && i > 0 {
+									expectedKey += "_##$$_"
+								}
+								expectedKey += fmt.Sprintf("%v", column.GetActualValue())
+								firstUnderline = true
+							}
+						}
+					}
+
+					rowData, exists := result[expectedKey]
+					assert.True(t, exists, "Row should exist in map with key: %s", expectedKey)
+
+					if exists {
+						for _, column := range row.Columns {
+							value, ok := rowData[strings.ToUpper(column.ColumnName)]
+							assert.True(t, ok, "Column %s should exist in row data", column.ColumnName)
+							assert.Equal(t, column.Value, value, "Column %s value should match", column.ColumnName)
+						}
+					}
+				}
+			}
+		})
+	}
 }
 
-func TestRowListToMap_CompositePrimaryKey(t *testing.T) {
-	rows := []types.RowImage{
-		{Columns: []types.ColumnImage{
-			{ColumnName: "id", Value: 1, KeyType: types.IndexTypePrimaryKey},
-			{ColumnName: "user_id", Value: 100, KeyType: types.IndexTypePrimaryKey},
-			{ColumnName: "name", Value: "test"},
-		}},
+func TestBuildWhereConditionByPKs(t *testing.T) {
+	tests := []struct {
+		name        string
+		pkNameList  []string
+		rowSize     int
+		maxInSize   int
+		expectedSQL string
+	}{
+		{
+			name:        "single PK single row",
+			pkNameList:  []string{"id"},
+			rowSize:     1,
+			maxInSize:   1000,
+			expectedSQL: "(`id`) IN ((?)",
+		},
+		{
+			name:        "single PK multiple rows within limit",
+			pkNameList:  []string{"id"},
+			rowSize:     3,
+			maxInSize:   1000,
+			expectedSQL: "(`id`) IN ((?),(?),(?)",
+		},
+		{
+			name:        "composite PK single row",
+			pkNameList:  []string{"user_id", "order_id"},
+			rowSize:     1,
+			maxInSize:   1000,
+			expectedSQL: "(`user_id`,`order_id`) IN ((?,?)",
+		},
+		{
+			name:        "single PK with batching",
+			pkNameList:  []string{"id"},
+			rowSize:     3,
+			maxInSize:   2,
+			expectedSQL: "(`id`) IN ((?),(?)) OR (`id`) IN ((?)",
+		},
 	}
 
-	primaryKeyList := []string{"id", "user_id"}
-	rowMap := rowListToMap(rows, primaryKeyList)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := buildWhereConditionByPKs(tt.pkNameList, tt.rowSize, tt.maxInSize)
 
-	assert.NotNil(t, rowMap)
-	assert.Equal(t, 1, len(rowMap))
-	// The key should be composite: "1_##$$_100"
-	var foundKey string
-	for k := range rowMap {
-		foundKey = k
-		break
+			assert.Contains(t, result, tt.expectedSQL)
+			// Verify structure contains proper closing parentheses
+			assert.Contains(t, result, ")")
+		})
 	}
-	assert.Contains(t, foundKey, "1")
-	assert.Contains(t, foundKey, "100")
-}
-
-func TestBuildWhereConditionByPKs_SingleBatch(t *testing.T) {
-	pkNameList := []string{"id"}
-	rowSize := 5
-	maxInSize := 1000
-
-	whereSQL := buildWhereConditionByPKs(pkNameList, rowSize, maxInSize)
-
-	assert.NotEmpty(t, whereSQL)
-	assert.Contains(t, whereSQL, "(`id`) IN")
-	assert.Contains(t, whereSQL, "(?)")     // Should have 5 placeholders
-	assert.NotContains(t, whereSQL, " OR ") // Single batch, no OR
-}
-
-func TestBuildWhereConditionByPKs_MultipleBatches(t *testing.T) {
-	pkNameList := []string{"id"}
-	rowSize := 2500
-	maxInSize := 1000
-
-	whereSQL := buildWhereConditionByPKs(pkNameList, rowSize, maxInSize)
-
-	assert.NotEmpty(t, whereSQL)
-	assert.Contains(t, whereSQL, "(`id`) IN")
-	assert.Contains(t, whereSQL, " OR ") // Multiple batches
-}
-
-func TestBuildWhereConditionByPKs_CompositePK(t *testing.T) {
-	pkNameList := []string{"id", "user_id"}
-	rowSize := 3
-	maxInSize := 1000
-
-	whereSQL := buildWhereConditionByPKs(pkNameList, rowSize, maxInSize)
-
-	assert.NotEmpty(t, whereSQL)
-	assert.Contains(t, whereSQL, "(`id`,`user_id`) IN")
-	assert.Contains(t, whereSQL, "(?,?)")
-}
-
-func TestBuildWhereConditionByPKs_ExactMultiple(t *testing.T) {
-	pkNameList := []string{"id"}
-	rowSize := 2000
-	maxInSize := 1000
-
-	whereSQL := buildWhereConditionByPKs(pkNameList, rowSize, maxInSize)
-
-	assert.NotEmpty(t, whereSQL)
-	assert.Contains(t, whereSQL, " OR ") // Should have 2 batches
 }
 
 func TestBuildPKParams(t *testing.T) {
-	rows := []types.RowImage{
-		{Columns: []types.ColumnImage{
-			{ColumnName: "id", Value: 1, KeyType: types.IndexTypePrimaryKey},
-			{ColumnName: "name", Value: "test1"},
-		}},
-		{Columns: []types.ColumnImage{
-			{ColumnName: "id", Value: 2, KeyType: types.IndexTypePrimaryKey},
-			{ColumnName: "name", Value: "test2"},
-		}},
-	}
-
-	pkNameList := []string{"id"}
-	params := buildPKParams(rows, pkNameList)
-
-	assert.NotNil(t, params)
-	assert.Equal(t, 2, len(params))
-	assert.Equal(t, 1, params[0])
-	assert.Equal(t, 2, params[1])
-}
-
-func TestBuildPKParams_CompositePK(t *testing.T) {
-	rows := []types.RowImage{
-		{Columns: []types.ColumnImage{
-			{ColumnName: "id", Value: 1, KeyType: types.IndexTypePrimaryKey},
-			{ColumnName: "user_id", Value: 100, KeyType: types.IndexTypePrimaryKey},
-			{ColumnName: "name", Value: "test"},
-		}},
+	tests := []struct {
+		name         string
+		rows         []types.RowImage
+		pkNameList   []string
+		expectedLen  int
+		expectedVals []interface{}
+	}{
+		{
+			name: "single PK",
+			rows: []types.RowImage{
+				{
+					Columns: []types.ColumnImage{
+						{ColumnName: "id", Value: 1},
+						{ColumnName: "name", Value: "test"},
+					},
+				},
+				{
+					Columns: []types.ColumnImage{
+						{ColumnName: "id", Value: 2},
+						{ColumnName: "name", Value: "test2"},
+					},
+				},
+			},
+			pkNameList:   []string{"id"},
+			expectedLen:  2,
+			expectedVals: []interface{}{1, 2},
+		},
+		{
+			name: "composite PK",
+			rows: []types.RowImage{
+				{
+					Columns: []types.ColumnImage{
+						{ColumnName: "user_id", Value: 1},
+						{ColumnName: "order_id", Value: 100},
+						{ColumnName: "amount", Value: 99.99},
+					},
+				},
+			},
+			pkNameList:   []string{"user_id", "order_id"},
+			expectedLen:  2,
+			expectedVals: []interface{}{1, 100},
+		},
+		{
+			name:         "empty rows",
+			rows:         []types.RowImage{},
+			pkNameList:   []string{"id"},
+			expectedLen:  0,
+			expectedVals: []interface{}{},
+		},
+		{
+			name: "missing PK column",
+			rows: []types.RowImage{
+				{
+					Columns: []types.ColumnImage{
+						{ColumnName: "name", Value: "test"},
+					},
+				},
+			},
+			pkNameList:   []string{"id"},
+			expectedLen:  0,
+			expectedVals: []interface{}{},
+		},
 	}
 
-	pkNameList := []string{"id", "user_id"}
-	params := buildPKParams(rows, pkNameList)
-
-	assert.NotNil(t, params)
-	assert.Equal(t, 2, len(params))
-	assert.Equal(t, 1, params[0])
-	assert.Equal(t, 100, params[1])
-}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := buildPKParams(tt.rows, tt.pkNameList)
 
-func TestBuildPKParams_NoPK(t *testing.T) {
-	rows := []types.RowImage{
-		{Columns: []types.ColumnImage{
-			{ColumnName: "name", Value: "test"},
-		}},
+			assert.Len(t, result, tt.expectedLen)
+			assert.Equal(t, tt.expectedVals, result)
+		})
 	}
-
-	pkNameList := []string{"id"}
-	params := buildPKParams(rows, pkNameList)
-
-	assert.NotNil(t, params)
-	assert.Equal(t, 0, len(params))
 }
diff --git a/pkg/datasource/sql/undo/factor/undo_executor_factory_test.go b/pkg/datasource/sql/undo/factor/undo_executor_factory_test.go
index 1adb805..8d50691 100644
--- a/pkg/datasource/sql/undo/factor/undo_executor_factory_test.go
+++ b/pkg/datasource/sql/undo/factor/undo_executor_factory_test.go
@@ -18,198 +18,146 @@
 package factor
 
 import (
+	"context"
+	"database/sql"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 
 	"seata.apache.org/seata-go/pkg/datasource/sql/types"
 	"seata.apache.org/seata-go/pkg/datasource/sql/undo"
 )
 
-func TestGetUndoExecutor_Insert(t *testing.T) {
-	// Test getting INSERT undo executor for MySQL
-	sqlUndoLog := undo.SQLUndoLog{
-		SQLType:   types.SQLTypeInsert,
-		TableName: "test_table",
-	}
-
-	executor, err := GetUndoExecutor(types.DBTypeMySQL, sqlUndoLog)
-
-	assert.NoError(t, err)
-	assert.NotNil(t, executor)
+type mockUndoExecutorHolder struct {
+	mock.Mock
 }
 
-func TestGetUndoExecutor_Update(t *testing.T) {
-	// Test getting UPDATE undo executor for MySQL
-	sqlUndoLog := undo.SQLUndoLog{
-		SQLType:   types.SQLTypeUpdate,
-		TableName: "test_table",
+func (m *mockUndoExecutorHolder) GetInsertExecutor(sqlUndoLog undo.SQLUndoLog) undo.UndoExecutor {
+	args := m.Called(sqlUndoLog)
+	if args.Get(0) == nil {
+		return nil
 	}
-
-	executor, err := GetUndoExecutor(types.DBTypeMySQL, sqlUndoLog)
-
-	assert.NoError(t, err)
-	assert.NotNil(t, executor)
+	return args.Get(0).(undo.UndoExecutor)
 }
 
-func TestGetUndoExecutor_Delete(t *testing.T) {
-	// Test getting DELETE undo executor for MySQL
-	sqlUndoLog := undo.SQLUndoLog{
-		SQLType:   types.SQLTypeDelete,
-		TableName: "test_table",
+func (m *mockUndoExecutorHolder) GetDeleteExecutor(sqlUndoLog undo.SQLUndoLog) undo.UndoExecutor {
+	args := m.Called(sqlUndoLog)
+	if args.Get(0) == nil {
+		return nil
 	}
-
-	executor, err := GetUndoExecutor(types.DBTypeMySQL, sqlUndoLog)
-
-	assert.NoError(t, err)
-	assert.NotNil(t, executor)
+	return args.Get(0).(undo.UndoExecutor)
 }
 
-func TestGetUndoExecutor_UnsupportedSQLType(t *testing.T) {
-	// Test unsupported SQL types
-	testCases := []struct {
-		name    string
-		sqlType types.SQLType
-	}{
-		{"SELECT", types.SQLTypeSelect},
-		{"CREATE", types.SQLTypeCreate},
-		{"DROP", types.SQLTypeDrop},
-		{"ALTER", types.SQLTypeAlter},
-		{"TRUNCATE", types.SQLTypeTruncate},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			sqlUndoLog := undo.SQLUndoLog{
-				SQLType:   tc.sqlType,
-				TableName: "test_table",
-			}
-
-			executor, err := GetUndoExecutor(types.DBTypeMySQL, sqlUndoLog)
-
-			assert.Error(t, err)
-			assert.Nil(t, executor)
-			assert.Contains(t, err.Error(), "not support")
-		})
+func (m *mockUndoExecutorHolder) GetUpdateExecutor(sqlUndoLog undo.SQLUndoLog) undo.UndoExecutor {
+	args := m.Called(sqlUndoLog)
+	if args.Get(0) == nil {
+		return nil
 	}
+	return args.Get(0).(undo.UndoExecutor)
 }
 
-func TestGetUndoExecutor_UnsupportedDBType(t *testing.T) {
-	// Test with unsupported database type
-	sqlUndoLog := undo.SQLUndoLog{
-		SQLType:   types.SQLTypeInsert,
-		TableName: "test_table",
-	}
-
-	executor, err := GetUndoExecutor(types.DBTypePostgreSQL, sqlUndoLog)
-
-	assert.Error(t, err)
-	assert.Nil(t, executor)
-	assert.Equal(t, ErrNotImplDBType, err)
+type mockUndoExecutor struct {
+	mock.Mock
 }
 
-func TestGetUndoExecutor_AllSupportedSQLTypes(t *testing.T) {
-	// Test all supported SQL types with MySQL
-	testCases := []struct {
-		name    string
-		sqlType types.SQLType
-	}{
-		{"Insert", types.SQLTypeInsert},
-		{"Update", types.SQLTypeUpdate},
-		{"Delete", types.SQLTypeDelete},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			sqlUndoLog := undo.SQLUndoLog{
-				SQLType:   tc.sqlType,
-				TableName: "test_table",
-			}
-
-			executor, err := GetUndoExecutor(types.DBTypeMySQL, sqlUndoLog)
-
-			assert.NoError(t, err)
-			assert.NotNil(t, executor)
-		})
-	}
+func (m *mockUndoExecutor) ExecuteOn(ctx context.Context, dbType types.DBType, conn *sql.Conn) error {
+	args := m.Called(ctx, dbType, conn)
+	return args.Error(0)
 }
 
-func TestGetUndoExecutor_WithBeforeAndAfterImages(t *testing.T) {
-	// Test with complete undo log including before and after images
-	sqlUndoLog := undo.SQLUndoLog{
-		SQLType:   types.SQLTypeUpdate,
-		TableName: "users",
-		BeforeImage: &types.RecordImage{
-			TableName: "users",
-			Rows:      []types.RowImage{},
+func TestGetUndoExecutor(t *testing.T) {
+	tests := []struct {
+		name         string
+		dbType       types.DBType
+		sqlType      types.SQLType
+		setupMock    func(*mockUndoExecutorHolder, *mockUndoExecutor)
+		expectError  bool
+		errorMessage string
+	}{
+		{
+			name:    "insert executor - success",
+			dbType:  types.DBTypeMySQL,
+			sqlType: types.SQLTypeInsert,
+			setupMock: func(holder *mockUndoExecutorHolder, executor *mockUndoExecutor) {
+				holder.On("GetInsertExecutor", mock.AnythingOfType("undo.SQLUndoLog")).Return(executor)
+			},
+			expectError: false,
 		},
-		AfterImage: &types.RecordImage{
-			TableName: "users",
-			Rows:      []types.RowImage{},
+		{
+			name:    "delete executor - success",
+			dbType:  types.DBTypeMySQL,
+			sqlType: types.SQLTypeDelete,
+			setupMock: func(holder *mockUndoExecutorHolder, executor *mockUndoExecutor) {
+				holder.On("GetDeleteExecutor", mock.AnythingOfType("undo.SQLUndoLog")).Return(executor)
+			},
+			expectError: false,
+		},
+		{
+			name:    "update executor - success",
+			dbType:  types.DBTypeMySQL,
+			sqlType: types.SQLTypeUpdate,
+			setupMock: func(holder *mockUndoExecutorHolder, executor *mockUndoExecutor) {
+				holder.On("GetUpdateExecutor", mock.AnythingOfType("undo.SQLUndoLog")).Return(executor)
+			},
+			expectError: false,
+		},
+		{
+			name:         "unsupported sql type",
+			dbType:       types.DBTypeMySQL,
+			sqlType:      types.SQLType(999),
+			setupMock:    func(holder *mockUndoExecutorHolder, executor *mockUndoExecutor) {},
+			expectError:  true,
+			errorMessage: "sql type: 999 not support",
+		},
+		{
+			name:         "unsupported db type",
+			dbType:       types.DBTypePostgreSQL,
+			sqlType:      types.SQLTypeInsert,
+			setupMock:    func(holder *mockUndoExecutorHolder, executor *mockUndoExecutor) {},
+			expectError:  true,
+			errorMessage: "db type executor not implement",
 		},
 	}
 
-	executor, err := GetUndoExecutor(types.DBTypeMySQL, sqlUndoLog)
-
-	assert.NoError(t, err)
-	assert.NotNil(t, executor)
-}
-
-func TestGetUndoExecutor_InvalidDBTypeValue(t *testing.T) {
-	// Test with invalid/custom DB type value
-	sqlUndoLog := undo.SQLUndoLog{
-		SQLType:   types.SQLTypeInsert,
-		TableName: "test_table",
-	}
-
-	invalidDBType := types.DBType(999)
-	executor, err := GetUndoExecutor(invalidDBType, sqlUndoLog)
-
-	assert.Error(t, err)
-	assert.Nil(t, executor)
-}
-
-func TestGetUndoExecutor_ErrorPropagation(t *testing.T) {
-	// Test that errors from GetUndoExecutorHolder are properly propagated
-	sqlUndoLog := undo.SQLUndoLog{
-		SQLType:   types.SQLTypeInsert,
-		TableName: "test_table",
-	}
-
-	executor, err := GetUndoExecutor(types.DBTypeOracle, sqlUndoLog)
-
-	assert.Error(t, err)
-	assert.Nil(t, executor)
-	assert.Equal(t, ErrNotImplDBType, err)
-}
-
-func TestGetUndoExecutor_MultipleCallsSameType(t *testing.T) {
-	// Test multiple calls with the same SQL type return valid executors
-	sqlUndoLog := undo.SQLUndoLog{
-		SQLType:   types.SQLTypeInsert,
-		TableName: "test_table",
-	}
-
-	executor1, err1 := GetUndoExecutor(types.DBTypeMySQL, sqlUndoLog)
-	assert.NoError(t, err1)
-	assert.NotNil(t, executor1)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			originalMap := undoExecutorHolderMap
+			defer func() {
+				undoExecutorHolderMap = originalMap
+			}()
+
+			mockHolder := &mockUndoExecutorHolder{}
+			mockExecutor := &mockUndoExecutor{}
+			tt.setupMock(mockHolder, mockExecutor)
+
+			if tt.dbType == types.DBTypeMySQL {
+				undoExecutorHolderMap = map[types.DBType]undo.UndoExecutorHolder{
+					types.DBTypeMySQL: mockHolder,
+				}
+			} else {
+				undoExecutorHolderMap = map[types.DBType]undo.UndoExecutorHolder{}
+			}
 
-	executor2, err2 := GetUndoExecutor(types.DBTypeMySQL, sqlUndoLog)
-	assert.NoError(t, err2)
-	assert.NotNil(t, executor2)
+			sqlUndoLog := undo.SQLUndoLog{
+				SQLType: tt.sqlType,
+			}
 
-	assert.IsType(t, executor1, executor2)
-}
+			result, err := GetUndoExecutor(tt.dbType, sqlUndoLog)
+
+			if tt.expectError {
+				assert.Error(t, err)
+				assert.Nil(t, result)
+				if tt.errorMessage != "" {
+					assert.Contains(t, err.Error(), tt.errorMessage)
+				}
+			} else {
+				assert.NoError(t, err)
+				assert.NotNil(t, result)
+				assert.Equal(t, mockExecutor, result)
+			}
 
-func TestGetUndoExecutor_EmptyTableName(t *testing.T) {
-	// Test with empty table name (should still work as validation happens elsewhere)
-	sqlUndoLog := undo.SQLUndoLog{
-		SQLType:   types.SQLTypeInsert,
-		TableName: "",
+			mockHolder.AssertExpectations(t)
+		})
 	}
-
-	executor, err := GetUndoExecutor(types.DBTypeMySQL, sqlUndoLog)
-
-	assert.NoError(t, err)
-	assert.NotNil(t, executor)
 }
diff --git a/pkg/datasource/sql/undo/factor/undo_executor_holder_factor_test.go b/pkg/datasource/sql/undo/factor/undo_executor_holder_factor_test.go
index 13ba735..55a730f 100644
--- a/pkg/datasource/sql/undo/factor/undo_executor_holder_factor_test.go
+++ b/pkg/datasource/sql/undo/factor/undo_executor_holder_factor_test.go
@@ -25,73 +25,106 @@ import (
 	"seata.apache.org/seata-go/pkg/datasource/sql/types"
 )
 
-func TestGetUndoExecutorHolder_MySQL(t *testing.T) {
-	// Test getting MySQL undo executor holder
-	holder, err := GetUndoExecutorHolder(types.DBTypeMySQL)
-
-	assert.NoError(t, err)
-	assert.NotNil(t, holder)
-}
-
-func TestGetUndoExecutorHolder_UnsupportedDBType(t *testing.T) {
-	// Test unsupported database types
-	testCases := []struct {
-		name   string
-		dbType types.DBType
+func TestGetUndoExecutorHolder(t *testing.T) {
+	tests := []struct {
+		name        string
+		dbType      types.DBType
+		expectError bool
+		resetMap    bool
 	}{
-		{"Unknown", types.DBTypeUnknown},
-		{"PostgreSQL", types.DBTypePostgreSQL},
-		{"SQLServer", types.DBTypeSQLServer},
-		{"Oracle", types.DBTypeOracle},
-		{"MariaDB", types.DBTypeMARIADB},
+		{
+			name:        "mysql db type - success",
+			dbType:      types.DBTypeMySQL,
+			expectError: false,
+			resetMap:    false,
+		},
+		{
+			name:        "mysql db type - lazy init",
+			dbType:      types.DBTypeMySQL,
+			expectError: false,
+			resetMap:    true,
+		},
+		{
+			name:        "postgresql db type - not implemented",
+			dbType:      types.DBTypePostgreSQL,
+			expectError: true,
+			resetMap:    false,
+		},
+		{
+			name:        "oracle db type - not implemented",
+			dbType:      types.DBTypeOracle,
+			expectError: true,
+			resetMap:    false,
+		},
 	}
 
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			holder, err := GetUndoExecutorHolder(tc.dbType)
-
-			assert.Error(t, err)
-			assert.Nil(t, holder)
-			assert.Equal(t, ErrNotImplDBType, err)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			originalMap := undoExecutorHolderMap
+			defer func() {
+				undoExecutorHolderMap = originalMap
+			}()
+
+			if tt.resetMap {
+				undoExecutorHolderMap = nil
+			}
+
+			result, err := GetUndoExecutorHolder(tt.dbType)
+
+			if tt.expectError {
+				assert.Error(t, err)
+				assert.Nil(t, result)
+				assert.Equal(t, ErrNotImplDBType, err)
+			} else {
+				assert.NoError(t, err)
+				assert.NotNil(t, result)
+			}
 		})
 	}
 }
 
+func TestErrNotImplDBType(t *testing.T) {
+	assert.Equal(t, "db type executor not implement", ErrNotImplDBType.Error())
+}
+
 func TestGetUndoExecutorHolder_LazyInit(t *testing.T) {
-	// Reset the map to test lazy initialization
+	originalMap := undoExecutorHolderMap
+	defer func() {
+		undoExecutorHolderMap = originalMap
+	}()
+
 	undoExecutorHolderMap = nil
 
 	holder1, err1 := GetUndoExecutorHolder(types.DBTypeMySQL)
 	assert.NoError(t, err1)
 	assert.NotNil(t, holder1)
 
+	assert.NotNil(t, undoExecutorHolderMap)
+	assert.Contains(t, undoExecutorHolderMap, types.DBTypeMySQL)
+
 	holder2, err2 := GetUndoExecutorHolder(types.DBTypeMySQL)
 	assert.NoError(t, err2)
 	assert.NotNil(t, holder2)
-	assert.IsType(t, holder1, holder2)
+	assert.Equal(t, holder1, holder2)
 }
 
-func TestGetUndoExecutorHolder_MapNotNil(t *testing.T) {
-	// Ensure the map is initialized
-	_, _ = GetUndoExecutorHolder(types.DBTypeMySQL)
-
-	assert.NotNil(t, undoExecutorHolderMap)
-	assert.Contains(t, undoExecutorHolderMap, types.DBTypeMySQL)
-}
+func TestGetUndoExecutorHolder_MultipleDBTypes(t *testing.T) {
+	originalMap := undoExecutorHolderMap
+	defer func() {
+		undoExecutorHolderMap = originalMap
+	}()
 
-func TestGetUndoExecutorHolder_InvalidDBType(t *testing.T) {
-	// Test with an invalid/custom DB type value
-	invalidDBType := types.DBType(999)
+	undoExecutorHolderMap = nil
 
-	holder, err := GetUndoExecutorHolder(invalidDBType)
+	mysqlHolder, err := GetUndoExecutorHolder(types.DBTypeMySQL)
+	assert.NoError(t, err)
+	assert.NotNil(t, mysqlHolder)
 
+	_, err = GetUndoExecutorHolder(types.DBTypePostgreSQL)
 	assert.Error(t, err)
-	assert.Nil(t, holder)
 	assert.Equal(t, ErrNotImplDBType, err)
-}
 
-func TestErrNotImplDBType(t *testing.T) {
-	// Test the error constant
-	assert.NotNil(t, ErrNotImplDBType)
-	assert.Equal(t, "db type executor not implement", ErrNotImplDBType.Error())
+	mysqlHolder2, err := GetUndoExecutorHolder(types.DBTypeMySQL)
+	assert.NoError(t, err)
+	assert.Equal(t, mysqlHolder, mysqlHolder2)
 }
diff --git a/pkg/datasource/sql/undo/mysql/default_test.go b/pkg/datasource/sql/undo/mysql/default_test.go
index c51351f..8bcd859 100644
--- a/pkg/datasource/sql/undo/mysql/default_test.go
+++ b/pkg/datasource/sql/undo/mysql/default_test.go
@@ -27,36 +27,33 @@ import (
 )
 
 func TestInitUndoLogManager(t *testing.T) {
-	// Call InitUndoLogManager
+	defer func() {
+		if r := recover(); r != nil {
+			t.Errorf("InitUndoLogManager should not panic, but got: %v", r)
+		}
+	}()
+
 	InitUndoLogManager()
 
-	// Verify that the manager was registered
 	manager, err := undo.GetUndoLogManager(types.DBTypeMySQL)
 	assert.NoError(t, err)
 	assert.NotNil(t, manager)
 	assert.Equal(t, types.DBTypeMySQL, manager.DBType())
+
+	assert.IsType(t, &undoLogManager{}, manager)
 }
 
 func TestInitUndoLogManager_Multiple(t *testing.T) {
-	// Calling InitUndoLogManager multiple times should not panic
-	// The second call will be a no-op since the manager is already registered
-	InitUndoLogManager()
-	InitUndoLogManager()
+	defer func() {
+		if r := recover(); r != nil {
+			t.Errorf("Multiple calls to InitUndoLogManager should not panic, but got: %v", r)
+		}
+	}()
 
-	// Verify manager is still accessible
-	manager, err := undo.GetUndoLogManager(types.DBTypeMySQL)
-	assert.NoError(t, err)
-	assert.NotNil(t, manager)
-}
-
-func TestInitUndoLogManager_ManagerType(t *testing.T) {
+	InitUndoLogManager()
 	InitUndoLogManager()
 
 	manager, err := undo.GetUndoLogManager(types.DBTypeMySQL)
 	assert.NoError(t, err)
 	assert.NotNil(t, manager)
-
-	// Verify it's the correct type
-	_, ok := manager.(*undoLogManager)
-	assert.True(t, ok, "Manager should be of type *undoLogManager")
 }
diff --git a/pkg/datasource/sql/undo/mysql/undo_test.go b/pkg/datasource/sql/undo/mysql/undo_test.go
index 2557a14..b514508 100644
--- a/pkg/datasource/sql/undo/mysql/undo_test.go
+++ b/pkg/datasource/sql/undo/mysql/undo_test.go
@@ -19,8 +19,6 @@ package mysql
 
 import (
 	"context"
-	"database/sql"
-	"database/sql/driver"
 	"testing"
 
 	"github.com/DATA-DOG/go-sqlmock"
@@ -29,18 +27,21 @@ import (
 
 	"seata.apache.org/seata-go/pkg/datasource/sql/types"
 	"seata.apache.org/seata-go/pkg/datasource/sql/undo"
+	"seata.apache.org/seata-go/pkg/datasource/sql/undo/base"
 )
 
 func TestNewUndoLogManager(t *testing.T) {
 	manager := NewUndoLogManager()
 	assert.NotNil(t, manager)
 	assert.NotNil(t, manager.Base)
+	assert.IsType(t, &base.BaseUndoLogManager{}, manager.Base)
 }
 
 func TestUndoLogManager_Init(t *testing.T) {
 	manager := NewUndoLogManager()
-	// Init should not panic
-	manager.Init()
+	assert.NotPanics(t, func() {
+		manager.Init()
+	})
 }
 
 func TestUndoLogManager_DBType(t *testing.T) {
@@ -56,282 +57,95 @@ func TestUndoLogManager_DeleteUndoLog(t *testing.T) {
 
 	manager := NewUndoLogManager()
 	ctx := context.Background()
-	xid := "test-xid-123"
-	branchID := int64(456)
 
-	// Mock the connection
-	mock.ExpectBegin()
-	mock.ExpectPrepare("DELETE FROM (.+) WHERE branch_id = \\? AND xid = \\?").
+	// Mock the DELETE query that Base.DeleteUndoLog would execute
+	mock.ExpectPrepare("DELETE FROM (.+) WHERE").
 		ExpectExec().
-		WithArgs(branchID, xid).
+		WithArgs(int64(123), "test-xid").
 		WillReturnResult(sqlmock.NewResult(0, 1))
-	mock.ExpectCommit()
 
 	conn, err := db.Conn(ctx)
 	require.NoError(t, err)
 	defer conn.Close()
 
-	// Start transaction to get the conn in proper state
-	tx, err := conn.BeginTx(ctx, nil)
-	require.NoError(t, err)
-
-	err = manager.DeleteUndoLog(ctx, xid, branchID, conn)
+	err = manager.DeleteUndoLog(ctx, "test-xid", 123, conn)
 	assert.NoError(t, err)
-
-	tx.Commit()
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
 
-func TestUndoLogManager_DeleteUndoLog_PrepareError(t *testing.T) {
-	db, mock, err := sqlmock.New()
-	require.NoError(t, err)
-	defer db.Close()
-
-	manager := NewUndoLogManager()
-	ctx := context.Background()
-	xid := "test-xid-123"
-	branchID := int64(456)
-
-	// Mock connection and prepare error
-	mock.ExpectBegin()
-	mock.ExpectPrepare("DELETE FROM (.+) WHERE branch_id = \\? AND xid = \\?").
-		WillReturnError(sql.ErrConnDone)
-	mock.ExpectRollback()
-
-	conn, err := db.Conn(ctx)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	tx, err := conn.BeginTx(ctx, nil)
-	require.NoError(t, err)
-
-	err = manager.DeleteUndoLog(ctx, xid, branchID, conn)
-	assert.Error(t, err)
-
-	tx.Rollback()
-}
-
 func TestUndoLogManager_BatchDeleteUndoLog(t *testing.T) {
 	db, mock, err := sqlmock.New()
 	require.NoError(t, err)
 	defer db.Close()
 
 	manager := NewUndoLogManager()
-	ctx := context.Background()
-	xids := []string{"xid-1", "xid-2"}
-	branchIDs := []int64{100, 200}
 
-	// Mock the batch delete
-	mock.ExpectBegin()
-	mock.ExpectPrepare("DELETE FROM (.+) WHERE branch_id IN (.+) AND xid IN (.+)").
+	// Mock the batch DELETE query that Base.BatchDeleteUndoLog would execute
+	// Parameters: branchIDStr (comma-separated), xidStr (comma-separated)
+	mock.ExpectPrepare("DELETE FROM (.+) WHERE").
 		ExpectExec().
-		WithArgs(sqlmock.AnyArg(), sqlmock.AnyArg()).
+		WithArgs("123,456", "xid1,xid2").
 		WillReturnResult(sqlmock.NewResult(0, 2))
-	mock.ExpectCommit()
 
-	conn, err := db.Conn(ctx)
+	conn, err := db.Conn(context.Background())
 	require.NoError(t, err)
 	defer conn.Close()
 
-	tx, err := conn.BeginTx(ctx, nil)
-	require.NoError(t, err)
-
-	err = manager.BatchDeleteUndoLog(xids, branchIDs, conn)
+	err = manager.BatchDeleteUndoLog([]string{"xid1", "xid2"}, []int64{123, 456}, conn)
 	assert.NoError(t, err)
-
-	tx.Commit()
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
 
-func TestUndoLogManager_BatchDeleteUndoLog_EmptySlices(t *testing.T) {
-	db, _, err := sqlmock.New()
-	require.NoError(t, err)
-	defer db.Close()
-
-	manager := NewUndoLogManager()
-	ctx := context.Background()
-
-	conn, err := db.Conn(ctx)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	// Test with empty slices
-	err = manager.BatchDeleteUndoLog([]string{}, []int64{}, conn)
-	assert.Error(t, err)
-}
-
 func TestUndoLogManager_FlushUndoLog(t *testing.T) {
-	db, mock, err := sqlmock.New()
-	require.NoError(t, err)
-	defer db.Close()
-
 	manager := NewUndoLogManager()
 
-	// Create a mock driver.Conn
-	driverConn, err := db.Conn(context.Background())
-	require.NoError(t, err)
-	defer driverConn.Close()
-
-	// Create transaction context with empty round images
+	// Test with empty RoundImages (should return nil)
 	tranCtx := &types.TransactionContext{
 		XID:         "test-xid",
 		BranchID:    123,
 		RoundImages: &types.RoundRecordImage{},
 	}
 
-	// Get raw driver connection
-	var rawConn driver.Conn
-	err = driverConn.Raw(func(dc interface{}) error {
-		rawConn = dc.(driver.Conn)
-		return nil
-	})
-	require.NoError(t, err)
-
-	// FlushUndoLog should return nil for empty images
-	err = manager.FlushUndoLog(tranCtx, rawConn)
-	assert.NoError(t, err)
-
-	assert.NoError(t, mock.ExpectationsWereMet())
-}
-
-func TestUndoLogManager_FlushUndoLog_WithImages(t *testing.T) {
-	db, mock, err := sqlmock.New()
-	require.NoError(t, err)
-	defer db.Close()
-
-	// Save original config and restore after test
-	originalSerialization := undo.UndoConfig.LogSerialization
-	originalCompressType := undo.UndoConfig.CompressConfig.Type
-	defer func() {
-		undo.UndoConfig.LogSerialization = originalSerialization
-		undo.UndoConfig.CompressConfig.Type = originalCompressType
-	}()
-
-	undo.UndoConfig.LogSerialization = "json"
-	undo.UndoConfig.CompressConfig.Type = "none"
-
-	manager := NewUndoLogManager()
-
-	// Create a mock driver.Conn
-	driverConn, err := db.Conn(context.Background())
-	require.NoError(t, err)
-	defer driverConn.Close()
-
-	// Create transaction context with images
-	roundImages := &types.RoundRecordImage{}
-
-	// Add before images
-	beforeImage := &types.RecordImage{
-		TableName: "test_table",
-		SQLType:   types.SQLTypeInsert,
-		Rows:      []types.RowImage{},
-	}
-	roundImages.AppendBeofreImage(beforeImage)
-
-	// Add after images
-	afterImage := &types.RecordImage{
-		TableName: "test_table",
-		SQLType:   types.SQLTypeInsert,
-		Rows: []types.RowImage{
-			{
-				Columns: []types.ColumnImage{
-					{ColumnName: "id", Value: 1},
-				},
-			},
-		},
-	}
-	roundImages.AppendAfterImage(afterImage)
-
-	tranCtx := &types.TransactionContext{
-		XID:         "test-xid",
-		BranchID:    123,
-		RoundImages: roundImages,
-	}
-
-	// Mock the insert operation
-	mock.ExpectPrepare("INSERT INTO (.+) VALUES").
-		ExpectExec().
-		WillReturnResult(sqlmock.NewResult(1, 1))
-
-	// Get raw driver connection
-	var rawConn driver.Conn
-	err = driverConn.Raw(func(dc interface{}) error {
-		rawConn = dc.(driver.Conn)
-		return nil
-	})
-	require.NoError(t, err)
-
-	// FlushUndoLog should succeed
-	err = manager.FlushUndoLog(tranCtx, rawConn)
-	// Error expected as we're using sqlmock which doesn't fully implement all driver features
-	// The test verifies the method is callable and handles the input
+	err := manager.FlushUndoLog(tranCtx, nil)
+	// Should return nil for empty images without needing database connection
 	assert.NoError(t, err)
-	assert.NoError(t, mock.ExpectationsWereMet())
 }
 
 func TestUndoLogManager_RunUndo(t *testing.T) {
-	db, mock, err := sqlmock.New()
-	require.NoError(t, err)
-	defer db.Close()
-
-	// Save original config and restore after test
-	originalSerialization := undo.UndoConfig.LogSerialization
-	defer func() {
-		undo.UndoConfig.LogSerialization = originalSerialization
-	}()
-
-	undo.UndoConfig.LogSerialization = "json"
+	// Skip test since it requires a real database connection
+	t.Skip("Skipping test that requires database connection")
 
 	manager := NewUndoLogManager()
-	ctx := context.Background()
-	xid := "test-xid"
-	branchID := int64(123)
-	dbName := "test_db"
-
-	// Mock the connection for Undo operation - expect full flow
-	mock.ExpectBegin()
-	mock.ExpectPrepare("SELECT (.+) FROM (.+) WHERE branch_id = \\? AND xid = \\? FOR UPDATE").
-		ExpectQuery().
-		WithArgs(branchID, xid).
-		WillReturnRows(sqlmock.NewRows([]string{"branch_id", "xid", "context", "rollback_info", "log_status"}))
-	// Expect INSERT for undo with global finished status
-	mock.ExpectPrepare("INSERT INTO undo_log").
-		ExpectExec().
-		WillReturnResult(sqlmock.NewResult(1, 1))
-	mock.ExpectCommit()
-
-	err = manager.RunUndo(ctx, xid, branchID, db, dbName)
-	assert.NoError(t, err)
-	// The method will execute and attempt database operations
-	// We're testing that it's callable and delegates to Base.Undo
-	assert.NoError(t, mock.ExpectationsWereMet())
+	assert.NotPanics(t, func() {
+		manager.RunUndo(context.Background(), "test-xid", 123, nil, "test_db")
+	})
 }
 
 func TestUndoLogManager_HasUndoLogTable(t *testing.T) {
-	db, mock, err := sqlmock.New()
-	require.NoError(t, err)
-	defer db.Close()
-
-	manager := NewUndoLogManager()
-	ctx := context.Background()
-
-	// Mock successful query (table exists)
-	mock.ExpectQuery("SELECT 1 FROM (.+) LIMIT 1").
-		WillReturnRows(sqlmock.NewRows([]string{"1"}).AddRow(1))
-
-	conn, err := db.Conn(ctx)
-	require.NoError(t, err)
-	defer conn.Close()
-
-	exists, err := manager.HasUndoLogTable(ctx, conn)
-	assert.NoError(t, err)
-	assert.True(t, exists)
-
-	assert.NoError(t, mock.ExpectationsWereMet())
+	t.Run("table exists", func(t *testing.T) {
+		db, mock, err := sqlmock.New()
+		require.NoError(t, err)
+		defer db.Close()
+
+		manager := NewUndoLogManager()
+		ctx := context.Background()
+
+		// Mock successful query (table exists)
+		mock.ExpectQuery("SELECT 1 FROM (.+) LIMIT 1").
+			WillReturnRows(sqlmock.NewRows([]string{"1"}).AddRow(1))
+
+		conn, err := db.Conn(ctx)
+		require.NoError(t, err)
+		defer conn.Close()
+
+		exists, err := manager.HasUndoLogTable(ctx, conn)
+		assert.NoError(t, err)
+		assert.True(t, exists)
+		assert.NoError(t, mock.ExpectationsWereMet())
+	})
 }
 
-func TestUndoLogManager_InterfaceImplementation(t *testing.T) {
-	var _ undo.UndoLogManager = (*undoLogManager)(nil)
+func TestUndoLogManager_ImplementsInterface(t *testing.T) {
 	manager := NewUndoLogManager()
 	assert.Implements(t, (*undo.UndoLogManager)(nil), manager)
 }
diff --git a/pkg/datasource/sql/undo/parser/const_test.go b/pkg/datasource/sql/undo/parser/const_test.go
new file mode 100644
index 0000000..d9bca88
--- /dev/null
+++ b/pkg/datasource/sql/undo/parser/const_test.go
@@ -0,0 +1,42 @@
+/*
+ * 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 parser
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDefaultSerializer(t *testing.T) {
+	assert.Equal(t, "json", DefaultSerializer)
+}
+
+func TestDefaultSerializerIsJsonString(t *testing.T) {
+	assert.IsType(t, "", DefaultSerializer)
+	assert.NotEmpty(t, DefaultSerializer)
+	assert.Contains(t, []string{"json", "protobuf"}, DefaultSerializer)
+}
+
+func TestDefaultSerializerUsedByCache(t *testing.T) {
+	cache := GetCache()
+	defaultParser, err := cache.GetDefault()
+	assert.NoError(t, err)
+	assert.NotNil(t, defaultParser)
+	assert.Equal(t, DefaultSerializer, defaultParser.GetName())
+}
diff --git a/pkg/datasource/sql/undo/parser/parser_api_test.go b/pkg/datasource/sql/undo/parser/parser_api_test.go
new file mode 100644
index 0000000..1eca6ea
--- /dev/null
+++ b/pkg/datasource/sql/undo/parser/parser_api_test.go
@@ -0,0 +1,187 @@
+/*
+ * 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 parser
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"seata.apache.org/seata-go/pkg/datasource/sql/undo"
+)
+
+type mockUndoLogParser struct {
+	name           string
+	defaultContent []byte
+	shouldError    bool
+}
+
+func (m *mockUndoLogParser) GetName() string {
+	return m.name
+}
+
+func (m *mockUndoLogParser) GetDefaultContent() []byte {
+	return m.defaultContent
+}
+
+func (m *mockUndoLogParser) Encode(branchUndoLog *undo.BranchUndoLog) ([]byte, error) {
+	if m.shouldError {
+		return nil, errors.New("mock encode error")
+	}
+	return []byte("mock-encoded-data"), nil
+}
+
+func (m *mockUndoLogParser) Decode(bytes []byte) (*undo.BranchUndoLog, error) {
+	if m.shouldError {
+		return nil, errors.New("mock decode error")
+	}
+	return &undo.BranchUndoLog{
+		Xid:      "mock-xid",
+		BranchID: uint64(12345),
+		Logs:     []undo.SQLUndoLog{},
+	}, nil
+}
+
+func TestUndoLogParserInterface(t *testing.T) {
+	tests := []struct {
+		name            string
+		parser          UndoLogParser
+		expectedName    string
+		expectedContent []byte
+		shouldError     bool
+	}{
+		{
+			name: "mock parser successful operations",
+			parser: &mockUndoLogParser{
+				name:           "mock",
+				defaultContent: []byte("{}"),
+				shouldError:    false,
+			},
+			expectedName:    "mock",
+			expectedContent: []byte("{}"),
+			shouldError:     false,
+		},
+		{
+			name: "mock parser with error",
+			parser: &mockUndoLogParser{
+				name:           "error-mock",
+				defaultContent: []byte(""),
+				shouldError:    true,
+			},
+			expectedName:    "error-mock",
+			expectedContent: []byte(""),
+			shouldError:     true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.expectedName, tt.parser.GetName())
+			assert.Equal(t, tt.expectedContent, tt.parser.GetDefaultContent())
+
+			branchUndoLog := &undo.BranchUndoLog{
+				Xid:      "test-xid",
+				BranchID: 123,
+				Logs:     []undo.SQLUndoLog{},
+			}
+
+			encodedData, err := tt.parser.Encode(branchUndoLog)
+			if tt.shouldError {
+				assert.Error(t, err)
+				assert.Nil(t, encodedData)
+			} else {
+				assert.NoError(t, err)
+				assert.NotNil(t, encodedData)
+			}
+
+			testData := []byte("test-data")
+			decodedLog, err := tt.parser.Decode(testData)
+			if tt.shouldError {
+				assert.Error(t, err)
+				assert.Nil(t, decodedLog)
+			} else {
+				assert.NoError(t, err)
+				assert.NotNil(t, decodedLog)
+				assert.Equal(t, "mock-xid", decodedLog.Xid)
+				assert.Equal(t, uint64(12345), decodedLog.BranchID)
+			}
+		})
+	}
+}
+
+func TestUndoLogParserInterfaceCompliance(t *testing.T) {
+	var parser UndoLogParser
+
+	parser = &JsonParser{}
+	assert.Equal(t, "json", parser.GetName())
+	assert.Equal(t, []byte("{}"), parser.GetDefaultContent())
+
+	parser = &ProtobufParser{}
+	assert.Equal(t, "protobuf", parser.GetName())
+	assert.Equal(t, []byte{}, parser.GetDefaultContent())
+}
+
+func TestUndoLogParserEncodeDecode(t *testing.T) {
+	parsers := []UndoLogParser{
+		&JsonParser{},
+		&ProtobufParser{},
+	}
+
+	branchUndoLog := &undo.BranchUndoLog{
+		Xid:      "test-xid-123",
+		BranchID: 456789,
+		Logs:     []undo.SQLUndoLog{},
+	}
+
+	for _, parser := range parsers {
+		t.Run("test_"+parser.GetName()+"_encode_decode", func(t *testing.T) {
+			encoded, err := parser.Encode(branchUndoLog)
+			assert.NoError(t, err)
+			assert.NotNil(t, encoded)
+			assert.True(t, len(encoded) > 0)
+
+			decoded, err := parser.Decode(encoded)
+			assert.NoError(t, err)
+			assert.NotNil(t, decoded)
+			assert.Equal(t, branchUndoLog.Xid, decoded.Xid)
+			assert.Equal(t, branchUndoLog.BranchID, decoded.BranchID)
+		})
+	}
+}
+
+func TestUndoLogParserInvalidData(t *testing.T) {
+	parsers := []UndoLogParser{
+		&JsonParser{},
+		&ProtobufParser{},
+	}
+
+	invalidData := []byte("invalid-data-that-cannot-be-parsed")
+
+	for _, parser := range parsers {
+		t.Run("test_"+parser.GetName()+"_decode_invalid_data", func(t *testing.T) {
+			decoded, err := parser.Decode(invalidData)
+			assert.Error(t, err)
+			if parser.GetName() == "json" {
+				assert.Nil(t, decoded)
+			} else {
+				assert.Nil(t, decoded)
+			}
+		})
+	}
+}
diff --git a/pkg/datasource/sql/undo/parser/parser_cache_test.go b/pkg/datasource/sql/undo/parser/parser_cache_test.go
index 5aebb39..109e087 100644
--- a/pkg/datasource/sql/undo/parser/parser_cache_test.go
+++ b/pkg/datasource/sql/undo/parser/parser_cache_test.go
@@ -24,6 +24,8 @@ import (
 )
 
 func TestInitCache(t *testing.T) {
+	// Reset cache to test initialization
+	cache = nil
 	assert.Nil(t, cache)
 	initCache()
 	assert.NotNil(t, cache)
@@ -46,34 +48,73 @@ func TestLoad(t *testing.T) {
 	assert.NotNil(t, jsonParser)
 }
 
-func TestLoad_NotFound(t *testing.T) {
-	parser, err := GetCache().Load("nonexistent")
-	assert.NotNil(t, err)
+func TestLoadJsonParser(t *testing.T) {
+	cache := GetCache()
+	parser, err := cache.Load("json")
+	assert.NoError(t, err)
+	assert.NotNil(t, parser)
+	assert.Equal(t, "json", parser.GetName())
+	assert.IsType(t, &JsonParser{}, parser)
+}
+
+func TestLoadProtobufParser(t *testing.T) {
+	cache := GetCache()
+	parser, err := cache.Load("protobuf")
+	assert.NoError(t, err)
+	assert.NotNil(t, parser)
+	assert.Equal(t, "protobuf", parser.GetName())
+	assert.IsType(t, &ProtobufParser{}, parser)
+}
+
+func TestLoadNonExistentParser(t *testing.T) {
+	cache := GetCache()
+	parser, err := cache.Load("non-existent")
+	assert.Error(t, err)
 	assert.Nil(t, parser)
-	assert.Contains(t, err.Error(), "not found")
+	assert.Contains(t, err.Error(), "undo log parser type non-existent not found")
 }
 
-func TestLoad_Protobuf(t *testing.T) {
-	protobufParser, err := GetCache().Load("protobuf")
-	assert.Nil(t, err)
-	assert.NotNil(t, protobufParser)
-	assert.Equal(t, "protobuf", protobufParser.GetName())
+func TestLoadEmptyName(t *testing.T) {
+	cache := GetCache()
+	parser, err := cache.Load("")
+	assert.Error(t, err)
+	assert.Nil(t, parser)
+	assert.Contains(t, err.Error(), "undo log parser type  not found")
 }
 
-func TestUndoLogParserCache_Store(t *testing.T) {
+func TestCacheInitialization(t *testing.T) {
 	cache := GetCache()
 	assert.NotNil(t, cache)
+	assert.NotNil(t, cache.serializerNameToParser)
+	assert.Len(t, cache.serializerNameToParser, 2)
 
-	// Verify both parsers are stored
-	jsonParser, err := cache.Load("json")
-	assert.NoError(t, err)
-	assert.NotNil(t, jsonParser)
+	_, existsJson := cache.serializerNameToParser["json"]
+	assert.True(t, existsJson)
 
-	protobufParser, err := cache.Load("protobuf")
-	assert.NoError(t, err)
-	assert.NotNil(t, protobufParser)
+	_, existsProtobuf := cache.serializerNameToParser["protobuf"]
+	assert.True(t, existsProtobuf)
+}
+
+func TestCacheSingleton(t *testing.T) {
+	cache1 := GetCache()
+	cache2 := GetCache()
+	assert.Same(t, cache1, cache2)
 }
 
-func TestDefaultSerializer(t *testing.T) {
-	assert.Equal(t, "json", DefaultSerializer)
+func TestStoreMethod(t *testing.T) {
+	cache := GetCache()
+
+	mockParser := &mockUndoLogParser{
+		name:           "test-mock",
+		defaultContent: []byte("test"),
+		shouldError:    false,
+	}
+
+	cache.store(mockParser)
+
+	loadedParser, err := cache.Load("test-mock")
+	assert.NoError(t, err)
+	assert.NotNil(t, loadedParser)
+	assert.Equal(t, "test-mock", loadedParser.GetName())
+	assert.Same(t, mockParser, loadedParser)
 }
diff --git a/pkg/datasource/sql/undo/parser/protobuf_conversion_test.go b/pkg/datasource/sql/undo/parser/protobuf_conversion_test.go
new file mode 100644
index 0000000..57f2252
--- /dev/null
+++ b/pkg/datasource/sql/undo/parser/protobuf_conversion_test.go
@@ -0,0 +1,422 @@
+/*
+ * 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 parser
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"seata.apache.org/seata-go/pkg/datasource/sql/types"
+	"seata.apache.org/seata-go/pkg/datasource/sql/undo"
+)
+
+func TestConvertToProto(t *testing.T) {
+	tests := []struct {
+		name        string
+		intreeLog   *undo.BranchUndoLog
+		expectError bool
+	}{
+		{
+			name: "convert simple branch undo log",
+			intreeLog: &undo.BranchUndoLog{
+				Xid:      "test-xid-123",
+				BranchID: 456789,
+				Logs: []undo.SQLUndoLog{
+					{
+						SQLType:   types.SQLTypeInsert,
+						TableName: "test_table",
+						AfterImage: &types.RecordImage{
+							TableName: "test_table",
+							SQLType:   types.SQLTypeInsert,
+							Rows: []types.RowImage{
+								{
+									Columns: []types.ColumnImage{
+										{
+											KeyType:    types.PrimaryKey.Number(),
+											ColumnName: "id",
+											ColumnType: types.JDBCTypeBigInt,
+											Value:      int64(123),
+										},
+										{
+											KeyType:    types.IndexTypeNull,
+											ColumnName: "name",
+											ColumnType: types.JDBCTypeVarchar,
+											Value:      "test_name",
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectError: false,
+		},
+		{
+			name: "convert with before image",
+			intreeLog: &undo.BranchUndoLog{
+				Xid:      "test-xid-456",
+				BranchID: 789123,
+				Logs: []undo.SQLUndoLog{
+					{
+						SQLType:   types.SQLTypeUpdate,
+						TableName: "user_table",
+						BeforeImage: &types.RecordImage{
+							TableName: "user_table",
+							SQLType:   types.SQLTypeUpdate,
+							Rows: []types.RowImage{
+								{
+									Columns: []types.ColumnImage{
+										{
+											KeyType:    types.PrimaryKey.Number(),
+											ColumnName: "user_id",
+											ColumnType: types.JDBCTypeBigInt,
+											Value:      int64(999),
+										},
+										{
+											KeyType:    types.IndexTypeNull,
+											ColumnName: "username",
+											ColumnType: types.JDBCTypeVarchar,
+											Value:      "old_username",
+										},
+									},
+								},
+							},
+						},
+						AfterImage: &types.RecordImage{
+							TableName: "user_table",
+							SQLType:   types.SQLTypeUpdate,
+							Rows: []types.RowImage{
+								{
+									Columns: []types.ColumnImage{
+										{
+											KeyType:    types.PrimaryKey.Number(),
+											ColumnName: "user_id",
+											ColumnType: types.JDBCTypeBigInt,
+											Value:      int64(999),
+										},
+										{
+											KeyType:    types.IndexTypeNull,
+											ColumnName: "username",
+											ColumnType: types.JDBCTypeVarchar,
+											Value:      "new_username",
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectError: false,
+		},
+		{
+			name: "convert with nil images",
+			intreeLog: &undo.BranchUndoLog{
+				Xid:      "test-xid-789",
+				BranchID: 111222,
+				Logs: []undo.SQLUndoLog{
+					{
+						SQLType:     types.SQLTypeDelete,
+						TableName:   "delete_table",
+						BeforeImage: nil,
+						AfterImage:  nil,
+					},
+				},
+			},
+			expectError: false,
+		},
+		{
+			name: "convert with multiple logs",
+			intreeLog: &undo.BranchUndoLog{
+				Xid:      "test-xid-multi",
+				BranchID: 333444,
+				Logs: []undo.SQLUndoLog{
+					{
+						SQLType:   types.SQLTypeInsert,
+						TableName: "table1",
+					},
+					{
+						SQLType:   types.SQLTypeUpdate,
+						TableName: "table2",
+					},
+				},
+			},
+			expectError: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			protoLog := ConvertToProto(tt.intreeLog)
+
+			assert.NotNil(t, protoLog)
+			assert.Equal(t, tt.intreeLog.Xid, protoLog.Xid)
+			assert.Equal(t, tt.intreeLog.BranchID, protoLog.BranchID)
+			assert.Equal(t, len(tt.intreeLog.Logs), len(protoLog.Logs))
+
+			for i, sqlLog := range tt.intreeLog.Logs {
+				protoSqlLog := protoLog.Logs[i]
+				assert.Equal(t, SQLType(sqlLog.SQLType), protoSqlLog.SQLType)
+				assert.Equal(t, sqlLog.TableName, protoSqlLog.TableName)
+
+				if sqlLog.BeforeImage != nil {
+					assert.NotNil(t, protoSqlLog.BeforeImage)
+					assert.Equal(t, sqlLog.BeforeImage.TableName, protoSqlLog.BeforeImage.TableName)
+					assert.Equal(t, SQLType(sqlLog.BeforeImage.SQLType), protoSqlLog.BeforeImage.SQLType)
+				} else {
+					assert.Nil(t, protoSqlLog.BeforeImage)
+				}
+
+				if sqlLog.AfterImage != nil {
+					assert.NotNil(t, protoSqlLog.AfterImage)
+					assert.Equal(t, sqlLog.AfterImage.TableName, protoSqlLog.AfterImage.TableName)
+					assert.Equal(t, SQLType(sqlLog.AfterImage.SQLType), protoSqlLog.AfterImage.SQLType)
+				} else {
+					assert.Nil(t, protoSqlLog.AfterImage)
+				}
+			}
+		})
+	}
+}
+
+func TestConvertToIntree(t *testing.T) {
+	tests := []struct {
+		name           string
+		protoLog       *BranchUndoLog
+		expectXid      string
+		expectBranchID uint64
+	}{
+		{
+			name: "convert simple protobuf log",
+			protoLog: &BranchUndoLog{
+				Xid:      "proto-xid-123",
+				BranchID: 654321,
+				Logs: []*SQLUndoLog{
+					{
+						SQLType:   SQLType_SQLTypeInsert,
+						TableName: "proto_table",
+						AfterImage: &RecordImage{
+							TableName: "proto_table",
+							SQLType:   SQLType_SQLTypeInsert,
+							Rows: []*RowImage{
+								{
+									Columns: []*ColumnImage{
+										{
+											KeyType:    IndexType_IndexTypePrimaryKey,
+											ColumnName: "id",
+											ColumnType: JDBCType_JDBCTypeBigInt,
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectXid:      "proto-xid-123",
+			expectBranchID: 654321,
+		},
+		{
+			name: "convert with multiple logs",
+			protoLog: &BranchUndoLog{
+				Xid:      "proto-xid-multi",
+				BranchID: 777888,
+				Logs: []*SQLUndoLog{
+					{
+						SQLType:   SQLType_SQLTypeUpdate,
+						TableName: "table_a",
+						BeforeImage: &RecordImage{
+							TableName: "table_a",
+							SQLType:   SQLType_SQLTypeUpdate,
+						},
+					},
+					{
+						SQLType:   SQLType_SQLTypeDelete,
+						TableName: "table_b",
+					},
+				},
+			},
+			expectXid:      "proto-xid-multi",
+			expectBranchID: 777888,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			intreeLog := ConvertToIntree(tt.protoLog)
+
+			assert.NotNil(t, intreeLog)
+			assert.Equal(t, tt.expectXid, intreeLog.Xid)
+			assert.Equal(t, tt.expectBranchID, intreeLog.BranchID)
+			assert.Equal(t, len(tt.protoLog.Logs), len(intreeLog.Logs))
+
+			for i, protoSqlLog := range tt.protoLog.Logs {
+				intreeSqlLog := intreeLog.Logs[i]
+				assert.Equal(t, types.SQLType(protoSqlLog.SQLType), intreeSqlLog.SQLType)
+				assert.Equal(t, protoSqlLog.TableName, intreeSqlLog.TableName)
+
+				if protoSqlLog.BeforeImage != nil {
+					assert.NotNil(t, intreeSqlLog.BeforeImage)
+					assert.Equal(t, protoSqlLog.BeforeImage.TableName, intreeSqlLog.BeforeImage.TableName)
+					assert.Equal(t, types.SQLType(protoSqlLog.BeforeImage.SQLType), intreeSqlLog.BeforeImage.SQLType)
+				} else {
+					assert.Nil(t, intreeSqlLog.BeforeImage)
+				}
+
+				if protoSqlLog.AfterImage != nil {
+					assert.NotNil(t, intreeSqlLog.AfterImage)
+					assert.Equal(t, protoSqlLog.AfterImage.TableName, intreeSqlLog.AfterImage.TableName)
+					assert.Equal(t, types.SQLType(protoSqlLog.AfterImage.SQLType), intreeSqlLog.AfterImage.SQLType)
+				} else {
+					assert.Nil(t, intreeSqlLog.AfterImage)
+				}
+			}
+		})
+	}
+}
+
+func TestConvertToProtoAndBack(t *testing.T) {
+	originalLog := &undo.BranchUndoLog{
+		Xid:      "round-trip-test",
+		BranchID: 999888,
+		Logs: []undo.SQLUndoLog{
+			{
+				SQLType:   types.SQLTypeUpdate,
+				TableName: "round_trip_table",
+				BeforeImage: &types.RecordImage{
+					TableName: "round_trip_table",
+					SQLType:   types.SQLTypeUpdate,
+					Rows: []types.RowImage{
+						{
+							Columns: []types.ColumnImage{
+								{
+									KeyType:    types.PrimaryKey.Number(),
+									ColumnName: "id",
+									ColumnType: types.JDBCTypeBigInt,
+									Value:      int64(111),
+								},
+								{
+									KeyType:    types.IndexTypeNull,
+									ColumnName: "status",
+									ColumnType: types.JDBCTypeVarchar,
+									Value:      "old_status",
+								},
+							},
+						},
+					},
+				},
+				AfterImage: &types.RecordImage{
+					TableName: "round_trip_table",
+					SQLType:   types.SQLTypeUpdate,
+					Rows: []types.RowImage{
+						{
+							Columns: []types.ColumnImage{
+								{
+									KeyType:    types.PrimaryKey.Number(),
+									ColumnName: "id",
+									ColumnType: types.JDBCTypeBigInt,
+									Value:      int64(111),
+								},
+								{
+									KeyType:    types.IndexTypeNull,
+									ColumnName: "status",
+									ColumnType: types.JDBCTypeVarchar,
+									Value:      "new_status",
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	protoLog := ConvertToProto(originalLog)
+	convertedBack := ConvertToIntree(protoLog)
+
+	assert.Equal(t, originalLog.Xid, convertedBack.Xid)
+	assert.Equal(t, originalLog.BranchID, convertedBack.BranchID)
+	assert.Equal(t, len(originalLog.Logs), len(convertedBack.Logs))
+}
+
+func TestConvertAnyToInterface(t *testing.T) {
+	testValues := []interface{}{
+		"string_value",
+		int64(12345),
+		float64(123.45),
+		true,
+		nil,
+	}
+
+	for i, value := range testValues {
+		t.Run("test_convert_any", func(t *testing.T) {
+			anyValue, err := convertInterfaceToAny(value)
+			require.NoError(t, err)
+			require.NotNil(t, anyValue)
+
+			convertedBack, err := convertAnyToInterface(anyValue)
+			if value == nil {
+				assert.NoError(t, err)
+				assert.Nil(t, convertedBack)
+			} else {
+				assert.NoError(t, err)
+				// JSON serialization converts int64 to float64, so we need to handle this
+				if i == 1 { // int64(12345) case
+					// Check if the value is the expected float64 representation
+					if floatVal, ok := convertedBack.(float64); ok {
+						assert.Equal(t, float64(12345), floatVal)
+					} else {
+						// If it's still int64, that's also acceptable
+						assert.Equal(t, value, convertedBack)
+					}
+				} else {
+					assert.Equal(t, value, convertedBack)
+				}
+			}
+		})
+	}
+}
+
+func TestConvertInterfaceToAnyWithComplexTypes(t *testing.T) {
+	complexValues := []interface{}{
+		map[string]interface{}{
+			"key1": "value1",
+			"key2": 123,
+		},
+		[]interface{}{1, 2, 3, "test"},
+		struct {
+			Name string
+			Age  int
+		}{Name: "test", Age: 25},
+	}
+
+	for i, value := range complexValues {
+		t.Run("test_complex_convert", func(t *testing.T) {
+			anyValue, err := convertInterfaceToAny(value)
+			assert.NoError(t, err, "Failed to convert complex value %d", i)
+			assert.NotNil(t, anyValue)
+
+			convertedBack, err := convertAnyToInterface(anyValue)
+			assert.NoError(t, err)
+			assert.NotNil(t, convertedBack)
+		})
+	}
+}
diff --git a/pkg/datasource/sql/undo/parser/protobuf_generated_test.go b/pkg/datasource/sql/undo/parser/protobuf_generated_test.go
new file mode 100644
index 0000000..95dbaf3
--- /dev/null
+++ b/pkg/datasource/sql/undo/parser/protobuf_generated_test.go
@@ -0,0 +1,362 @@
+/*
+ * 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 parser
+
+import (
+	"testing"
+
+	"github.com/golang/protobuf/ptypes/any"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBranchUndoLogMethods(t *testing.T) {
+	log := &BranchUndoLog{
+		Xid:      "test-xid",
+		BranchID: 12345,
+		Logs: []*SQLUndoLog{
+			{
+				SQLType:   SQLType_SQLTypeInsert,
+				TableName: "test_table",
+			},
+		},
+	}
+
+	// Test getter methods
+	assert.Equal(t, "test-xid", log.GetXid())
+	assert.Equal(t, uint64(12345), log.GetBranchID())
+	assert.Len(t, log.GetLogs(), 1)
+
+	// Test String method
+	str := log.String()
+	assert.NotEmpty(t, str)
+
+	// Test Reset method
+	log.Reset()
+	assert.Empty(t, log.GetXid())
+	assert.Equal(t, uint64(0), log.GetBranchID())
+	assert.Len(t, log.GetLogs(), 0)
+
+	// Test ProtoMessage method
+	log.ProtoMessage()
+
+	// Test ProtoReflect method
+	msg := log.ProtoReflect()
+	assert.NotNil(t, msg)
+
+	// Test Descriptor method
+	_, _ = log.Descriptor()
+}
+
+func TestSQLUndoLogMethods(t *testing.T) {
+	undoLog := &SQLUndoLog{
+		SQLType:   SQLType_SQLTypeUpdate,
+		TableName: "users",
+		BeforeImage: &RecordImage{
+			TableName: "users",
+			SQLType:   SQLType_SQLTypeUpdate,
+		},
+		AfterImage: &RecordImage{
+			TableName: "users",
+			SQLType:   SQLType_SQLTypeUpdate,
+		},
+	}
+
+	// Test getter methods
+	assert.Equal(t, SQLType_SQLTypeUpdate, undoLog.GetSQLType())
+	assert.Equal(t, "users", undoLog.GetTableName())
+	assert.NotNil(t, undoLog.GetBeforeImage())
+	assert.NotNil(t, undoLog.GetAfterImage())
+
+	// Test String method
+	str := undoLog.String()
+	assert.NotEmpty(t, str)
+
+	// Test Reset method
+	undoLog.Reset()
+	assert.Equal(t, SQLType_SQLTypeSelect, undoLog.GetSQLType()) // Default enum value
+	assert.Empty(t, undoLog.GetTableName())
+
+	// Test ProtoMessage method
+	undoLog.ProtoMessage()
+
+	// Test ProtoReflect method
+	msg := undoLog.ProtoReflect()
+	assert.NotNil(t, msg)
+
+	// Test Descriptor method
+	_, _ = undoLog.Descriptor()
+}
+
+func TestRecordImageMethods(t *testing.T) {
+	recordImage := &RecordImage{
+		Index:     1,
+		TableName: "test_table",
+		SQLType:   SQLType_SQLTypeInsert,
+		Rows: []*RowImage{
+			{
+				Columns: []*ColumnImage{
+					{
+						KeyType:    IndexType_IndexTypePrimaryKey,
+						ColumnName: "id",
+						ColumnType: JDBCType_JDBCTypeBigInt,
+					},
+				},
+			},
+		},
+		TableMeta: &TableMeta{
+			TableName: "test_table",
+		},
+	}
+
+	// Test getter methods
+	assert.Equal(t, int32(1), recordImage.GetIndex()) // Use int32 instead of uint32
+	assert.Equal(t, "test_table", recordImage.GetTableName())
+	assert.Equal(t, SQLType_SQLTypeInsert, recordImage.GetSQLType())
+	assert.Len(t, recordImage.GetRows(), 1)
+	assert.NotNil(t, recordImage.GetTableMeta())
+
+	// Test String method
+	str := recordImage.String()
+	assert.NotEmpty(t, str)
+
+	// Test Reset method
+	recordImage.Reset()
+	assert.Equal(t, int32(0), recordImage.GetIndex()) // Use int32 instead of uint32
+	assert.Empty(t, recordImage.GetTableName())
+
+	// Test ProtoMessage method
+	recordImage.ProtoMessage()
+
+	// Test ProtoReflect method
+	msg := recordImage.ProtoReflect()
+	assert.NotNil(t, msg)
+
+	// Test Descriptor method
+	_, _ = recordImage.Descriptor()
+}
+
+func TestRowImageMethods(t *testing.T) {
+	rowImage := &RowImage{
+		Columns: []*ColumnImage{
+			{
+				KeyType:    IndexType_IndexTypePrimaryKey,
+				ColumnName: "id",
+				ColumnType: JDBCType_JDBCTypeBigInt,
+				Value:      &any.Any{},
+			},
+		},
+	}
+
+	// Test getter methods
+	assert.Len(t, rowImage.GetColumns(), 1)
+
+	// Test String method
+	str := rowImage.String()
+	assert.NotEmpty(t, str)
+
+	// Test Reset method
+	rowImage.Reset()
+	assert.Len(t, rowImage.GetColumns(), 0)
+
+	// Test ProtoMessage method
+	rowImage.ProtoMessage()
+
+	// Test ProtoReflect method
+	msg := rowImage.ProtoReflect()
+	assert.NotNil(t, msg)
+
+	// Test Descriptor method
+	_, _ = rowImage.Descriptor()
+}
+
+func TestColumnImageMethods(t *testing.T) {
+	columnImage := &ColumnImage{
+		KeyType:    IndexType_IndexTypePrimaryKey,
+		ColumnName: "id",
+		ColumnType: JDBCType_JDBCTypeBigInt,
+		Value:      &any.Any{},
+	}
+
+	// Test getter methods
+	assert.Equal(t, IndexType_IndexTypePrimaryKey, columnImage.GetKeyType())
+	assert.Equal(t, "id", columnImage.GetColumnName())
+	assert.Equal(t, JDBCType_JDBCTypeBigInt, columnImage.GetColumnType())
+	assert.NotNil(t, columnImage.GetValue())
+
+	// Test String method
+	str := columnImage.String()
+	assert.NotEmpty(t, str)
+
+	// Test Reset method
+	columnImage.Reset()
+	assert.Equal(t, IndexType_IndexTypeNull, columnImage.GetKeyType()) // Default enum value
+	assert.Empty(t, columnImage.GetColumnName())
+
+	// Test ProtoMessage method
+	columnImage.ProtoMessage()
+
+	// Test ProtoReflect method
+	msg := columnImage.ProtoReflect()
+	assert.NotNil(t, msg)
+
+	// Test Descriptor method
+	_, _ = columnImage.Descriptor()
+}
+
+func TestTableMetaMethods(t *testing.T) {
+	tableMeta := &TableMeta{
+		TableName:   "test_table",
+		ColumnNames: []string{"id", "name"},
+	}
+
+	// Test getter methods
+	assert.Equal(t, "test_table", tableMeta.GetTableName())
+	assert.Len(t, tableMeta.GetColumnNames(), 2)
+	// These return nil when empty, so test appropriately
+	columns := tableMeta.GetColumns()
+	indexs := tableMeta.GetIndexs()
+	_ = columns // May be nil, just test it doesn't panic
+	_ = indexs  // May be nil, just test it doesn't panic
+
+	// Test String method
+	str := tableMeta.String()
+	assert.NotEmpty(t, str)
+
+	// Test Reset method
+	tableMeta.Reset()
+	assert.Empty(t, tableMeta.GetTableName())
+
+	// Test ProtoMessage method
+	tableMeta.ProtoMessage()
+
+	// Test ProtoReflect method
+	msg := tableMeta.ProtoReflect()
+	assert.NotNil(t, msg)
+
+	// Test Descriptor method
+	_, _ = tableMeta.Descriptor()
+}
+
+func TestColumnMetaMethods(t *testing.T) {
+	columnMeta := &ColumnMeta{
+		Schema:             "test_schema",
+		Table:              "test_table",
+		ColumnName:         "id",
+		ColumnType:         "bigint",
+		DatabaseTypeString: "mysql",
+	}
+
+	// Test getter methods
+	assert.Equal(t, "test_schema", columnMeta.GetSchema())
+	assert.Equal(t, "test_table", columnMeta.GetTable())
+	assert.Equal(t, "id", columnMeta.GetColumnName())
+	assert.Equal(t, "bigint", columnMeta.GetColumnType())
+	assert.Equal(t, "mysql", columnMeta.GetDatabaseTypeString())
+	// GetColumnDef returns []byte which may be nil when empty
+	columnDef := columnMeta.GetColumnDef()
+	_ = columnDef // May be nil, just test it doesn't panic
+	assert.Equal(t, false, columnMeta.GetAutoincrement())
+	assert.Equal(t, int32(0), columnMeta.GetDatabaseType())
+	assert.Empty(t, columnMeta.GetColumnKey())
+	assert.Equal(t, int32(0), columnMeta.GetIsNullable())
+	assert.Empty(t, columnMeta.GetExtra())
+
+	// Test String method
+	str := columnMeta.String()
+	assert.NotEmpty(t, str)
+
+	// Test Reset method
+	columnMeta.Reset()
+	assert.Empty(t, columnMeta.GetSchema())
+	assert.Empty(t, columnMeta.GetTable())
+
+	// Test ProtoMessage method
+	columnMeta.ProtoMessage()
+
+	// Test ProtoReflect method
+	msg := columnMeta.ProtoReflect()
+	assert.NotNil(t, msg)
+
+	// Test Descriptor method
+	_, _ = columnMeta.Descriptor()
+}
+
+func TestIndexMetaMethods(t *testing.T) {
+	indexMeta := &IndexMeta{
+		Schema:     "test_schema",
+		Table:      "test_table",
+		Name:       "pk_id",
+		ColumnName: "id",
+		IType:      IndexType_IndexTypePrimaryKey,
+		Columns: []*ColumnMeta{
+			{
+				ColumnName: "id",
+			},
+		},
+	}
+
+	// Test getter methods
+	assert.Equal(t, "test_schema", indexMeta.GetSchema())
+	assert.Equal(t, "test_table", indexMeta.GetTable())
+	assert.Equal(t, "pk_id", indexMeta.GetName())
+	assert.Equal(t, "id", indexMeta.GetColumnName())
+	assert.Equal(t, IndexType_IndexTypePrimaryKey, indexMeta.GetIType())
+	assert.Len(t, indexMeta.GetColumns(), 1)
+
+	// Test String method
+	str := indexMeta.String()
+	assert.NotEmpty(t, str)
+
+	// Test Reset method
+	indexMeta.Reset()
+	assert.Empty(t, indexMeta.GetSchema())
+	assert.Empty(t, indexMeta.GetTable())
+
+	// Test ProtoMessage method
+	indexMeta.ProtoMessage()
+
+	// Test ProtoReflect method
+	msg := indexMeta.ProtoReflect()
+	assert.NotNil(t, msg)
+
+	// Test Descriptor method
+	_, _ = indexMeta.Descriptor()
+}
+
+func TestEnumMethods(t *testing.T) {
+	// Test SQLType enum methods
+	sqlType := SQLType_SQLTypeInsert
+	assert.Equal(t, "SQLTypeInsert", sqlType.String())
+	assert.Equal(t, int32(1), int32(sqlType.Number()))
+	assert.NotNil(t, sqlType.Type())
+	assert.NotNil(t, sqlType.Descriptor())
+
+	// Test IndexType enum methods
+	indexType := IndexType_IndexTypePrimaryKey
+	assert.Equal(t, "IndexTypePrimaryKey", indexType.String())
+	assert.Equal(t, int32(1), int32(indexType.Number()))
+	assert.NotNil(t, indexType.Type())
+	assert.NotNil(t, indexType.Descriptor())
+
+	// Test JDBCType enum methods
+	jdbcType := JDBCType_JDBCTypeBigInt
+	assert.Equal(t, "JDBCTypeBigInt", jdbcType.String())
+	assert.Equal(t, int32(-5), int32(jdbcType.Number()))
+	assert.NotNil(t, jdbcType.Type())
+	assert.NotNil(t, jdbcType.Descriptor())
+}
diff --git a/pkg/datasource/sql/undo/undo_executor_holder_test.go b/pkg/datasource/sql/undo/undo_executor_holder_test.go
new file mode 100644
index 0000000..5b7e210
--- /dev/null
+++ b/pkg/datasource/sql/undo/undo_executor_holder_test.go
@@ -0,0 +1,117 @@
+/*
+ * 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 undo
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+type mockUndoExecutorHolderImpl struct {
+	mock.Mock
+}
+
+func (m *mockUndoExecutorHolderImpl) GetInsertExecutor(sqlUndoLog SQLUndoLog) UndoExecutor {
+	args := m.Called(sqlUndoLog)
+	if args.Get(0) == nil {
+		return nil
+	}
+	return args.Get(0).(UndoExecutor)
+}
+
+func (m *mockUndoExecutorHolderImpl) GetDeleteExecutor(sqlUndoLog SQLUndoLog) UndoExecutor {
+	args := m.Called(sqlUndoLog)
+	if args.Get(0) == nil {
+		return nil
+	}
+	return args.Get(0).(UndoExecutor)
+}
+
+func (m *mockUndoExecutorHolderImpl) GetUpdateExecutor(sqlUndoLog SQLUndoLog) UndoExecutor {
+	args := m.Called(sqlUndoLog)
+	if args.Get(0) == nil {
+		return nil
+	}
+	return args.Get(0).(UndoExecutor)
+}
+
+func TestUndoExecutorHolder_Interface(t *testing.T) {
+	holder := &mockUndoExecutorHolderImpl{}
+	assert.Implements(t, (*UndoExecutorHolder)(nil), holder)
+}
+
+func TestUndoExecutorHolder_GetInsertExecutor(t *testing.T) {
+	holder := &mockUndoExecutorHolderImpl{}
+	executor := &mockUndoExecutorImpl{}
+	sqlUndoLog := SQLUndoLog{TableName: "test_table"}
+
+	holder.On("GetInsertExecutor", sqlUndoLog).Return(executor)
+
+	result := holder.GetInsertExecutor(sqlUndoLog)
+	assert.Equal(t, executor, result)
+
+	holder.AssertExpectations(t)
+}
+
+func TestUndoExecutorHolder_GetDeleteExecutor(t *testing.T) {
+	holder := &mockUndoExecutorHolderImpl{}
+	executor := &mockUndoExecutorImpl{}
+	sqlUndoLog := SQLUndoLog{TableName: "test_table"}
+
+	holder.On("GetDeleteExecutor", sqlUndoLog).Return(executor)
+
+	result := holder.GetDeleteExecutor(sqlUndoLog)
+	assert.Equal(t, executor, result)
+
+	holder.AssertExpectations(t)
+}
+
+func TestUndoExecutorHolder_GetUpdateExecutor(t *testing.T) {
+	holder := &mockUndoExecutorHolderImpl{}
+	executor := &mockUndoExecutorImpl{}
+	sqlUndoLog := SQLUndoLog{TableName: "test_table"}
+
+	holder.On("GetUpdateExecutor", sqlUndoLog).Return(executor)
+
+	result := holder.GetUpdateExecutor(sqlUndoLog)
+	assert.Equal(t, executor, result)
+
+	holder.AssertExpectations(t)
+}
+
+func TestUndoExecutorHolder_ReturnsNil(t *testing.T) {
+	holder := &mockUndoExecutorHolderImpl{}
+	sqlUndoLog := SQLUndoLog{TableName: "test_table"}
+
+	holder.On("GetInsertExecutor", sqlUndoLog).Return(nil)
+	holder.On("GetDeleteExecutor", sqlUndoLog).Return(nil)
+	holder.On("GetUpdateExecutor", sqlUndoLog).Return(nil)
+
+	insertResult := holder.GetInsertExecutor(sqlUndoLog)
+	assert.Nil(t, insertResult)
+
+	deleteResult := holder.GetDeleteExecutor(sqlUndoLog)
+	assert.Nil(t, deleteResult)
+
+	updateResult := holder.GetUpdateExecutor(sqlUndoLog)
+	assert.Nil(t, updateResult)
+
+	holder.AssertExpectations(t)
+}
diff --git a/pkg/datasource/sql/undo/undo_executor_test.go b/pkg/datasource/sql/undo/undo_executor_test.go
new file mode 100644
index 0000000..6426a10
--- /dev/null
+++ b/pkg/datasource/sql/undo/undo_executor_test.go
@@ -0,0 +1,57 @@
+/*
+ * 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 undo
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+
+	"seata.apache.org/seata-go/pkg/datasource/sql/types"
+)
+
+type mockUndoExecutorImpl struct {
+	mock.Mock
+}
+
+func (m *mockUndoExecutorImpl) ExecuteOn(ctx context.Context, dbType types.DBType, conn *sql.Conn) error {
+	args := m.Called(ctx, dbType, conn)
+	return args.Error(0)
+}
+
+func TestUndoExecutor_Interface(t *testing.T) {
+	executor := &mockUndoExecutorImpl{}
+	assert.Implements(t, (*UndoExecutor)(nil), executor)
+}
+
+func TestUndoExecutor_ExecuteOn(t *testing.T) {
+	executor := &mockUndoExecutorImpl{}
+	ctx := context.Background()
+	dbType := types.DBTypeMySQL
+	var conn *sql.Conn
+
+	executor.On("ExecuteOn", ctx, dbType, conn).Return(nil)
+
+	err := executor.ExecuteOn(ctx, dbType, conn)
+	assert.NoError(t, err)
+
+	executor.AssertExpectations(t)
+}
diff --git a/pkg/datasource/sql/undo/undo_test.go b/pkg/datasource/sql/undo/undo_test.go
new file mode 100644
index 0000000..43a7317
--- /dev/null
+++ b/pkg/datasource/sql/undo/undo_test.go
@@ -0,0 +1,367 @@
+/*
+ * 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 undo
+
+import (
+	"context"
+	"database/sql"
+	"database/sql/driver"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+
+	"seata.apache.org/seata-go/pkg/datasource/sql/types"
+)
+
+type mockUndoLogManager struct {
+	mock.Mock
+}
+
+func (m *mockUndoLogManager) Init() {
+	m.Called()
+}
+
+func (m *mockUndoLogManager) DeleteUndoLog(ctx context.Context, xid string, branchID int64, conn *sql.Conn) error {
+	args := m.Called(ctx, xid, branchID, conn)
+	return args.Error(0)
+}
+
+func (m *mockUndoLogManager) BatchDeleteUndoLog(xid []string, branchID []int64, conn *sql.Conn) error {
+	args := m.Called(xid, branchID, conn)
+	return args.Error(0)
+}
+
+func (m *mockUndoLogManager) FlushUndoLog(tranCtx *types.TransactionContext, conn driver.Conn) error {
+	args := m.Called(tranCtx, conn)
+	return args.Error(0)
+}
+
+func (m *mockUndoLogManager) RunUndo(ctx context.Context, xid string, branchID int64, conn *sql.DB, dbName string) error {
+	args := m.Called(ctx, xid, branchID, conn, dbName)
+	return args.Error(0)
+}
+
+func (m *mockUndoLogManager) DBType() types.DBType {
+	args := m.Called()
+	return args.Get(0).(types.DBType)
+}
+
+func (m *mockUndoLogManager) HasUndoLogTable(ctx context.Context, conn *sql.Conn) (bool, error) {
+	args := m.Called(ctx, conn)
+	return args.Bool(0), args.Error(1)
+}
+
+type mockUndoLogBuilder struct {
+	mock.Mock
+}
+
+func (m *mockUndoLogBuilder) BeforeImage(ctx context.Context, execCtx *types.ExecContext) ([]*types.RecordImage, error) {
+	args := m.Called(ctx, execCtx)
+	return args.Get(0).([]*types.RecordImage), args.Error(1)
+}
+
+func (m *mockUndoLogBuilder) AfterImage(ctx context.Context, execCtx *types.ExecContext, beforeImages []*types.RecordImage) ([]*types.RecordImage, error) {
+	args := m.Called(ctx, execCtx, beforeImages)
+	return args.Get(0).([]*types.RecordImage), args.Error(1)
+}
+
+func (m *mockUndoLogBuilder) GetExecutorType() types.ExecutorType {
+	args := m.Called()
+	return args.Get(0).(types.ExecutorType)
+}
+
+func TestRegisterUndoLogManager(t *testing.T) {
+	tests := []struct {
+		name    string
+		manager UndoLogManager
+		dbType  types.DBType
+		wantErr bool
+	}{
+		{
+			name:    "register mysql manager",
+			manager: &mockUndoLogManager{},
+			dbType:  types.DBTypeMySQL,
+			wantErr: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mockMgr := tt.manager.(*mockUndoLogManager)
+			mockMgr.On("DBType").Return(tt.dbType)
+			mockMgr.On("Init").Return()
+
+			err := RegisterUndoLogManager(tt.manager)
+			if tt.wantErr {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+
+			err = RegisterUndoLogManager(tt.manager)
+			assert.NoError(t, err)
+		})
+	}
+}
+
+func TestRegisterUndoLogBuilder(t *testing.T) {
+	executorType := types.InsertExecutor
+	builderFunc := func() UndoLogBuilder {
+		return &mockUndoLogBuilder{}
+	}
+
+	RegisterUndoLogBuilder(executorType, builderFunc)
+
+	builder := GetUndologBuilder(executorType)
+	assert.NotNil(t, builder)
+
+	RegisterUndoLogBuilder(executorType, builderFunc)
+	builder2 := GetUndologBuilder(executorType)
+	assert.NotNil(t, builder2)
+}
+
+func TestGetUndologBuilder(t *testing.T) {
+	tests := []struct {
+		name         string
+		executorType types.ExecutorType
+		setup        func()
+		want         bool
+	}{
+		{
+			name:         "get existing builder",
+			executorType: types.UpdateExecutor,
+			setup: func() {
+				RegisterUndoLogBuilder(types.UpdateExecutor, func() UndoLogBuilder {
+					return &mockUndoLogBuilder{}
+				})
+			},
+			want: true,
+		},
+		{
+			name:         "get non-existing builder",
+			executorType: types.DeleteExecutor,
+			setup:        func() {},
+			want:         false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tt.setup()
+			builder := GetUndologBuilder(tt.executorType)
+			if tt.want {
+				assert.NotNil(t, builder)
+			} else {
+				assert.Nil(t, builder)
+			}
+		})
+	}
+}
+
+func TestGetUndoLogManager(t *testing.T) {
+	mockMgr := &mockUndoLogManager{}
+	mockMgr.On("DBType").Return(types.DBTypeMySQL)
+	mockMgr.On("Init").Return()
+
+	err := RegisterUndoLogManager(mockMgr)
+	assert.NoError(t, err)
+
+	tests := []struct {
+		name    string
+		dbType  types.DBType
+		wantErr bool
+	}{
+		{
+			name:    "get existing manager",
+			dbType:  types.DBTypeMySQL,
+			wantErr: false,
+		},
+		{
+			name:    "get non-existing manager",
+			dbType:  types.DBTypePostgreSQL,
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			manager, err := GetUndoLogManager(tt.dbType)
+			if tt.wantErr {
+				assert.Error(t, err)
+				assert.Nil(t, manager)
+				assert.Contains(t, err.Error(), "not found UndoLogManager")
+			} else {
+				assert.NoError(t, err)
+				assert.NotNil(t, manager)
+			}
+		})
+	}
+}
+
+func TestUndologRecord_CanUndo(t *testing.T) {
+	tests := []struct {
+		name      string
+		logStatus UndoLogStatus
+		want      bool
+	}{
+		{
+			name:      "can undo - normal status",
+			logStatus: UndoLogStatusNormal,
+			want:      true,
+		},
+		{
+			name:      "cannot undo - global finished",
+			logStatus: UndoLogStatusGlobalFinished,
+			want:      false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			record := &UndologRecord{
+				LogStatus: tt.logStatus,
+			}
+			assert.Equal(t, tt.want, record.CanUndo())
+		})
+	}
+}
+
+func TestBranchUndoLog_Marshal(t *testing.T) {
+	branchLog := &BranchUndoLog{
+		Xid:      "test-xid",
+		BranchID: 123,
+		Logs:     []SQLUndoLog{},
+	}
+
+	result := branchLog.Marshal()
+	assert.Nil(t, result)
+}
+
+func TestBranchUndoLog_Reverse(t *testing.T) {
+	tests := []struct {
+		name     string
+		logs     []SQLUndoLog
+		expected []string
+	}{
+		{
+			name: "empty logs",
+			logs: []SQLUndoLog{},
+		},
+		{
+			name: "single log",
+			logs: []SQLUndoLog{
+				{TableName: "table1"},
+			},
+			expected: []string{"table1"},
+		},
+		{
+			name: "multiple logs",
+			logs: []SQLUndoLog{
+				{TableName: "table1"},
+				{TableName: "table2"},
+				{TableName: "table3"},
+			},
+			expected: []string{"table3", "table2", "table1"},
+		},
+		{
+			name: "even number of logs",
+			logs: []SQLUndoLog{
+				{TableName: "table1"},
+				{TableName: "table2"},
+				{TableName: "table3"},
+				{TableName: "table4"},
+			},
+			expected: []string{"table4", "table3", "table2", "table1"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			branchLog := &BranchUndoLog{
+				Logs: tt.logs,
+			}
+			branchLog.Reverse()
+
+			if len(tt.expected) == 0 {
+				assert.Equal(t, 0, len(branchLog.Logs))
+			} else {
+				assert.Equal(t, len(tt.expected), len(branchLog.Logs))
+				for i, expectedTable := range tt.expected {
+					assert.Equal(t, expectedTable, branchLog.Logs[i].TableName)
+				}
+			}
+		})
+	}
+}
+
+func TestSQLUndoLog_SetTableMeta(t *testing.T) {
+	tableMeta := &types.TableMeta{
+		TableName: "test_table",
+	}
+
+	beforeImage := &types.RecordImage{}
+	afterImage := &types.RecordImage{}
+
+	tests := []struct {
+		name        string
+		beforeImage *types.RecordImage
+		afterImage  *types.RecordImage
+	}{
+		{
+			name:        "both images exist",
+			beforeImage: beforeImage,
+			afterImage:  afterImage,
+		},
+		{
+			name:        "only before image",
+			beforeImage: beforeImage,
+			afterImage:  nil,
+		},
+		{
+			name:        "only after image",
+			beforeImage: nil,
+			afterImage:  afterImage,
+		},
+		{
+			name:        "no images",
+			beforeImage: nil,
+			afterImage:  nil,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			sqlLog := SQLUndoLog{
+				BeforeImage: tt.beforeImage,
+				AfterImage:  tt.afterImage,
+			}
+
+			sqlLog.SetTableMeta(tableMeta)
+
+			if tt.beforeImage != nil {
+				assert.Equal(t, tableMeta, tt.beforeImage.TableMeta)
+				assert.Equal(t, tableMeta.TableName, tt.beforeImage.TableName)
+			}
+			if tt.afterImage != nil {
+				assert.Equal(t, tableMeta, tt.afterImage.TableMeta)
+				assert.Equal(t, tableMeta.TableName, tt.afterImage.TableName)
+			}
+		})
+	}
+}