thunguo commented on code in PR #1141:
URL:
https://github.com/apache/incubator-seata-go/pull/1141#discussion_r3851256218
##########
pkg/datasource/sql/undo/executor/mysql_undo_insert_executor.go:
##########
@@ -43,13 +43,19 @@ func newMySQLUndoInsertExecutor(sqlUndoLog undo.SQLUndoLog)
*mySQLUndoInsertExec
// ExecuteOn execute insert undo logic
func (m *mySQLUndoInsertExecutor) ExecuteOn(ctx context.Context, dbType
types.DBType, conn *sql.Conn) error {
m.BaseExecutor.dbType = dbType
-
- if err := m.BaseExecutor.ExecuteOn(ctx, dbType, conn); err != nil {
+ ok, err := m.BaseExecutor.dataValidationAndGoOn(ctx, conn)
Review Comment:
When `dataValidationAndGoOn` is enabled here, standard `INSERT` operations
involving `DECIMAL` types fail to roll back correctly. The Phase 1 image scans
`DECIMAL` values as `float64` based on table metadata, whereas
`queryCurrentRecords` scans them as strings (e.g., `"13.370000"`) based on
`go-sql-driver`'s `ColumnTypes`, leading to a mismatch (e.g., `float64(13.37)`
vs. `"13.370000"`). Since the current `DeepEqual` logic does not perform
string-to-numeric conversion, it incorrectly triggers a `SQLUndoDirtyError`. It
is recommended to standardize the scanning/normalization of `DECIMAL` values
based on the undo image's `JDBCType` and to add test cases specifically for
`INSERT` rollbacks against a real MySQL instance; however, one should avoid
parsing all numeric strings during general comparisons to prevent `VARCHAR`
values from being misidentified.
##########
pkg/datasource/sql/undo/parser/parser_protobuf.go:
##########
@@ -244,6 +246,55 @@ func convertAnyToInterface(anyValue *any.Any)
(interface{}, error) {
return value, nil
}
+func convertAnyToColumnValue(anyValue *any.Any, columnType types.JDBCType)
(interface{}, error) {
+ bytesValue := &wrappers.BytesValue{}
+ if err := anypb.UnmarshalTo(anyValue, bytesValue,
proto.UnmarshalOptions{}); err != nil {
+ return nil, err
+ }
+ if bytes.Equal(bytesValue.Value, []byte("null")) {
+ return nil, nil
+ }
+
+ switch columnType {
+ case types.JDBCTypeReal, types.JDBCTypeDecimal, types.JDBCTypeDouble,
+ types.JDBCTypeTinyInt, types.JDBCTypeSmallInt,
types.JDBCTypeInteger, types.JDBCTypeBigInt:
+ decoder := json.NewDecoder(bytes.NewReader(bytesValue.Value))
+ decoder.UseNumber()
+ var value json.Number
+ if err := decoder.Decode(&value); err != nil {
+ return nil, err
+ }
+ switch columnType {
+ case types.JDBCTypeReal:
+ parsed, err := strconv.ParseFloat(value.String(), 32)
+ return float32(parsed), err
+ case types.JDBCTypeDecimal, types.JDBCTypeDouble:
+ return strconv.ParseFloat(value.String(), 64)
+ case types.JDBCTypeTinyInt:
+ parsed, err := strconv.ParseInt(value.String(), 10, 8)
Review Comment:
Here, `ParseInt` is called based on JDBC signed width; however, MySQL's
unsigned types—`TINYINT UNSIGNED` (255), `SMALLINT UNSIGNED` (65535), and `INT
UNSIGNED` (4294967295)—cannot be parsed as 8-bit, 16-bit, or 32-bit signed
integers, respectively. The JSON undo log decoding fails outright, while the
protobuf path silently skips the column (via a `continue` statement),
potentially resulting in incomplete undo SQL. Since `ColumnImage` currently
lacks the unsigned attribute, the type cannot be narrowed based solely on
`JDBCType`. It is recommended to either retain the type as `int64`/`uint64` or
pass the unsigned metadata into the decoding process, while also prohibiting
the silent skipping of columns upon conversion failure. Additionally, please
add JSON/protobuf round-trip tests covering the boundary values for these
three unsigned types.
##########
pkg/datasource/sql/exec/at/insert_executor.go:
##########
@@ -385,57 +485,146 @@ func (i *insertExecutor) buildAfterImageSQL(ctx
context.Context) (string, []driv
for _, column := range i.parserCtx.InsertStmt.Columns {
insertColumns = append(insertColumns, column.Name.O)
}
- sb.WriteString("SELECT " + strings.Join(i.getNeedColumns(meta,
insertColumns, dbType), ", "))
+ selectColumns, err := buildImageSelectColumns(meta, insertColumns,
dbType, undo.UndoConfig.OnlyCareUpdateColumns)
+ if err != nil {
+ return "", nil, err
+ }
+ sb.WriteString("SELECT " + strings.Join(selectColumns, ", "))
suffix.WriteString(" FROM " + tableName)
whereSQL := i.buildWhereConditionByPKs(pkColumnNameList, rowSize,
dbType, maxInSize)
suffix.WriteString(" WHERE " + whereSQL + " ")
sb.WriteString(suffix.String())
+ i.resolvedPKRows = pkRowImages
return sb.String(), i.buildPKParams(pkRowImages, pkColumnNameList,
dbType), nil
}
-func (i *insertExecutor) getPkValues(ctx context.Context, execCtx
*types.ExecContext, parseCtx *types.ParseContext, meta types.TableMeta)
(map[string][]interface{}, error) {
- pkColumnNameList := meta.GetPrimaryKeyOnlyName()
- pkValuesMap := make(map[string][]interface{})
- var err error
- // when there is only one pk in the table
- if len(pkColumnNameList) == 1 {
- if i.containsPK(meta, parseCtx) {
- // the insert sql contain pk value
- pkValuesMap, err = i.getPkValuesByColumn(ctx, execCtx)
- if err != nil {
- return nil, err
- }
- } else if containsColumns(parseCtx) {
- // the insert table pk auto generated
- pkValuesMap, err = i.getPkValuesByAuto(ctx, execCtx)
- if err != nil {
- return nil, err
- }
- } else {
- pkValuesMap, err = i.getPkValuesByColumn(ctx, execCtx)
- if err != nil {
- return nil, err
- }
- }
- } else {
- // when there is multiple pk in the table
- // 1,all pk columns are filled value.
- // 2,the auto increment pk column value is null, and other pk
value are not null.
- pkValuesMap, err = i.getPkValuesByColumn(ctx, execCtx)
+func (i *insertExecutor) getPkValues(execCtx *types.ExecContext, meta
types.TableMeta) (map[string][]interface{}, error) {
+ if i.keyPlan == nil {
+ plan, err := i.buildInsertKeyPlan(&meta)
if err != nil {
return nil, err
}
- for _, columnName := range pkColumnNameList {
- if _, ok := pkValuesMap[columnName]; !ok {
- curPkValuesMap, err := i.getPkValuesByAuto(ctx,
execCtx)
- if err != nil {
- return nil, err
+ i.keyPlan = plan
+ }
+ return i.resolveInsertKeyPlan(execCtx)
+}
+
+func (i *insertExecutor) buildInsertKeyPlan(meta *types.TableMeta)
(*insertKeyPlan, error) {
+ if meta == nil || !i.isAstStmtValid() {
+ return nil, fmt.Errorf("invalid insert metadata or statement")
+ }
+ stmt := i.parserCtx.InsertStmt
+ if stmt.Select != nil || len(stmt.Lists) == 0 {
+ return nil, fmt.Errorf("insert source other than VALUES is
unsupported")
+ }
+ if stmt.IgnoreErr && len(stmt.Lists) > 1 {
+ return nil, fmt.Errorf("multi-values INSERT IGNORE is
unsupported because successful rows cannot be determined")
+ }
+
+ pkNames := meta.GetPrimaryKeyOnlyName()
+ if len(pkNames) == 0 {
+ return nil, fmt.Errorf("pk columnName size is zero")
+ }
+ pkValues, err := i.parsePkValuesFromStatement(stmt, *meta,
i.execContext.NamedValues)
+ if err != nil {
+ return nil, err
+ }
+
+ plan := &insertKeyPlan{rowCount: len(stmt.Lists), pkValues:
make(map[string][]interface{})}
+ pkMeta := meta.GetPrimaryKeyMap()
+ for _, pkName := range pkNames {
+ columnMeta, ok := pkMeta[pkName]
+ if !ok {
+ return nil, fmt.Errorf("primary key metadata not found
for %s", pkName)
+ }
+ values, present := pkValues[pkName]
+ generatedCount := 0
+ if present {
+ if len(values) != plan.rowCount {
+ return nil, fmt.Errorf("insert primary key %s
has %d values, want %d", pkName, len(values), plan.rowCount)
+ }
+ for _, value := range values {
+ switch value.(type) {
+ case nil, *ast.DefaultExpr, ast.DefaultExpr:
Review Comment:
By default, MySQL treats a value of `0` in an `AUTO_INCREMENT` column as a
signal to generate a new value; the literal `0` is stored only when the session
has `NO_AUTO_VALUE_ON_ZERO` enabled. Since only `nil` or `DEFAULT` are
classified here as "generated," an `INSERT ... VALUES (0, ...)` statement
causes the system to look up the "after-image" using `id=0`, whereas the actual
inserted row uses a newly generated ID; this ultimately results in an
after-image validation error after the business SQL has executed. It is
recommended to read `@@SESSION.sql_mode` for the current connection and—when
that mode is disabled—route literal or bound zeros through the `LastInsertId`
path, while also ensuring test coverage for both `sql_mode` settings and
prepared statement parameters.
##########
pkg/datasource/sql/exec/at/insert_executor.go:
##########
@@ -540,94 +715,54 @@ func (i *insertExecutor) matchPKColumnName(columnName
string, meta types.TableMe
// 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) {
if insertStmt == nil {
- return nil, nil
- }
- pkIndexMap := i.getPkIndex(insertStmt, meta)
- if pkIndexMap == nil || len(pkIndexMap) == 0 {
- return nil, fmt.Errorf("pkIndex is not found")
+ return nil, fmt.Errorf("insert statement is nil")
}
- var pkIndexArray []int
- for _, val := range pkIndexMap {
- tmpVal := val
- pkIndexArray = append(pkIndexArray, tmpVal)
+ if len(insertStmt.Lists) == 0 {
+ return nil, fmt.Errorf("insert VALUES list is empty")
}
- if insertStmt == nil || len(insertStmt.Lists) == 0 {
- return nil, fmt.Errorf("parCtx is nil, perhaps InsertStmt is
empty")
+ expectedColumns := len(insertStmt.Columns)
Review Comment:
Both `INSERT INTO t () VALUES ()` and `INSERT INTO t VALUES ()` are
represented by the current parser as a `VALUES` list with a length of zero.
Since the expected width defaults to the total number of table columns when no
explicit column list is provided, a row-width error inevitably occurs. However,
this is valid MySQL syntax for "using default values for all columns"—often
used to generate only an auto-incrementing primary key. It is recommended to
treat an empty `VALUES` row as all `DEFAULT`s and construct the primary key
plan using `LastInsertId`, while also adding test cases for these two syntaxes.
##########
pkg/datasource/sql/exec/at/insert_executor.go:
##########
@@ -385,57 +485,146 @@ func (i *insertExecutor) buildAfterImageSQL(ctx
context.Context) (string, []driv
for _, column := range i.parserCtx.InsertStmt.Columns {
insertColumns = append(insertColumns, column.Name.O)
}
- sb.WriteString("SELECT " + strings.Join(i.getNeedColumns(meta,
insertColumns, dbType), ", "))
+ selectColumns, err := buildImageSelectColumns(meta, insertColumns,
dbType, undo.UndoConfig.OnlyCareUpdateColumns)
+ if err != nil {
+ return "", nil, err
+ }
+ sb.WriteString("SELECT " + strings.Join(selectColumns, ", "))
suffix.WriteString(" FROM " + tableName)
whereSQL := i.buildWhereConditionByPKs(pkColumnNameList, rowSize,
dbType, maxInSize)
suffix.WriteString(" WHERE " + whereSQL + " ")
sb.WriteString(suffix.String())
+ i.resolvedPKRows = pkRowImages
return sb.String(), i.buildPKParams(pkRowImages, pkColumnNameList,
dbType), nil
}
-func (i *insertExecutor) getPkValues(ctx context.Context, execCtx
*types.ExecContext, parseCtx *types.ParseContext, meta types.TableMeta)
(map[string][]interface{}, error) {
- pkColumnNameList := meta.GetPrimaryKeyOnlyName()
- pkValuesMap := make(map[string][]interface{})
- var err error
- // when there is only one pk in the table
- if len(pkColumnNameList) == 1 {
- if i.containsPK(meta, parseCtx) {
- // the insert sql contain pk value
- pkValuesMap, err = i.getPkValuesByColumn(ctx, execCtx)
- if err != nil {
- return nil, err
- }
- } else if containsColumns(parseCtx) {
- // the insert table pk auto generated
- pkValuesMap, err = i.getPkValuesByAuto(ctx, execCtx)
- if err != nil {
- return nil, err
- }
- } else {
- pkValuesMap, err = i.getPkValuesByColumn(ctx, execCtx)
- if err != nil {
- return nil, err
- }
- }
- } else {
- // when there is multiple pk in the table
- // 1,all pk columns are filled value.
- // 2,the auto increment pk column value is null, and other pk
value are not null.
- pkValuesMap, err = i.getPkValuesByColumn(ctx, execCtx)
+func (i *insertExecutor) getPkValues(execCtx *types.ExecContext, meta
types.TableMeta) (map[string][]interface{}, error) {
+ if i.keyPlan == nil {
+ plan, err := i.buildInsertKeyPlan(&meta)
if err != nil {
return nil, err
}
- for _, columnName := range pkColumnNameList {
- if _, ok := pkValuesMap[columnName]; !ok {
- curPkValuesMap, err := i.getPkValuesByAuto(ctx,
execCtx)
- if err != nil {
- return nil, err
+ i.keyPlan = plan
+ }
+ return i.resolveInsertKeyPlan(execCtx)
+}
+
+func (i *insertExecutor) buildInsertKeyPlan(meta *types.TableMeta)
(*insertKeyPlan, error) {
+ if meta == nil || !i.isAstStmtValid() {
+ return nil, fmt.Errorf("invalid insert metadata or statement")
+ }
+ stmt := i.parserCtx.InsertStmt
+ if stmt.Select != nil || len(stmt.Lists) == 0 {
+ return nil, fmt.Errorf("insert source other than VALUES is
unsupported")
+ }
+ if stmt.IgnoreErr && len(stmt.Lists) > 1 {
+ return nil, fmt.Errorf("multi-values INSERT IGNORE is
unsupported because successful rows cannot be determined")
+ }
+
+ pkNames := meta.GetPrimaryKeyOnlyName()
+ if len(pkNames) == 0 {
+ return nil, fmt.Errorf("pk columnName size is zero")
+ }
+ pkValues, err := i.parsePkValuesFromStatement(stmt, *meta,
i.execContext.NamedValues)
+ if err != nil {
+ return nil, err
+ }
+
+ plan := &insertKeyPlan{rowCount: len(stmt.Lists), pkValues:
make(map[string][]interface{})}
+ pkMeta := meta.GetPrimaryKeyMap()
+ for _, pkName := range pkNames {
+ columnMeta, ok := pkMeta[pkName]
+ if !ok {
+ return nil, fmt.Errorf("primary key metadata not found
for %s", pkName)
+ }
+ values, present := pkValues[pkName]
+ generatedCount := 0
+ if present {
+ if len(values) != plan.rowCount {
+ return nil, fmt.Errorf("insert primary key %s
has %d values, want %d", pkName, len(values), plan.rowCount)
+ }
+ for _, value := range values {
+ switch value.(type) {
+ case nil, *ast.DefaultExpr, ast.DefaultExpr:
Review Comment:
https://dev.mysql.com/doc/refman/8.0/en/example-auto-increment.html
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
