roptimize: efactor delete and insert executor (apache#409)

* refactor delete and insert executor

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

@@ -57,10 +57,12 @@ func (e *AtExecutor) ExecWithNamedValue(ctx context.Context, execCtx *types.Exec
 		exec = NewPlainExecutor(parser, execCtx)
 	} else {
 		switch parser.SQLType {
-		//case types.SQLTypeInsert:
+		case types.SQLTypeInsert:
+			exec = NewInsertExecutor(parser, execCtx, e.hooks)
 		case types.SQLTypeUpdate:
 			exec = NewUpdateExecutor(parser, execCtx, e.hooks)
-		//case types.SQLTypeDelete:
+		case types.SQLTypeDelete:
+			exec = NewDeleteExecutor(parser, execCtx, e.hooks)
 		//case types.SQLTypeSelectForUpdate:
 		//case types.SQLTypeMultiDelete:
 		//case types.SQLTypeMultiUpdate:

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

@@ -61,7 +61,7 @@ func (*baseExecutor) GetScanSlice(columnNames []string, tableMeta *types.TableMe
 		)
 		switch strings.ToUpper(columnMeta.DatabaseTypeString) {
 		case "VARCHAR", "NVARCHAR", "VARCHAR2", "CHAR", "TEXT", "JSON", "TINYTEXT":
-			var scanVal string
+			var scanVal sql.NullString
 			scanSlice = append(scanSlice, &scanVal)
 		case "BIT", "INT", "LONGBLOB", "SMALLINT", "TINYINT", "BIGINT", "MEDIUMINT":
 			if columnMeta.IsNullable == 0 {
@@ -72,7 +72,7 @@ func (*baseExecutor) GetScanSlice(columnNames []string, tableMeta *types.TableMe
 				scanSlice = append(scanSlice, &scanVal)
 			}
 		case "DATE", "DATETIME", "TIME", "TIMESTAMP", "YEAR":
-			scanVal := sql.NullTime{}
+			var scanVal sql.NullTime
 			scanSlice = append(scanSlice, &scanVal)
 		case "DECIMAL", "DOUBLE", "FLOAT":
 			if columnMeta.IsNullable == 0 {
@@ -177,7 +177,7 @@ func (b *baseExecutor) buildRecordImages(rowsi driver.Rows, tableMetaData *types
 				KeyType:    keyType,
 				ColumnName: name,
 				ColumnType: jdbcType,
-				Value:      reflectx.GetElemDataValue(ss[i]),
+				Value:      getSqlNullValue(reflectx.GetElemDataValue(ss[i])),
 			})
 		}
 		rowImages = append(rowImages, types.RowImage{Columns: columns})
@@ -186,6 +186,61 @@ func (b *baseExecutor) buildRecordImages(rowsi driver.Rows, tableMetaData *types
 	return &types.RecordImage{TableName: tableMetaData.TableName, Rows: rowImages}, nil
 }
 
+func getSqlNullValue(value interface{}) interface{} {
+	if value == nil {
+		return nil
+	}
+	if v, ok := value.(sql.NullString); ok {
+		if v.Valid {
+			return v.String
+		}
+		return nil
+	}
+	if v, ok := value.(sql.NullFloat64); ok {
+		if v.Valid {
+			return v.Float64
+		}
+		return nil
+	}
+	if v, ok := value.(sql.NullBool); ok {
+		if v.Valid {
+			return v.Bool
+		}
+		return nil
+	}
+	if v, ok := value.(sql.NullTime); ok {
+		if v.Valid {
+			return v.Time
+		}
+		return nil
+	}
+	if v, ok := value.(sql.NullByte); ok {
+		if v.Valid {
+			return v.Byte
+		}
+		return nil
+	}
+	if v, ok := value.(sql.NullInt16); ok {
+		if v.Valid {
+			return v.Int16
+		}
+		return nil
+	}
+	if v, ok := value.(sql.NullInt32); ok {
+		if v.Valid {
+			return v.Int32
+		}
+		return nil
+	}
+	if v, ok := value.(sql.NullInt64); ok {
+		if v.Valid {
+			return v.Int64
+		}
+		return nil
+	}
+	return value
+}
+
 // buildWhereConditionByPKs build where condition by primary keys
 // each pk is a condition.the result will like :" (id,userCode) in ((?,?),(?,?)) or (id,userCode) in ((?,?),(?,?) ) or (id,userCode) in ((?,?))"
 func (b *baseExecutor) buildWhereConditionByPKs(pkNameList []string, rowSize int, dbType string, maxInSize int) string {

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

@@ -0,0 +1,159 @@
+/*
+ * 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"
+	"fmt"
+
+	"github.com/arana-db/parser/ast"
+	"github.com/arana-db/parser/format"
+	"github.com/seata/seata-go/pkg/datasource/sql/datasource"
+	"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/util"
+	"github.com/seata/seata-go/pkg/util/bytes"
+	"github.com/seata/seata-go/pkg/util/log"
+)
+
+// deleteExecutor execute delete SQL
+type deleteExecutor struct {
+	baseExecutor
+	parserCtx   *types.ParseContext
+	execContent *types.ExecContext
+}
+
+//NewDeleteExecutor get delete executor
+func NewDeleteExecutor(parserCtx *types.ParseContext, execContent *types.ExecContext, hooks []exec.SQLHook) executor {
+	return &deleteExecutor{parserCtx: parserCtx, execContent: execContent, baseExecutor: baseExecutor{hooks: hooks}}
+}
+
+//ExecContext exec SQL, and generate before image and after image
+func (d deleteExecutor) ExecContext(ctx context.Context, f exec.CallbackWithNamedValue) (types.ExecResult, error) {
+	d.beforeHooks(ctx, d.execContent)
+	defer func() {
+		d.afterHooks(ctx, d.execContent)
+	}()
+
+	beforeImage, err := d.beforeImage(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	res, err := f(ctx, d.execContent.Query, d.execContent.NamedValues)
+	if err != nil {
+		return nil, err
+	}
+
+	afterImage, err := d.afterImage(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	d.execContent.TxCtx.RoundImages.AppendBeofreImage(beforeImage)
+	d.execContent.TxCtx.RoundImages.AppendAfterImage(afterImage)
+	return res, nil
+}
+
+//beforeImage build before image
+func (d *deleteExecutor) beforeImage(ctx context.Context) (*types.RecordImage, error) {
+	selectSQL, selectArgs, err := d.buildBeforeImageSQL(d.execContent.Query, d.execContent.NamedValues)
+	if err != nil {
+		return nil, err
+	}
+
+	var rowsi driver.Rows
+	queryerCtx, ok := d.execContent.Conn.(driver.QueryerContext)
+	var queryer driver.Queryer
+	if !ok {
+		queryer, ok = d.execContent.Conn.(driver.Queryer)
+	}
+	if ok {
+		rowsi, err = util.CtxDriverQuery(ctx, queryerCtx, queryer, selectSQL, selectArgs)
+		if err != nil {
+			log.Errorf("ctx driver query: %+v", err)
+			return nil, err
+		}
+	} else {
+		log.Errorf("target conn should been driver.QueryerContext or driver.Queryer")
+		return nil, fmt.Errorf("invalid conn")
+	}
+
+	tableName, _ := d.parserCtx.GteTableName()
+	metaData, err := datasource.GetTableCache(types.DBTypeMySQL).GetTableMeta(ctx, d.execContent.DBName, tableName)
+	if err != nil {
+		return nil, err
+	}
+
+	image, err := d.buildRecordImages(rowsi, metaData)
+	if err != nil {
+		return nil, err
+	}
+	image.SQLType = types.SQLTypeDelete
+	image.TableMeta = metaData
+
+	lockKey := d.buildLockKey(image, *metaData)
+	d.execContent.TxCtx.LockKeys[lockKey] = struct{}{}
+
+	return image, nil
+}
+
+// buildBeforeImageSQL build delete sql from delete sql
+func (d *deleteExecutor) buildBeforeImageSQL(query string, args []driver.NamedValue) (string, []driver.NamedValue, error) {
+	p, err := parser.DoParser(query)
+	if err != nil {
+		return "", nil, err
+	}
+
+	if p.DeleteStmt == nil {
+		log.Errorf("invalid delete stmt")
+		return "", nil, fmt.Errorf("invalid delete stmt")
+	}
+
+	selStmt := ast.SelectStmt{
+		SelectStmtOpts: &ast.SelectStmtOpts{},
+		From:           p.DeleteStmt.TableRefs,
+		Where:          p.DeleteStmt.Where,
+		Fields:         &ast.FieldList{Fields: []*ast.SelectField{{WildCard: &ast.WildCardField{}}}},
+		OrderBy:        p.DeleteStmt.Order,
+		Limit:          p.DeleteStmt.Limit,
+		TableHints:     p.DeleteStmt.TableHints,
+		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 delete sourceQuery, sql {%s}", sql)
+
+	return sql, d.buildSelectArgs(&selStmt, args), nil
+}
+
+//afterImage build after image
+func (d *deleteExecutor) afterImage(ctx context.Context) (*types.RecordImage, error) {
+	tableName, _ := d.parserCtx.GteTableName()
+	metaData, err := datasource.GetTableCache(types.DBTypeMySQL).GetTableMeta(ctx, d.execContent.DBName, tableName)
+	if err != nil {
+		return nil, err
+	}
+	return types.NewEmptyRecordImage(metaData, types.SQLTypeDelete), nil
+}

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

@@ -0,0 +1,81 @@
+/*
+ * 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 (
+	"database/sql/driver"
+	"testing"
+
+	"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/util"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewDeleteExecutor(t *testing.T) {
+	executor := NewDeleteExecutor(nil, nil, nil)
+	_, ok := executor.(*deleteExecutor)
+	assert.Equalf(t, true, ok, "should be *deleteExecutor")
+}
+
+func Test_deleteExecutor_buildBeforeImageSQL(t *testing.T) {
+	tests := []struct {
+		name            string
+		sourceQuery     string
+		sourceQueryArgs []driver.Value
+		expectQuery     string
+		expectQueryArgs []driver.Value
+	}{
+		{
+			sourceQuery:     "delete from t_user where id = ?",
+			sourceQueryArgs: []driver.Value{100},
+			expectQuery:     "SELECT SQL_NO_CACHE * FROM t_user WHERE id=? FOR UPDATE",
+			expectQueryArgs: []driver.Value{100},
+		},
+		{
+			sourceQuery:     "delete from t_user where id = ? and name = 'Jack' and age between ? and ?",
+			sourceQueryArgs: []driver.Value{100, 18, 28},
+			expectQuery:     "SELECT SQL_NO_CACHE * FROM t_user WHERE id=? AND name=_UTF8MB4Jack AND age BETWEEN ? AND ? FOR UPDATE",
+			expectQueryArgs: []driver.Value{100, 18, 28},
+		},
+		{
+			sourceQuery:     "delete from t_user where id = ? and name = 'Jack' and age in (?,?)",
+			sourceQueryArgs: []driver.Value{100, 18, 28},
+			expectQuery:     "SELECT SQL_NO_CACHE * FROM t_user WHERE id=? AND name=_UTF8MB4Jack AND age IN (?,?) FOR UPDATE",
+			expectQueryArgs: []driver.Value{100, 18, 28},
+		},
+		{
+			sourceQuery:     "delete from t_user where kk between ? and ? and id = ? and addr in(?,?) and age > ? order by name desc limit ?",
+			sourceQueryArgs: []driver.Value{10, 20, 17, "Beijing", "Guangzhou", 18, 2},
+			expectQuery:     "SELECT SQL_NO_CACHE * FROM t_user WHERE kk BETWEEN ? AND ? AND id=? AND addr IN (?,?) AND age>? ORDER BY name DESC LIMIT ? FOR UPDATE",
+			expectQueryArgs: []driver.Value{10, 20, 17, "Beijing", "Guangzhou", 18, 2},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c, err := parser.DoParser(tt.sourceQuery)
+			assert.Nil(t, err)
+			executor := NewDeleteExecutor(c, &types.ExecContext{Values: tt.sourceQueryArgs, NamedValues: util.ValueToNamedValue(tt.sourceQueryArgs)}, []exec.SQLHook{})
+			query, args, err := executor.(*deleteExecutor).buildBeforeImageSQL(tt.sourceQuery, util.ValueToNamedValue(tt.sourceQueryArgs))
+			assert.Nil(t, err)
+			assert.Equal(t, tt.expectQuery, query)
+			assert.Equal(t, tt.expectQueryArgs, util.NamedValueToValue(args))
+		})
+	}
+}