db, sstable: disallow invalid empty keys in comparer functions

checkpoint_test.go

@@ -139,7 +139,7 @@ func testCheckpointImpl(t *testing.T, ddFile string, createOnShared bool) {
 			}
 			memLog.Reset()
 			d := dbs[td.CmdArgs[0].String()]
-			if err := d.Compact(context.Background(), nil, []byte("\xff"), false); err != nil {
+			if err := d.Compact(context.Background(), []byte{0}, []byte("\xff"), false); err != nil {
 				return err.Error()
 			}
 			d.TestOnlyWaitForCleaning()

cockroachkvs/cockroachkvs.go

@@ -312,6 +312,9 @@ func Split(key []byte) int {
 // timestamps).
 func Compare(a, b []byte) int {
 	if len(a) == 0 || len(b) == 0 {
+		if invariants.Enabled {
+			panic(errors.AssertionFailedf("empty key passed to Compare"))
+		}
 		return cmp.Compare(len(a), len(b))
 	}
 	if invariants.Enabled {
@@ -390,6 +393,9 @@ func compareVersions(version, firstRowUntypedVer []byte) bool {
 // Equal implements base.Equal for Cockroach keys.
 func Equal(a, b []byte) bool {
 	if len(a) == 0 || len(b) == 0 {
+		if invariants.Enabled {
+			panic(errors.AssertionFailedf("empty key passed to Equal"))
+		}
 		return len(a) == len(b)
 	}
 	if invariants.Enabled {

cockroachkvs/cockroachkvs_test.go

@@ -178,11 +178,13 @@ func TestKeySchema_KeyWriter(t *testing.T) {
 				k := parseUserKey(line)
 				fmt.Fprintf(&buf, "Parse(%s) = hex:%x\n", line, k)
 				kcmp := kw.ComparePrev(k)
-				if v := Compare(k, keyBuf); v < 0 {
-					t.Fatalf("line %d: Compare(%s, hex:%x) = %d", i, line, keyBuf, v)
-				} else if v != int(kcmp.UserKeyComparison) {
-					t.Fatalf("line %d: Compare(%s, hex:%x) = %d; kcmp.UserKeyComparison = %d",
-						i, line, keyBuf, v, kcmp.UserKeyComparison)
+				if len(keyBuf) > 0 {
+					if v := Compare(k, keyBuf); v < 0 {
+						t.Fatalf("line %d: Compare(%s, hex:%x) = %d", i, line, keyBuf, v)
+					} else if v != int(kcmp.UserKeyComparison) {
+						t.Fatalf("line %d: Compare(%s, hex:%x) = %d; kcmp.UserKeyComparison = %d",
+							i, line, keyBuf, v, kcmp.UserKeyComparison)
+					}
 				}
 
 				fmt.Fprintf(&buf, "%02d: ComparePrev(%s): PrefixLen=%d; CommonPrefixLen=%d; UserKeyComparison=%d\n",

compaction.go

@@ -918,7 +918,9 @@ func newFlush(
 	if opts.FlushSplitBytes > 0 {
 		c.maxOutputFileSize = uint64(opts.TargetFileSizes[0])
 		c.maxOverlapBytes = maxGrandparentOverlapBytes(opts.TargetFileSizes[0])
-		c.grandparents = cur.Overlaps(baseLevel, c.bounds)
+		if c.bounds.Start != nil && c.bounds.End.Key != nil {
+			c.grandparents = cur.Overlaps(baseLevel, c.bounds)
+		}
 		adjustGrandparentOverlapBytesForFlush(c, flushingBytes)
 	}
 
@@ -1789,18 +1791,22 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 			// c.kind == compactionKindIngestedFlushable && we could have deleted files due
 			// to ingest-time splits or excises.
 			ingestFlushable := c.flush.flushables[0].flushable.(*ingestedFlushable)
-			exciseBounds := ingestFlushable.exciseSpan.UserKeyBounds()
-			for c2 := range d.mu.compact.inProgress {
-				// Check if this compaction overlaps with the excise span. Note that just
-				// checking if the inputs individually overlap with the excise span
-				// isn't sufficient; for instance, a compaction could have [a,b] and [e,f]
-				// as inputs and write it all out as [a,b,e,f] in one sstable. If we're
-				// doing a [c,d) excise at the same time as this compaction, we will have
-				// to error out the whole compaction as we can't guarantee it hasn't/won't
-				// write a file overlapping with the excise span.
-				bounds := c2.Bounds()
-				if bounds != nil && bounds.Overlaps(d.cmp, exciseBounds) {
-					c2.Cancel()
+			if ingestFlushable.exciseSpan.Valid() {
+				exciseBounds := ingestFlushable.exciseSpan.UserKeyBounds()
+				for c2 := range d.mu.compact.inProgress {
+					// Check if this compaction overlaps with the excise span. Note that just
+					// checking if the inputs individually overlap with the excise span
+					// isn't sufficient; for instance, a compaction could have [a,b] and [e,f]
+					// as inputs and write it all out as [a,b,e,f] in one sstable. If we're
+					// doing a [c,d) excise at the same time as this compaction, we will have
+					// to error out the whole compaction as we can't guarantee it hasn't/won't
+					// write a file overlapping with the excise span.
+					if bounds := c2.Bounds(); bounds != nil {
+						if (bounds.Start == nil || exciseBounds.End.IsUpperBoundFor(d.cmp, bounds.Start)) &&
+							(bounds.End.Key == nil || bounds.End.IsUpperBoundFor(d.cmp, exciseBounds.Start)) {
+							c2.Cancel()
+						}
+					}
 				}
 			}
 

db.go

@@ -1690,6 +1690,9 @@ func (d *DB) Compact(ctx context.Context, start, end []byte, parallelize bool) e
 	if d.opts.ReadOnly {
 		return ErrReadOnly
 	}
+	if start == nil || end == nil {
+		return errors.Errorf("Compact start/end keys cannot be nil")
+	}
 	if d.cmp(start, end) >= 0 {
 		return errors.Errorf("Compact start %s is not less than end %s",
 			d.opts.Comparer.FormatKey(start), d.opts.Comparer.FormatKey(end))
@@ -2793,7 +2796,7 @@ func (d *DB) ScanStatistics(
 			// pinned by a snapshot.
 			size := uint64(key.Size())
 			kind := key.Kind()
-			sameKey := d.equal(prevKey.UserKey, key.UserKey)
+			sameKey := prevKey.UserKey != nil && d.equal(prevKey.UserKey, key.UserKey)
 			if iterInfo.Kind == IteratorLevelLSM && sameKey {
 				stats.Levels[iterInfo.Level].SnapshotPinnedKeys++
 				stats.Levels[iterInfo.Level].SnapshotPinnedKeysBytes += size

db_test.go

@@ -1194,7 +1194,7 @@ func TestDBConcurrentCommitCompactFlush(t *testing.T) {
 			var err error
 			switch i % 3 {
 			case 0:
-				err = d.Compact(context.Background(), nil, []byte("\xff"), false)
+				err = d.Compact(context.Background(), []byte{0}, []byte("\xff"), false)
 			case 1:
 				err = d.Flush()
 			case 2:

error_test.go

@@ -135,7 +135,7 @@ func TestErrors(t *testing.T) {
 		if err := d.Flush(); err != nil {
 			return err
 		}
-		if err := d.Compact(context.Background(), nil, []byte("\xff"), false); err != nil {
+		if err := d.Compact(context.Background(), []byte{0}, []byte("\xff"), false); err != nil {
 			return err
 		}
 

excise.go

@@ -429,6 +429,13 @@ func determineRightTableBounds(
 func determineExcisedTableSize(
 	fc *fileCacheHandle, originalTable, excisedTable *manifest.TableMetadata,
 ) error {
+	if !originalTable.HasPointKeys {
+		// The original table has no point keys (only range keys); the index
+		// block only has a placeholder entry for the forced empty data block.
+		// Skip estimating size through the index block.
+		excisedTable.Size = 1
+		return nil
+	}
 	size, err := fc.estimateSize(originalTable, excisedTable.Smallest().UserKey, excisedTable.Largest().UserKey)
 	if err != nil {
 		return err

flush_test.go

@@ -111,9 +111,9 @@ func TestFlushEmptyKey(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	d, err := Open("", &Options{FS: vfs.NewMem()})
 	require.NoError(t, err)
-	require.NoError(t, d.Set(nil, []byte("hello"), nil))
+	require.NoError(t, d.Set([]byte{}, []byte("hello"), nil))
 	require.NoError(t, d.Flush())
-	val, closer, err := d.Get(nil)
+	val, closer, err := d.Get([]byte{})
 	require.NoError(t, err)
 	require.Equal(t, val, []byte("hello"))
 	require.NoError(t, closer.Close())

flushable.go

@@ -485,7 +485,12 @@ func determineOverlapAllIters(
 func determineOverlapPointIterator(
 	cmp base.Compare, bounds base.UserKeyBounds, iter internalIterator,
 ) (bool, error) {
-	kv := iter.SeekGE(bounds.Start, base.SeekGEFlagsNone)
+	var kv *base.InternalKV
+	if bounds.Start != nil {
+		kv = iter.SeekGE(bounds.Start, base.SeekGEFlagsNone)
+	} else {
+		kv = iter.First()
+	}
 	if kv == nil {
 		return false, iter.Error()
 	}
@@ -496,7 +501,13 @@ func determineOverlapKeyspanIterator(
 	cmp base.Compare, bounds base.UserKeyBounds, iter keyspan.FragmentIterator,
 ) (bool, error) {
 	// NB: The spans surfaced by the fragment iterator are non-overlapping.
-	span, err := iter.SeekGE(bounds.Start)
+	var span *keyspan.Span
+	var err error
+	if bounds.Start != nil {
+		span, err = iter.SeekGE(bounds.Start)
+	} else {
+		span, err = iter.First()
+	}
 	if err != nil {
 		return false, err
 	}

format_major_version_test.go

@@ -109,7 +109,7 @@ func testBasicDB(d *DB) error {
 	if err := d.Flush(); err != nil {
 		return err
 	}
-	if err := d.Compact(context.Background(), nil, []byte("\xff"), false); err != nil {
+	if err := d.Compact(context.Background(), []byte{0}, []byte("\xff"), false); err != nil {
 		return err
 	}
 

ingest.go

@@ -2463,7 +2463,9 @@ func (d *DB) ingestApply(
 				// to error out the whole compaction as we can't guarantee it hasn't/won't
 				// write a file overlapping with the excise span.
 				bounds := c.Bounds()
-				if bounds != nil && bounds.Overlaps(d.cmp, exciseBounds) {
+				if exciseSpan.Valid() && bounds != nil &&
+					(bounds.Start == nil || exciseBounds.End.IsUpperBoundFor(d.cmp, bounds.Start)) &&
+					(bounds.End.Key == nil || bounds.End.IsUpperBoundFor(d.cmp, exciseBounds.Start)) {
 					c.Cancel()
 				}
 				// Check if this compaction's inputs have been replaced due to an

internal/base/comparer_test.go

@@ -113,7 +113,7 @@ func TestAbbreviatedKey(t *testing.T) {
 
 	keys := make([][]byte, 10000)
 	for i := range keys {
-		keys[i] = randBytes(rng.IntN(16))
+		keys[i] = randBytes(1 + rng.IntN(15))
 	}
 	slices.SortFunc(keys, DefaultComparer.Compare)
 

internal/base/internal.go

@@ -470,7 +470,12 @@ 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(k.UserKey) == 0 || len(buf) <= len(k.UserKey)) && cmp(k.UserKey, buf) < 0 {
+	if len(k.UserKey) == 0 {
+		if len(buf) == 0 {
+			// The empty slice might not be a valid key, while buf is always valid.
+			return MakeInternalKey(buf, SeqNumMax, InternalKeyKindSeparator)
+		}
+	} else if 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

internal/base/key_bounds.go

@@ -176,6 +176,9 @@ func UserKeyBoundsFromInternal(smallest, largest InternalKey) UserKeyBounds {
 
 // Valid returns true if the bounds contain at least a user key.
 func (b *UserKeyBounds) Valid(cmp Compare) bool {
+	if b.Start == nil {
+		return false
+	}
 	return b.End.IsUpperBoundFor(cmp, b.Start)
 }
 

internal/batchskl/skl_test.go

@@ -254,7 +254,7 @@ func TestIteratorSeekLT(t *testing.T) {
 		require.Nil(t, l.Add(d.add(fmt.Sprintf("%05d", i*10+1000))))
 	}
 
-	require.Nil(t, it.SeekLT(makeKey("")))
+	require.Nil(t, it.SeekLT(makeKey("\x00")))
 	require.Nil(t, it.SeekLT(makeKey("01000")))
 	assertKey(t, "01000", it.SeekLT(makeKey("01001")))
 	assertKey(t, "01000", it.SeekLT(makeKey("01005")))

internal/manifest/table_metadata.go

@@ -965,12 +965,12 @@ func ParseTableMetadataDebug(s string) (_ *TableMetadata, err error) {
 	}
 
 	cmp := base.DefaultComparer.Compare
-	if base.InternalCompare(cmp, smallest, m.PointKeyBounds.Smallest()) == 0 {
+	if m.HasPointKeys && base.InternalCompare(cmp, smallest, m.PointKeyBounds.Smallest()) == 0 {
 		m.boundTypeSmallest = boundTypePointKey
 	} else if m.HasRangeKeys && base.InternalCompare(cmp, smallest, m.RangeKeyBounds.Smallest()) == 0 {
 		m.boundTypeSmallest = boundTypeRangeKey
 	}
-	if base.InternalCompare(cmp, largest, m.PointKeyBounds.Largest()) == 0 {
+	if m.HasPointKeys && base.InternalCompare(cmp, largest, m.PointKeyBounds.Largest()) == 0 {
 		m.boundTypeLargest = boundTypePointKey
 	} else if m.HasRangeKeys && base.InternalCompare(cmp, largest, m.RangeKeyBounds.Largest()) == 0 {
 		m.boundTypeLargest = boundTypeRangeKey

internal/testkeys/testkeys.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/invariants"
 )
 
 const alpha = "abcdefghijklmnopqrstuvwxyz"
@@ -129,6 +130,9 @@ var Comparer = &base.Comparer{
 //   - when both keys have a suffix, the key with the larger (decoded) suffix
 //     value is smaller.
 func compare(a, b []byte) int {
+	if invariants.Enabled && (len(a) == 0 || len(b) == 0) {
+		panic(errors.AssertionFailedf("empty key passed to Compare"))
+	}
 	ai, bi := split(a), split(b)
 	if v := bytes.Compare(a[:ai], b[:bi]); v != 0 {
 		return v

iterator.go

@@ -2581,8 +2581,8 @@ func (i *Iterator) SetBounds(lower, upper []byte) {
 
 	if ((i.opts.LowerBound == nil) == (lower == nil)) &&
 		((i.opts.UpperBound == nil) == (upper == nil)) &&
-		i.equal(i.opts.LowerBound, lower) &&
-		i.equal(i.opts.UpperBound, upper) {
+		(i.opts.LowerBound == nil || i.equal(i.opts.LowerBound, lower)) &&
+		(i.opts.UpperBound == nil || i.equal(i.opts.UpperBound, upper)) {
 		// Unchanged, noop.
 		return
 	}
@@ -2799,8 +2799,8 @@ func (i *Iterator) SetOptions(o *IterOptions) {
 
 	boundsEqual := ((i.opts.LowerBound == nil) == (o.LowerBound == nil)) &&
 		((i.opts.UpperBound == nil) == (o.UpperBound == nil)) &&
-		i.equal(i.opts.LowerBound, o.LowerBound) &&
-		i.equal(i.opts.UpperBound, o.UpperBound)
+		(i.opts.LowerBound == nil || i.equal(i.opts.LowerBound, o.LowerBound)) &&
+		(i.opts.UpperBound == nil || i.equal(i.opts.UpperBound, o.UpperBound))
 
 	if boundsEqual && o.KeyTypes == i.opts.KeyTypes &&
 		(i.pointIter != nil || !i.opts.pointKeys()) &&

iterator_test.go

@@ -1499,7 +1499,7 @@ func testSetOptionsEquivalence(t *testing.T, seed uint64) {
 					o.UpperBound = testkeys.KeyAt(ks, rng.Uint64N(ks.Count()), rng.Int64N(int64(ks.Count())))
 				}
 			}
-			if testkeys.Comparer.Compare(o.LowerBound, o.UpperBound) > 0 {
+			if len(o.LowerBound) > 0 && len(o.UpperBound) > 0 && testkeys.Comparer.Compare(o.LowerBound, o.UpperBound) > 0 {
 				o.LowerBound, o.UpperBound = o.UpperBound, o.LowerBound
 			}
 		}
@@ -2067,7 +2067,7 @@ func BenchmarkBlockPropertyFilter(b *testing.B) {
 			}
 			require.NoError(b, batch.Commit(nil))
 			require.NoError(b, d.Flush())
-			require.NoError(b, d.Compact(context.Background(), nil, []byte{0xFF}, false))
+			require.NoError(b, d.Compact(context.Background(), []byte{0}, []byte{0xFF}, false))
 
 			for _, filter := range []bool{false, true} {
 				b.Run(fmt.Sprintf("filter=%t", filter), func(b *testing.B) {

level_checker.go

@@ -184,7 +184,7 @@ func (m *simpleMergingIter) handleVisiblePoint(
 	item *simpleMergingIterItem, l *simpleMergingIterLevel,
 ) (ok bool) {
 	m.numPoints++
-	keyChanged := m.heap.cmp(item.kv.K.UserKey, m.lastKey.UserKey) != 0
+	keyChanged := m.lastKey.UserKey == nil || m.heap.cmp(item.kv.K.UserKey, m.lastKey.UserKey) != 0
 	if !keyChanged {
 		// At the same user key. We will see them in decreasing seqnum
 		// order so the lastLevel must not be lower.
@@ -315,7 +315,7 @@ func iterateAndCheckTombstones(
 	// in non-decreasing level order.
 	lastTombstone := tombstoneWithLevel{}
 	for _, t := range tombstones {
-		if cmp(lastTombstone.Start, t.Start) == 0 && lastTombstone.level > t.level {
+		if lastTombstone.Start != nil && cmp(lastTombstone.Start, t.Start) == 0 && lastTombstone.level > t.level {
 			return errors.Errorf("encountered tombstone %s in %s"+
 				" that has a lower seqnum than the same tombstone in %s",
 				t.Span.Pretty(formatKey), levelOrMemtable(t.lsmLevel, t.tableNum),
@@ -443,7 +443,7 @@ func addTombstonesFromIter(
 		// This is mainly a test for rangeDelV2 formatted blocks which are expected to
 		// be ordered and fragmented on disk. But we anyways check for memtables,
 		// rangeDelV1 as well.
-		if cmp(prevTombstone.End, t.Start) > 0 {
+		if prevTombstone.End != nil && cmp(prevTombstone.End, t.Start) > 0 {
 			return nil, errors.Errorf("unordered or unfragmented range delete tombstones %s, %s in %s",
 				prevTombstone.Pretty(formatKey), t.Pretty(formatKey), levelOrMemtable(lsmLevel, tableNum))
 		}

metamorphic/generator.go

@@ -569,7 +569,7 @@ func (g *generator) newIter() {
 			// Generate a new key with a .1% probability.
 			opts.upper = g.keyGenerator.RandKey(0.001)
 		}
-		if g.cmp(opts.lower, opts.upper) > 0 {
+		if opts.lower != nil && opts.upper != nil && g.cmp(opts.lower, opts.upper) > 0 {
 			opts.lower, opts.upper = opts.upper, opts.lower
 		}
 	}
@@ -728,7 +728,7 @@ func (g *generator) iterSetBounds(iterID objID) {
 				// Generate a new key with a .1% probability.
 				upper = g.keyGenerator.RandKey(0.001)
 			}
-			if g.cmp(lower, upper) > 0 {
+			if lower != nil && upper != nil && g.cmp(lower, upper) > 0 {
 				lower, upper = upper, lower
 			}
 			if ensureLowerGE && g.cmp(iterLastOpts.upper, lower) > 0 {
@@ -1565,7 +1565,7 @@ func (g *generator) maybeMutateOptions(readerID objID, opts *iterOpts) {
 				// Generate a new key with a .1% probability.
 				opts.upper = g.keyGenerator.RandKey(0.001)
 			}
-			if g.cmp(opts.lower, opts.upper) > 0 {
+			if opts.lower != nil && opts.upper != nil && g.cmp(opts.lower, opts.upper) > 0 {
 				opts.lower, opts.upper = opts.upper, opts.lower
 			}
 		}

metamorphic/ops.go

@@ -831,7 +831,7 @@ func (o *ingestOp) collapseBatch(
 			// sequence number precedence determine which of the keys "wins".
 			// But the code to build the ingested sstable will only keep the
 			// most recent internal key and will not merge across internal keys.
-			if equal(lastUserKey, kv.K.UserKey) {
+			if lastUserKey != nil && equal(lastUserKey, kv.K.UserKey) {
 				continue
 			}
 			// NB: We don't have to copy the key or value since we're reading from a

obsolete_files_test.go

@@ -53,7 +53,7 @@ func TestCleaner(t *testing.T) {
 				return "compact <db>"
 			}
 			d := dbs[td.CmdArgs[0].String()]
-			if err := d.Compact(context.Background(), nil, []byte("\xff"), false); err != nil {
+			if err := d.Compact(context.Background(), []byte{0}, []byte("\xff"), false); err != nil {
 				return err.Error()
 			}
 			return memLog.String()