diff --git a/pkg/datasource/sql/conn.go b/pkg/datasource/sql/conn.go
index 7a2b0423..f316292b 100644
--- a/pkg/datasource/sql/conn.go
+++ b/pkg/datasource/sql/conn.go
@@ -244,6 +244,10 @@ func (c *Conn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, e
 	)
 }
 
+func (c *Conn) GetDbVersion() string {
+	return c.res.GetDbVersion()
+}
+
 func (c *Conn) GetAutoCommit() bool {
 	return c.autoCommit
 }
diff --git a/pkg/datasource/sql/conn_at.go b/pkg/datasource/sql/conn_at.go
index f1d0f5ed..abc31f26 100644
--- a/pkg/datasource/sql/conn_at.go
+++ b/pkg/datasource/sql/conn_at.go
@@ -63,6 +63,7 @@ func (c *ATConn) QueryContext(ctx context.Context, query string, args []driver.N
 			NamedValues:          args,
 			Conn:                 c.targetConn,
 			DBName:               c.dbName,
+			DbVersion:            c.GetDbVersion(),
 			IsSupportsSavepoints: true,
 			IsAutoCommit:         c.GetAutoCommit(),
 		}
@@ -102,6 +103,7 @@ func (c *ATConn) ExecContext(ctx context.Context, query string, args []driver.Na
 			NamedValues:          args,
 			Conn:                 c.targetConn,
 			DBName:               c.dbName,
+			DbVersion:            c.GetDbVersion(),
 			IsSupportsSavepoints: true,
 			IsAutoCommit:         c.GetAutoCommit(),
 		}
diff --git a/pkg/datasource/sql/exec/at/base_executor.go b/pkg/datasource/sql/exec/at/base_executor.go
index 05f44057..a2595be8 100644
--- a/pkg/datasource/sql/exec/at/base_executor.go
+++ b/pkg/datasource/sql/exec/at/base_executor.go
@@ -23,15 +23,17 @@ import (
 	"database/sql"
 	"database/sql/driver"
 	"fmt"
-	"seata.apache.org/seata-go/pkg/datasource/sql/undo"
 	"strings"
 
 	"github.com/arana-db/parser/ast"
+	"github.com/arana-db/parser/model"
 	"github.com/arana-db/parser/test_driver"
 	gxsort "github.com/dubbogo/gost/sort"
+	"github.com/pkg/errors"
 
 	"seata.apache.org/seata-go/pkg/datasource/sql/exec"
 	"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/util"
 	"seata.apache.org/seata-go/pkg/util/reflectx"
 )
@@ -98,7 +100,13 @@ func (b *baseExecutor) buildSelectArgs(stmt *ast.SelectStmt, args []driver.Named
 		selectArgs       = make([]driver.NamedValue, 0)
 	)
 
+	b.traversalArgs(stmt.From.TableRefs, &selectArgsIndexs)
 	b.traversalArgs(stmt.Where, &selectArgsIndexs)
+	if stmt.GroupBy != nil {
+		for _, item := range stmt.GroupBy.Items {
+			b.traversalArgs(item, &selectArgsIndexs)
+		}
+	}
 	if stmt.OrderBy != nil {
 		for _, item := range stmt.OrderBy.Items {
 			b.traversalArgs(item, &selectArgsIndexs)
@@ -143,6 +151,16 @@ func (b *baseExecutor) traversalArgs(node ast.Node, argsIndex *[]int32) {
 			b.traversalArgs(exprs[i], argsIndex)
 		}
 		break
+	case *ast.Join:
+		exprs := node.(*ast.Join)
+		b.traversalArgs(exprs.Left, argsIndex)
+		if exprs.Right != nil {
+			b.traversalArgs(exprs.Right, argsIndex)
+		}
+		if exprs.On != nil {
+			b.traversalArgs(exprs.On.Expr, argsIndex)
+		}
+		break
 	case *test_driver.ParamMarkerExpr:
 		*argsIndex = append(*argsIndex, int32(node.(*test_driver.ParamMarkerExpr).Order))
 		break
@@ -230,6 +248,64 @@ func (b *baseExecutor) containsPKByName(meta *types.TableMeta, columns []string)
 	return matchCounter == len(pkColumnNameList)
 }
 
+func (u *baseExecutor) buildSelectFields(ctx context.Context, tableMeta *types.TableMeta, tableAliases string, inUseFields []*ast.Assignment) ([]*ast.SelectField, error) {
+	fields := make([]*ast.SelectField, 0, len(inUseFields))
+
+	tableName := tableAliases
+	if tableAliases == "" {
+		tableName = tableMeta.TableName
+	}
+	if undo.UndoConfig.OnlyCareUpdateColumns {
+		for _, column := range inUseFields {
+			tn := column.Column.Table.O
+			if tn != "" && tn != tableName {
+				continue
+			}
+
+			fields = append(fields, &ast.SelectField{
+				Expr: &ast.ColumnNameExpr{
+					Name: column.Column,
+				},
+			})
+		}
+
+		if len(fields) == 0 {
+			return fields, nil
+		}
+
+		// select indexes columns
+		for _, columnName := range tableMeta.GetPrimaryKeyOnlyName() {
+			fields = append(fields, &ast.SelectField{
+				Expr: &ast.ColumnNameExpr{
+					Name: &ast.ColumnName{
+						Table: model.CIStr{
+							O: tableName,
+							L: tableName,
+						},
+						Name: model.CIStr{
+							O: columnName,
+							L: columnName,
+						},
+					},
+				},
+			})
+		}
+	} else {
+		fields = append(fields, &ast.SelectField{
+			Expr: &ast.ColumnNameExpr{
+				Name: &ast.ColumnName{
+					Name: model.CIStr{
+						O: "*",
+						L: "*",
+					},
+				},
+			},
+		})
+	}
+
+	return fields, nil
+}
+
 func getSqlNullValue(value interface{}) interface{} {
 	if value == nil {
 		return nil
@@ -393,3 +469,23 @@ func (b *baseExecutor) buildLockKey(records *types.RecordImage, meta types.Table
 
 	return lockKeys.String()
 }
+
+func (b *baseExecutor) rowsPrepare(ctx context.Context, conn driver.Conn, selectSQL string, selectArgs []driver.NamedValue) (driver.Rows, error) {
+	var queryer driver.Queryer
+
+	queryerContext, ok := conn.(driver.QueryerContext)
+	if !ok {
+		queryer, ok = conn.(driver.Queryer)
+	}
+	if ok {
+		var err error
+		rows, err = util.CtxDriverQuery(ctx, queryerContext, queryer, selectSQL, selectArgs)
+
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		return nil, errors.New("target conn should been driver.QueryerContext or driver.Queryer")
+	}
+	return rows, nil
+}
diff --git a/pkg/datasource/sql/exec/at/update_executor.go b/pkg/datasource/sql/exec/at/update_executor.go
index 0ac9b049..0f14e97b 100644
--- a/pkg/datasource/sql/exec/at/update_executor.go
+++ b/pkg/datasource/sql/exec/at/update_executor.go
@@ -21,17 +21,17 @@ import (
 	"context"
 	"database/sql/driver"
 	"fmt"
+	"github.com/arana-db/parser/model"
+	"seata.apache.org/seata-go/pkg/datasource/sql/util"
 	"strings"
 
 	"github.com/arana-db/parser/ast"
 	"github.com/arana-db/parser/format"
-	"github.com/arana-db/parser/model"
 
 	"seata.apache.org/seata-go/pkg/datasource/sql/datasource"
 	"seata.apache.org/seata-go/pkg/datasource/sql/exec"
 	"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/util"
 	"seata.apache.org/seata-go/pkg/util/bytes"
 	"seata.apache.org/seata-go/pkg/util/log"
 )
@@ -49,6 +49,10 @@ type updateExecutor struct {
 
 // NewUpdateExecutor get update executor
 func NewUpdateExecutor(parserCtx *types.ParseContext, execContent *types.ExecContext, hooks []exec.SQLHook) executor {
+	// Because update join cannot be clearly identified when SQL cannot be parsed
+	if parserCtx.UpdateStmt.TableRefs.TableRefs.Right != nil {
+		return NewUpdateJoinExecutor(parserCtx, execContent, hooks)
+	}
 	return &updateExecutor{parserCtx: parserCtx, execContext: execContent, baseExecutor: baseExecutor{hooks: hooks}}
 }
 
diff --git a/pkg/datasource/sql/exec/at/update_join_executor.go b/pkg/datasource/sql/exec/at/update_join_executor.go
new file mode 100644
index 00000000..55261064
--- /dev/null
+++ b/pkg/datasource/sql/exec/at/update_join_executor.go
@@ -0,0 +1,344 @@
+/*
+* 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 at
+
+import (
+	"context"
+	"database/sql/driver"
+	"errors"
+	"io"
+	"reflect"
+	"strings"
+
+	"github.com/arana-db/parser/ast"
+	"github.com/arana-db/parser/format"
+	"github.com/arana-db/parser/model"
+
+	"seata.apache.org/seata-go/pkg/datasource/sql/datasource"
+	"seata.apache.org/seata-go/pkg/datasource/sql/exec"
+	"seata.apache.org/seata-go/pkg/datasource/sql/types"
+	"seata.apache.org/seata-go/pkg/datasource/sql/util"
+	"seata.apache.org/seata-go/pkg/util/bytes"
+	"seata.apache.org/seata-go/pkg/util/log"
+)
+
+// updateJoinExecutor execute update SQL
+type updateJoinExecutor struct {
+	baseExecutor
+	parserCtx                       *types.ParseContext
+	execContext                     *types.ExecContext
+	isLowerSupportGroupByPksVersion bool
+	sqlMode                         string
+	tableAliasesMap                 map[string]string
+}
+
+// NewUpdateJoinExecutor get executor
+func NewUpdateJoinExecutor(parserCtx *types.ParseContext, execContent *types.ExecContext, hooks []exec.SQLHook) executor {
+	minimumVersion, _ := util.ConvertDbVersion("5.7.5")
+	currentVersion, _ := util.ConvertDbVersion(execContent.DbVersion)
+	return &updateJoinExecutor{
+		parserCtx:                       parserCtx,
+		execContext:                     execContent,
+		baseExecutor:                    baseExecutor{hooks: hooks},
+		isLowerSupportGroupByPksVersion: currentVersion < minimumVersion,
+		tableAliasesMap:                 make(map[string]string, 0),
+	}
+}
+
+// ExecContext exec SQL, and generate before image and after image
+func (u *updateJoinExecutor) ExecContext(ctx context.Context, f exec.CallbackWithNamedValue) (types.ExecResult, error) {
+	u.beforeHooks(ctx, u.execContext)
+	defer func() {
+		u.afterHooks(ctx, u.execContext)
+	}()
+
+	if u.isAstStmtValid() {
+		u.tableAliasesMap = u.parseTableName(u.parserCtx.UpdateStmt.TableRefs.TableRefs)
+	}
+
+	beforeImages, err := u.beforeImage(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	res, err := f(ctx, u.execContext.Query, u.execContext.NamedValues)
+	if err != nil {
+		return nil, err
+	}
+
+	afterImages, err := u.afterImage(ctx, beforeImages)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(afterImages) != len(beforeImages) {
+		return nil, errors.New("Before image size is not equaled to after image size, probably because you updated the primary keys.")
+	}
+
+	u.execContext.TxCtx.RoundImages.AppendBeofreImages(beforeImages)
+	u.execContext.TxCtx.RoundImages.AppendAfterImages(afterImages)
+
+	return res, nil
+}
+
+func (u *updateJoinExecutor) isAstStmtValid() bool {
+	return u.parserCtx != nil && u.parserCtx.UpdateStmt != nil && u.parserCtx.UpdateStmt.TableRefs.TableRefs.Right != nil
+}
+
+func (u *updateJoinExecutor) beforeImage(ctx context.Context) ([]*types.RecordImage, error) {
+	if !u.isAstStmtValid() {
+		return nil, nil
+	}
+
+	var recordImages []*types.RecordImage
+
+	for tbName, tableAliases := range u.tableAliasesMap {
+		metaData, err := datasource.GetTableCache(types.DBTypeMySQL).GetTableMeta(ctx, u.execContext.DBName, tbName)
+		if err != nil {
+			return nil, err
+		}
+		selectSQL, selectArgs, err := u.buildBeforeImageSQL(ctx, metaData, tableAliases, u.execContext.NamedValues)
+		if err != nil {
+			return nil, err
+		}
+		if selectSQL == "" {
+			log.Debugf("Skip unused table [{%s}] when build select sql by update sourceQuery", tbName)
+			continue
+		}
+
+		var image *types.RecordImage
+		rowsi, err := u.rowsPrepare(ctx, u.execContext.Conn, selectSQL, selectArgs)
+		if err == nil {
+			image, err = u.buildRecordImages(rowsi, metaData, types.SQLTypeUpdate)
+		}
+		if rowsi != nil {
+			if rowerr := rows.Close(); rowerr != nil {
+				log.Errorf("rows close fail, err:%v", rowerr)
+				return nil, rowerr
+			}
+		}
+		if err != nil {
+			// If one fail, all fails
+			return nil, err
+		}
+
+		lockKey := u.buildLockKey(image, *metaData)
+		u.execContext.TxCtx.LockKeys[lockKey] = struct{}{}
+		image.SQLType = u.parserCtx.SQLType
+
+		recordImages = append(recordImages, image)
+	}
+
+	return recordImages, nil
+}
+
+func (u *updateJoinExecutor) afterImage(ctx context.Context, beforeImages []*types.RecordImage) ([]*types.RecordImage, error) {
+	if !u.isAstStmtValid() {
+		return nil, nil
+	}
+
+	if len(beforeImages) == 0 {
+		return nil, errors.New("empty beforeImages")
+	}
+
+	var recordImages []*types.RecordImage
+	for _, beforeImage := range beforeImages {
+		metaData, err := datasource.GetTableCache(types.DBTypeMySQL).GetTableMeta(ctx, u.execContext.DBName, beforeImage.TableName)
+		if err != nil {
+			return nil, err
+		}
+
+		selectSQL, selectArgs, err := u.buildAfterImageSQL(ctx, *beforeImage, metaData, u.tableAliasesMap[beforeImage.TableName])
+		if err != nil {
+			return nil, err
+		}
+
+		var image *types.RecordImage
+		rowsi, err := u.rowsPrepare(ctx, u.execContext.Conn, selectSQL, selectArgs)
+		if err == nil {
+			image, err = u.buildRecordImages(rowsi, metaData, types.SQLTypeUpdate)
+		}
+		if rowsi != nil {
+			if rowerr := rowsi.Close(); rowerr != nil {
+				log.Errorf("rows close fail, err:%v", rowerr)
+				return nil, rowerr
+			}
+		}
+		if err != nil {
+			// If one fail, all fails
+			return nil, err
+		}
+
+		image.SQLType = u.parserCtx.SQLType
+		recordImages = append(recordImages, image)
+	}
+
+	return recordImages, nil
+}
+
+// buildAfterImageSQL build the SQL to query before image data
+func (u *updateJoinExecutor) buildBeforeImageSQL(ctx context.Context, tableMeta *types.TableMeta, tableAliases string, args []driver.NamedValue) (string, []driver.NamedValue, error) {
+	updateStmt := u.parserCtx.UpdateStmt
+	fields, err := u.buildSelectFields(ctx, tableMeta, tableAliases, updateStmt.List)
+	if err != nil {
+		return "", nil, err
+	}
+	if len(fields) == 0 {
+		return "", nil, err
+	}
+
+	selStmt := ast.SelectStmt{
+		SelectStmtOpts: &ast.SelectStmtOpts{},
+		From:           updateStmt.TableRefs,
+		Where:          updateStmt.Where,
+		Fields:         &ast.FieldList{Fields: fields},
+		OrderBy:        updateStmt.Order,
+		Limit:          updateStmt.Limit,
+		TableHints:     updateStmt.TableHints,
+		// maybe duplicate row for select join sql.remove duplicate row by 'group by' condition
+		GroupBy: &ast.GroupByClause{
+			Items: u.buildGroupByClause(ctx, tableMeta.TableName, tableAliases, tableMeta.GetPrimaryKeyOnlyName(), fields),
+		},
+		LockInfo: &ast.SelectLockInfo{
+			LockType: ast.SelectLockForUpdate,
+		},
+	}
+
+	b := bytes.NewByteBuffer([]byte{})
+	_ = selStmt.Restore(format.NewRestoreCtx(format.RestoreKeyWordUppercase, b))
+	sql := string(b.Bytes())
+	log.Infof("build select sql by update sourceQuery, sql {%s}", sql)
+
+	return sql, u.buildSelectArgs(&selStmt, args), nil
+}
+
+func (u *updateJoinExecutor) buildAfterImageSQL(ctx context.Context, beforeImage types.RecordImage, meta *types.TableMeta, tableAliases string) (string, []driver.NamedValue, error) {
+	if len(beforeImage.Rows) == 0 {
+		return "", nil, nil
+	}
+
+	fields, err := u.buildSelectFields(ctx, meta, tableAliases, u.parserCtx.UpdateStmt.List)
+	if err != nil {
+		return "", nil, err
+	}
+	if len(fields) == 0 {
+		return "", nil, err
+	}
+
+	updateStmt := u.parserCtx.UpdateStmt
+	selStmt := ast.SelectStmt{
+		SelectStmtOpts: &ast.SelectStmtOpts{},
+		From:           updateStmt.TableRefs,
+		Where:          updateStmt.Where,
+		Fields:         &ast.FieldList{Fields: fields},
+		OrderBy:        updateStmt.Order,
+		Limit:          updateStmt.Limit,
+		TableHints:     updateStmt.TableHints,
+		// maybe duplicate row for select join sql.remove duplicate row by 'group by' condition
+		GroupBy: &ast.GroupByClause{
+			Items: u.buildGroupByClause(ctx, meta.TableName, tableAliases, meta.GetPrimaryKeyOnlyName(), fields),
+		},
+	}
+
+	b := bytes.NewByteBuffer([]byte{})
+	_ = selStmt.Restore(format.NewRestoreCtx(format.RestoreKeyWordUppercase, b))
+	sql := string(b.Bytes())
+	log.Infof("build select sql by update sourceQuery, sql {%s}", sql)
+
+	return sql, u.buildPKParams(beforeImage.Rows, meta.GetPrimaryKeyOnlyName()), nil
+}
+
+func (u *updateJoinExecutor) parseTableName(joinMate *ast.Join) map[string]string {
+	tableNames := make(map[string]string, 0)
+	if item, ok := joinMate.Left.(*ast.Join); ok {
+		tableNames = u.parseTableName(item)
+	} else {
+		leftTableSource := joinMate.Left.(*ast.TableSource)
+		leftName := leftTableSource.Source.(*ast.TableName)
+		tableNames[leftName.Name.O] = leftTableSource.AsName.O
+	}
+
+	rightTableSource := joinMate.Right.(*ast.TableSource)
+	rightName := rightTableSource.Source.(*ast.TableName)
+	tableNames[rightName.Name.O] = rightTableSource.AsName.O
+	return tableNames
+}
+
+// build group by condition which used for removing duplicate row in select join sql
+func (u *updateJoinExecutor) buildGroupByClause(ctx context.Context, tableName string, tableAliases string, pkColumns []string, allSelectColumns []*ast.SelectField) []*ast.ByItem {
+	var groupByPks = true
+	if tableAliases != "" {
+		tableName = tableAliases
+	}
+	//only pks group by is valid when db version >= 5.7.5
+	if u.isLowerSupportGroupByPksVersion {
+		if u.sqlMode == "" {
+			rowsi, err := u.rowsPrepare(ctx, u.execContext.Conn, "SELECT @@SQL_MODE", nil)
+			defer func() {
+				if rowsi != nil {
+					if rowerr := rowsi.Close(); rowerr != nil {
+						log.Errorf("rows close fail, err:%v", rowerr)
+					}
+				}
+			}()
+			if err != nil {
+				groupByPks = false
+				log.Warnf("determine group by pks or all columns error:%s", err)
+			} else {
+				// getString("@@SQL_MODE")
+				mode := make([]driver.Value, 1)
+				if err = rowsi.Next(mode); err != nil {
+					if err != io.EOF && len(mode) == 1 {
+						u.sqlMode = reflect.ValueOf(mode[0]).String()
+					}
+				}
+			}
+		}
+
+		if strings.Contains(u.sqlMode, "ONLY_FULL_GROUP_BY") {
+			groupByPks = false
+		}
+	}
+
+	groupByColumns := make([]*ast.ByItem, 0)
+	if groupByPks {
+		for _, column := range pkColumns {
+			groupByColumns = append(groupByColumns, &ast.ByItem{
+				Expr: &ast.ColumnNameExpr{
+					Name: &ast.ColumnName{
+						Table: model.CIStr{
+							O: tableName,
+							L: strings.ToLower(tableName),
+						},
+						Name: model.CIStr{
+							O: column,
+							L: strings.ToLower(column),
+						},
+					},
+				},
+			})
+		}
+	} else {
+		for _, column := range allSelectColumns {
+			groupByColumns = append(groupByColumns, &ast.ByItem{
+				Expr: column.Expr,
+			})
+		}
+	}
+	return groupByColumns
+}
diff --git a/pkg/datasource/sql/exec/at/update_join_executor_test.go b/pkg/datasource/sql/exec/at/update_join_executor_test.go
new file mode 100644
index 00000000..0ef30da3
--- /dev/null
+++ b/pkg/datasource/sql/exec/at/update_join_executor_test.go
@@ -0,0 +1,231 @@
+/*
+* 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 at
+
+import (
+	"context"
+	"database/sql/driver"
+	"seata.apache.org/seata-go/pkg/datasource/sql/undo"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"seata.apache.org/seata-go/pkg/datasource/sql/exec"
+	"seata.apache.org/seata-go/pkg/datasource/sql/parser"
+	"seata.apache.org/seata-go/pkg/datasource/sql/types"
+	"seata.apache.org/seata-go/pkg/datasource/sql/util"
+	_ "seata.apache.org/seata-go/pkg/util/log"
+)
+
+func TestBuildSelectSQLByUpdateJoin(t *testing.T) {
+	MetaDataMap := map[string]*types.TableMeta{
+		"table1": {
+			TableName: "table1",
+			Indexs: map[string]types.IndexMeta{
+				"id": {
+					IType: types.IndexTypePrimaryKey,
+					Columns: []types.ColumnMeta{
+						{ColumnName: "id"},
+					},
+				},
+			},
+			Columns: map[string]types.ColumnMeta{
+				"id": {
+					ColumnDef:  nil,
+					ColumnName: "id",
+				},
+				"name": {
+					ColumnDef:  nil,
+					ColumnName: "name",
+				},
+				"age": {
+					ColumnDef:  nil,
+					ColumnName: "age",
+				},
+			},
+			ColumnNames: []string{"id", "name", "age"},
+		},
+		"table2": {
+			TableName: "table2",
+			Indexs: map[string]types.IndexMeta{
+				"id": {
+					IType: types.IndexTypePrimaryKey,
+					Columns: []types.ColumnMeta{
+						{ColumnName: "id"},
+					},
+				},
+			},
+			Columns: map[string]types.ColumnMeta{
+				"id": {
+					ColumnDef:  nil,
+					ColumnName: "id",
+				},
+				"name": {
+					ColumnDef:  nil,
+					ColumnName: "name",
+				},
+				"age": {
+					ColumnDef:  nil,
+					ColumnName: "age",
+				},
+				"kk": {
+					ColumnDef:  nil,
+					ColumnName: "kk",
+				},
+				"addr": {
+					ColumnDef:  nil,
+					ColumnName: "addr",
+				},
+			},
+			ColumnNames: []string{"id", "name", "age", "kk", "addr"},
+		},
+		"table3": {
+			TableName: "table3",
+			Indexs: map[string]types.IndexMeta{
+				"id": {
+					IType: types.IndexTypePrimaryKey,
+					Columns: []types.ColumnMeta{
+						{ColumnName: "id"},
+					},
+				},
+			},
+			Columns: map[string]types.ColumnMeta{
+				"id": {
+					ColumnDef:  nil,
+					ColumnName: "id",
+				},
+				"age": {
+					ColumnDef:  nil,
+					ColumnName: "age",
+				},
+			},
+			ColumnNames: []string{"id", "age"},
+		},
+		"table4": {
+			TableName: "table4",
+			Indexs: map[string]types.IndexMeta{
+				"id": {
+					IType: types.IndexTypePrimaryKey,
+					Columns: []types.ColumnMeta{
+						{ColumnName: "id"},
+					},
+				},
+			},
+			Columns: map[string]types.ColumnMeta{
+				"id": {
+					ColumnDef:  nil,
+					ColumnName: "id",
+				},
+				"age": {
+					ColumnDef:  nil,
+					ColumnName: "age",
+				},
+			},
+			ColumnNames: []string{"id", "age"},
+		},
+	}
+
+	undo.InitUndoConfig(undo.Config{OnlyCareUpdateColumns: true})
+
+	tests := []struct {
+		name            string
+		sourceQuery     string
+		sourceQueryArgs []driver.Value
+		expectQuery     map[string]string
+		expectQueryArgs []driver.Value
+	}{
+		{
+			sourceQuery:     "update table1 t1 left join table2 t2 on t1.id = t2.id and t1.age=? set t1.name = 'WILL',t2.name = ?",
+			sourceQueryArgs: []driver.Value{18, "Jack"},
+			expectQuery: map[string]string{
+				"table1": "SELECT SQL_NO_CACHE t1.name,t1.id FROM table1 AS t1 LEFT JOIN table2 AS t2 ON t1.id=t2.id AND t1.age=? GROUP BY t1.name,t1.id FOR UPDATE",
+				"table2": "SELECT SQL_NO_CACHE t2.name,t2.id FROM table1 AS t1 LEFT JOIN table2 AS t2 ON t1.id=t2.id AND t1.age=? GROUP BY t2.name,t2.id FOR UPDATE",
+			},
+			expectQueryArgs: []driver.Value{18},
+		},
+		{
+			sourceQuery:     "update table1 AS t1 inner join table2 AS t2 on t1.id = t2.id set t1.name = 'WILL',t2.name = 'WILL' where t1.id=?",
+			sourceQueryArgs: []driver.Value{1},
+			expectQuery: map[string]string{
+				"table1": "SELECT SQL_NO_CACHE t1.name,t1.id FROM table1 AS t1 JOIN table2 AS t2 ON t1.id=t2.id WHERE t1.id=? GROUP BY t1.name,t1.id FOR UPDATE",
+				"table2": "SELECT SQL_NO_CACHE t2.name,t2.id FROM table1 AS t1 JOIN table2 AS t2 ON t1.id=t2.id WHERE t1.id=? GROUP BY t2.name,t2.id FOR UPDATE",
+			},
+			expectQueryArgs: []driver.Value{1},
+		},
+		{
+			sourceQuery:     "update table1 AS t1 right join table2 AS t2 on t1.id = t2.id set t1.name = 'WILL',t2.name = 'WILL' where t1.id=?",
+			sourceQueryArgs: []driver.Value{1},
+			expectQuery: map[string]string{
+				"table1": "SELECT SQL_NO_CACHE t1.name,t1.id FROM table1 AS t1 RIGHT JOIN table2 AS t2 ON t1.id=t2.id WHERE t1.id=? GROUP BY t1.name,t1.id FOR UPDATE",
+				"table2": "SELECT SQL_NO_CACHE t2.name,t2.id FROM table1 AS t1 RIGHT JOIN table2 AS t2 ON t1.id=t2.id WHERE t1.id=? GROUP BY t2.name,t2.id FOR UPDATE",
+			},
+			expectQueryArgs: []driver.Value{1},
+		},
+		{
+			sourceQuery:     "update table1 t1 inner join table2 t2 on t1.id = t2.id set t1.name = ?, t1.age = ? where t1.id = ? and t1.name = ? and t2.age between ? and ?",
+			sourceQueryArgs: []driver.Value{"newJack", 38, 1, "Jack", 18, 28},
+			expectQuery: map[string]string{
+				"table1": "SELECT SQL_NO_CACHE t1.name,t1.age,t1.id FROM table1 AS t1 JOIN table2 AS t2 ON t1.id=t2.id WHERE t1.id=? AND t1.name=? AND t2.age BETWEEN ? AND ? GROUP BY t1.name,t1.age,t1.id FOR UPDATE",
+			},
+			expectQueryArgs: []driver.Value{1, "Jack", 18, 28},
+		},
+		{
+			sourceQuery:     "update table1 t1 left join table2 t2 on t1.id = t2.id set t1.name = ?, t1.age = ? where t1.id=? and t2.id is null and t1.age IN (?,?)",
+			sourceQueryArgs: []driver.Value{"newJack", 38, 1, 18, 28},
+			expectQuery: map[string]string{
+				"table1": "SELECT SQL_NO_CACHE t1.name,t1.age,t1.id FROM table1 AS t1 LEFT JOIN table2 AS t2 ON t1.id=t2.id WHERE t1.id=? AND t2.id IS NULL AND t1.age IN (?,?) GROUP BY t1.name,t1.age,t1.id FOR UPDATE",
+			},
+			expectQueryArgs: []driver.Value{1, 18, 28},
+		},
+		{
+			sourceQuery:     "update table1 t1 inner join table2 t2 on t1.id = t2.id set t1.name = ?, t2.age = ? where t2.kk between ? and ? and t2.addr in(?,?) and t2.age > ? order by t1.name desc limit ?",
+			sourceQueryArgs: []driver.Value{"Jack", 18, 10, 20, "Beijing", "Guangzhou", 18, 2},
+			expectQuery: map[string]string{
+				"table1": "SELECT SQL_NO_CACHE t1.name,t1.id FROM table1 AS t1 JOIN table2 AS t2 ON t1.id=t2.id WHERE t2.kk BETWEEN ? AND ? AND t2.addr IN (?,?) AND t2.age>? GROUP BY t1.name,t1.id ORDER BY t1.name DESC LIMIT ? FOR UPDATE",
+				"table2": "SELECT SQL_NO_CACHE t2.age,t2.id FROM table1 AS t1 JOIN table2 AS t2 ON t1.id=t2.id WHERE t2.kk BETWEEN ? AND ? AND t2.addr IN (?,?) AND t2.age>? GROUP BY t2.age,t2.id ORDER BY t1.name DESC LIMIT ? FOR UPDATE",
+			},
+			expectQueryArgs: []driver.Value{10, 20, "Beijing", "Guangzhou", 18, 2},
+		},
+		{
+			sourceQuery:     "update table1 t1 left join table2 t2 on t1.id = t2.id inner join table3 t3 on t3.id = t2.id right join table4 t4 on t4.id = t2.id set t1.name = ?,t2.name = ? where t1.id=? and t3.age=? and t4.age>30",
+			sourceQueryArgs: []driver.Value{"Jack", "WILL", 1, 10},
+			expectQuery: map[string]string{
+				"table1": "SELECT SQL_NO_CACHE t1.name,t1.id FROM ((table1 AS t1 LEFT JOIN table2 AS t2 ON t1.id=t2.id) JOIN table3 AS t3 ON t3.id=t2.id) RIGHT JOIN table4 AS t4 ON t4.id=t2.id WHERE t1.id=? AND t3.age=? AND t4.age>30 GROUP BY t1.name,t1.id FOR UPDATE",
+				"table2": "SELECT SQL_NO_CACHE t2.name,t2.id FROM ((table1 AS t1 LEFT JOIN table2 AS t2 ON t1.id=t2.id) JOIN table3 AS t3 ON t3.id=t2.id) RIGHT JOIN table4 AS t4 ON t4.id=t2.id WHERE t1.id=? AND t3.age=? AND t4.age>30 GROUP BY t2.name,t2.id FOR UPDATE",
+			},
+			expectQueryArgs: []driver.Value{1, 10},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c, err := parser.DoParser(tt.sourceQuery)
+			assert.Nil(t, err)
+			executor := NewUpdateJoinExecutor(c, &types.ExecContext{Values: tt.sourceQueryArgs, NamedValues: util.ValueToNamedValue(tt.sourceQueryArgs)}, []exec.SQLHook{})
+			tableNames := executor.(*updateJoinExecutor).parseTableName(c.UpdateStmt.TableRefs.TableRefs)
+			for tbName, tableAliases := range tableNames {
+				query, args, err := executor.(*updateJoinExecutor).buildBeforeImageSQL(context.Background(), MetaDataMap[tbName], tableAliases, util.ValueToNamedValue(tt.sourceQueryArgs))
+				assert.Nil(t, err)
+				if query == "" {
+					continue
+				}
+				assert.Equal(t, tt.expectQuery[tbName], query)
+				assert.Equal(t, tt.expectQueryArgs, util.NamedValueToValue(args))
+			}
+		})
+	}
+}
diff --git a/pkg/datasource/sql/types/types.go b/pkg/datasource/sql/types/types.go
index 36b44a2e..aa911285 100644
--- a/pkg/datasource/sql/types/types.go
+++ b/pkg/datasource/sql/types/types.go
@@ -159,6 +159,7 @@ type ExecContext struct {
 	Conn        driver.Conn
 	DBName      string
 	DBType      DBType
+	DbVersion   string
 	// todo set values for these 4 param
 	IsAutoCommit         bool
 	IsSupportsSavepoints bool