*: 'load data' should not use latches

executor/executor_test.go

@@ -71,6 +71,7 @@ func TestT(t *testing.T) {
 
 var _ = Suite(&testSuite{})
 var _ = Suite(&testContextOptionSuite{})
+var _ = Suite(&testBypassSuite{})
 
 type testSuite struct {
 	cluster   *mocktikv.Cluster

executor/write_test.go

@@ -24,12 +24,19 @@ import (
 	"github.com/pingcap/tidb/executor"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/model"
+	"github.com/pingcap/tidb/session"
 	"github.com/pingcap/tidb/sessionctx"
+	"github.com/pingcap/tidb/store/mockstore"
 	"github.com/pingcap/tidb/table"
 	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tidb/util/testkit"
 )
 
+type testBypassSuite struct{}
+
+func (s *testBypassSuite) SetUpSuite(c *C) {
+}
+
 func (s *testSuite) TestInsert(c *C) {
 	tk := testkit.NewTestKit(c, s.store)
 	tk.MustExec("use test")
@@ -1467,6 +1474,53 @@ func assertEqualStrings(c *C, got []string, expect []string) {
 	}
 }
 
+func (s *testBypassSuite) TestBypassLatch(c *C) {
+	store, err := mockstore.NewMockTikvStore(
+		// Small latch slot size to make conflicts.
+		mockstore.WithTxnLocalLatches(64),
+	)
+	c.Assert(err, IsNil)
+	defer store.Close()
+
+	dom, err1 := session.BootstrapSession(store)
+	c.Assert(err1, IsNil)
+	defer dom.Close()
+
+	tk1 := testkit.NewTestKit(c, store)
+	tk1.MustExec("use test")
+	tk1.MustExec("drop table if exists t")
+	tk1.MustExec("create table t (id int)")
+	tk1.MustExec("set @@tidb_disable_txn_auto_retry = true")
+
+	tk2 := testkit.NewTestKit(c, store)
+	tk2.MustExec("use test")
+	tk1.MustExec("set @@tidb_disable_txn_auto_retry = true")
+
+	fn := func() {
+		tk1.MustExec("begin")
+		for i := 0; i < 100; i++ {
+			tk1.MustExec(fmt.Sprintf("insert into t values (%d)", i))
+		}
+		tk2.MustExec("begin")
+		for i := 100; i < 200; i++ {
+			tk1.MustExec(fmt.Sprintf("insert into t values (%d)", i))
+		}
+		tk2.MustExec("commit")
+	}
+
+	// txn1 and txn2 data range do not overlap, but using latches result in txn conflict.
+	fn()
+	_, err = tk1.Exec("commit")
+	c.Assert(err, NotNil)
+
+	tk1.MustExec("truncate table t")
+	fn()
+	txn := tk1.Se.Txn()
+	txn.SetOption(kv.BypassLatch, true)
+	// Bypass latch, there will be no conflicts.
+	tk1.MustExec("commit")
+}
+
 // TestIssue4067 Test issue https://github.com/pingcap/tidb/issues/4067
 func (s *testSuite) TestIssue4067(c *C) {
 	tk := testkit.NewTestKit(c, s.store)

kv/kv.go

@@ -42,8 +42,9 @@ const (
 	NotFillCache
 	// SyncLog decides whether the WAL(write-ahead log) of this request should be synchronized.
 	SyncLog
-	// ForUpdate option would be set to true in 'select for update' transaction.
-	ForUpdate
+	// BypassLatch option tells 2PC commit to bypass latches, it would be true when the
+	// transaction is not conflict-retryable, for example: 'select for update', 'load data'.
+	BypassLatch
 )
 
 // Priority value for transaction priority.

server/conn.go

@@ -750,6 +750,11 @@ func insertDataWithCommit(ctx context.Context, prevData, curData []byte, loadDat
 			break
 		}
 		loadDataInfo.Ctx.StmtCommit()
+		// Load data should not use latches, because:
+		// 1. latches may result in false positive transaction conflicts.
+		// 2. load data is not retryable when it meets conflicts.
+		// 3. load data will abort abnormally under condition 1 + 2.
+		loadDataInfo.Ctx.Txn().SetOption(kv.BypassLatch, true)
 		// Make sure that there are no retries when committing.
 		if err = loadDataInfo.Ctx.RefreshTxnCtx(ctx); err != nil {
 			return nil, errors.Trace(err)

session/session.go

@@ -341,7 +341,7 @@ func (s *session) doCommit(ctx context.Context) error {
 		relatedTableIDs: tableIDs,
 	})
 	if s.sessionVars.TxnCtx.ForUpdate {
-		s.txn.SetOption(kv.ForUpdate, true)
+		s.txn.SetOption(kv.BypassLatch, true)
 	}
 
 	if err := s.txn.Commit(sessionctx.SetCommitCtx(ctx, s)); err != nil {

store/mockstore/tikv.go

@@ -41,12 +41,12 @@ func (d MockDriver) Open(path string) (kv.Storage, error) {
 }
 
 type mockOptions struct {
-	cluster                *mocktikv.Cluster
-	mvccStore              mocktikv.MVCCStore
-	clientHijack           func(tikv.Client) tikv.Client
-	pdClientHijack         func(pd.Client) pd.Client
-	path                   string
-	disableTxnLocalLatches bool
+	cluster         *mocktikv.Cluster
+	mvccStore       mocktikv.MVCCStore
+	clientHijack    func(tikv.Client) tikv.Client
+	pdClientHijack  func(pd.Client) pd.Client
+	path            string
+	txnLocalLatches uint
 }
 
 // MockTiKVStoreOption is used to control some behavior of mock tikv.
@@ -81,10 +81,10 @@ func WithPath(path string) MockTiKVStoreOption {
 	}
 }
 
-// WithTxnLocalLatches enable or disble txnLocalLatches.
-func WithTxnLocalLatches(enable bool) MockTiKVStoreOption {
+// WithTxnLocalLatches enable txnLocalLatches, when capacity > 0.
+func WithTxnLocalLatches(capacity uint) MockTiKVStoreOption {
 	return func(c *mockOptions) {
-		c.disableTxnLocalLatches = !enable
+		c.txnLocalLatches = capacity
 	}
 }
 
@@ -101,5 +101,5 @@ func NewMockTikvStore(options ...MockTiKVStoreOption) (kv.Storage, error) {
 		return nil, errors.Trace(err)
 	}
 
-	return tikv.NewTestTiKVStore(client, pdClient, opt.clientHijack, opt.pdClientHijack, opt.disableTxnLocalLatches)
+	return tikv.NewTestTiKVStore(client, pdClient, opt.clientHijack, opt.pdClientHijack, opt.txnLocalLatches)
 }

store/tikv/delete_range_test.go

@@ -37,7 +37,7 @@ func (s *testDeleteRangeSuite) SetUpTest(c *C) {
 	client, pdClient, err := mocktikv.NewTestClient(s.cluster, nil, "")
 	c.Assert(err, IsNil)
 
-	store, err := NewTestTiKVStore(client, pdClient, nil, nil, false)
+	store, err := NewTestTiKVStore(client, pdClient, nil, nil, 0)
 	c.Check(err, IsNil)
 	s.store = store.(*tikvStore)
 }

store/tikv/split_test.go

@@ -35,7 +35,7 @@ func (s *testSplitSuite) SetUpTest(c *C) {
 	client, pdClient, err := mocktikv.NewTestClient(s.cluster, nil, "")
 	c.Assert(err, IsNil)
 
-	store, err := NewTestTiKVStore(client, pdClient, nil, nil, false)
+	store, err := NewTestTiKVStore(client, pdClient, nil, nil, 0)
 	c.Check(err, IsNil)
 	s.store = store.(*tikvStore)
 	s.bo = NewBackoffer(context.Background(), 5000)

store/tikv/test_util.go

@@ -21,7 +21,7 @@ import (
 )
 
 // NewTestTiKVStore creates a test store with Option
-func NewTestTiKVStore(client Client, pdClient pd.Client, clientHijack func(Client) Client, pdClientHijack func(pd.Client) pd.Client, disableTxnLocalLatches bool) (kv.Storage, error) {
+func NewTestTiKVStore(client Client, pdClient pd.Client, clientHijack func(Client) Client, pdClientHijack func(pd.Client) pd.Client, txnLocalLatches uint) (kv.Storage, error) {
 	if clientHijack != nil {
 		client = clientHijack(client)
 	}
@@ -36,8 +36,8 @@ func NewTestTiKVStore(client Client, pdClient pd.Client, clientHijack func(Clien
 	spkv := NewMockSafePointKV()
 	tikvStore, err := newTikvStore(uid, pdCli, spkv, client, false)
 
-	if !disableTxnLocalLatches {
-		tikvStore.EnableTxnLocalLatches(1024000)
+	if txnLocalLatches > 0 {
+		tikvStore.EnableTxnLocalLatches(txnLocalLatches)
 	}
 
 	tikvStore.mock = true

store/tikv/ticlient_test.go

@@ -51,7 +51,7 @@ func NewTestStore(c *C) kv.Storage {
 	client, pdClient, err := mocktikv.NewTestClient(nil, nil, "")
 	c.Assert(err, IsNil)
 
-	store, err := NewTestTiKVStore(client, pdClient, nil, nil, false)
+	store, err := NewTestTiKVStore(client, pdClient, nil, nil, 0)
 	c.Assert(err, IsNil)
 	return store
 }

store/tikv/txn.go

@@ -195,12 +195,12 @@ func (txn *tikvTxn) Commit(ctx context.Context) error {
 	}
 
 	// latches enabled
-	var forUpdate bool
-	if option := txn.us.GetOption(kv.ForUpdate); option != nil {
-		forUpdate = option.(bool)
+	var bypassLatch bool
+	if option := txn.us.GetOption(kv.BypassLatch); option != nil {
+		bypassLatch = option.(bool)
 	}
-	// For update transaction is not retryable, commit directly.
-	if forUpdate {
+	// When bypassLatch flag is true, commit directly.
+	if bypassLatch {
 		err = committer.executeAndWriteFinishBinlog(ctx)
 		if err == nil {
 			txn.store.txnLatches.RefreshCommitTS(committer.keys, committer.commitTS)

util/kvencoder/kv_encoder.go

@@ -192,7 +192,7 @@ func (e *kvEncoder) GetSystemVariable(name string) (string, bool) {
 }
 
 func newMockTikvWithBootstrap() (kv.Storage, *domain.Domain, error) {
-	store, err := mockstore.NewMockTikvStore(mockstore.WithTxnLocalLatches(false))
+	store, err := mockstore.NewMockTikvStore(mockstore.WithTxnLocalLatches(0))
 	if err != nil {
 		return nil, nil, errors.Trace(err)
 	}