storage: fix EngineComparer ordering of bare suffixes

pkg/storage/pebble.go

@@ -93,9 +93,15 @@ func EngineKeyCompare(a, b []byte) int {
 		return bytes.Compare(a, b)
 	}
 
-	// Compute the index of the separator between the key and the version.
+	// Compute the index of the separator between the key and the version. If the
+	// separator is found to be at -1 for both keys, then we are comparing bare
+	// suffixes without a user key part. Pebble requires bare suffixes to be
+	// comparable with the same ordering as if they had a common user key.
 	aSep := aEnd - int(a[aEnd])
 	bSep := bEnd - int(b[bEnd])
+	if aSep == -1 && bSep == -1 {
+		aSep, bSep = 0, 0 // comparing bare suffixes
+	}
 	if aSep < 0 || bSep < 0 {
 		// This should never happen unless there is some sort of corruption of
 		// the keys.

pkg/storage/pebble_test.go

@@ -44,6 +44,40 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestEngineComparer(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	keyAMetadata := MVCCKey{
+		Key: []byte("a"),
+	}
+	keyA2 := MVCCKey{
+		Key:       []byte("a"),
+		Timestamp: hlc.Timestamp{WallTime: 2},
+	}
+	keyA1 := MVCCKey{
+		Key:       []byte("a"),
+		Timestamp: hlc.Timestamp{WallTime: 1},
+	}
+	keyB2 := MVCCKey{
+		Key:       []byte("b"),
+		Timestamp: hlc.Timestamp{WallTime: 2},
+	}
+
+	require.Equal(t, -1, EngineComparer.Compare(EncodeKey(keyAMetadata), EncodeKey(keyA1)),
+		"expected key metadata to sort first")
+	require.Equal(t, -1, EngineComparer.Compare(EncodeKey(keyA2), EncodeKey(keyA1)),
+		"expected higher timestamp to sort first")
+	require.Equal(t, -1, EngineComparer.Compare(EncodeKey(keyA2), EncodeKey(keyB2)),
+		"expected lower key to sort first")
+
+	suffix := func(key []byte) []byte {
+		return key[EngineComparer.Split(key):]
+	}
+	require.Equal(t, -1, EngineComparer.Compare(suffix(EncodeKey(keyA2)), suffix(EncodeKey(keyA1))),
+		"expected bare suffix with higher timestamp to sort first")
+}
+
 func TestPebbleTimeBoundPropCollector(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

pkg/storage/sst_iterator_test.go

@@ -126,34 +126,3 @@ func TestSSTIterator(t *testing.T) {
 		runTestSSTIterator(t, iter, allKVs)
 	})
 }
-
-func TestCockroachComparer(t *testing.T) {
-	defer leaktest.AfterTest(t)()
-	defer log.Scope(t).Close(t)
-
-	keyAMetadata := MVCCKey{
-		Key: []byte("a"),
-	}
-	keyA2 := MVCCKey{
-		Key:       []byte("a"),
-		Timestamp: hlc.Timestamp{WallTime: 2},
-	}
-	keyA1 := MVCCKey{
-		Key:       []byte("a"),
-		Timestamp: hlc.Timestamp{WallTime: 1},
-	}
-	keyB2 := MVCCKey{
-		Key:       []byte("b"),
-		Timestamp: hlc.Timestamp{WallTime: 2},
-	}
-
-	if x := EngineComparer.Compare(EncodeKey(keyAMetadata), EncodeKey(keyA1)); x != -1 {
-		t.Errorf("expected key metadata to sort first got: %d", x)
-	}
-	if x := EngineComparer.Compare(EncodeKey(keyA2), EncodeKey(keyA1)); x != -1 {
-		t.Errorf("expected higher timestamp to sort first got: %d", x)
-	}
-	if x := EngineComparer.Compare(EncodeKey(keyA2), EncodeKey(keyB2)); x != -1 {
-		t.Errorf("expected lower key to sort first got: %d", x)
-	}
-}