[DNM] *: add some injections to simulate mutation errors

kv/kv.go

@@ -173,6 +173,9 @@ type MemBuffer interface {
 
 	// Size returns sum of keys and values length.
 	Size() int
+
+	// RemoveFromBuffer removes the entry from the buffer. It's used for testing.
+	RemoveFromBuffer(Key)
 }
 
 // LockCtx contains information for LockKeys method.

session/session.go

@@ -37,22 +37,6 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/kvproto/pkg/kvrpcpb"
-	"github.com/pingcap/tidb/parser"
-	"github.com/pingcap/tidb/parser/ast"
-	"github.com/pingcap/tidb/parser/auth"
-	"github.com/pingcap/tidb/parser/charset"
-	"github.com/pingcap/tidb/parser/model"
-	"github.com/pingcap/tidb/parser/mysql"
-	"github.com/pingcap/tidb/parser/terror"
-	"github.com/pingcap/tidb/sessiontxn"
-	"github.com/pingcap/tidb/store/driver/txn"
-	"github.com/pingcap/tidb/table/tables"
-	"github.com/pingcap/tidb/table/temptable"
-	"github.com/pingcap/tidb/util/topsql"
-	topsqlstate "github.com/pingcap/tidb/util/topsql/state"
-	"github.com/pingcap/tipb/go-binlog"
-	"go.uber.org/zap"
-
 	"github.com/pingcap/tidb/bindinfo"
 	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/ddl"
@@ -65,6 +49,13 @@ import (
 	"github.com/pingcap/tidb/meta"
 	"github.com/pingcap/tidb/metrics"
 	"github.com/pingcap/tidb/owner"
+	"github.com/pingcap/tidb/parser"
+	"github.com/pingcap/tidb/parser/ast"
+	"github.com/pingcap/tidb/parser/auth"
+	"github.com/pingcap/tidb/parser/charset"
+	"github.com/pingcap/tidb/parser/model"
+	"github.com/pingcap/tidb/parser/mysql"
+	"github.com/pingcap/tidb/parser/terror"
 	"github.com/pingcap/tidb/planner"
 	plannercore "github.com/pingcap/tidb/planner/core"
 	"github.com/pingcap/tidb/plugin"
@@ -75,9 +66,13 @@ import (
 	"github.com/pingcap/tidb/sessionctx/binloginfo"
 	"github.com/pingcap/tidb/sessionctx/stmtctx"
 	"github.com/pingcap/tidb/sessionctx/variable"
+	"github.com/pingcap/tidb/sessiontxn"
 	"github.com/pingcap/tidb/statistics"
 	"github.com/pingcap/tidb/statistics/handle"
 	storeerr "github.com/pingcap/tidb/store/driver/error"
+	"github.com/pingcap/tidb/store/driver/txn"
+	"github.com/pingcap/tidb/table/tables"
+	"github.com/pingcap/tidb/table/temptable"
 	"github.com/pingcap/tidb/tablecodec"
 	"github.com/pingcap/tidb/telemetry"
 	"github.com/pingcap/tidb/types"
@@ -92,10 +87,14 @@ import (
 	"github.com/pingcap/tidb/util/sqlexec"
 	"github.com/pingcap/tidb/util/tableutil"
 	"github.com/pingcap/tidb/util/timeutil"
+	"github.com/pingcap/tidb/util/topsql"
+	topsqlstate "github.com/pingcap/tidb/util/topsql/state"
+	"github.com/pingcap/tipb/go-binlog"
 	tikvstore "github.com/tikv/client-go/v2/kv"
 	"github.com/tikv/client-go/v2/oracle"
 	"github.com/tikv/client-go/v2/tikv"
 	tikvutil "github.com/tikv/client-go/v2/util"
+	"go.uber.org/zap"
 )
 
 var (
@@ -679,6 +678,7 @@ func (c *cachedTableRenewLease) commitTSCheck(commitTS uint64) bool {
 
 func (s *session) commitTxnWithTemporaryData(ctx context.Context, txn kv.Transaction) error {
 	sessVars := s.sessionVars
+
 	txnTempTables := sessVars.TxnCtx.TemporaryTables
 	if len(txnTempTables) == 0 {
 		return txn.Commit(ctx)

store/driver/txn/unionstore_driver.go

@@ -43,6 +43,10 @@ func (m *memBuffer) Delete(k kv.Key) error {
 	return m.MemDB.Delete(k)
 }
 
+func (m *memBuffer) RemoveFromBuffer(k kv.Key) {
+	m.MemDB.RemoveFromBuffer(k)
+}
+
 func (m *memBuffer) DeleteWithFlags(k kv.Key, ops ...kv.FlagsOp) error {
 	err := m.MemDB.DeleteWithFlags(k, getTiKVFlagsOps(ops)...)
 	return derr.ToTiDBErr(err)

table/tables/mutation_checker.go

@@ -15,7 +15,11 @@
 package tables
 
 import (
+	"fmt"
+	"strings"
+
 	"github.com/pingcap/errors"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/errno"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/parser/model"
@@ -381,3 +385,100 @@ func getOrBuildColumnMaps(
 	}
 	return maps
 }
+
+// only used in tests
+// commands is a comma separated string, each representing a type of corruptions to the mutations
+// The injection depends on actual encoding rules.
+func corruptMutations(t *TableCommon, txn kv.Transaction, sh kv.StagingHandle, cmds string) error {
+	commands := strings.Split(cmds, ",")
+	memBuffer := txn.GetMemBuffer()
+
+	indexMutations, _, err := collectTableMutationsFromBufferStage(t, memBuffer, sh)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	for _, cmd := range commands {
+		switch cmd {
+		case "extraIndex":
+			// an extra index mutation
+			// TODO: distinguish which part to corrupt, value or handle
+			{
+				if len(indexMutations) == 0 {
+					continue
+				}
+				indexMutation := indexMutations[0]
+				key := make([]byte, len(indexMutation.key))
+				copy(key, indexMutation.key)
+				key[len(key)-1] += 1
+				if len(indexMutation.value) == 0 {
+					if err := memBuffer.Delete(key); err != nil {
+						return errors.Trace(err)
+					}
+				} else {
+					if err := memBuffer.Set(key, indexMutation.value); err != nil {
+						return errors.Trace(err)
+					}
+				}
+			}
+		//case "extraIndexByHandle":
+		//	{
+		//	}
+		case "missingIndex":
+			// an index mutation is missing
+			// "missIndex" should be placed in front of "extraIndex"es,
+			// in case it removes the mutation that was just added
+			{
+				indexMutation := indexMutations[0]
+				memBuffer.RemoveFromBuffer(indexMutation.key)
+			}
+		case "corruptIndexKey":
+			// a corrupted index mutation.
+			// TODO: distinguish which part is corrupted, value or handle
+			{
+				indexMutation := indexMutations[0]
+				key := indexMutation.key
+				memBuffer.RemoveFromBuffer(key)
+				key[len(key)-1] += 1
+				if err := memBuffer.Set(key, indexMutation.value); err != nil {
+					return errors.Trace(err)
+				}
+			}
+		//case "corruptIndexKeyByHandle":
+		//	{
+		//	}
+		case "corruptIndexValue":
+			// TODO: distinguish which part to corrupt, int handle, common handle, or restored data?
+			// It doesn't make much sense to always corrupt the last byte
+			{
+				if len(indexMutations) == 0 {
+					continue
+				}
+				indexMutation := indexMutations[0]
+				value := indexMutation.value
+				if len(value) > 0 {
+					value[len(value)-1] += 1
+				}
+				if err := memBuffer.Set(indexMutation.key, value); err != nil {
+					return errors.Trace(err)
+				}
+			}
+		//case "corruptIndexValueCommonHandle":
+		//	{
+		//	}
+		//case "missIndexValueRestoredData":
+		//	{
+		//	}
+		default:
+			return errors.New(fmt.Sprintf("unknown command to corrupt mutation: %s", cmd))
+		}
+	}
+	return nil
+}
+
+func injectMutationError(t *TableCommon, txn kv.Transaction, sh kv.StagingHandle) error {
+	failpoint.Inject("corruptMutations", func(commands failpoint.Value) {
+		failpoint.Return(corruptMutations(t, txn, sh, commands.(string)))
+	})
+	return nil
+}

table/tables/tables.go

@@ -446,11 +446,16 @@ func (t *TableCommon) UpdateRecord(ctx context.Context, sctx sessionctx.Context,
 		return err
 	}
 
+
+	if err = injectMutationError(t, txn, sh); err != nil {
+		return err
+	}
 	if sessVars.EnableMutationChecker {
 		if err = CheckDataConsistency(txn, sessVars, t, newData, oldData, memBuffer, sh); err != nil {
 			return errors.Trace(err)
 		}
 	}
+
 	memBuffer.Release(sh)
 	if shouldWriteBinlog(sctx, t.meta) {
 		if !t.meta.PKIsHandle && !t.meta.IsCommonHandle {
@@ -886,6 +891,9 @@ func (t *TableCommon) AddRecord(sctx sessionctx.Context, r []types.Datum, opts .
 		return h, err
 	}
 
+	if err = injectMutationError(t, txn, sh); err != nil {
+		return nil, err
+	}
 	if sessVars.EnableMutationChecker {
 		if err = CheckDataConsistency(txn, sessVars, t, r, nil, memBuffer, sh); err != nil {
 			return nil, errors.Trace(err)
@@ -1146,6 +1154,9 @@ func (t *TableCommon) RemoveRecord(ctx sessionctx.Context, h kv.Handle, r []type
 
 	sessVars := ctx.GetSessionVars()
 	sc := sessVars.StmtCtx
+	if err = injectMutationError(t, txn, sh); err != nil {
+		return err
+	}
 	if sessVars.EnableMutationChecker {
 		if err = CheckDataConsistency(txn, sessVars, t, nil, r, memBuffer, sh); err != nil {
 			return errors.Trace(err)