This is an automated email from the ASF dual-hosted git repository.
Ethan-Xingyue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-seata-go.git
The following commit(s) were added to refs/heads/master by this push:
new 2167ed1c fix: support MySQL multi-value inserts with composite primary
keys (#1136)
2167ed1c is described below
commit 2167ed1cbe765af7a76e3e863c1308f1f489ecfa
Author: AsperforMias <[email protected]>
AuthorDate: Sat Jul 18 20:07:06 2026 +0800
fix: support MySQL multi-value inserts with composite primary keys (#1136)
* fix: support composite pk multi-value inserts
* test: cover multi-value insert composite primary keys
* docs: register multi-value insert change
* test: avoid race in DayValue YAML test
* test: cover primary key type fallback
* fix: handle insert values without column list
* refactor: clarify insert primary key handling
* fix: close auto increment step query rows
---
changes/dev.md | 1 +
changes/dev_zh.md | 1 +
pkg/datasource/sql/exec/at/insert_executor.go | 71 +++--
pkg/datasource/sql/exec/at/insert_executor_test.go | 285 +++++++++++++++++++++
pkg/datasource/sql/types/meta.go | 6 +-
pkg/datasource/sql/types/meta_test.go | 33 +++
6 files changed, 378 insertions(+), 19 deletions(-)
diff --git a/changes/dev.md b/changes/dev.md
index 487c6fab..e837c1a0 100755
--- a/changes/dev.md
+++ b/changes/dev.md
@@ -27,6 +27,7 @@
- [[#123](https://github.com/apache/incubator-seata-go/pull/123)] add two
phase and dubbo
- support PostgreSQL XA via pgx driver
+ - [[#1130](https://github.com/apache/incubator-seata-go/issues/1130)]
support MySQL multi-value INSERT in AT mode for composite and mixed primary keys
### bugfix：
diff --git a/changes/dev_zh.md b/changes/dev_zh.md
index 623e53c0..53615803 100644
--- a/changes/dev_zh.md
+++ b/changes/dev_zh.md
@@ -28,6 +28,7 @@ Seata-go 是一款开源的分布式事务解决方案，提供高性能和简
- [[#123](https://github.com/apache/incubator-seata-go/pull/123)]
添加二阶段事务接口，以及dubbo集成
- 支持基于 pgx 驱动的 PostgreSQL XA
+- [[#1130](https://github.com/apache/incubator-seata-go/issues/1130)] 支持 AT
模式下 MySQL 多值 INSERT 的复合主键与混合主键场景
### bugfix：
diff --git a/pkg/datasource/sql/exec/at/insert_executor.go
b/pkg/datasource/sql/exec/at/insert_executor.go
index b6cf3977..0bb77068 100644
--- a/pkg/datasource/sql/exec/at/insert_executor.go
+++ b/pkg/datasource/sql/exec/at/insert_executor.go
@@ -22,6 +22,7 @@ import (
"database/sql/driver"
"errors"
"fmt"
+ "strconv"
"strings"
"github.com/arana-db/parser/ast"
@@ -487,17 +488,26 @@ func (i *insertExecutor) getPkIndex(InsertStmt
*ast.InsertStmt, meta types.Table
return pkIndexMap
}
insertColumnsSize := len(InsertStmt.Columns)
- if insertColumnsSize == 0 {
+ if meta.ColumnNames == nil {
return pkIndexMap
}
- if meta.ColumnNames == nil {
+ if insertColumnsSize == 0 {
+ if len(InsertStmt.Lists) == 0 {
+ return pkIndexMap
+ }
+ // INSERT without a column list follows the physical table
column order.
+ for idx, columnName := range meta.ColumnNames {
+ if pkColumnName, ok := i.matchPKColumnName(columnName,
meta); ok {
+ pkIndexMap[pkColumnName] = idx
+ }
+ }
return pkIndexMap
}
if len(meta.Columns) > 0 {
for paramIdx := 0; paramIdx < insertColumnsSize; paramIdx++ {
sqlColumnName := InsertStmt.Columns[paramIdx].Name.O
- if i.containPK(sqlColumnName, meta) {
- pkIndexMap[sqlColumnName] = paramIdx
+ if pkColumnName, ok :=
i.matchPKColumnName(sqlColumnName, meta); ok {
+ pkIndexMap[pkColumnName] = paramIdx
}
}
return pkIndexMap
@@ -516,6 +526,16 @@ func (i *insertExecutor) getPkIndex(InsertStmt
*ast.InsertStmt, meta types.Table
return pkIndexMap
}
+func (i *insertExecutor) matchPKColumnName(columnName string, meta
types.TableMeta) (string, bool) {
+ newColumnName := util.DelEscape(columnName, i.dbType())
+ for _, name := range meta.GetPrimaryKeyOnlyName() {
+ if strings.EqualFold(name, newColumnName) {
+ return name, true
+ }
+ }
+ return "", false
+}
+
// parsePkValuesFromStatement parse primary key value from statement.
// return the primary key and values<key:primary key,value:primary key
values></key:primary>
func (i *insertExecutor) parsePkValuesFromStatement(insertStmt
*ast.InsertStmt, meta types.TableMeta, nameValues []driver.NamedValue)
(map[string][]interface{}, error) {
@@ -590,9 +610,7 @@ func (i *insertExecutor)
parsePkValuesFromStatement(insertStmt *ast.InsertStmt,
} else {
pkValues = append(pkValues, pkValue)
}
- if _, ok := pkValuesMap[pkKey]; !ok {
- pkValuesMap[pkKey] = pkValues
- }
+ pkValuesMap[pkKey] = pkValues
}
}
} else {
@@ -707,7 +725,7 @@ func (i *insertExecutor) getPkValuesByAuto(ctx
context.Context, execCtx *types.E
// If there is batch insert
// do auto increment base LAST_INSERT_ID and variable
`auto_increment_increment`
- if lastInsertId > 0 && updateCount > 1 && canAutoIncrement(pkMetaMap) {
+ if lastInsertId > 0 && updateCount > 1 && canAutoGeneratePKs(pkMetaMap)
{
return i.autoGeneratePks(execCtx, autoColumnName, lastInsertId,
updateCount)
}
@@ -721,12 +739,11 @@ func (i *insertExecutor) getPkValuesByAuto(ctx
context.Context, execCtx *types.E
return nil, nil
}
-func canAutoIncrement(pkMetaMap map[string]types.ColumnMeta) bool {
- if len(pkMetaMap) != 1 {
- return false
- }
+func canAutoGeneratePKs(pkMetaMap map[string]types.ColumnMeta) bool {
for _, meta := range pkMetaMap {
- return meta.Autoincrement
+ if meta.Autoincrement {
+ return true
+ }
}
return false
}
@@ -746,22 +763,27 @@ func (i *insertExecutor) autoGeneratePks(execCtx
*types.ExecContext, autoColumnN
log.Errorf("build prepare stmt: %+v", err)
return nil, err
}
+ defer stmt.Close()
rows, err := stmt.Query(nil)
if err != nil {
log.Errorf("stmt query: %+v", err)
return nil, err
}
+ defer rows.Close()
- if len(rows.Columns()) > 0 {
- var curStep []driver.Value
+ columns := rows.Columns()
+ if len(columns) > 1 {
+ curStep := make([]driver.Value, len(columns))
if err := rows.Next(curStep); err != nil {
return nil, err
}
- if curStepInt, ok := curStep[0].(int64); ok {
- step = curStepInt
+ curStepInt, err := parseAutoIncrementStep(curStep[1])
+ if err != nil {
+ return nil, err
}
+ step = curStepInt
} else {
return nil, fmt.Errorf("query is empty")
}
@@ -780,11 +802,24 @@ func (i *insertExecutor) autoGeneratePks(execCtx
*types.ExecContext, autoColumnN
return pkValuesMap, nil
}
+func parseAutoIncrementStep(value driver.Value) (int64, error) {
+ switch v := value.(type) {
+ case int64:
+ return v, nil
+ case []byte:
+ return strconv.ParseInt(string(v), 10, 64)
+ case string:
+ return strconv.ParseInt(v, 10, 64)
+ default:
+ return 0, fmt.Errorf("unsupported auto_increment_increment
value type %T", value)
+ }
+}
+
func pkValuesMapMerge(dest *map[string][]interface{}, src
map[string][]interface{}) {
for k, v := range src {
tmpK := k
tmpV := v
- (*dest)[tmpK] = append((*dest)[tmpK], tmpV)
+ (*dest)[tmpK] = tmpV
}
}
diff --git a/pkg/datasource/sql/exec/at/insert_executor_test.go
b/pkg/datasource/sql/exec/at/insert_executor_test.go
index 7a26dbfe..b3a2463e 100644
--- a/pkg/datasource/sql/exec/at/insert_executor_test.go
+++ b/pkg/datasource/sql/exec/at/insert_executor_test.go
@@ -107,6 +107,225 @@ func TestBuildSelectSQLByInsert(t *testing.T) {
expectQuery: "SELECT user_id, name FROM user WHERE
(`user_id`) IN ((?)) ",
expectQueryArgs: []driver.Value{int64(20)},
},
+ {
+ name: "single pk without explicit columns",
+ query: "insert into user values
(19,'Tony'),(21,'tony')",
+ metaData: types.TableMeta{
+ ColumnNames: []string{"id", "name"},
+ Indexs: map[string]types.IndexMeta{
+ "id": {
+ IType:
types.IndexTypePrimaryKey,
+ ColumnName: "id",
+ Columns: []types.ColumnMeta{
+ {
+ ColumnName:
"id",
+ DatabaseType:
types.GetSqlDataType("BIGINT"),
+
DatabaseTypeString: "BIGINT",
+ },
+ },
+ },
+ },
+ Columns: map[string]types.ColumnMeta{
+ "id": {ColumnName: "id"},
+ "name": {ColumnName: "name"},
+ },
+ },
+ expectQuery: "SELECT id, name FROM user WHERE
(`id`) IN ((?),(?)) ",
+ expectQueryArgs: []driver.Value{int64(19), int64(21)},
+ },
+ {
+ name: "composite pk explicit values",
+ query: "insert into user(id,tenant_id,name) values
(19,100,'Tony'),(21,101,'tony')",
+ metaData: types.TableMeta{
+ ColumnNames: []string{"id", "tenant_id",
"name"},
+ Indexs: map[string]types.IndexMeta{
+ "PRIMARY": {
+ IType:
types.IndexTypePrimaryKey,
+ Columns: []types.ColumnMeta{
+ {
+ ColumnName:
"id",
+ DatabaseType:
types.GetSqlDataType("BIGINT"),
+
DatabaseTypeString: "BIGINT",
+ },
+ {
+ ColumnName:
"tenant_id",
+ DatabaseType:
types.GetSqlDataType("BIGINT"),
+
DatabaseTypeString: "BIGINT",
+ },
+ },
+ },
+ },
+ Columns: map[string]types.ColumnMeta{
+ "id": {ColumnName: "id"},
+ "tenant_id": {ColumnName: "tenant_id"},
+ "name": {ColumnName: "name"},
+ },
+ },
+ expectQuery: "SELECT id, tenant_id, name FROM user
WHERE (`id`,`tenant_id`) IN ((?,?),(?,?)) ",
+ expectQueryArgs: []driver.Value{int64(19), int64(100),
int64(21), int64(101)},
+ },
+ {
+ name: "composite pk without explicit columns",
+ query: "insert into user values
(19,100,'Tony'),(21,101,'tony')",
+ metaData: types.TableMeta{
+ ColumnNames: []string{"id", "tenant_id",
"name"},
+ Indexs: map[string]types.IndexMeta{
+ "PRIMARY": {
+ IType:
types.IndexTypePrimaryKey,
+ Columns: []types.ColumnMeta{
+ {
+ ColumnName:
"id",
+ DatabaseType:
types.GetSqlDataType("BIGINT"),
+
DatabaseTypeString: "BIGINT",
+ },
+ {
+ ColumnName:
"tenant_id",
+ DatabaseType:
types.GetSqlDataType("BIGINT"),
+
DatabaseTypeString: "BIGINT",
+ },
+ },
+ },
+ },
+ Columns: map[string]types.ColumnMeta{
+ "id": {ColumnName: "id"},
+ "tenant_id": {ColumnName: "tenant_id"},
+ "name": {ColumnName: "name"},
+ },
+ },
+ expectQuery: "SELECT id, tenant_id, name FROM user
WHERE (`id`,`tenant_id`) IN ((?,?),(?,?)) ",
+ expectQueryArgs: []driver.Value{int64(19), int64(100),
int64(21), int64(101)},
+ },
+ {
+ name: "composite pk escaped explicit values",
+ query: "insert into user(`id`,`tenant_id`,name) values
(19,100,'Tony'),(21,101,'tony')",
+ metaData: types.TableMeta{
+ ColumnNames: []string{"id", "tenant_id",
"name"},
+ Indexs: map[string]types.IndexMeta{
+ "PRIMARY": {
+ IType:
types.IndexTypePrimaryKey,
+ Columns: []types.ColumnMeta{
+ {
+ ColumnName:
"id",
+ DatabaseType:
types.GetSqlDataType("BIGINT"),
+
DatabaseTypeString: "BIGINT",
+ },
+ {
+ ColumnName:
"tenant_id",
+ DatabaseType:
types.GetSqlDataType("BIGINT"),
+
DatabaseTypeString: "BIGINT",
+ },
+ },
+ },
+ },
+ Columns: map[string]types.ColumnMeta{
+ "id": {ColumnName: "id"},
+ "tenant_id": {ColumnName: "tenant_id"},
+ "name": {ColumnName: "name"},
+ },
+ },
+ expectQuery: "SELECT id, tenant_id, name FROM user
WHERE (`id`,`tenant_id`) IN ((?,?),(?,?)) ",
+ expectQueryArgs: []driver.Value{int64(19), int64(100),
int64(21), int64(101)},
+ },
+ {
+ name: "composite pk prepared explicit values",
+ query: "insert into user(id,tenant_id,name)
values (?,?,?),(?,?,?)",
+ NamedValues: []driver.NamedValue{{Ordinal: 1, Value:
int64(19)}, {Ordinal: 2, Value: int64(100)}, {Ordinal: 3, Value: "Tony"},
{Ordinal: 4, Value: int64(21)}, {Ordinal: 5, Value: int64(101)}, {Ordinal: 6,
Value: "tony"}},
+ metaData: types.TableMeta{
+ ColumnNames: []string{"id", "tenant_id",
"name"},
+ Indexs: map[string]types.IndexMeta{
+ "PRIMARY": {
+ IType:
types.IndexTypePrimaryKey,
+ Columns: []types.ColumnMeta{
+ {
+ ColumnName:
"id",
+ DatabaseType:
types.GetSqlDataType("BIGINT"),
+
DatabaseTypeString: "BIGINT",
+ },
+ {
+ ColumnName:
"tenant_id",
+ DatabaseType:
types.GetSqlDataType("BIGINT"),
+
DatabaseTypeString: "BIGINT",
+ },
+ },
+ },
+ },
+ Columns: map[string]types.ColumnMeta{
+ "id": {ColumnName: "id"},
+ "tenant_id": {ColumnName: "tenant_id"},
+ "name": {ColumnName: "name"},
+ },
+ },
+ expectQuery: "SELECT id, tenant_id, name FROM user
WHERE (`id`,`tenant_id`) IN ((?,?),(?,?)) ",
+ expectQueryArgs: []driver.Value{int64(19), int64(100),
int64(21), int64(101)},
+ },
+ {
+ name: "composite pk omitted auto increment column",
+ query: "insert into user(tenant_id,name) values
(100,'Tony'),(101,'tony')",
+ metaData: types.TableMeta{
+ ColumnNames: []string{"id", "tenant_id",
"name"},
+ Indexs: map[string]types.IndexMeta{
+ "PRIMARY": {
+ IType:
types.IndexTypePrimaryKey,
+ Columns: []types.ColumnMeta{
+ {
+ ColumnName:
"id",
+ DatabaseType:
types.GetSqlDataType("BIGINT"),
+
DatabaseTypeString: "BIGINT",
+ Autoincrement:
true,
+ },
+ {
+ ColumnName:
"tenant_id",
+ DatabaseType:
types.GetSqlDataType("BIGINT"),
+
DatabaseTypeString: "BIGINT",
+ },
+ },
+ },
+ },
+ Columns: map[string]types.ColumnMeta{
+ "id": {ColumnName: "id"},
+ "tenant_id": {ColumnName: "tenant_id"},
+ "name": {ColumnName: "name"},
+ },
+ },
+ mockInsertResult: mockInsertResult{lastInsertID: 19,
rowsAffected: 2},
+ IncrementStep: 2,
+ expectQuery: "SELECT id, tenant_id, name FROM user
WHERE (`id`,`tenant_id`) IN ((?,?),(?,?)) ",
+ expectQueryArgs: []driver.Value{int64(19), int64(100),
int64(21), int64(101)},
+ },
+ {
+ name: "composite pk null auto increment column",
+ query: "insert into user(id,tenant_id,name) values
(NULL,100,'Tony'),(NULL,101,'tony')",
+ metaData: types.TableMeta{
+ ColumnNames: []string{"id", "tenant_id",
"name"},
+ Indexs: map[string]types.IndexMeta{
+ "PRIMARY": {
+ IType:
types.IndexTypePrimaryKey,
+ Columns: []types.ColumnMeta{
+ {
+ ColumnName:
"id",
+ DatabaseType:
types.GetSqlDataType("BIGINT"),
+
DatabaseTypeString: "BIGINT",
+ Autoincrement:
true,
+ },
+ {
+ ColumnName:
"tenant_id",
+ DatabaseType:
types.GetSqlDataType("BIGINT"),
+
DatabaseTypeString: "BIGINT",
+ },
+ },
+ },
+ },
+ Columns: map[string]types.ColumnMeta{
+ "id": {ColumnName: "id"},
+ "tenant_id": {ColumnName: "tenant_id"},
+ "name": {ColumnName: "name"},
+ },
+ },
+ mockInsertResult: mockInsertResult{lastInsertID: 19,
rowsAffected: 2},
+ IncrementStep: 2,
+ expectQuery: "SELECT id, tenant_id, name FROM user
WHERE (`id`,`tenant_id`) IN ((?,?),(?,?)) ",
+ expectQueryArgs: []driver.Value{int64(19), int64(100),
int64(21), int64(101)},
+ },
}
for _, test := range tests {
@@ -871,6 +1090,18 @@ func TestMySQLInsertUndoLogBuilder_autoGeneratePks(t
*testing.T) {
}, want: map[string][]interface{}{
"id": {int64(100)},
}},
+ {name: "query auto increment step", fields: fields{
+ IncrementStep: 0,
+ }, args: args{
+ execCtx: &types.ExecContext{
+ Conn: &autoIncrementStepConn{value:
[]byte("2")},
+ },
+ autoColumnName: "id",
+ lastInsetId: 100,
+ updateCount: 3,
+ }, want: map[string][]interface{}{
+ "id": {int64(100), int64(102), int64(104)},
+ }},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
@@ -886,3 +1117,57 @@ func TestMySQLInsertUndoLogBuilder_autoGeneratePks(t
*testing.T) {
})
}
}
+
+type autoIncrementStepConn struct {
+ value driver.Value
+}
+
+func (c *autoIncrementStepConn) Prepare(query string) (driver.Stmt, error) {
+ return &autoIncrementStepStmt{value: c.value}, nil
+}
+
+func (c *autoIncrementStepConn) Close() error {
+ return nil
+}
+
+func (c *autoIncrementStepConn) Begin() (driver.Tx, error) {
+ return nil, nil
+}
+
+type autoIncrementStepStmt struct {
+ value driver.Value
+}
+
+func (s *autoIncrementStepStmt) Close() error {
+ return nil
+}
+
+func (s *autoIncrementStepStmt) NumInput() int {
+ return 0
+}
+
+func (s *autoIncrementStepStmt) Exec(args []driver.Value) (driver.Result,
error) {
+ return nil, nil
+}
+
+func (s *autoIncrementStepStmt) Query(args []driver.Value) (driver.Rows,
error) {
+ return &autoIncrementStepRows{value: s.value}, nil
+}
+
+type autoIncrementStepRows struct {
+ value driver.Value
+}
+
+func (r *autoIncrementStepRows) Columns() []string {
+ return []string{"Variable_name", "Value"}
+}
+
+func (r *autoIncrementStepRows) Close() error {
+ return nil
+}
+
+func (r *autoIncrementStepRows) Next(dest []driver.Value) error {
+ dest[0] = "auto_increment_increment"
+ dest[1] = r.value
+ return nil
+}
diff --git a/pkg/datasource/sql/types/meta.go b/pkg/datasource/sql/types/meta.go
index 90cf53cd..3dc9b7a8 100644
--- a/pkg/datasource/sql/types/meta.go
+++ b/pkg/datasource/sql/types/meta.go
@@ -150,7 +150,11 @@ func (m TableMeta) GetPrimaryKeyTypeStrMap()
(map[string]string, error) {
for _, index := range m.Indexs {
if index.IType == IndexTypePrimaryKey {
for i := range index.Columns {
- pkMap[index.ColumnName] =
index.Columns[i].DatabaseTypeString
+ columnName := index.Columns[i].ColumnName
+ if columnName == "" {
+ columnName = index.ColumnName
+ }
+ pkMap[columnName] =
index.Columns[i].DatabaseTypeString
}
}
}
diff --git a/pkg/datasource/sql/types/meta_test.go
b/pkg/datasource/sql/types/meta_test.go
index c44f3406..d6ec45cf 100644
--- a/pkg/datasource/sql/types/meta_test.go
+++ b/pkg/datasource/sql/types/meta_test.go
@@ -51,6 +51,39 @@ func TestTableMeta_GetPrimaryKeyTypeStrMap(t *testing.T) {
}}, want: map[string]string{
"id": "BIGINT",
}},
+ {name: "composite primary key", fields: fields{TableName:
"test", Indexs: map[string]IndexMeta{
+ "PRIMARY": {
+ Name: "PRIMARY",
+ IType: IndexTypePrimaryKey,
+ Columns: []ColumnMeta{
+ {
+ ColumnName: "id",
+ DatabaseTypeString: "BIGINT",
+ },
+ {
+ ColumnName: "tenant_id",
+ DatabaseTypeString: "BIGINT",
+ },
+ },
+ },
+ }}, want: map[string]string{
+ "id": "BIGINT",
+ "tenant_id": "BIGINT",
+ }},
+ {name: "fallback to index column name", fields:
fields{TableName: "test", Indexs: map[string]IndexMeta{
+ "id": {
+ Name: "id",
+ ColumnName: "id",
+ IType: IndexTypePrimaryKey,
+ Columns: []ColumnMeta{
+ {
+ DatabaseTypeString: "BIGINT",
+ },
+ },
+ },
+ }}, want: map[string]string{
+ "id": "BIGINT",
+ }},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
