cockroachkvs: always use non-empty index separators

Adapt the cockroachkv Successor implementation to always return non-empty keys
with a version length byte. Previously it was possible for Successor to return
an empty slice if passed an empty slice, resulting in an empty string index
separator. When combined with a synthetic prefix, this produced an invalid key:
a non-empty string without a version length byte. Additionally, document that
Successor may receive an empty key as an operand. Document and enforce under
invariants that Successor must never receive empty keys as operands.

Author: jbowens

cockroachkvs/cockroachkvs.go

@@ -70,6 +70,9 @@ var Comparer = base.Comparer{
 	FormatKey: FormatKey,
 
 	Separator: func(dst, a, b []byte) []byte {
+		if len(a) == 0 || len(b) == 0 {
+			panic(errors.AssertionFailedf("cannot separate empty keys"))
+		}
 		aKey, ok := getKeyPartFromEngineKey(a)
 		if !ok {
 			return append(dst, a...)
@@ -94,6 +97,13 @@ var Comparer = base.Comparer{
 	},
 
 	Successor: func(dst, a []byte) []byte {
+		// If a is empty, the key consisting of just a single 0x00 byte is the
+		// smallest non-empty key ≥ a. We don't return an empty key because
+		// empty keys are problematic when combined with a synthetic prefix
+		// (they form a key that has no trailing sentinel byte).
+		if len(a) == 0 {
+			return append(dst, 0x00)
+		}
 		aKey, ok := getKeyPartFromEngineKey(a)
 		if !ok {
 			return append(dst, a...)
@@ -1035,6 +1045,10 @@ func FormatKey(key []byte) fmt.Formatter {
 type formatKey []byte
 
 func (fk formatKey) Format(f fmt.State, c rune) {
+	if len(fk) == 0 {
+		fmt.Fprint(f, "<empty>")
+		return
+	}
 	i := Split(fk)
 	fmt.Fprintf(f, "%s", []byte(fk)[:i-1])
 	if i == len(fk) {
@@ -1083,7 +1097,15 @@ func (fk formatKeySuffix) Format(f fmt.State, c rune) {
 //     c. <WALLTIME-SECONDS>,<LOGICAL>
 //     d. <WALLTIME-SECONDS>.<WALLTIME-NANOS>,<LOGICAL>
 //  2. A lock table key in the format: <STRENGTH-BYTE>,<TXNUUID>
+//  3. A raw hex-encoded string prefixed with "hex:".
 func ParseFormattedKey(formattedKey string) []byte {
+	if strings.HasPrefix(formattedKey, "hex:") {
+		decoded, err := hex.DecodeString(formattedKey[4:])
+		if err != nil {
+			panic(errors.Newf("invalid hex-encoded key: %w", err))
+		}
+		return decoded
+	}
 	i := strings.IndexByte(formattedKey, '@')
 	if i == -1 {
 		// Suffix-less.

cockroachkvs/cockroachkvs_test.go

@@ -79,6 +79,46 @@ func TestComparer(t *testing.T) {
 	}
 }
 
+func TestComparerFuncs(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	var buf bytes.Buffer
+	datadriven.RunTest(t, "testdata/comparer_funcs",
+		func(t *testing.T, td *datadriven.TestData) string {
+			buf.Reset()
+			switch td.Cmd {
+			case "separator":
+				var keys [][]byte
+				var dst []byte
+				for _, line := range crstrings.Lines(td.Input) {
+					keys = keys[:0]
+					for _, formattedKey := range strings.Fields(line) {
+						k := ParseFormattedKey(formattedKey)
+						keys = append(keys, k)
+					}
+					if len(keys) != 2 {
+						t.Fatalf("expected two keys, but found %d", len(keys))
+					}
+					dst = Comparer.Separator(dst[:0], keys[0], keys[1])
+					fmt.Fprintf(&buf, "Separator(%q [%x], %q [%x]) = %q [%x]\n",
+						FormatKey(keys[0]), keys[0], FormatKey(keys[1]), keys[1], FormatKey(dst), dst)
+				}
+				return buf.String()
+			case "successor":
+				var dst []byte
+				for _, line := range crstrings.Lines(td.Input) {
+					k := ParseFormattedKey(line)
+					dst = Comparer.Successor(dst[:0], k)
+					fmt.Fprintf(&buf, "Successor(%q [%x]) = %q [%x]\n",
+						FormatKey(k), k, FormatKey(dst), dst)
+				}
+				return buf.String()
+			default:
+				panic(fmt.Sprintf("unknown command: %s", td.Cmd))
+			}
+		})
+}
+
 func TestKeySchema_KeyWriter(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 

cockroachkvs/testdata/comparer_funcs

@@ -0,0 +1,36 @@
+successor
+foo
+bar
+foo@1000000,0
+[email protected],0
+foo@1000000,8
+[email protected],7
+foo@01,cf379e85-6371-42e6-acc8-ed5259fa177a
+----
+Successor(foo [666f6f00]) = g [6700]
+Successor(bar [62617200]) = c [6300]
+Successor(foo@1000000,0 [666f6f0000038d7ea4c6800009]) = g [6700]
+Successor([email protected],0 [666f6f0000038d7ea4c7480c09]) = g [6700]
+Successor(foo@1000000,8 [666f6f0000038d7ea4c68000000000080d]) = g [6700]
+Successor([email protected],7 [666f6f0000038d7ea4c7480c000000070d]) = g [6700]
+Successor(foo@01,cf379e85-6371-42e6-acc8-ed5259fa177a [666f6f0001cf379e85637142e6acc8ed5259fa177a12]) = g [6700]
+
+# The successor for an empty string should be a key consisting only of the 0x00
+# byte.
+successor
+hex:
+----
+Successor(<empty> []) =  [00]
+
+separator
+bar foo
+foo@1000000,0 foo@1000000,0
+foo@1000000,1 foo@1000000,0
+foo@01,a65a118c-5dcb-452b-bd82-7dd17e121138 foo@01,cf379e85-6371-42e6-acc8-ed5259fa177a
+fax@01,a65a118c-5dcb-452b-bd82-7dd17e121138 foo@01,cf379e85-6371-42e6-acc8-ed5259fa177a
+----
+Separator(bar [62617200], foo [666f6f00]) = c [6300]
+Separator(foo@1000000,0 [666f6f0000038d7ea4c6800009], foo@1000000,0 [666f6f0000038d7ea4c6800009]) = foo@1000000,0 [666f6f0000038d7ea4c6800009]
+Separator(foo@1000000,1 [666f6f0000038d7ea4c68000000000010d], foo@1000000,0 [666f6f0000038d7ea4c6800009]) = foo@1000000,1 [666f6f0000038d7ea4c68000000000010d]
+Separator(foo@01,a65a118c-5dcb-452b-bd82-7dd17e121138 [666f6f0001a65a118c5dcb452bbd827dd17e12113812], foo@01,cf379e85-6371-42e6-acc8-ed5259fa177a [666f6f0001cf379e85637142e6acc8ed5259fa177a12]) = foo@01,a65a118c-5dcb-452b-bd82-7dd17e121138 [666f6f0001a65a118c5dcb452bbd827dd17e12113812]
+Separator(fax@01,a65a118c-5dcb-452b-bd82-7dd17e121138 [6661780001a65a118c5dcb452bbd827dd17e12113812], foo@01,cf379e85-6371-42e6-acc8-ed5259fa177a [666f6f0001cf379e85637142e6acc8ed5259fa177a12]) = fb [666200]

internal/base/comparer.go

@@ -12,6 +12,7 @@ import (
 	"strconv"
 	"unicode/utf8"
 
+	"github.com/cockroachdb/crlib/crbytes"
 	"github.com/cockroachdb/errors"
 )
 
@@ -124,11 +125,18 @@ type FormatValue func(key, value []byte) fmt.Formatter
 //
 // For example, if a and b are the []byte equivalents of the strings "black" and
 // "blue", then the function may append "blb" to dst.
+//
+// Callers must guarantee that len(a) > 0 and len(b) > 0.
 type Separator func(dst, a, b []byte) []byte
 
 // Successor appends to dst a shortened key k given a key a such that
 // Compare(a, k) <= 0. A simple implementation may return a unchanged.
 // The appended key k must be valid to pass to Compare.
+//
+// The parameter a may be an empty slice even if an empty slice was never
+// committed to Pebble and is not a valid key representation. If a is the empty
+// slice, Successor must return a valid key but otherwise has no other
+// constraints.
 type Successor func(dst, a []byte) []byte
 
 // ImmediateSuccessor is invoked with a prefix key ([Split(a) == len(a)]) and
@@ -312,14 +320,15 @@ var DefaultComparer = &Comparer{
 	FormatKey: DefaultFormatter,
 
 	Separator: func(dst, a, b []byte) []byte {
-		i, n := SharedPrefixLen(a, b), len(dst)
+		if len(a) == 0 || len(b) == 0 {
+			panic(errors.AssertionFailedf("empty keys"))
+		}
+
+		i := crbytes.CommonPrefix(a, b)
+		n := len(dst)
 		dst = append(dst, a...)
 
-		min := len(a)
-		if min > len(b) {
-			min = len(b)
-		}
-		if i >= min {
+		if i == len(a) || i == len(b) {
 			// Do not shorten if one string is a prefix of the other.
 			return dst
 		}
@@ -366,25 +375,6 @@ var DefaultComparer = &Comparer{
 	Name: "leveldb.BytewiseComparator",
 }
 
-// SharedPrefixLen returns the largest i such that a[:i] equals b[:i].
-// This function can be useful in implementing the Comparer interface.
-func SharedPrefixLen(a, b []byte) int {
-	i, n := 0, len(a)
-	if n > len(b) {
-		n = len(b)
-	}
-	asUint64 := func(c []byte, i int) uint64 {
-		return binary.LittleEndian.Uint64(c[i:])
-	}
-	for i < n-7 && asUint64(a, i) == asUint64(b, i) {
-		i += 8
-	}
-	for i < n && a[i] == b[i] {
-		i++
-	}
-	return i
-}
-
 // MinUserKey returns the smaller of two user keys. If one of the keys is nil,
 // the other one is returned.
 func MinUserKey(cmp Compare, a, b []byte) []byte {
@@ -444,6 +434,9 @@ func MakeAssertComparer(c Comparer) Comparer {
 		CompareRangeSuffixes: c.CompareRangeSuffixes,
 		AbbreviatedKey:       c.AbbreviatedKey,
 		Separator: func(dst, a, b []byte) []byte {
+			if len(a) == 0 || len(b) == 0 {
+				panic(errors.AssertionFailedf("empty keys"))
+			}
 			ret := c.Separator(dst, a, b)
 			// The Separator func must return a valid key.
 			c.ValidateKey.MustValidate(ret)

internal/base/comparer_test.go

@@ -12,15 +12,11 @@ import (
 	"time"
 )
 
-func TestDefAppendSeparator(t *testing.T) {
+func TestDefaultComparer_Separator(t *testing.T) {
 	testCases := []struct {
 		a, b, want string
 	}{
-		// Examples from the doc comments.
 		{"black", "blue", "blb"},
-		{"green", "", "green"},
-		// Non-empty b values. The C++ Level-DB code calls these separators.
-		{"", "2", ""},
 		{"1", "2", "1"},
 		{"1", "29", "2"},
 		{"13", "19", "14"},
@@ -33,22 +29,37 @@ func TestDefAppendSeparator(t *testing.T) {
 		{"1\xff\xff", "19", "1\xff\xff"},
 		{"1\xff\xff", "2", "1\xff\xff"},
 		{"1\xff\xff", "9", "2"},
-		// Empty b values. The C++ Level-DB code calls these successors.
-		{"", "", ""},
-		{"1", "", "1"},
-		{"11", "", "11"},
-		{"11\xff", "", "11\xff"},
-		{"1\xff", "", "1\xff"},
-		{"1\xff\xff", "", "1\xff\xff"},
-		{"\xff", "", "\xff"},
-		{"\xff\xff", "", "\xff\xff"},
-		{"\xff\xff\xff", "", "\xff\xff\xff"},
 	}
 	for _, tc := range testCases {
 		t.Run("", func(t *testing.T) {
 			got := string(DefaultComparer.Separator(nil, []byte(tc.a), []byte(tc.b)))
 			if got != tc.want {
-				t.Errorf("a, b = %q, %q: got %q, want %q", tc.a, tc.b, got, tc.want)
+				t.Errorf("DefaultComparer.Separator(nil, %q, %q) = %q, want %q", tc.a, tc.b, got, tc.want)
+			}
+		})
+	}
+}
+
+func TestDefaultComparer_Successor(t *testing.T) {
+	testCases := []struct {
+		a, want string
+	}{
+		{"green", "h"},
+		{"", ""},
+		{"1", "2"},
+		{"11", "2"},
+		{"11\xff", "2"},
+		{"1\xff", "2"},
+		{"1\xff\xff", "2"},
+		{"\xff", "\xff"},
+		{"\xff\xff", "\xff\xff"},
+		{"\xff\xff\xff", "\xff\xff\xff"},
+	}
+	for _, tc := range testCases {
+		t.Run("", func(t *testing.T) {
+			got := string(DefaultComparer.Successor(nil, []byte(tc.a)))
+			if got != tc.want {
+				t.Errorf("DefaultComparer.Successor(nil, %q) = %q, want %q", tc.a, got, tc.want)
 			}
 		})
 	}

internal/base/internal.go

@@ -12,6 +12,8 @@ import (
 	"strings"
 	"sync/atomic"
 
+	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/redact"
 )
 
@@ -398,6 +400,9 @@ func (k InternalKey) EncodeTrailer() [8]byte {
 func (k InternalKey) Separator(
 	cmp Compare, sep Separator, buf []byte, other InternalKey,
 ) InternalKey {
+	if invariants.Enabled && (len(k.UserKey) == 0 || len(other.UserKey) == 0) {
+		panic(errors.AssertionFailedf("empty keys passed to Separator: %s, %s", k, other))
+	}
 	buf = sep(buf, k.UserKey, other.UserKey)
 	if len(buf) <= len(k.UserKey) && cmp(k.UserKey, buf) < 0 {
 		// The separator user key is physically shorter than k.UserKey (if it is
@@ -416,7 +421,7 @@ func (k InternalKey) Separator(
 // InternalKey.UserKey, though it is valid to pass a nil.
 func (k InternalKey) Successor(cmp Compare, succ Successor, buf []byte) InternalKey {
 	buf = succ(buf, k.UserKey)
-	if len(buf) <= len(k.UserKey) && cmp(k.UserKey, buf) < 0 {
+	if (len(k.UserKey) == 0 || len(buf) <= len(k.UserKey)) && cmp(k.UserKey, buf) < 0 {
 		// The successor user key is physically shorter that k.UserKey (if it is
 		// longer, we'll continue to use "k"), but logically after. Tack on the max
 		// sequence number to the shortened user key. Note that we could tack on