*: Reduce data inconsistencies

br/pkg/lightning/backend/kv/session.go

@@ -218,6 +218,11 @@ func (t *transaction) GetTableInfo(id int64) *model.TableInfo {
 func (t *transaction) CacheTableInfo(id int64, info *model.TableInfo) {
 }
 
+// SetAssertion implements the kv.Transaction interface.
+func (t *transaction) SetAssertion(key []byte, assertion ...kv.FlagsOp) error {
+	return nil
+}
+
 // session is a trimmed down Session type which only wraps our own trimmed-down
 // transaction type and provides the session variables to the TiDB library
 // optimized for Lightning.

ddl/db_integration_test.go

@@ -2718,7 +2718,7 @@ func (s *testIntegrationSuite3) TestAutoIncrementForce(c *C) {
 	tk.MustGetErrCode("alter table t force auto_increment = 0;", errno.ErrAutoincReadFailed)
 	tk.MustExec("alter table t force auto_increment = 1;")
 	c.Assert(getNextGlobalID(), Equals, uint64(1))
-	// inserting new rows can overwrite the existing data.
+	// Inserting new rows will cause duplicate entry error.
 	tk.MustExec("insert into t values (3);")
 	c.Assert(tk.ExecToErr("insert into t values (3);").Error(), Equals, "[kv:1062]Duplicate entry '2' for key 'PRIMARY'")
 	tk.MustQuery("select a, _tidb_rowid from t;").Check(testkit.Rows("3 1", "1 2", "2 3"))

ddl/index.go

@@ -1337,7 +1337,7 @@ func (w *addIndexWorker) BackfillDataInTxn(handleRange reorgBackfillTask) (taskC
 			}
 
 			// Create the index.
-			handle, err := w.index.Create(w.sessCtx, txn, idxRecord.vals, idxRecord.handle, idxRecord.rsData)
+			handle, err := w.index.Create(w.sessCtx, txn, idxRecord.vals, idxRecord.handle, idxRecord.rsData, table.WithIgnoreAssertion)
 			if err != nil {
 				if kv.ErrKeyExists.Equal(err) && idxRecord.handle.Equal(handle) {
 					// Index already exists, skip it.

docs/tidb_http_api.md

@@ -542,6 +542,13 @@ timezone.*
     curl -X POST -d "tidb_enable_1pc=0" http://{TiDBIP}:10080/settings
     ```
 
+1. Enable/disable the mutation checker
+
+    ```shell
+    curl -X POST -d "tidb_enable_mutation_checker=1" http://{TiDBIP}:10080/settings
+    curl -X POST -d "tidb_enable_mutation_checker=0" http://{TiDBIP}:10080/settings
+    ```
+
 1. Get/Set the size of the Ballast Object
 
     ```shell

errno/errcode.go

@@ -1007,11 +1007,15 @@ const (
 	ErrMultiStatementDisabled              = 8130
 	ErrPartitionStatsMissing               = 8131
 	ErrNotSupportedWithSem                 = 8132
-	ErrDataInConsistentExtraIndex          = 8133
-	ErrDataInConsistentMisMatchIndex       = 8134
+	ErrDataInconsistentMismatchCount       = 8133
+	ErrDataInconsistentMismatchIndex       = 8134
 	ErrAsOf                                = 8135
 	ErrVariableNoLongerSupported           = 8136
 	ErrAnalyzeMissColumn                   = 8137
+	ErrInconsistentRowValue                = 8138
+	ErrInconsistentHandle                  = 8139
+	ErrInconsistentIndexedValue            = 8140
+	ErrAssertionFailed                     = 8141
 
 	// Error codes used by TiDB ddl package
 	ErrUnsupportedDDLOperation            = 8200
@@ -1037,7 +1041,7 @@ const (
 	ErrWriteOnSnapshot                    = 8220
 	ErrInvalidKey                         = 8221
 	ErrInvalidIndexKey                    = 8222
-	ErrDataInConsistent                   = 8223
+	ErrDataInconsistent                   = 8223
 	ErrDDLJobNotFound                     = 8224
 	ErrCancelFinishedDDLJob               = 8225
 	ErrCannotCancelDDLJob                 = 8226

errno/errname.go

@@ -1000,16 +1000,20 @@ var MySQLErrName = map[uint16]*mysql.ErrMessage{
 	ErrWriteOnSnapshot:               mysql.Message("write on snapshot", nil),
 	ErrInvalidKey:                    mysql.Message("invalid key", nil),
 	ErrInvalidIndexKey:               mysql.Message("invalid index key", nil),
-	ErrDataInConsistent:              mysql.Message("index:%#v != record:%#v", []int{0, 1}),
+	ErrDataInconsistent:              mysql.Message("data inconsistency in table: %s, index: %s, handle: %s, index-values:%#v != record-values:%#v", []int{2, 3, 4}),
 	ErrDDLReorgElementNotExist:       mysql.Message("DDL reorg element does not exist", nil),
 	ErrDDLJobNotFound:                mysql.Message("DDL Job:%v not found", nil),
 	ErrCancelFinishedDDLJob:          mysql.Message("This job:%v is finished, so can't be cancelled", nil),
 	ErrCannotCancelDDLJob:            mysql.Message("This job:%v is almost finished, can't be cancelled now", nil),
 	ErrUnknownAllocatorType:          mysql.Message("Invalid allocator type", nil),
 	ErrAutoRandReadFailed:            mysql.Message("Failed to read auto-random value from storage engine", nil),
 	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),
-	ErrDataInConsistentExtraIndex:    mysql.Message("handle %#v, index:%#v != record:%#v", []int{0, 1, 2}),
-	ErrDataInConsistentMisMatchIndex: mysql.Message("col %s, handle %#v, index:%#v != record:%#v, compare err:%#v", []int{1, 2, 3, 4}),
+	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}),
+	ErrAssertionFailed:               mysql.Message("assertion failed: key: %s, assertion: %s, start_ts: %v, existing start ts: %v, existing commit ts: %v", []int{0}),
 
 	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

@@ -233,7 +233,7 @@ TiDB admin check table failed.
 
 ["admin:8223"]
 error = '''
-index:%#v != record:%#v
+data inconsistency in table: %s, index: %s, handle: %s, index-values:%#v != record-values:%#v
 '''
 
 ["admin:8224"]
@@ -918,12 +918,12 @@ Export failed: %s
 
 ["executor:8133"]
 error = '''
-handle %#v, index:%#v != record:%#v
+data inconsistency in table: %s, index: %s, index-count:%d != record-count:%d
 '''
 
 ["executor:8134"]
 error = '''
-col %s, handle %#v, index:%#v != record:%#v, compare err:%#v
+data inconsistency in table: %s, index: %s, col: %s, handle: %#v, index-values:%#v != record-values:%#v, compare err:%#v
 '''
 
 ["executor:8212"]
@@ -1811,6 +1811,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
@@ -1831,6 +1846,11 @@ error = '''
 Statement aborted because lock(s) could not be acquired immediately and NOWAIT is set.
 '''
 
+["tikv:8141"]
+error = '''
+assertion failed: key: %s, assertion: %s, start_ts: %v, existing start ts: %v, existing commit ts: %v
+'''
+
 ["tikv:8229"]
 error = '''
 TTL manager has timed out, pessimistic locks may expire, please commit or rollback this transaction

executor/adapter.go

@@ -677,7 +677,7 @@ func (a *ExecStmt) handlePessimisticDML(ctx context.Context, e Executor) error {
 		keys = filterTemporaryTableKeys(sctx.GetSessionVars(), keys)
 		seVars := sctx.GetSessionVars()
 		keys = filterLockTableKeys(seVars.StmtCtx, keys)
-		lockCtx := newLockCtx(seVars, seVars.LockWaitTimeout)
+		lockCtx := newLockCtx(seVars, seVars.LockWaitTimeout, len(keys))
 		var lockKeyStats *util.LockKeysDetails
 		ctx = context.WithValue(ctx, util.LockKeysDetailCtxKey, &lockKeyStats)
 		startLocking := time.Now()

executor/admin.go

@@ -467,7 +467,7 @@ func (e *RecoverIndexExec) backfillIndexInTxn(ctx context.Context, txn kv.Transa
 			return result, err
 		}
 
-		_, err = e.index.Create(e.ctx, txn, row.idxVals, row.handle, row.rsData)
+		_, err = e.index.Create(e.ctx, txn, row.idxVals, row.handle, row.rsData, table.WithIgnoreAssertion)
 		if err != nil {
 			return result, err
 		}