reactor:multiple update executor

pkg/datasource/sql/exec/at/at_executor.go

@@ -43,40 +43,38 @@ func (e *ATExecutor) Interceptors(hooks []exec.SQLHook) {
 	e.hooks = hooks
 }
 
-// ExecWithNamedValue
+// ExecWithNamedValue find the executor by sql type
 func (e *ATExecutor) ExecWithNamedValue(ctx context.Context, execCtx *types.ExecContext, f exec.CallbackWithNamedValue) (types.ExecResult, error) {
-	parser, err := parser.DoParser(execCtx.Query)
+	queryParser, err := parser.DoParser(execCtx.Query)
 	if err != nil {
 		return nil, err
 	}
 
-	var exec executor
+	var executor executor
 
 	if !tm.IsGlobalTx(ctx) {
-		exec = NewPlainExecutor(parser, execCtx)
+		executor = NewPlainExecutor(queryParser, execCtx)
 	} else {
-		switch parser.SQLType {
+		switch queryParser.SQLType {
 		case types.SQLTypeInsert:
-			exec = NewInsertExecutor(parser, execCtx, e.hooks)
+			executor = NewInsertExecutor(queryParser, execCtx, e.hooks)
 		case types.SQLTypeUpdate:
-			exec = NewUpdateExecutor(parser, execCtx, e.hooks)
+			executor = NewUpdateExecutor(queryParser, execCtx, e.hooks)
 		case types.SQLTypeDelete:
-			exec = NewDeleteExecutor(parser, execCtx, e.hooks)
+			executor = NewDeleteExecutor(queryParser, execCtx, e.hooks)
 		case types.SQLTypeSelectForUpdate:
-			exec = NewSelectForUpdateExecutor(parser, execCtx, e.hooks)
-		case types.SQLTypeInsertOnUpdate:
-			exec = NewInsertOnUpdateExecutor(parser, execCtx, e.hooks)
-		// case types.SQLTypeMultiDelete:
-		// case types.SQLTypeMultiUpdate:
+			executor = NewSelectForUpdateExecutor(queryParser, execCtx, e.hooks)
+		case types.SQLTypeMulti:
+			executor = NewMultiExecutor(queryParser, execCtx, e.hooks)
 		default:
-			exec = NewPlainExecutor(parser, execCtx)
+			executor = NewPlainExecutor(queryParser, execCtx)
 		}
 	}
 
-	return exec.ExecContext(ctx, f)
+	return executor.ExecContext(ctx, f)
 }
 
-// ExecWithValue
+// ExecWithValue transfer value to nameValue execute
 func (e *ATExecutor) ExecWithValue(ctx context.Context, execCtx *types.ExecContext, f exec.CallbackWithNamedValue) (types.ExecResult, error) {
 	execCtx.NamedValues = util.ValueToNamedValue(execCtx.Values)
 	return e.ExecWithNamedValue(ctx, execCtx, f)

pkg/datasource/sql/exec/at/multi_executor.go

@@ -0,0 +1,132 @@
+/*
+ * 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"
+	"fmt"
+
+	"github.com/seata/seata-go/pkg/datasource/sql/exec"
+	"github.com/seata/seata-go/pkg/datasource/sql/types"
+)
+
+type multiExecutor struct {
+	baseExecutor
+	parserCtx   *types.ParseContext
+	execContext *types.ExecContext
+}
+
+// NewMultiExecutor get new multi executor
+func NewMultiExecutor(parserCtx *types.ParseContext, execContext *types.ExecContext, hooks []exec.SQLHook) executor {
+	return &multiExecutor{parserCtx: parserCtx, execContext: execContext, baseExecutor: baseExecutor{hooks: hooks}}
+}
+
+// ExecContext exec SQL, and generate before image and after image
+func (m *multiExecutor) ExecContext(ctx context.Context, f exec.CallbackWithNamedValue) (types.ExecResult, error) {
+	m.beforeHooks(ctx, m.execContext)
+
+	defer func() {
+		m.afterHooks(ctx, m.execContext)
+	}()
+
+	beforeImages, err := m.beforeImage(ctx, m.parserCtx)
+	if err != nil {
+		return nil, err
+	}
+
+	res, err := f(ctx, m.execContext.Query, m.execContext.NamedValues)
+	if err != nil {
+		return nil, err
+	}
+
+	afterImages, err := m.afterImage(ctx, m.parserCtx, beforeImages)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(afterImages) != len(beforeImages) {
+		return nil, fmt.Errorf("Before image size is not equaled to after image size, probably because you updated the primary keys.")
+	}
+
+	for i, afterImage := range afterImages {
+		beforeImage := beforeImages[i]
+		if len(beforeImage.Rows) != len(afterImage.Rows) {
+			return nil, fmt.Errorf("Before image size is not equaled to after image size, probably because you updated the primary keys.")
+		}
+
+		m.execContext.TxCtx.RoundImages.AppendBeofreImage(beforeImage)
+		m.execContext.TxCtx.RoundImages.AppendAfterImage(afterImage)
+	}
+
+	return res, nil
+}
+func (m *multiExecutor) beforeImage(ctx context.Context, parseContext *types.ParseContext) ([]*types.RecordImage, error) {
+	if len(parseContext.MultiStmt) == 0 {
+		return nil, nil
+	}
+	tmpImages := make([]*types.RecordImage, 0)
+	var err error
+
+	var beforeImages = make([]*types.RecordImage, 0)
+	for _, multiStmt := range parseContext.MultiStmt {
+		switch multiStmt.ExecutorType {
+		case types.UpdateExecutor:
+			multiUpdateExec := NewMultiUpdateExecutor(m.parserCtx, m.execContext, m.hooks)
+			tmpImages, err = multiUpdateExec.beforeImage(ctx)
+			break
+		case types.DeleteExecutor:
+			//todo use MultiDeleteExecutor
+			break
+		default:
+			return nil, fmt.Errorf("not support sql %s", m.execContext.Query)
+		}
+
+		if err != nil {
+			return nil, err
+		}
+		beforeImages = append(beforeImages, tmpImages...)
+	}
+
+	return beforeImages, err
+}
+
+func (m *multiExecutor) afterImage(ctx context.Context, parseContext *types.ParseContext, beforeImages []*types.RecordImage) ([]*types.RecordImage, error) {
+	if len(parseContext.MultiStmt) == 0 {
+		return nil, nil
+	}
+	tmpImages := make([]*types.RecordImage, 0)
+	var err error
+
+	var afterImages = make([]*types.RecordImage, 0)
+	for _, multiStmt := range parseContext.MultiStmt {
+		switch multiStmt.ExecutorType {
+		case types.UpdateExecutor:
+			multiUpdateExec := NewMultiUpdateExecutor(m.parserCtx, m.execContext, m.hooks)
+			tmpImages, err = multiUpdateExec.afterImage(ctx, beforeImages)
+			break
+		case types.DeleteExecutor:
+			// todo use MultiDeleteExecutor
+			break
+		}
+		afterImages = append(afterImages, tmpImages...)
+	}
+	if err != nil {
+		return nil, err
+	}
+	return afterImages, err
+}

pkg/datasource/sql/exec/at/multi_executor_test.go

@@ -0,0 +1,76 @@
+/*
+ * 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"
+	"github.com/seata/seata-go/pkg/datasource/sql/datasource"
+	"github.com/seata/seata-go/pkg/datasource/sql/datasource/mysql"
+	"github.com/seata/seata-go/pkg/datasource/sql/exec"
+	"github.com/seata/seata-go/pkg/datasource/sql/parser"
+	"github.com/seata/seata-go/pkg/datasource/sql/types"
+	"github.com/seata/seata-go/pkg/datasource/sql/undo"
+	"github.com/seata/seata-go/pkg/datasource/sql/util"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestMultiExecutor_beforeImage(t *testing.T) {
+	undo.InitUndoConfig(undo.Config{OnlyCareUpdateColumns: true})
+	datasource.RegisterTableCache(types.DBTypeMySQL, mysql.NewTableMetaInstance(nil))
+
+	tests := []struct {
+		name            string
+		sourceQuery     string
+		sourceQueryArgs []driver.Value
+		expectQuery     string
+		expectQueryArgs []driver.Value
+	}{
+		{
+			sourceQuery:     "update t_user set name = ?, age = ? where id = ?;update t_user set name = ?, age = ? where id = ?;",
+			sourceQueryArgs: []driver.Value{"Jack", 1, 100, "TOM", 2, 200},
+			expectQuery:     "SELECT SQL_NO_CACHE name,age FROM t_user WHERE id=? OR id=? FOR UPDATE",
+			expectQueryArgs: []driver.Value{100, 200},
+		},
+		{
+			sourceQuery:     "update t_user set name = ?, age = ? where id = ? and name = 'Jack' and age between ? and ?;update t_user set name = ?, age = ? where id = ? and name = 'Jack2' and age between ? and ?",
+			sourceQueryArgs: []driver.Value{"Jack", 1, 100, 18, 28, "Jack2", 2, 200, 28, 38},
+			expectQuery:     "SELECT SQL_NO_CACHE name,age FROM t_user WHERE id=? AND name=_UTF8MB4Jack AND age BETWEEN ? AND ? OR id=? AND name=_UTF8MB4Jack2 AND age BETWEEN ? AND ? FOR UPDATE",
+			expectQueryArgs: []driver.Value{100, 18, 28, 200, 28, 38},
+		},
+		{
+			sourceQuery:     "update t_user set name = ?, age = ? where id = ? and name = 'Jack' and age in (?,?);update t_user set name = ?, age = ? where id = ? and name = 'Jack2' and age in (?,?)",
+			sourceQueryArgs: []driver.Value{"Jack", 1, 100, 18, 28, "Jack2", 2, 200, 48, 58},
+			expectQuery:     "SELECT SQL_NO_CACHE name,age FROM t_user WHERE id=? AND name=_UTF8MB4Jack AND age IN (?,?) OR id=? AND name=_UTF8MB4Jack2 AND age IN (?,?) FOR UPDATE",
+			expectQueryArgs: []driver.Value{100, 18, 28, 200, 48, 58},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c, err := parser.DoParser(tt.sourceQuery)
+			assert.Nil(t, err)
+
+			executor := NewMultiExecutor(c, &types.ExecContext{NamedValues: util.ValueToNamedValue(tt.sourceQueryArgs)}, []exec.SQLHook{})
+			beforeImage, err := executor.(*multiExecutor).beforeImage(context.Background(), c)
+			assert.NoError(t, err)
+			assert.NotNil(t, beforeImage)
+		})
+	}
+}