Repository: petermattis/pebble (go)

Issue:
perf: skip user-key comparisons if prev seqnum is zero
`db.Iterator.Next` currently performs a comparison of the user-key when iterating to the next entry. This comparison can be elided if the previous seqnum is zero. In Pebble this is currently rarely the case, but RocksDB contains an optimization to set the seqnum to zero for records stored in the bottommost level (modulo snapshot requirements).

A quick test shows that skipping the user-key comparison provides a ~5% speedup when iterating over sstables in the bottommost level.

Pull request description:
Zero seqnums in sstables where possible
Zero seqnums in sstables in order to improve sstable compression and
enable an optimization to avoid a user-key comparison during forward
iteration. The seqnum for an entry can be zeroed if the key does not
exist at lower levels in the tree and if the entry is in the last
snapshot stripe. Rather than performing the check on a per-key basis, a
single check is done to see if the bounds of a compaction overlaps with
any sstables at lower-levels in the tree. This is slightly
pessimistic (it disallows zeroing of some keys when they could be
zeroed), but fast.

Forward iteration normally requires a user-key comparison to determine
if we've moved to the next user-key. We can skip this comparison if the
seqnum for an entry is 0 as that is the oldest seqnum for a particular
user-key which guarantees the next entry is for a different
user-key. This produces a ~5% speedup in forward iteration performance
when iterating over data contained in the bottommost sstables. Note the
rule of thumb is that ~90% of data in an LSM is in the bottommost level.

Fixes #87

Patch:
diff --git a/compaction.go b/compaction.go
index ecb2419..e685100 100644
--- a/compaction.go
+++ b/compaction.go
@@ -198,6 +198,29 @@ func (c *compaction) shouldStopBefore(key db.InternalKey) bool {
 	return false
 }
 
+// allowZeroSeqNum returns true if seqnum's can be zeroed if there are no
+// snapshots requiring them to be kept. It performs this determination by
+// looking for an sstable which overlaps the bounds of the compaction at a
+// lower level in the LSM.
+func (c *compaction) allowZeroSeqNum() bool {
+	var lower, upper []byte
+	for i := range c.inputs {
+		files := c.inputs[i]
+		for j := range files {
+			f := &files[j]
+			if lower == nil || c.cmp(lower, f.smallest.UserKey) > 0 {
+				lower = f.smallest.UserKey
+			}
+			if upper == nil || c.cmp(upper, f.largest.UserKey) < 0 {
+				upper = f.largest.UserKey
+			}
+		}
+	}
+	// [lower,upper] now cover the bounds of the compaction inputs. Check to see
+	// if those bounds overlap an sstable at a lower level.
+	return c.elideRangeTombstone(lower, upper)
+}
+
 // elideTombstone returns true if it is ok to elide a tombstone for the
 // specified key. A return value of true guarantees that there are no key/value
 // pairs at c.level+2 or higher that possibly contain the specified user key.
@@ -550,8 +573,30 @@ func (d *DB) writeLevel0Table(
 		return true
 	}
 
+	// Allow zeroing of seqnum if the bounds of the compaction do not overlap
+	// with any lower level LSM. Note that this treats range tombstones as point
+	// entries. That is fine for this usage, but beware of adapting this for
+	// another purpose.
+	allowZeroSeqNum := func() bool {
+		if len(version.files[0]) > 0 {
+			// We can only allow zeroing of seqnum for L0 tables if no other L0
+			// tables exist. Otherwise we may violate the invariant that L0 tables
+			// are ordered by increasing seqnum. This could be relaxed with a bit
+			// more intelligence in how a new L0 table is merged into the existing
+			// set of L0 tables.
+			return false
+		}
+		lower, _ := iiter.First()
+		upper, _ := iiter.Last()
+		if lower == nil || upper == nil {
+			return false
+		}
+		return elideRangeTombstone(lower.UserKey, upper.UserKey)
+	}()
+
 	iter := newCompactionIter(
 		d.cmp, d.merge, iiter, snapshots,
+		allowZeroSeqNum,
 		func([]byte) bool { return false },
 		elideRangeTombstone,
 	)
@@ -621,14 +666,6 @@ func (d *DB) writeLevel0Table(
 	meta.largestSeqNum = writerMeta.LargestSeqNum
 	tw = nil
 
-	// TODO(peter): After a flush we set the commit rate to 110% of the flush
-	// rate. The rationale behind the 110% is to account for slack. Investigate a
-	// more principled way of setting this.
-	// d.commitController.limiter.SetLimit(rate.Limit(d.flushController.sensor.Rate()))
-	// if false {
-	// 	fmt.Printf("flush: %.1f MB/s\n", d.flushController.sensor.Rate()/float64(1<<20))
-	// }
-
 	// TODO(peter): compaction stats.
 
 	return meta, nil
@@ -788,7 +825,7 @@ func (d *DB) compactDiskTables(c *compaction) (ve *versionEdit, pendingOutputs [
 		return nil, pendingOutputs, err
 	}
 	iter := newCompactionIter(d.cmp, d.merge, iiter, snapshots,
-		c.elideTombstone, c.elideRangeTombstone)
+		c.allowZeroSeqNum(), c.elideTombstone, c.elideRangeTombstone)
 
 	var (
 		filenames []string
diff --git a/compaction_iter.go b/compaction_iter.go
index cb76bb6..119e03a 100644
--- a/compaction_iter.go
+++ b/compaction_iter.go
@@ -155,6 +155,7 @@ type compactionIter struct {
 	tombstones []rangedel.Tombstone
 	// Byte allocator for the tombstone keys.
 	alloc               bytealloc.A
+	allowZeroSeqNum     bool
 	elideTombstone      func(key []byte) bool
 	elideRangeTombstone func(start, end []byte) bool
 }
@@ -164,6 +165,7 @@ func newCompactionIter(
 	merge db.Merge,
 	iter internalIterator,
 	snapshots []uint64,
+	allowZeroSeqNum bool,
 	elideTombstone func(key []byte) bool,
 	elideRangeTombstone func(start, end []byte) bool,
 ) *compactionIter {
@@ -172,6 +174,7 @@ func newCompactionIter(
 		merge:               merge,
 		iter:                iter,
 		snapshots:           snapshots,
+		allowZeroSeqNum:     allowZeroSeqNum,
 		elideTombstone:      elideTombstone,
 		elideRangeTombstone: elideRangeTombstone,
 	}
@@ -237,6 +240,7 @@ func (i *compactionIter) Next() (*db.InternalKey, []byte) {
 			i.value = i.iterValue
 			i.valid = true
 			i.skip = true
+			i.maybeZeroSeqnum()
 			return &i.key, i.value
 
 		case db.InternalKeyKindMerge:
@@ -246,6 +250,11 @@ func (i *compactionIter) Next() (*db.InternalKey, []byte) {
 				continue
 			}
 
+			// NB: it is important to call maybeZeroSeqnum before mergeNext as
+			// merging advances the iterator, adjusting curSnapshotIdx and thus
+			// invalidating the state that maybeZeroSeqnum uses to make its
+			// determination.
+			i.maybeZeroSeqnum()
 			return i.mergeNext()
 
 		case db.InternalKeyKindInvalid:
@@ -321,7 +330,7 @@ func (i *compactionIter) mergeNext() (*db.InternalKey, []byte) {
 	i.saveValue()
 	i.valid = true
 
-	// Loop looking for older values in the current snapshot stripe and merging
+	// Loop looking for older values in the current snapshot stripe and merge
 	// them.
 	for {
 		if !i.nextInStripe() {
@@ -449,3 +458,22 @@ func (i *compactionIter) emitRangeDelChunk(fragmented []rangedel.Tombstone) {
 		currentIdx = idx
 	}
 }
+
+// maybeZeroSeqnum attempts to set the seqnum for the current key to 0. Doing
+// so improves compression and enables an optimization during forward iteration
+// to skip some key comparisons. The seqnum for an entry can be zeroed if the
+// entry is on the bottom snapshot stripe and on the bottom level of the LSM.
+func (i *compactionIter) maybeZeroSeqnum() {
+	if !i.allowZeroSeqNum {
+		// TODO(peter): allowZeroSeqNum applies to the entire compaction. We could
+		// make the determination on a key by key basis, similar to what is done
+		// for elideTombstone. Need to add a benchmark for compactionIter to verify
+		// that isn't too expensive.
+		return
+	}
+	if i.curSnapshotIdx > 0 {
+		// This is not the last snapshot
+		return
+	}
+	i.key.SetSeqNum(0)
+}
diff --git a/iterator.go b/iterator.go
index a93ded1..9f71b7b 100644
--- a/iterator.go
+++ b/iterator.go
@@ -87,15 +87,17 @@ func (i *Iterator) findNextEntry() bool {
 
 func (i *Iterator) nextUserKey() {
 	if i.iterKey != nil {
+		done := i.iterKey.SeqNum() == 0
 		if !i.valid {
 			i.keyBuf = append(i.keyBuf[:0], i.iterKey.UserKey...)
 			i.key = i.keyBuf
 		}
 		for {
 			i.iterKey, i.iterValue = i.iter.Next()
-			if i.iterKey == nil || !i.equal(i.key, i.iterKey.UserKey) {
+			if done || i.iterKey == nil || !i.equal(i.key, i.iterKey.UserKey) {
 				break
 			}
+			done = i.iterKey.SeqNum() == 0
 		}
 	} else {
 		i.iterKey, i.iterValue = i.iter.First()
@@ -174,6 +176,8 @@ func (i *Iterator) findPrevEntry() bool {
 func (i *Iterator) prevUserKey() {
 	if i.iterKey != nil {
 		if !i.valid {
+			// If we're going to compare against the prev key, we need to save the
+			// current key.
 			i.keyBuf = append(i.keyBuf[:0], i.iterKey.UserKey...)
 			i.key = i.keyBuf
 		}
diff --git a/version.go b/version.go
index 56434bd..6c5e033 100644
--- a/version.go
+++ b/version.go
@@ -249,6 +249,9 @@ func (v *version) overlaps(
 	upper := sort.Search(len(files), func(i int) bool {
 		return cmp(files[i].smallest.UserKey, end) > 0
 	})
+	if lower >= upper {
+		return nil
+	}
 	return files[lower:upper]
 }
 


Test patch:
diff --git a/compaction_iter_test.go b/compaction_iter_test.go
index e4fcf35..39d9876 100644
--- a/compaction_iter_test.go
+++ b/compaction_iter_test.go
@@ -58,6 +58,7 @@ func TestCompactionIter(t *testing.T) {
 			db.DefaultMerger.Merge,
 			&fakeIter{keys: keys, vals: vals},
 			snapshots,
+			false, /* allowZeroSeqNum */
 			func([]byte) bool {
 				return elideTombstones
 			},
diff --git a/compaction_test.go b/compaction_test.go
index eb5295c..4375b9c 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -7,6 +7,7 @@ package pebble
 import (
 	"bytes"
 	"fmt"
+	"regexp"
 	"sort"
 	"strconv"
 	"strings"
@@ -953,3 +954,79 @@ func TestCompactionAtomicUnitBounds(t *testing.T) {
 			}
 		})
 }
+
+func TestCompactionAllowZeroSeqNum(t *testing.T) {
+	var d *DB
+
+	metaRE := regexp.MustCompile(`^L([0-9]+):([^-]+)-(.+)$`)
+	parseMeta := func(s string) (level int, meta fileMetadata) {
+		match := metaRE.FindStringSubmatch(s)
+		if match == nil {
+			t.Fatalf("malformed table spec: %s", s)
+		}
+		level, err := strconv.Atoi(match[1])
+		if err != nil {
+			t.Fatalf("malformed table spec: %s: %s", s, err)
+		}
+		meta = fileMetadata{
+			smallest: db.InternalKey{UserKey: []byte(match[2])},
+			largest:  db.InternalKey{UserKey: []byte(match[3])},
+		}
+		return level, meta
+	}
+
+	datadriven.RunTest(t, "testdata/compaction_allow_zero_seqnum",
+		func(td *datadriven.TestData) string {
+			switch td.Cmd {
+			case "define":
+				var err error
+				if d, err = runDBDefineCmd(td); err != nil {
+					return err.Error()
+				}
+
+				d.mu.Lock()
+				s := d.mu.versions.currentVersion().String()
+				d.mu.Unlock()
+				return s
+
+			case "allow-zero-seqnum":
+				d.mu.Lock()
+				c := &compaction{
+					cmp:     d.cmp,
+					version: d.mu.versions.currentVersion(),
+				}
+				d.mu.Unlock()
+
+				var buf bytes.Buffer
+				for _, line := range strings.Split(td.Input, "\n") {
+					parts := strings.Fields(line)
+					if len(parts) == 0 {
+						continue
+					}
+					c.inputs[0] = nil
+					c.inputs[1] = nil
+					c.level = -1
+
+					for _, p := range parts {
+						level, meta := parseMeta(p)
+						i := 0
+						switch {
+						case c.level == -1:
+							c.level = level
+						case c.level+1 == level:
+							i = 1
+						case c.level != level:
+							return fmt.Sprintf("invalid level %d: expected %d or %d",
+								level, c.level, c.level+1)
+						}
+						c.inputs[i] = append(c.inputs[i], meta)
+					}
+					fmt.Fprintf(&buf, "%t\n", c.allowZeroSeqNum())
+				}
+				return buf.String()
+
+			default:
+				return fmt.Sprintf("unknown command: %s", td.Cmd)
+			}
+		})
+}
diff --git a/testdata/compaction_allow_zero_seqnum b/testdata/compaction_allow_zero_seqnum
new file mode 100644
index 0000000..37a334c
--- /dev/null
+++ b/testdata/compaction_allow_zero_seqnum
@@ -0,0 +1,35 @@
+define
+L2
+  c.SET.2:2
+----
+2: c-c
+
+allow-zero-seqnum
+L0:b-b
+L0:c-c
+L0:d-d
+----
+true
+false
+true
+
+allow-zero-seqnum
+L0:c-c L0:d-d
+L0:c-c L1:d-d
+L0:b-b L0:b-c
+L0:b-b L1:b-c
+----
+false
+false
+false
+false
+
+# We only look for overlaps at L<N+2> as it isn't valid for a
+# compaction rooted at L<N> to not include overlapping tables at
+# L<N+1>.
+
+allow-zero-seqnum
+L1:c-c
+----
+true
+
diff --git a/testdata/manual_flush b/testdata/manual_flush
index 64882b0..98bbff5 100644
--- a/testdata/manual_flush
+++ b/testdata/manual_flush
@@ -3,9 +3,10 @@ set a 1
 set b 2
 ----
 
+# The first L0 table can have its seqnums zeroed.
 flush
 ----
-0: a#0,1-b#1,1
+0: a#0,1-b#0,1
 
 reset
 ----
@@ -21,6 +22,24 @@ flush
 ----
 0: a#2,0-b#3,0
 
+batch
+set a 3
+----
+
+# A second (overlapping) L0 table will have non-zero seqnums.
+flush
+----
+0: a#2,0-b#3,0 a#4,1-a#4,1
+
+batch
+set c 4
+----
+
+# A third (non-overlapping) L0 table will have non-zero seqnums.
+flush
+----
+0: a#2,0-b#3,0 a#4,1-a#4,1 c#5,1-c#5,1
+
 reset
 ----
 
@@ -43,4 +62,4 @@ set b 2
 
 async-flush
 ----
-0: a#0,1-b#1,1
+0: a#0,1-b#0,1