sstable: assert KeyComparison correctness under invariants

When the invariants build tag is enabled, assert that the KeyComparison
returned by the columnar data block writer is accurate by recomputing the
KeyComparison from a buffered previous user key.

Informs #4103.

internal/invariants/off.go

@@ -27,3 +27,18 @@ func (d *CloseChecker) AssertClosed() {}
 
 // AssertNotClosed panics in invariant builds if Close was called.
 func (d *CloseChecker) AssertNotClosed() {}
+
+// Value is a generic container for a value that should only exist in invariant
+// builds. In non-invariant builds, storing a value is a no-op, retrieving a
+// value returns the type parameter's zero value, and the Value struct takes up
+// no space.
+type Value[V any] struct{}
+
+// Get returns the current value, or the zero-value if invariants are disabled.
+func (*Value[V]) Get() V {
+	var v V // zero value
+	return v
+}
+
+// Store stores the value.
+func (*Value[V]) Store(v V) {}

internal/invariants/on.go

@@ -40,3 +40,21 @@ func (d *CloseChecker) AssertNotClosed() {
 		panic("closed")
 	}
 }
+
+// Value is a generic container for a value that should only exist in invariant
+// builds. In non-invariant builds, storing a value is a no-op, retrieving a
+// value returns the type parameter's zero value, and the Value struct takes up
+// no space.
+type Value[V any] struct {
+	v V
+}
+
+// Get returns the current value, or the zero-value if invariants are disabled.
+func (v *Value[V]) Get() V {
+	return v.v
+}
+
+// Store stores the value.
+func (v *Value[V]) Store(inner V) {
+	v.v = inner
+}

sstable/colblk/data_block.go

@@ -97,6 +97,24 @@ type KeyWriter interface {
 	FinishHeader(dst []byte)
 }
 
+// AssertKeyCompare compares two keys using the provided comparer, ensuring the
+// provided KeyComparison accurately describing the result. Panics if the
+// assertion does not hold.
+func AssertKeyCompare(comparer *base.Comparer, a, b []byte, kcmp KeyComparison) {
+	bi := comparer.Split(b)
+	var recomputed KeyComparison
+	recomputed.PrefixLen = int32(comparer.Split(a))
+	recomputed.CommonPrefixLen = int32(crbytes.CommonPrefix(a[:recomputed.PrefixLen], b[:bi]))
+	recomputed.UserKeyComparison = int32(comparer.Compare(a, b))
+	if recomputed.PrefixEqual() != bytes.Equal(a[:recomputed.PrefixLen], b[:bi]) {
+		panic(errors.AssertionFailedf("PrefixEqual()=%t doesn't hold: %q, %q", kcmp.PrefixEqual(), a, b))
+	}
+	if recomputed != kcmp {
+		panic(errors.AssertionFailedf("KeyComparison of (%q, %q) = %s, ComparePrev gave %s",
+			a, b, recomputed, kcmp))
+	}
+}
+
 // KeyComparison holds information about a key and its comparison to another a
 // key.
 type KeyComparison struct {
@@ -116,6 +134,12 @@ type KeyComparison struct {
 	UserKeyComparison int32
 }
 
+// String returns a string representation of the KeyComparison.
+func (kcmp KeyComparison) String() string {
+	return fmt.Sprintf("(prefix={%d,common=%d} cmp=%d)",
+		kcmp.PrefixLen, kcmp.CommonPrefixLen, kcmp.UserKeyComparison)
+}
+
 // PrefixEqual returns true if the key comparison determined that the keys have
 // equal prefixes.
 func (kcmp KeyComparison) PrefixEqual() bool { return kcmp.PrefixLen == kcmp.CommonPrefixLen }

sstable/colblk_writer.go

@@ -16,6 +16,7 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/bytealloc"
+	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/sstable/block"
@@ -95,6 +96,7 @@ type RawColumnWriter struct {
 
 	separatorBuf          []byte
 	tmp                   [blockHandleLikelyMaxLen]byte
+	previousUserKey       invariants.Value[[]byte]
 	disableKeyOrderChecks bool
 }
 
@@ -469,6 +471,13 @@ func (w *RawColumnWriter) evaluatePoint(
 	key base.InternalKey, valueLen int,
 ) (eval pointKeyEvaluation, err error) {
 	eval.kcmp = w.dataBlock.KeyWriter.ComparePrev(key.UserKey)
+
+	// When invariants are enabled, validate kcmp.
+	if invariants.Enabled {
+		colblk.AssertKeyCompare(w.comparer, key.UserKey, w.previousUserKey.Get(), eval.kcmp)
+		w.previousUserKey.Store(append(w.previousUserKey.Get()[:0], key.UserKey...))
+	}
+
 	if !w.meta.HasPointKeys {
 		return eval, nil
 	}