Enforce ValueThreshold max to uint16.

Keep ValueThreshold to below uint16, because that value size limit is
baked into the header of the key-value pairs in SSTable. Keeps most of
the simplifications of the previous change by Andy Kimball, while
bringing the uint16 limits back. In fact, this change enforces that
limit by returning an explicit error for invalid ValueThreshold.

db.go

@@ -167,6 +167,10 @@ func Open(opt Options) (db *DB, err error) {
 	opt.maxBatchSize = (15 * opt.MaxTableSize) / 100
 	opt.maxBatchCount = opt.maxBatchSize / int64(skl.MaxNodeSize)
 
+	if opt.ValueThreshold > math.MaxUint16-16 {
+		return nil, ErrValueThreshold
+	}
+
 	if opt.ReadOnly {
 		// Can't truncate if the DB is read only.
 		opt.Truncate = false

db_test.go

@@ -1621,7 +1621,12 @@ func TestLSMOnly(t *testing.T) {
 	require.NotEqual(t, dopts.ValueLogLoadingMode, opts.ValueLogLoadingMode)
 	require.NotEqual(t, dopts.ValueLogFileSize, opts.ValueLogFileSize)
 
+	dopts.ValueThreshold = 1 << 16
+	_, err = Open(dopts)
+	require.Equal(t, ErrValueThreshold, err)
+
 	db, err := Open(opts)
+	require.NoError(t, err)
 	if err != nil {
 		t.Fatal(err)
 	}

errors.go

@@ -27,6 +27,10 @@ var (
 	// range.
 	ErrValueLogSize = errors.New("Invalid ValueLogFileSize, must be between 1MB and 2GB")
 
+	// ErrValueThreshold is returned when ValueThreshold is set to a value close to or greater than
+	// uint16.
+	ErrValueThreshold = errors.New("Invalid ValueThreshold, must be lower than uint16.")
+
 	// ErrKeyNotFound is returned when key isn't found on a txn.Get.
 	ErrKeyNotFound = errors.New("Key not found")
 

options.go

@@ -129,9 +129,7 @@ var LSMOnlyOptions = Options{}
 func init() {
 	LSMOnlyOptions = DefaultOptions
 
-	// TODO: Once we switch Skiplist code to store uint32 value sizes, we can increase this.
-	// But for now, ensure that the value length fits in uint16.
-	LSMOnlyOptions.ValueThreshold = 65001
+	LSMOnlyOptions.ValueThreshold = 65500      // Max value length which fits in uint16.
 	LSMOnlyOptions.ValueLogFileSize = 64 << 20 // Allow easy space reclamation.
 	LSMOnlyOptions.ValueLogLoadingMode = options.FileIO
 }

skl/arena.go

@@ -30,7 +30,7 @@ const (
 	// so that the node.value field is 64-bit aligned. This is necessary because
 	// node.getValueOffset uses atomic.LoadUint64, which expects its input
 	// pointer to be 64-bit aligned.
-	nodeAlign  = int(unsafe.Sizeof(uint64(0))) - 1
+	nodeAlign = int(unsafe.Sizeof(uint64(0))) - 1
 )
 
 // Arena should be lock-free.
@@ -120,8 +120,8 @@ func (s *Arena) getKey(offset uint32, size uint16) []byte {
 
 // getVal returns byte slice at offset. The given size should be just the value
 // size and should NOT include the meta bytes.
-func (s *Arena) getVal(offset uint32, size uint32) (ret y.ValueStruct) {
-	ret.Decode(s.buf[offset : offset+size])
+func (s *Arena) getVal(offset uint32, size uint16) (ret y.ValueStruct) {
+	ret.Decode(s.buf[offset : offset+uint32(size)])
 	return
 }
 

skl/skl.go

@@ -53,7 +53,7 @@ type node struct {
 	// Multiple parts of the value are encoded as a single uint64 so that it
 	// can be atomically loaded and stored:
 	//   value offset: uint32 (bits 0-31)
-	//   value size  : uint32 (bits 32-63)
+	//   value size  : uint16 (bits 32-47)
 	value uint64
 
 	// A byte slice is 24 bytes. We are trying to save space here.
@@ -112,13 +112,13 @@ func newNode(arena *Arena, key []byte, v y.ValueStruct, height int) *node {
 	return node
 }
 
-func encodeValue(valOffset uint32, valSize uint32) uint64 {
+func encodeValue(valOffset uint32, valSize uint16) uint64 {
 	return uint64(valSize)<<32 | uint64(valOffset)
 }
 
-func decodeValue(value uint64) (valOffset uint32, valSize uint32) {
+func decodeValue(value uint64) (valOffset uint32, valSize uint16) {
 	valOffset = uint32(value)
-	valSize = uint32(value >> 32)
+	valSize = uint16(value >> 32)
 	return
 }
 
@@ -134,7 +134,7 @@ func NewSkiplist(arenaSize int64) *Skiplist {
 	}
 }
 
-func (s *node) getValueOffset() (uint32, uint32) {
+func (s *node) getValueOffset() (uint32, uint16) {
 	value := atomic.LoadUint64(&s.value)
 	return decodeValue(value)
 }

y/iterator.go

@@ -47,14 +47,14 @@ func sizeVarint(x uint64) (n int) {
 }
 
 // EncodedSize is the size of the ValueStruct when encoded
-func (v *ValueStruct) EncodedSize() uint32 {
+func (v *ValueStruct) EncodedSize() uint16 {
 	sz := len(v.Value) + 2 // meta, usermeta.
 	if v.ExpiresAt == 0 {
-		return uint32(sz + 1)
+		return uint16(sz + 1)
 	}
 
 	enc := sizeVarint(v.ExpiresAt)
-	return uint32(sz + enc)
+	return uint16(sz + enc)
 }
 
 // Decode uses the length of the slice to infer the length of the Value field.