Fix a uint16 overflow.

- This fixes #308.
- Fix  uint16 overflow when resizing key slice in table/iterator.go.
- Key shouldn't exceed uint16.
- Avoid conversion to MiB when reporting key and value length errors.

db_test.go

@@ -1167,3 +1167,57 @@ func ExampleTxn_NewIterator() {
 	// Output:
 	// Counted 1000 elements
 }
+
+func randBytes(n int) []byte {
+	recv := make([]byte, n)
+	in, err := rand.Read(recv)
+	if err != nil {
+		log.Fatal(err)
+	}
+	return recv[:in]
+}
+
+var benchmarkData = []struct {
+	key, value []byte
+}{
+	{randBytes(100), nil},
+	{randBytes(1000), []byte("foo")},
+	{[]byte("foo"), randBytes(1000)},
+	{[]byte(""), randBytes(1000)},
+	{nil, randBytes(1000000)},
+	{randBytes(100000), nil},
+	{randBytes(1000000), nil},
+}
+
+func TestLargeKeys(t *testing.T) {
+	dir, err := ioutil.TempDir("", "badger")
+	require.NoError(t, err)
+	defer os.RemoveAll(dir)
+
+	opts := new(Options)
+	*opts = DefaultOptions
+	opts.ValueLogFileSize = 1024 * 1024 * 1024
+	opts.Dir = dir
+	opts.ValueDir = dir
+
+	db, err := Open(*opts)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < 1000; i++ {
+		tx := db.NewTransaction(true)
+		for _, kv := range benchmarkData {
+			k := make([]byte, len(kv.key))
+			copy(k, kv.key)
+
+			v := make([]byte, len(kv.value))
+			copy(v, kv.value)
+			if err := tx.Set(k, v); err != nil {
+				// Skip over this record.
+			}
+		}
+		if err := tx.Commit(nil); err != nil {
+			t.Fatalf("#%d: batchSet err: %v", i, err)
+		}
+	}
+}

errors.go

@@ -80,14 +80,14 @@ var (
 	ErrInvalidDump = errors.New("Data dump cannot be read")
 )
 
-const maxKeySize = 1 << 20
+const maxKeySize = 1 << 16 // Key length can't be more than uint16, as determined by table::header.
 
 func exceedsMaxKeySizeError(key []byte) error {
-	return errors.Errorf("Key with size %d exceeded %dMiB limit. Key:\n%s",
-		len(key), maxKeySize>>20, hex.Dump(key[:1<<10]))
+	return errors.Errorf("Key with size %d exceeded %d limit. Key:\n%s",
+		len(key), maxKeySize, hex.Dump(key[:1<<10]))
 }
 
 func exceedsMaxValueSizeError(value []byte, maxValueSize int64) error {
-	return errors.Errorf("Value with size %d exceeded ValueLogFileSize (%dMiB). Key:\n%s",
-		len(value), maxValueSize>>20, hex.Dump(value[:1<<10]))
+	return errors.Errorf("Value with size %d exceeded ValueLogFileSize (%d). Key:\n%s",
+		len(value), maxValueSize, hex.Dump(value[:1<<10]))
 }

table/iterator.go

@@ -110,7 +110,8 @@ func (itr *blockIterator) SeekToLast() {
 // parseKV would allocate a new byte slice for key and for value.
 func (itr *blockIterator) parseKV(h header) {
 	if cap(itr.key) < int(h.plen+h.klen) {
-		itr.key = make([]byte, 2*(h.plen+h.klen))
+		sz := int(h.plen) + int(h.klen) // Convert to int before adding to avoid uint16 overflow.
+		itr.key = make([]byte, 2*sz)
 	}
 	itr.key = itr.key[:h.plen+h.klen]
 	copy(itr.key, itr.baseKey[:h.plen])