*: Improve errors for mutation checker

errno/errcode.go

@@ -1011,6 +1011,9 @@ const (
 	ErrAsOf                                = 8135
 	ErrVariableNoLongerSupported           = 8136
 	ErrAnalyzeMissColumn                   = 8137
+	ErrInconsistentRowValue                = 8138
+	ErrInconsistentHandle                  = 8139
+	ErrInconsistentIndexedValue            = 8140
 
 	// Error codes used by TiDB ddl package
 	ErrUnsupportedDDLOperation            = 8200

errno/errname.go

@@ -1009,6 +1009,9 @@ var MySQLErrName = map[uint16]*mysql.ErrMessage{
 	ErrInvalidIncrementAndOffset:     mysql.Message("Invalid auto_increment settings: auto_increment_increment: %d, auto_increment_offset: %d, both of them must be in range [1..65535]", nil),
 	ErrDataInconsistentMismatchCount: mysql.Message("data inconsistency in table: %s, index: %s, index-count:%d != record-count:%d", nil),
 	ErrDataInconsistentMismatchIndex: mysql.Message("data inconsistency in table: %s, index: %s, col: %s, handle: %#v, index-values:%#v != record-values:%#v, compare err:%#v", []int{3, 4, 5, 6}),
+	ErrInconsistentRowValue:          mysql.Message("writing inconsistent data in table: %s, expected-values:{%s} != record-values:{%s}", []int{1, 2}),
+	ErrInconsistentHandle:            mysql.Message("writing inconsistent data in table: %s, index: %s, index-handle:%#v != record-handle:%#v, index: %#v, record: %#v", []int{2, 3, 4, 5}),
+	ErrInconsistentIndexedValue:      mysql.Message("writing inconsistent data in table: %s, index: %s, col: %s, indexed-value:{%s} != record-value:{%s}", []int{3, 4}),
 
 	ErrWarnOptimizerHintInvalidInteger:  mysql.Message("integer value is out of range in '%s'", nil),
 	ErrWarnOptimizerHintUnsupportedHint: mysql.Message("Optimizer hint %s is not supported by TiDB and is ignored", nil),

errors.toml

@@ -1801,6 +1801,21 @@ error = '''
 column %s can't be in none state
 '''
 
+["table:8138"]
+error = '''
+writing inconsistent data in table: %s, expected-values:{%s} != record-values:{%s}
+'''
+
+["table:8139"]
+error = '''
+writing inconsistent data in table: %s, index: %s, index-handle:%#v != record-handle:%#v, index: %#v, record: %#v
+'''
+
+["table:8140"]
+error = '''
+writing inconsistent data in table: %s, index: %s, col: %s, indexed-value:{%s} != record-value:{%s}
+'''
+
 ["tikv:1105"]
 error = '''
 Unknown error

table/tables/mutation_checker.go

@@ -15,21 +15,30 @@
 package tables
 
 import (
-	"fmt"
-
 	"github.com/pingcap/errors"
+	"github.com/pingcap/tidb/errno"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tidb/sessionctx/stmtctx"
 	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/table"
 	"github.com/pingcap/tidb/tablecodec"
 	"github.com/pingcap/tidb/types"
+	"github.com/pingcap/tidb/util/dbterror"
 	"github.com/pingcap/tidb/util/logutil"
 	"github.com/pingcap/tidb/util/rowcodec"
 	"go.uber.org/zap"
 )
 
+var (
+	// ErrInconsistentRowValue is the error when values in a row insertion does not match the expected ones.
+	ErrInconsistentRowValue = dbterror.ClassTable.NewStd(errno.ErrInconsistentRowValue)
+	// ErrInconsistentHandle is the error when the handle in the row/index insertions does not match.
+	ErrInconsistentHandle = dbterror.ClassTable.NewStd(errno.ErrInconsistentHandle)
+	// ErrInconsistentIndexedValue is the error when decoded values from the index mutation cannot match row value
+	ErrInconsistentIndexedValue = dbterror.ClassTable.NewStd(errno.ErrInconsistentIndexedValue)
+)
+
 type mutation struct {
 	key     kv.Key
 	flags   kv.KeyFlags
@@ -82,7 +91,7 @@ func CheckDataConsistency(
 
 	if rowToInsert != nil {
 		if err := checkRowInsertionConsistency(
-			sessVars, rowToInsert, rowInsertion, columnMaps.ColumnIDToInfo, columnMaps.ColumnIDToFieldType,
+			sessVars, rowToInsert, rowInsertion, columnMaps.ColumnIDToInfo, columnMaps.ColumnIDToFieldType, t.Meta().Name.O,
 		); err != nil {
 			return errors.Trace(err)
 		}
@@ -93,7 +102,7 @@ func CheckDataConsistency(
 	}
 
 	if rowInsertion.key != nil {
-		if err = checkHandleConsistency(rowInsertion, indexMutations, columnMaps.IndexIDToInfo); err != nil {
+		if err = checkHandleConsistency(rowInsertion, indexMutations, columnMaps.IndexIDToInfo, t.Meta().Name.O); err != nil {
 			return errors.Trace(err)
 		}
 	}
@@ -110,7 +119,7 @@ func CheckDataConsistency(
 // in row insertions and index insertions are consistent.
 // A PUT_index implies a PUT_row with the same handle.
 // Deletions are not checked since the values of deletions are unknown
-func checkHandleConsistency(rowInsertion mutation, indexMutations []mutation, indexIDToInfo map[int64]*model.IndexInfo) error {
+func checkHandleConsistency(rowInsertion mutation, indexMutations []mutation, indexIDToInfo map[int64]*model.IndexInfo, tableName string) error {
 	var insertionHandle kv.Handle
 	var err error
 
@@ -138,9 +147,9 @@ func checkHandleConsistency(rowInsertion mutation, indexMutations []mutation, in
 		}
 		// NOTE: handle type can be different, see issue 29520
 		if indexHandle.IsInt() == insertionHandle.IsInt() && indexHandle.Compare(insertionHandle) != 0 {
-			return errors.Errorf("inconsistent handles in row and index insertions. index handle = %v, "+
-				"row handle = %v, index = %+v, row = %+v",
-				indexHandle, insertionHandle, m, rowInsertion)
+			err = ErrInconsistentHandle.GenWithStackByArgs(tableName, indexInfo.Name.O, indexHandle, insertionHandle, m, rowInsertion)
+			logutil.BgLogger().Error("inconsistent handle in index and record insertions", zap.Error(err))
+			return err
 		}
 	}
 
@@ -204,9 +213,9 @@ func checkIndexKeys(
 		}
 
 		if len(m.value) == 0 {
-			err = compareIndexData(sessVars.StmtCtx, t.Columns, indexData, rowToRemove, indexInfo)
+			err = compareIndexData(sessVars.StmtCtx, t.Columns, indexData, rowToRemove, indexInfo, t.Meta())
 		} else {
-			err = compareIndexData(sessVars.StmtCtx, t.Columns, indexData, rowToInsert, indexInfo)
+			err = compareIndexData(sessVars.StmtCtx, t.Columns, indexData, rowToInsert, indexInfo, t.Meta())
 		}
 		if err != nil {
 			return errors.Trace(err)
@@ -219,7 +228,7 @@ func checkIndexKeys(
 // We only check data added since a deletion of a row doesn't care about its value (and we cannot know it)
 func checkRowInsertionConsistency(
 	sessVars *variable.SessionVars, rowToInsert []types.Datum, rowInsertion mutation,
-	columnIDToInfo map[int64]*model.ColumnInfo, columnIDToFieldType map[int64]*types.FieldType,
+	columnIDToInfo map[int64]*model.ColumnInfo, columnIDToFieldType map[int64]*types.FieldType, tableName string,
 ) error {
 	if rowToInsert == nil {
 		// it's a deletion
@@ -241,14 +250,9 @@ func checkRowInsertionConsistency(
 			return errors.Trace(err)
 		}
 		if cmp != 0 {
-			logutil.BgLogger().Error(
-				"inconsistent row mutation", zap.String("decoded datum", decodedDatum.String()),
-				zap.String("input datum", inputDatum.String()),
-			)
-			return errors.Errorf(
-				"inconsistent row mutation, row datum = {%v}, input datum = {%v}", decodedDatum.String(),
-				inputDatum.String(),
-			)
+			err = ErrInconsistentRowValue.GenWithStackByArgs(tableName, inputDatum.String(), decodedDatum.String())
+			logutil.BgLogger().Error("inconsistent row value in row insertion", zap.Error(err))
+			return err
 		}
 	}
 	return nil
@@ -295,6 +299,7 @@ func collectTableMutationsFromBufferStage(t *TableCommon, memBuffer kv.MemBuffer
 // Returns error if the index data is not a subset of the input data.
 func compareIndexData(
 	sc *stmtctx.StatementContext, cols []*table.Column, indexData, input []types.Datum, indexInfo *model.IndexInfo,
+	tableInfo *model.TableInfo,
 ) error {
 	for i := range indexData {
 		decodedMutationDatum := indexData[i]
@@ -315,12 +320,12 @@ func compareIndexData(
 		}
 
 		if comparison != 0 {
-			logutil.BgLogger().Error(
-				"inconsistent index values",
-				zap.String("truncated mutation datum", fmt.Sprintf("%v", decodedMutationDatum)),
-				zap.String("truncated expected datum", fmt.Sprintf("%v", expectedDatum)),
+			err = ErrInconsistentIndexedValue.GenWithStackByArgs(
+				tableInfo.Name.O, indexInfo.Name.O, cols[indexInfo.Columns[i].Offset].ColumnInfo.Name.O,
+				decodedMutationDatum.String(), expectedDatum.String(),
 			)
-			return errors.New("inconsistent index values")
+			logutil.BgLogger().Error("inconsistent indexed value in index insertion", zap.Error(err))
+			return err
 		}
 	}
 	return nil

table/tables/mutation_checker_test.go

@@ -15,6 +15,7 @@
 package tables
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
@@ -75,12 +76,12 @@ func TestCompareIndexData(t *testing.T) {
 		cols := make([]*table.Column, 0)
 		indexCols := make([]*model.IndexColumn, 0)
 		for i, ft := range data.fts {
-			cols = append(cols, &table.Column{ColumnInfo: &model.ColumnInfo{FieldType: *ft}})
+			cols = append(cols, &table.Column{ColumnInfo: &model.ColumnInfo{Name: model.NewCIStr(fmt.Sprintf("c%d", i)), FieldType: *ft}})
 			indexCols = append(indexCols, &model.IndexColumn{Offset: i, Length: data.indexLength[i]})
 		}
-		indexInfo := &model.IndexInfo{Columns: indexCols}
+		indexInfo := &model.IndexInfo{Name: model.NewCIStr("i0"), Columns: indexCols}
 
-		err := compareIndexData(sc, cols, data.indexData, data.inputData, indexInfo)
+		err := compareIndexData(sc, cols, data.indexData, data.inputData, indexInfo, &model.TableInfo{Name: model.NewCIStr("t")})
 		require.Equal(t, data.correct, err == nil, "case id = %v", caseID)
 	}
 }
@@ -166,7 +167,7 @@ func TestCheckRowInsertionConsistency(t *testing.T) {
 
 	for caseID, data := range testData {
 		err := checkRowInsertionConsistency(
-			sessVars, data.rowToInsert, data.rowInsertion, data.columnIDToInfo, data.columnIDToFieldType,
+			sessVars, data.rowToInsert, data.rowInsertion, data.columnIDToInfo, data.columnIDToFieldType, "t",
 		)
 		require.Equal(t, data.correct, err == nil, "case id = %v", caseID)
 	}
@@ -309,9 +310,9 @@ func TestCheckIndexKeysAndCheckHandleConsistency(t *testing.T) {
 					require.Nil(t, err)
 					rowMutation := mutation{key: rowKey, value: rowValue}
 					corruptedRowMutation := mutation{key: corruptedRowKey, value: rowValue}
-					err = checkHandleConsistency(rowMutation, indexMutations, maps.IndexIDToInfo)
+					err = checkHandleConsistency(rowMutation, indexMutations, maps.IndexIDToInfo, "t")
 					require.Nil(t, err)
-					err = checkHandleConsistency(corruptedRowMutation, indexMutations, maps.IndexIDToInfo)
+					err = checkHandleConsistency(corruptedRowMutation, indexMutations, maps.IndexIDToInfo, "t")
 					require.NotNil(t, err)
 				}
 			}