executor: fix the incorrect untouch used in optimistic transactions (#30447)

executor/write_concurrent_test.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/testkit"
+	"github.com/stretchr/testify/require"
 )
 
 func TestBatchInsertWithOnDuplicate(t *testing.T) {
@@ -73,3 +74,29 @@ func permInt(n int) []interface{} {
 	}
 	return v
 }
+
+func TestConstraintCheckForOptimisticUntouched(t *testing.T) {
+	t.Parallel()
+	store, clean := testkit.CreateMockStore(t)
+	defer clean()
+
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists test_optimistic_untouched_flag;")
+	tk.MustExec(`create table test_optimistic_untouched_flag(c0 int, c1 varchar(20), c2 varchar(20), unique key uk(c0));`)
+	tk.MustExec(`insert into test_optimistic_untouched_flag(c0, c1, c2) values (1, null, 'green');`)
+
+	// Insert a row with duplicated entry on the unique key, the commit should fail.
+	tk.MustExec("begin optimistic;")
+	tk.MustExec(`insert into test_optimistic_untouched_flag(c0, c1, c2) values (1, 'red', 'white');`)
+	tk.MustExec(`delete from test_optimistic_untouched_flag where c1 is null;`)
+	tk.MustExec("update test_optimistic_untouched_flag set c2 = 'green' where c2 between 'purple' and 'white';")
+	_, err := tk.Exec("commit")
+	require.Error(t, err)
+
+	tk.MustExec("begin optimistic;")
+	tk.MustExec(`insert into test_optimistic_untouched_flag(c0, c1, c2) values (1, 'red', 'white');`)
+	tk.MustExec("update test_optimistic_untouched_flag set c2 = 'green' where c2 between 'purple' and 'white';")
+	_, err = tk.Exec("commit")
+	require.Error(t, err)
+}

go.mod

@@ -66,7 +66,7 @@ require (
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.7.0
 	github.com/tiancaiamao/appdash v0.0.0-20181126055449-889f96f722a2
-	github.com/tikv/client-go/v2 v2.0.0-alpha.0.20210831090540-391fcd842dc8
+	github.com/tikv/client-go/v2 v2.0.0-alpha.0.20211221092530-2f48d74d6949
 	github.com/tikv/pd v1.1.0-beta.0.20210818112400-0c5667766690
 	github.com/twmb/murmur3 v1.1.3
 	github.com/uber-go/atomic v1.4.0

go.sum

@@ -694,8 +694,8 @@ github.com/tiancaiamao/appdash v0.0.0-20181126055449-889f96f722a2/go.mod h1:2PfK
 github.com/tidwall/gjson v1.3.5/go.mod h1:P256ACg0Mn+j1RXIDXoss50DeIABTYK1PULOJHhxOls=
 github.com/tidwall/match v1.0.1/go.mod h1:LujAq0jyVjBy028G1WhWfIzbpQfMO8bBZ6Tyb0+pL9E=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
-github.com/tikv/client-go/v2 v2.0.0-alpha.0.20210831090540-391fcd842dc8 h1:8nv1BaOlTfYtK3Pyb0fHQ0zm9mRMPlXFbAWUhTEvRdo=
-github.com/tikv/client-go/v2 v2.0.0-alpha.0.20210831090540-391fcd842dc8/go.mod h1:KwtZXt0JD+bP9bWW2ka0ir3Wp3oTEfZUTh22bs2sI4o=
+github.com/tikv/client-go/v2 v2.0.0-alpha.0.20211221092530-2f48d74d6949 h1:XY9qiqy9cHgPZ0E2zI8WfDK2wf8Gh+Yamuh48z55Rro=
+github.com/tikv/client-go/v2 v2.0.0-alpha.0.20211221092530-2f48d74d6949/go.mod h1:KwtZXt0JD+bP9bWW2ka0ir3Wp3oTEfZUTh22bs2sI4o=
 github.com/tikv/pd v1.1.0-beta.0.20210323121136-78679e5e209d/go.mod h1:Jw9KG11C/23Rr7DW4XWQ7H5xOgGZo6DFL1OKAF4+Igw=
 github.com/tikv/pd v1.1.0-beta.0.20210818112400-0c5667766690 h1:qGn7fDqj7IZ5dozy7QVkoj+0bama92ruVGHqoCBg7W4=
 github.com/tikv/pd v1.1.0-beta.0.20210818112400-0c5667766690/go.mod h1:rammPjeZgpvfrQRPkijcx8tlxF1XM5+m6kRXrkDzCAA=

session/pessimistic_test.go

@@ -69,13 +69,11 @@ func (s *testPessimisticSuite) SetUpSuite(c *C) {
 	// Set it to 300ms for testing lock resolve.
 	atomic.StoreUint64(&transaction.ManagedLockTTL, 300)
 	transaction.PrewriteMaxBackoff = 500
-	transaction.VeryLongMaxBackoff = 500
 }
 
 func (s *testPessimisticSuite) TearDownSuite(c *C) {
 	s.testSessionSuiteBase.TearDownSuite(c)
 	transaction.PrewriteMaxBackoff = 20000
-	transaction.VeryLongMaxBackoff = 600000
 }
 
 func (s *testPessimisticSuite) TestPessimisticTxn(c *C) {

session/session.go

@@ -44,6 +44,7 @@ import (
 	"github.com/pingcap/parser/model"
 	"github.com/pingcap/parser/mysql"
 	"github.com/pingcap/parser/terror"
+	"github.com/pingcap/tidb/store/driver/txn"
 	"github.com/pingcap/tidb/util/topsql"
 	"github.com/pingcap/tipb/go-binlog"
 	"go.uber.org/zap"
@@ -661,14 +662,15 @@ func (s *session) commitTxnWithTemporaryData(ctx context.Context, txn kv.Transac
 
 type temporaryTableKVFilter map[int64]tableutil.TempTable
 
-func (m temporaryTableKVFilter) IsUnnecessaryKeyValue(key, value []byte, flags tikvstore.KeyFlags) bool {
+func (m temporaryTableKVFilter) IsUnnecessaryKeyValue(key, value []byte, flags tikvstore.KeyFlags) (bool, error) {
 	tid := tablecodec.DecodeTableID(key)
 	if _, ok := m[tid]; ok {
-		return true
+		return true, nil
 	}
 
 	// This is the default filter for all tables.
-	return tablecodec.IsUntouchedIndexKValue(key, value)
+	defaultFilter := txn.TiDBKVFilter{}
+	return defaultFilter.IsUnnecessaryKeyValue(key, value, flags)
 }
 
 // errIsNoisy is used to filter DUPLCATE KEY errors.

store/driver/txn/txn_driver.go

@@ -27,10 +27,12 @@ import (
 	derr "github.com/pingcap/tidb/store/driver/error"
 	"github.com/pingcap/tidb/store/driver/options"
 	"github.com/pingcap/tidb/tablecodec"
+	"github.com/pingcap/tidb/util/logutil"
 	tikverr "github.com/tikv/client-go/v2/error"
 	tikvstore "github.com/tikv/client-go/v2/kv"
 	"github.com/tikv/client-go/v2/tikv"
 	"github.com/tikv/client-go/v2/txnkv/txnsnapshot"
+	"go.uber.org/zap"
 )
 
 type tikvTxn struct {
@@ -240,6 +242,15 @@ func (txn *tikvTxn) extractKeyExistsErr(key kv.Key) error {
 type TiDBKVFilter struct{}
 
 // IsUnnecessaryKeyValue defines which kinds of KV pairs from TiDB needn't be committed.
-func (f TiDBKVFilter) IsUnnecessaryKeyValue(key, value []byte, flags tikvstore.KeyFlags) bool {
-	return tablecodec.IsUntouchedIndexKValue(key, value)
+func (f TiDBKVFilter) IsUnnecessaryKeyValue(key, value []byte, flags tikvstore.KeyFlags) (bool, error) {
+	isUntouchedValue := tablecodec.IsUntouchedIndexKValue(key, value)
+	if isUntouchedValue && flags.HasPresumeKeyNotExists() {
+		logutil.BgLogger().Error("unexpected path the untouched key value with PresumeKeyNotExists flag",
+			zap.Stringer("key", kv.Key(key)), zap.Stringer("value", kv.Key(value)),
+			zap.Uint16("flags", uint16(flags)), zap.Stack("stack"))
+		return false, errors.Errorf(
+			"unexpected path the untouched key=%s value=%s contains PresumeKeyNotExists flag keyFlags=%v",
+			kv.Key(key).String(), kv.Key(value).String(), flags)
+	}
+	return isUntouchedValue, nil
 }

table/tables/index.go

@@ -168,8 +168,22 @@ func (c *index) Create(sctx sessionctx.Context, txn kv.Transaction, indexedValue
 		// should not overwrite the key with un-commit flag.
 		// So if the key exists, just do nothing and return.
 		v, err := txn.GetMemBuffer().Get(ctx, key)
-		if err == nil && len(v) != 0 {
-			return nil, nil
+		if err == nil {
+			if len(v) != 0 {
+				return nil, nil
+			}
+			// The key is marked as deleted in the memory buffer, as the existence check is done lazily
+			// for optimistic transactions by default. The "untouched" key could still exist in the store,
+			// it's needed to commit this key to do the existence check so unset the untouched flag.
+			if !txn.IsPessimistic() {
+				keyFlags, err := txn.GetMemBuffer().GetFlags(key)
+				if err != nil {
+					return nil, err
+				}
+				if keyFlags.HasPresumeKeyNotExists() {
+					opt.Untouched = false
+				}
+			}
 		}
 	}