reactor:multiple update executor

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

@@ -43,40 +43,40 @@ 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)
+			executor = NewSelectForUpdateExecutor(queryParser, execCtx, e.hooks)
 		case types.SQLTypeInsertOnUpdate:
-			exec = NewInsertOnUpdateExecutor(parser, execCtx, e.hooks)
-		// case types.SQLTypeMultiDelete:
-		// case types.SQLTypeMultiUpdate:
+			executor = NewInsertOnUpdateExecutor(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,129 @@
+/*
+ * 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"
+	"errors"
+	"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, errors.New("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, errors.New("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)
+		case types.DeleteExecutor:
+			//todo use MultiDeleteExecutor
+		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)
+		case types.DeleteExecutor:
+			// todo use MultiDeleteExecutor
+		}
+		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,18 @@
+/*
+ * 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