From 01dcf5752d56dfb7ded1255b3639cba6026d0fad Mon Sep 17 00:00:00 2001
From: Radu Berinde <radu@cockroachlabs.com>
Date: Tue, 24 Sep 2024 16:07:44 -0700
Subject: [PATCH] base: make comparer tolerate empty keys

When synthetic prefix is used, we can trim the synthetic prefix from a
given seek key and compare it with the "bare" key in a table. After
trimming the key prefix can become "empty" (with just the terminator
left). This change extends the comparer test suite to check that we
can remove leading bytes from prefixes and adjusts the testkeys and
crdb comparers.

Fixes #3906
---
 internal/base/comparer.go     | 14 +++++++++++++-
 internal/crdbtest/crdbtest.go | 13 ++-----------
 internal/testkeys/testkeys.go | 11 +----------
 3 files changed, 16 insertions(+), 22 deletions(-)

diff --git a/internal/base/comparer.go b/internal/base/comparer.go
index cad7347cf2..d2e76a8fe1 100644
--- a/internal/base/comparer.go
+++ b/internal/base/comparer.go
@@ -18,7 +18,11 @@ import (
 // Compare returns -1, 0, or +1 depending on whether a is 'less than', 'equal
 // to' or 'greater than' b.
 //
-// Both a and b must be valid keys.
+// Both a and b must be valid keys. Note that because of synthetic prefix
+// functionality, the Compare function can be called on a key (either from the
+// database or passed as an argument for an iterator operation) after the
+// synthetic prefix has been removed. In general, this implies that removing any
+// leading bytes from a prefix must yield another valid prefix.
 //
 // A key a is less than b if a's prefix is byte-wise less than b's prefix, or if
 // the prefixes are equal and a's suffix is less than b's suffix (according to
@@ -433,6 +437,14 @@ func CheckComparer(c *Comparer, prefixes [][]byte, suffixes [][]byte) error {
 		return errors.Errorf("CompareSuffixes is inconsistent")
 	}
 
+	n := len(prefixes)
+	// Removing leading bytes from prefixes must yield valid prefixes.
+	for i := 0; i < n; i++ {
+		for j := 1; j < len(prefixes[i]); j++ {
+			prefixes = append(prefixes, prefixes[i][j:])
+		}
+	}
+
 	// Check the split function.
 	for _, p := range prefixes {
 		for _, s := range suffixes {
diff --git a/internal/crdbtest/crdbtest.go b/internal/crdbtest/crdbtest.go
index 17f4cedc4e..dca7176069 100644
--- a/internal/crdbtest/crdbtest.go
+++ b/internal/crdbtest/crdbtest.go
@@ -178,8 +178,6 @@ func DecodeTimestamp(mvccKey []byte) ([]byte, []byte, uint64, uint32) {
 
 // Split implements base.Split for CockroachDB keys.
 func Split(key []byte) int {
-	// TODO(radu): Pebble sometimes passes empty "keys" and we have to tolerate
-	// them until we fix that.
 	if len(key) == 0 {
 		return 0
 	}
@@ -187,8 +185,8 @@ func Split(key []byte) int {
 	// Last byte is the version length + 1 when there is a version, else it is
 	// 0.
 	versionLen := int(key[len(key)-1])
-	if versionLen >= len(key) {
-		panic(errors.AssertionFailedf("empty key"))
+	if versionLen > len(key) {
+		panic(errors.AssertionFailedf("invalid version length"))
 	}
 	return len(key) - versionLen
 }
@@ -196,8 +194,6 @@ func Split(key []byte) int {
 // Compare compares cockroach keys, including the version (which could be MVCC
 // timestamps).
 func Compare(a, b []byte) int {
-	// TODO(radu): Pebble sometimes passes empty "keys" and we have to tolerate
-	// them until we fix that.
 	if len(a) == 0 || len(b) == 0 {
 		return cmp.Compare(len(a), len(b))
 	}
@@ -208,11 +204,6 @@ func Compare(a, b []byte) int {
 	// SplitMVCCKey instead of doing this.
 	aEnd := len(a) - 1
 	bEnd := len(b) - 1
-	if aEnd < 0 || bEnd < 0 {
-		// This should never happen unless there is some sort of corruption of
-		// the keys.
-		panic(errors.AssertionFailedf("malformed key: %x, %x", a, b))
-	}
 
 	// 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
diff --git a/internal/testkeys/testkeys.go b/internal/testkeys/testkeys.go
index 789ddfde8b..31519eb969 100644
--- a/internal/testkeys/testkeys.go
+++ b/internal/testkeys/testkeys.go
@@ -115,12 +115,6 @@ var Comparer = &base.Comparer{
 //     value is smaller.
 func compare(a, b []byte) int {
 	ai, bi := split(a), split(b)
-	if ai == 0 && len(a) > 0 {
-		panic(fmt.Sprintf("Compare called with bare suffix %s", a))
-	}
-	if bi == 0 && len(b) > 0 {
-		panic(fmt.Sprintf("Compare called with bare suffix %s", b))
-	}
 	if v := bytes.Compare(a[:ai], b[:bi]); v != 0 {
 		return v
 	}
@@ -129,10 +123,7 @@ func compare(a, b []byte) int {
 
 func split(a []byte) int {
 	i := bytes.LastIndexByte(a, suffixDelim)
-	if i == 0 {
-		panic(fmt.Sprintf("Split called on bare suffix %q", a))
-	}
-	if i > 0 {
+	if i >= 0 {
 		return i
 	}
 	return len(a)