unistore: Added error if row is too long to be locked

pkg/executor/insert_test.go

@@ -17,6 +17,7 @@ package executor_test
 import (
 	"fmt"
 	"strconv"
+	"strings"
 	"testing"
 	"time"
 
@@ -703,3 +704,18 @@ func TestInsertNullInNonStrictMode(t *testing.T) {
 	tk.MustExec("insert ignore t1 VALUES (4, 4) ON DUPLICATE KEY UPDATE col1 = null")
 	tk.MustQuery("select * from t1").Check(testkit.RowsWithSep("|", "1|", "2|", "3|", "4|", "5|"))
 }
+
+func TestInsertLargeRow(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	ver := tk.MustQuery("select tidb_version()").Rows()[0][0].(string)
+	if !strings.Contains(ver, "Store: unistore") {
+		t.Skipf("Only support 'Store: unistore'\n%s", ver)
+	}
+	tk.MustExec("use test")
+	tk.MustExec("create table t (id int primary key, b longtext)")
+	tk.MustExec("set tidb_txn_entry_size_limit = 1<<23")
+	// the unistore arena blocksize is 8MB (8388608 bytes), so Unistore cannot handle larger rows than that!
+	// since a row cannot span multiple arena blocks.
+	tk.MustContainErrMsg("insert into t values (1, REPEAT('t',8388493))", "unistore lock entry too big")
+}

pkg/store/mockstore/unistore/lockstore/lockstore.go

@@ -353,6 +353,12 @@ func (ls *MemStore) replace(key, v []byte, hint *Hint, old *node) {
 	ls.getArena().free(old.addr)
 }
 
+// MaxEntrySize will return the maximum entry size for the MemStore
+// Any entry larger than this will likely result in failure.
+func (ls *MemStore) MaxEntrySize() int {
+	return ls.getArena().blockSize - nodeHeaderSize - ls.getHeight()*8
+}
+
 func (ls *MemStore) newNode(arena *arena, key []byte, v []byte, height int) *node {
 	// The base level is already allocated in the node struct.
 	nodeSize := nodeHeaderSize + height*8 + len(key) + len(v)

pkg/store/mockstore/unistore/tikv/write.go

@@ -16,6 +16,7 @@ package tikv
 
 import (
 	"bytes"
+	"fmt"
 	"math"
 	"sync"
 	"sync/atomic"
@@ -166,6 +167,11 @@ func (w writeLockWorker) run() {
 					// Ignore if the key doesn't exist
 					ls.DeleteWithHint(entry.Key.UserKey, hint)
 				default:
+					// Make sure it fits into an arena block.
+					if len(entry.Key.UserKey)+len(entry.Value) > ls.MaxEntrySize() {
+						batch.err = fmt.Errorf("unistore lock entry too big %d > %d", len(entry.Key.UserKey)+len(entry.Value), ls.MaxEntrySize())
+						break
+					}
 					insertCnt++
 					ls.PutWithHint(entry.Key.UserKey, entry.Value, hint)
 				}