Repository: petermattis/pebble
Pull request #139 (2019-05-14 20:49:28)

Add support for prefix iterators
This is a continuation of https://github.com/petermattis/pebble/pull/35, which added support for writing prefix bloom filters.

Some things to keep note of are:
- `First()`, `Last()`, and `SeekLT()` are not supported in prefix iteration mode
- Prefix iteration mode **requires** a `Split()` function (aka `PrefixExtractor()` in RocksDB) to be passed into the reader options
- `{Lower,Upper}Bound` are ignored in prefix iteration mode
- There can be multiple calls to `SeekGE()`, each of which reset the prefix being iterated on
- The prefix is extracted on calls to `SeekGE()` from the key which is passed in, not the first key encountered

Since prefix iterators behave differently from regular iterators, the `./sstable/testdata/reader/` tests were separated into `./sstable/testdata/prefixreader/` to test just prefix iterators. 



petermattis: 

<!-- Reviewable:start -->
This change is [<img src="https://reviewable.io/review_button.png" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/petermattis/pebble/139)
<!-- Reviewable:end -->

petermattis: > I've compared pulling the upper bound logic into withinUpperBound and lower bound logic into withinLowerBound.

Benchmark numbers are much easier to view using the [`benchstat`](https://godoc.org/golang.org/x/perf/cmd/benchstat) tool. The way I usually do this is run each benchmark 10x (e.g. `go test -run - -bench Foo -count 10`) outputting to two separate files. Then run `benchstat old new`.

Patch:
diff --git a/batch.go b/batch.go
index 7801b63..9dbd6ea 100644
--- a/batch.go
+++ b/batch.go
@@ -473,8 +473,8 @@ func (b *Batch) Repr() []byte {
 }
 
 // NewIter returns an iterator that is unpositioned (Iterator.Valid() will
-// return false). The iterator can be positioned via a call to SeekGE, SeekLT,
-// First or Last. Only indexed batches support iterators.
+// return false). The iterator can be positioned via a call to SeekGE,
+// SeekPrefixGE, SeekLT, First or Last. Only indexed batches support iterators.
 func (b *Batch) NewIter(o *db.IterOptions) *Iterator {
 	if b.index == nil {
 		return &Iterator{err: ErrNotIndexed}
@@ -752,6 +752,10 @@ func (i *batchIter) SeekGE(key []byte) (*db.InternalKey, []byte) {
 	return ikey, i.Value()
 }
 
+func (i *batchIter) SeekPrefixGE(prefix, key []byte) (*db.InternalKey, []byte) {
+	return i.SeekGE(key)
+}
+
 func (i *batchIter) SeekLT(key []byte) (*db.InternalKey, []byte) {
 	ikey := i.iter.SeekLT(key)
 	if ikey == nil {
@@ -1006,6 +1010,10 @@ func (i *flushableBatchIter) SeekGE(key []byte) (*db.InternalKey, []byte) {
 	return &i.key, i.Value()
 }
 
+func (i *flushableBatchIter) SeekPrefixGE(prefix, key []byte) (*db.InternalKey, []byte) {
+	return i.SeekGE(key)
+}
+
 func (i *flushableBatchIter) SeekLT(key []byte) (*db.InternalKey, []byte) {
 	ikey := db.MakeSearchKey(key)
 	i.index = sort.Search(len(i.offsets), func(j int) bool {
diff --git a/compaction.go b/compaction.go
index cde4ea8..9ee3314 100644
--- a/compaction.go
+++ b/compaction.go
@@ -8,7 +8,6 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
-	"math"
 	"os"
 	"sort"
 	"unsafe"
@@ -58,13 +57,7 @@ type compaction struct {
 	// a compaction results in a larger size, the original compaction is used
 	// instead.
 	maxExpandedBytes uint64
-	// disableRangeTombstoneElision disables elision of range tombstones. Used by
-	// tests to allow range tombstones to be added to tables where they would
-	// otherwise be elided.
-	disableRangeTombstoneElision bool
 
-	// flushing contains the flushables (aka memtables) that are being flushed.
-	flushing []flushable
 	// inputs are the tables to be compacted.
 	inputs [2][]fileMetadata
 
@@ -93,7 +86,7 @@ func newCompaction(opts *db.Options, cur *version, startLevel, baseLevel int) *c
 	// bytes, we want to adjust the range to [1,numLevels].
 	adjustedOutputLevel := 1 + outputLevel - baseLevel
 
-	return &compaction{
+	c := &compaction{
 		cmp:               opts.Comparer.Compare,
 		version:           cur,
 		startLevel:        startLevel,
@@ -102,70 +95,6 @@ func newCompaction(opts *db.Options, cur *version, startLevel, baseLevel int) *c
 		maxOverlapBytes:   maxGrandparentOverlapBytes(opts, adjustedOutputLevel),
 		maxExpandedBytes:  expandedCompactionByteSizeLimit(opts, adjustedOutputLevel),
 	}
-}
-
-func newFlush(opts *db.Options, cur *version, baseLevel int, flushing []flushable) *compaction {
-	c := &compaction{
-		cmp:               opts.Comparer.Compare,
-		version:           cur,
-		startLevel:        -1,
-		outputLevel:       0,
-		maxOutputFileSize: math.MaxUint64,
-		maxOverlapBytes:   math.MaxUint64,
-		maxExpandedBytes:  math.MaxUint64,
-		flushing:          flushing,
-	}
-
-	// TODO(peter): When we allow flushing to create multiple tables we'll want
-	// to choose sstable boundaries based on the grandparents. But for now we
-	// want to create a single table during flushing so this is all commented
-	// out.
-	if false {
-		c.maxOutputFileSize = uint64(opts.Level(0).TargetFileSize)
-		c.maxOverlapBytes = maxGrandparentOverlapBytes(opts, 0)
-		c.maxExpandedBytes = expandedCompactionByteSizeLimit(opts, 0)
-
-		var smallest db.InternalKey
-		var largest db.InternalKey
-		smallestSet, largestSet := false, false
-
-		updatePointBounds := func(iter internalIterator) {
-			if key, _ := iter.First(); key != nil {
-				if !smallestSet ||
-					db.InternalCompare(c.cmp, smallest, *key) > 0 {
-					smallestSet = true
-					smallest = key.Clone()
-				}
-			}
-			if key, _ := iter.Last(); key != nil {
-				if !largestSet ||
-					db.InternalCompare(c.cmp, largest, *key) < 0 {
-					largestSet = true
-					largest = key.Clone()
-				}
-			}
-		}
-
-		updateRangeBounds := func(iter internalIterator) {
-			if key, _ := iter.First(); key != nil {
-				if !smallestSet ||
-					db.InternalCompare(c.cmp, smallest, *key) > 0 {
-					smallestSet = true
-					smallest = key.Clone()
-				}
-			}
-		}
-
-		for i := range flushing {
-			f := flushing[i]
-			updatePointBounds(f.newIter(nil))
-			if rangeDelIter := f.newRangeDelIter(nil); rangeDelIter != nil {
-				updateRangeBounds(rangeDelIter)
-			}
-		}
-
-		c.grandparents = c.version.overlaps(baseLevel, c.cmp, smallest.UserKey, largest.UserKey)
-	}
 	return c
 }
 
@@ -256,21 +185,6 @@ func (c *compaction) grow(sm, la db.InternalKey) bool {
 	return true
 }
 
-func (c *compaction) trivialMove() bool {
-	if len(c.flushing) != 0 {
-		return false
-	}
-	// Check for a trivial move of one table from one level to the next. We avoid
-	// such a move if there is lots of overlapping grandparent data. Otherwise,
-	// the move could create a parent file that will require a very expensive
-	// merge later on.
-	if len(c.inputs[0]) == 1 && len(c.inputs[1]) == 0 &&
-		totalSize(c.grandparents) <= c.maxOverlapBytes {
-		return true
-	}
-	return false
-}
-
 // shouldStopBefore returns true if the output to the current table should be
 // finished and a new table started before adding the specified key. This is
 // done in order to prevent a table at level N from overlapping too much data
@@ -309,23 +223,7 @@ func (c *compaction) shouldStopBefore(key db.InternalKey) bool {
 // 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(iter internalIterator) bool {
-	if len(c.flushing) != 0 {
-		if len(c.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, _ := iter.First()
-		upper, _ := iter.Last()
-		if lower == nil || upper == nil {
-			return false
-		}
-		return c.elideRangeTombstone(lower.UserKey, upper.UserKey)
-	}
-
+func (c *compaction) allowZeroSeqNum() bool {
 	var lower, upper []byte
 	for i := range c.inputs {
 		files := c.inputs[i]
@@ -348,21 +246,10 @@ func (c *compaction) allowZeroSeqNum(iter internalIterator) bool {
 // 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.
 func (c *compaction) elideTombstone(key []byte) bool {
-	if len(c.flushing) != 0 {
-		return false
-	}
-
-	level := c.outputLevel + 1
-	if c.outputLevel == 0 {
-		// Level 0 can contain overlapping sstables so we need to check it for
-		// overlaps.
-		level = 0
-	}
-
-	// TODO(peter): this can be faster if key is always increasing between
+	// TODO(peter): this can be faster if ukey is always increasing between
 	// successive elideTombstones calls and we can keep some state in between
 	// calls.
-	for ; level < numLevels; level++ {
+	for level := c.outputLevel + 1; level < numLevels; level++ {
 		for _, f := range c.version.files[level] {
 			if c.cmp(key, f.largest.UserKey) <= 0 {
 				if c.cmp(key, f.smallest.UserKey) >= 0 {
@@ -382,18 +269,7 @@ func (c *compaction) elideTombstone(key []byte) bool {
 // pairs at c.outputLevel+1 or higher that possibly overlap the specified
 // tombstone.
 func (c *compaction) elideRangeTombstone(start, end []byte) bool {
-	if c.disableRangeTombstoneElision {
-		return false
-	}
-
-	level := c.outputLevel + 1
-	if c.outputLevel == 0 {
-		// Level 0 can contain overlapping sstables so we need to check it for
-		// overlaps.
-		level = 0
-	}
-
-	for ; level < numLevels; level++ {
+	for level := c.outputLevel + 1; level < numLevels; level++ {
 		overlaps := c.version.overlaps(level, c.cmp, start, end)
 		if len(overlaps) > 0 {
 			return false
@@ -457,27 +333,6 @@ func (c *compaction) atomicUnitBounds(f *fileMetadata) (lower, upper []byte) {
 func (c *compaction) newInputIter(
 	newIters tableNewIters,
 ) (_ internalIterator, retErr error) {
-	if len(c.flushing) != 0 {
-		if len(c.flushing) == 1 {
-			f := c.flushing[0]
-			iter := f.newIter(nil)
-			if rangeDelIter := f.newRangeDelIter(nil); rangeDelIter != nil {
-				return newMergingIter(c.cmp, iter, rangeDelIter), nil
-			}
-			return iter, nil
-		}
-		iters := make([]internalIterator, 0, 2*len(c.flushing))
-		for i := range c.flushing {
-			f := c.flushing[i]
-			iters = append(iters, f.newIter(nil))
-			rangeDelIter := f.newRangeDelIter(nil)
-			if rangeDelIter != nil {
-				iters = append(iters, rangeDelIter)
-			}
-		}
-		return newMergingIter(c.cmp, iters...), nil
-	}
-
 	iters := make([]internalIterator, 0, 2*len(c.inputs[0])+1)
 	defer func() {
 		if retErr != nil {
@@ -545,10 +400,6 @@ func (c *compaction) newInputIter(
 }
 
 func (c *compaction) String() string {
-	if len(c.flushing) != 0 {
-		return "flush\n"
-	}
-
 	var buf bytes.Buffer
 	for i := range c.inputs {
 		level := c.startLevel
@@ -594,10 +445,8 @@ func (d *DB) flush() {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 	if err := d.flush1(); err != nil {
-		// TODO(peter): count consecutive flush errors and backoff.
-		if d.opts.EventListener.BackgroundError != nil {
-			d.opts.EventListener.BackgroundError(err)
-		}
+		// TODO(peter): count consecutive compaction errors and backoff.
+		_ = err
 	}
 	d.mu.compact.flushing = false
 	// More flush work may have arrived while we were flushing, so schedule
@@ -626,8 +475,28 @@ func (d *DB) flush1() error {
 		return nil
 	}
 
-	c := newFlush(d.opts, d.mu.versions.currentVersion(),
-		d.mu.versions.picker.baseLevel, d.mu.mem.queue[:n])
+	// TODO(peter,rangedel): test that range tombstones are properly included in
+	// the output sstable. Should propbably pull out the code below into a method
+	// that can be separately tested.
+	var iter internalIterator
+	if n == 1 {
+		mem := d.mu.mem.queue[0]
+		iter = mem.newIter(nil)
+		if rangeDelIter := mem.newRangeDelIter(nil); rangeDelIter != nil {
+			iter = newMergingIter(d.cmp, iter, rangeDelIter)
+		}
+	} else {
+		iters := make([]internalIterator, 0, 2*n)
+		for i := 0; i < n; i++ {
+			mem := d.mu.mem.queue[i]
+			iters = append(iters, mem.newIter(nil))
+			rangeDelIter := mem.newRangeDelIter(nil)
+			if rangeDelIter != nil {
+				iters = append(iters, rangeDelIter)
+			}
+		}
+		iter = newMergingIter(d.cmp, iters...)
+	}
 
 	jobID := d.mu.nextJobID
 	d.mu.nextJobID++
@@ -637,7 +506,7 @@ func (d *DB) flush1() error {
 		})
 	}
 
-	ve, pendingOutputs, err := d.runCompaction(c)
+	meta, err := d.writeLevel0Table(iter, true /* allowRangeTombstoneElision */)
 
 	if d.opts.EventListener.FlushEnd != nil {
 		info := db.FlushInfo{
@@ -645,36 +514,42 @@ func (d *DB) flush1() error {
 			Err:   err,
 		}
 		if err == nil {
-			for i := range ve.newFiles {
-				e := &ve.newFiles[i]
-				info.Output = append(info.Output, e.meta.tableInfo(d.dirname))
-			}
-			if len(ve.newFiles) == 0 {
-				info.Err = errEmptyTable
-			}
+			info.Output = meta.tableInfo(d.dirname)
 		}
 		d.opts.EventListener.FlushEnd(info)
 	}
 
-	if err != nil {
+	if err != nil && err != errEmptyTable {
 		return err
 	}
 
 	// The flush succeeded or it produced an empty sstable. In either case we
 	// want to bump the log number.
+	metrics := &LevelMetrics{}
+	ve := &versionEdit{
+		metrics: map[int]*LevelMetrics{
+			0: metrics,
+		},
+	}
 	ve.logNumber, _ = d.mu.mem.queue[n].logInfo()
-	metrics := ve.metrics[0]
 	for i := 0; i < n; i++ {
 		_, size := d.mu.mem.queue[i].logInfo()
 		metrics.BytesIn += size
 	}
+	if err != errEmptyTable {
+		ve.newFiles = []newFileEntry{
+			{level: 0, meta: meta},
+		}
+		metrics.BytesWritten = meta.size
+	}
 
 	err = d.mu.versions.logAndApply(jobID, ve, d.dataDir)
-	for _, fileNum := range pendingOutputs {
-		if _, ok := d.mu.compact.pendingOutputs[fileNum]; !ok {
+	for i := range ve.newFiles {
+		f := &ve.newFiles[i]
+		if _, ok := d.mu.compact.pendingOutputs[f.meta.fileNum]; !ok {
 			panic("pebble: expected pending output not present")
 		}
-		delete(d.mu.compact.pendingOutputs, fileNum)
+		delete(d.mu.compact.pendingOutputs, f.meta.fileNum)
 	}
 	if err != nil {
 		return err
@@ -697,6 +572,147 @@ func (d *DB) flush1() error {
 	return nil
 }
 
+// writeLevel0Table writes a memtable to a level-0 on-disk table.
+//
+// If no error is returned, it adds the file number of that on-disk table to
+// d.pendingOutputs. It is the caller's responsibility to remove that fileNum
+// from that set when it has been applied to d.mu.versions.
+//
+// d.mu must be held when calling this, but the mutex may be dropped and
+// re-acquired during the course of this method.
+func (d *DB) writeLevel0Table(
+	iiter internalIterator, allowRangeTombstoneElision bool,
+) (meta fileMetadata, err error) {
+	meta.fileNum = d.mu.versions.nextFileNum()
+	filename := dbFilename(d.dirname, fileTypeTable, meta.fileNum)
+	d.mu.compact.pendingOutputs[meta.fileNum] = struct{}{}
+	defer func(fileNum uint64) {
+		if err != nil {
+			delete(d.mu.compact.pendingOutputs, fileNum)
+		}
+	}(meta.fileNum)
+
+	snapshots := d.mu.snapshots.toSlice()
+	version := d.mu.versions.currentVersion()
+
+	// Release the d.mu lock while doing I/O.
+	// Note the unusual order: Unlock and then Lock.
+	d.mu.Unlock()
+	defer d.mu.Lock()
+
+	elideRangeTombstone := func(start, end []byte) bool {
+		if !allowRangeTombstoneElision {
+			return false
+		}
+		for level := 0; level < numLevels; level++ {
+			overlaps := version.overlaps(level, d.cmp, start, end)
+			if len(overlaps) > 0 {
+				return false
+			}
+		}
+		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,
+	)
+	var (
+		file vfs.File
+		tw   *sstable.Writer
+	)
+	defer func() {
+		if iter != nil {
+			err = firstError(err, iter.Close())
+		}
+		if tw != nil {
+			err = firstError(err, tw.Close())
+		}
+		if err != nil {
+			d.opts.FS.Remove(filename)
+			meta = fileMetadata{}
+		}
+	}()
+
+	file, err = d.opts.FS.Create(filename)
+	if err != nil {
+		return fileMetadata{}, err
+	}
+	tw = sstable.NewWriter(file, d.opts, d.opts.Level(0))
+
+	var count int
+	for key, val := iter.First(); key != nil; key, val = iter.Next() {
+		if err1 := tw.Add(*key, val); err1 != nil {
+			return fileMetadata{}, err1
+		}
+		count++
+	}
+
+	for _, v := range iter.Tombstones(nil) {
+		if err1 := tw.Add(v.Start, v.End); err1 != nil {
+			return fileMetadata{}, err1
+		}
+		count++
+	}
+
+	if err1 := iter.Close(); err1 != nil {
+		iter = nil
+		return fileMetadata{}, err1
+	}
+	iter = nil
+
+	if err1 := tw.Close(); err1 != nil {
+		tw = nil
+		return fileMetadata{}, err1
+	}
+
+	if count == 0 {
+		// The flush may have produced an empty table if a range tombstone deleted
+		// all the entries in the table and the range tombstone could be elided.
+		return fileMetadata{}, errEmptyTable
+	}
+
+	if err := d.dataDir.Sync(); err != nil {
+		return fileMetadata{}, err
+	}
+
+	writerMeta, err := tw.Metadata()
+	if err != nil {
+		return fileMetadata{}, err
+	}
+	meta.size = writerMeta.Size
+	meta.smallest = writerMeta.Smallest(d.cmp)
+	meta.largest = writerMeta.Largest(d.cmp)
+	meta.smallestSeqNum = writerMeta.SmallestSeqNum
+	meta.largestSeqNum = writerMeta.LargestSeqNum
+	tw = nil
+
+	return meta, nil
+}
+
 // maybeScheduleCompaction schedules a compaction if necessary.
 //
 // d.mu must be held when calling this.
@@ -776,7 +792,7 @@ func (d *DB) compact1() (err error) {
 		d.opts.EventListener.CompactionBegin(info)
 	}
 
-	ve, pendingOutputs, err := d.runCompaction(c)
+	ve, pendingOutputs, err := d.compactDiskTables(c)
 
 	if d.opts.EventListener.CompactionEnd != nil {
 		info.Err = err
@@ -807,19 +823,18 @@ func (d *DB) compact1() (err error) {
 	return nil
 }
 
-// runCompactions runs a compaction that produces new on-disk tables from
-// memtables or old on-disk tables.
+// compactDiskTables runs a compaction that produces new on-disk tables from
+// old on-disk tables.
 //
 // d.mu must be held when calling this, but the mutex may be dropped and
 // re-acquired during the course of this method.
-func (d *DB) runCompaction(c *compaction) (
-	ve *versionEdit, pendingOutputs []uint64, retErr error,
-) {
+func (d *DB) compactDiskTables(c *compaction) (ve *versionEdit, pendingOutputs []uint64, retErr error) {
 	// Check for a trivial move of one table from one level to the next. We avoid
 	// such a move if there is lots of overlapping grandparent data. Otherwise,
 	// the move could create a parent file that will require a very expensive
 	// merge later on.
-	if c.trivialMove() {
+	if len(c.inputs[0]) == 1 && len(c.inputs[1]) == 0 &&
+		totalSize(c.grandparents) <= maxGrandparentOverlapBytes(d.opts, c.outputLevel) {
 		meta := &c.inputs[0][0]
 		return &versionEdit{
 			deletedFiles: map[deletedFileEntry]bool{
@@ -852,12 +867,13 @@ func (d *DB) runCompaction(c *compaction) (
 	d.mu.Unlock()
 	defer d.mu.Lock()
 
+	c.cmp = d.cmp
 	iiter, err := c.newInputIter(d.newIters)
 	if err != nil {
 		return nil, pendingOutputs, err
 	}
-	iter := newCompactionIter(c.cmp, d.merge, iiter, snapshots,
-		c.allowZeroSeqNum(iiter), c.elideTombstone, c.elideRangeTombstone)
+	iter := newCompactionIter(d.cmp, d.merge, iiter, snapshots,
+		c.allowZeroSeqNum(), c.elideTombstone, c.elideRangeTombstone)
 
 	var (
 		filenames []string
@@ -902,9 +918,6 @@ func (d *DB) runCompaction(c *compaction) (
 		if err != nil {
 			return err
 		}
-		file = vfs.NewSyncingFile(file, vfs.SyncingFileOptions{
-			BytesPerSync: d.opts.BytesPerSync,
-		})
 		filenames = append(filenames, filename)
 		tw = sstable.NewWriter(file, d.opts, d.opts.Level(c.outputLevel))
 
@@ -918,24 +931,19 @@ func (d *DB) runCompaction(c *compaction) (
 	}
 
 	finishOutput := func(key db.InternalKey) error {
+		if tw == nil {
+			return nil
+		}
+
 		// NB: clone the key because the data can be held on to by the call to
 		// compactionIter.Tombstones via rangedel.Fragmenter.FlushTo.
 		key = key.Clone()
 		for _, v := range iter.Tombstones(key.UserKey) {
-			if tw == nil {
-				if err := newOutput(); err != nil {
-					return err
-				}
-			}
 			if err := tw.Add(v.Start, v.End); err != nil {
 				return err
 			}
 		}
 
-		if tw == nil {
-			return nil
-		}
-
 		if err := tw.Close(); err != nil {
 			tw = nil
 			return err
@@ -986,6 +994,7 @@ func (d *DB) runCompaction(c *compaction) (
 
 		meta.smallest = writerMeta.Smallest(d.cmp)
 		meta.largest = writerMeta.Largest(d.cmp)
+
 		return nil
 	}
 
diff --git a/db.go b/db.go
index 70c7686..36e4b36 100644
--- a/db.go
+++ b/db.go
@@ -135,6 +135,7 @@ type DB struct {
 	cmp            db.Compare
 	equal          db.Equal
 	merge          db.Merge
+	split          db.Split
 	abbreviatedKey db.AbbreviatedKey
 
 	dataDir vfs.File
@@ -254,6 +255,7 @@ func (d *DB) getInternal(key []byte, b *Batch, s *Snapshot) ([]byte, error) {
 	i.cmp = d.cmp
 	i.equal = d.equal
 	i.merge = d.merge
+	i.split = d.split
 	i.iter = get
 	i.readState = readState
 
@@ -452,6 +454,7 @@ func (d *DB) newIterInternal(
 	dbi.cmp = d.cmp
 	dbi.equal = d.equal
 	dbi.merge = d.merge
+	dbi.split = d.split
 	dbi.readState = readState
 
 	iters := buf.iters[:0]
diff --git a/db/comparer.go b/db/comparer.go
index d1e02f8..8502bc2 100644
--- a/db/comparer.go
+++ b/db/comparer.go
@@ -80,9 +80,10 @@ type Successor func(dst, a []byte) []byte
 // The returned prefix must have the following properties (where a and b are
 // feasible):
 //
-// 1) Compare(prefix(a), a) <= 0,
-// 2) If Compare(a, b) <= 0, then Compare(prefix(a), prefix(b)) <= 0
-// 3) if b begins with a, then prefix(b) = prefix(a).
+// 1) bytes.HasPrefix(a, prefix(a))
+// 2) Compare(prefix(a), a) <= 0,
+// 3) If Compare(a, b) <= 0, then Compare(prefix(a), prefix(b)) <= 0
+// 4) if b begins with a, then prefix(b) = prefix(a).
 type Split func(a []byte) int
 
 // Comparer defines a total ordering over the space of []byte keys: a 'less
diff --git a/db/event.go b/db/event.go
index e6b8533..60a5f6c 100644
--- a/db/event.go
+++ b/db/event.go
@@ -91,7 +91,7 @@ type FlushInfo struct {
 	Reason string
 	// Output contains the ouptut table generated by the flush. The output info
 	// is empty for the flush begin event.
-	Output []TableInfo
+	Output TableInfo
 	Err    error
 }
 
@@ -100,12 +100,12 @@ func (i FlushInfo) String() string {
 		return fmt.Sprintf("[JOB %d] flush error: %s", i.JobID, i.Err)
 	}
 
-	if len(i.Output) == 0 {
+	if i.Output.FileNum == 0 {
 		return fmt.Sprintf("[JOB %d] flushing to L0", i.JobID)
 	}
 
-	return fmt.Sprintf("[JOB %d] flushed to L0: %d (%s)", i.JobID,
-		len(i.Output), humanize.Uint64(totalSize(i.Output)))
+	return fmt.Sprintf("[JOB %d] flushed to L0 (%s)", i.JobID,
+		humanize.Uint64(i.Output.Size))
 }
 
 // ManifestCreateInfo contains info about a manifest creation event.
diff --git a/db/options.go b/db/options.go
index 4f908cb..59b1514 100644
--- a/db/options.go
+++ b/db/options.go
@@ -483,20 +483,6 @@ type IterOptions struct {
 	//
 	// TODO(peter): unimplemented.
 	// TableFilter func(userProps map[string]string) bool
-
-	// If Prefix is true, the iterator will only be used to iterate over keys
-	// matching that of the key it is first positioned at. If the Comparer was
-	// supplied with a user-defined Split function and bloom filters are
-	// enabled, this allows for improved performance by skipping SSTables known
-	// not to contain the given prefix. The iterator will not properly observe
-	// keys not matching the prefix.
-	//
-	// TODO(tbg): should an assertion trip if the first key's prefix is unstable?
-	// TODO(tbg): should Prefix override (or sharpen) {Lower,Upper}Bound? When
-	// we see the first key, we get the prefix and a separator which should be
-	// a good {Lower,Upper}Bound.
-	// TODO(tbg): unimplemented.
-	// Prefix bool
 }
 
 // GetLowerBound returns the LowerBound or nil if the receiver is nil.
diff --git a/error_iter.go b/error_iter.go
index e7d6d06..1eb5617 100644
--- a/error_iter.go
+++ b/error_iter.go
@@ -20,6 +20,10 @@ func (c *errorIter) SeekGE(key []byte) (*db.InternalKey, []byte) {
 	return nil, nil
 }
 
+func (c *errorIter) SeekPrefixGE(prefix, key []byte) (*db.InternalKey, []byte) {
+	return nil, nil
+}
+
 func (c *errorIter) SeekLT(key []byte) (*db.InternalKey, []byte) {
 	return nil, nil
 }
diff --git a/get_iter.go b/get_iter.go
index 62cb5f1..5b4d66a 100644
--- a/get_iter.go
+++ b/get_iter.go
@@ -40,6 +40,10 @@ func (g *getIter) SeekGE(key []byte) (*db.InternalKey, []byte) {
 	panic("pebble: SeekGE unimplemented")
 }
 
+func (g *getIter) SeekPrefixGE(prefix, key []byte) (*db.InternalKey, []byte) {
+	panic("pebble: SeekPrefixGE unimplemented")
+}
+
 func (g *getIter) SeekLT(key []byte) (*db.InternalKey, []byte) {
 	panic("pebble: SeekLT unimplemented")
 }
diff --git a/internal.go b/internal.go
index ce635f4..78dd6a4 100644
--- a/internal.go
+++ b/internal.go
@@ -32,6 +32,14 @@ type internalIterator interface {
 	// is pointing at a valid entry, and (nil, nil) otherwise.
 	SeekGE(key []byte) (*db.InternalKey, []byte)
 
+	// SeekPrefixGE moves the iterator to the first key/value pair whose key
+	// starts with the given prefix and is greater than or equal to the given
+	// key. Returns the key and value if the iterator is pointing at a valid
+	// entry, and (nil, nil) otherwise. Note that the iterator will still observe
+	// keys not matching the prefix. It is up to the user to check if the prefix
+	// matches, and iteration beyond the prefix is undefined.
+	SeekPrefixGE(prefix, key []byte) (*db.InternalKey, []byte)
+
 	// SeekLT moves the iterator to the last key/value pair whose key is less
 	// than the given key. Returns the key and value if the iterator is pointing
 	// at a valid entry, and (nil, nil) otherwise.
diff --git a/internal/arenaskl/iterator.go b/internal/arenaskl/iterator.go
index 5529ea8..7f1374c 100644
--- a/internal/arenaskl/iterator.go
+++ b/internal/arenaskl/iterator.go
@@ -84,6 +84,10 @@ func (it *Iterator) SeekGE(key []byte) (*db.InternalKey, []byte) {
 	return &it.key, it.Value()
 }
 
+func (it *Iterator) SeekPrefixGE(prefix, key []byte) (*db.InternalKey, []byte) {
+	return it.SeekGE(key)
+}
+
 // SeekLT moves the iterator to the last entry whose key is less than the given
 // key. Returns the key and value if the iterator is pointing at a valid entry,
 // and (nil, nil) otherwise. Note that SeekLT only checks the lower bound. It
diff --git a/internal/rangedel/iter.go b/internal/rangedel/iter.go
index ca96aa3..3a4fe06 100644
--- a/internal/rangedel/iter.go
+++ b/internal/rangedel/iter.go
@@ -52,6 +52,11 @@ func (i *Iter) SeekGE(key []byte) (*db.InternalKey, []byte) {
 	return &t.Start, t.End
 }
 
+func (i *Iter) SeekPrefixGE(prefix, key []byte) (*db.InternalKey, []byte) {
+	// This should never be called as prefix iteration is only done for point records.
+	panic("pebble: SeekPrefixGE unimplemented")
+}
+
 // SeekLT implements internalIterator.SeekLT, as documented in the pebble
 // package.
 func (i *Iter) SeekLT(key []byte) (*db.InternalKey, []byte) {
diff --git a/iterator.go b/iterator.go
index 97380d5..9414997 100644
--- a/iterator.go
+++ b/iterator.go
@@ -5,6 +5,8 @@
 package pebble
 
 import (
+	"bytes"
+	"errors"
 	"fmt"
 
 	"github.com/petermattis/pebble/db"
@@ -35,6 +37,7 @@ type Iterator struct {
 	cmp       db.Compare
 	equal     db.Equal
 	merge     db.Merge
+	split     db.Split
 	iter      internalIterator
 	readState *readState
 	err       error
@@ -48,6 +51,7 @@ type Iterator struct {
 	iterValue []byte
 	pos       iterPos
 	alloc     *iterAlloc
+	prefix    []byte
 }
 
 func (i *Iterator) findNextEntry() bool {
@@ -68,6 +72,9 @@ func (i *Iterator) findNextEntry() bool {
 			continue
 
 		case db.InternalKeyKindSet:
+			if i.prefix != nil && !bytes.HasPrefix(key.UserKey, i.prefix) {
+				return false
+			}
 			i.keyBuf = append(i.keyBuf[:0], key.UserKey...)
 			i.key = i.keyBuf
 			i.value = i.iterValue
@@ -75,6 +82,9 @@ func (i *Iterator) findNextEntry() bool {
 			return true
 
 		case db.InternalKeyKindMerge:
+			if i.prefix != nil && !bytes.HasPrefix(key.UserKey, i.prefix) {
+				return false
+			}
 			return i.mergeNext(key)
 
 		default:
@@ -133,6 +143,9 @@ func (i *Iterator) findPrevEntry() bool {
 			continue
 
 		case db.InternalKeyKindSet:
+			if i.prefix != nil && !bytes.HasPrefix(key.UserKey, i.prefix) {
+				return false
+			}
 			i.keyBuf = append(i.keyBuf[:0], key.UserKey...)
 			i.key = i.keyBuf
 			i.value = i.iterValue
@@ -142,6 +155,9 @@ func (i *Iterator) findPrevEntry() bool {
 
 		case db.InternalKeyKindMerge:
 			if !i.valid {
+				if i.prefix != nil && !bytes.HasPrefix(key.UserKey, i.prefix) {
+					return false
+				}
 				i.keyBuf = append(i.keyBuf[:0], key.UserKey...)
 				i.key = i.keyBuf
 				i.value = i.iterValue
@@ -251,6 +267,7 @@ func (i *Iterator) SeekGE(key []byte) bool {
 		return false
 	}
 
+	i.prefix = nil
 	if lowerBound := i.opts.GetLowerBound(); lowerBound != nil && i.cmp(key, lowerBound) < 0 {
 		key = lowerBound
 	}
@@ -259,6 +276,39 @@ func (i *Iterator) SeekGE(key []byte) bool {
 	return i.findNextEntry()
 }
 
+// SeekPrefixGE moves the iterator to the first key/value pair whose key is
+// greater than or equal to the given key and shares a common prefix with the
+// given key. Returns true if the iterator is pointing at a valid entry and
+// false otherwise. Note that a user-defined Split function must be supplied to
+// the Comparer. Also note that the iterator will not observe keys not matching
+// the prefix.
+func (i *Iterator) SeekPrefixGE(key []byte) bool {
+	if i.split == nil {
+		panic("pebble: split must be provided for SeekPrefixGE")
+	}
+
+	if i.err != nil {
+		return false
+	}
+
+	// Make a copy of the prefix so that modifications to the key after
+	// SeekPrefixGE returns does not affect the stored prefix.
+	prefixLen := i.split(key)
+	i.prefix = make([]byte, prefixLen)
+	copy(i.prefix, key[:prefixLen])
+
+	if lowerBound := i.opts.GetLowerBound(); lowerBound != nil && i.cmp(key, lowerBound) < 0 {
+		if !bytes.HasPrefix(lowerBound, i.prefix) {
+			i.err = errors.New("pebble: SeekPrefixGE supplied with key outside of lower bound")
+		} else {
+			key = lowerBound
+		}
+	}
+
+	i.iterKey, i.iterValue = i.iter.SeekPrefixGE(i.prefix, key)
+	return i.findNextEntry()
+}
+
 // SeekLT moves the iterator to the last key/value pair whose key is less than
 // the given key. Returns true if the iterator is pointing at a valid entry and
 // false otherwise.
@@ -267,6 +317,7 @@ func (i *Iterator) SeekLT(key []byte) bool {
 		return false
 	}
 
+	i.prefix = nil
 	if upperBound := i.opts.GetUpperBound(); upperBound != nil && i.cmp(key, upperBound) >= 0 {
 		key = upperBound
 	}
@@ -282,6 +333,7 @@ func (i *Iterator) First() bool {
 		return false
 	}
 
+	i.prefix = nil
 	if lowerBound := i.opts.GetLowerBound(); lowerBound != nil {
 		i.iterKey, i.iterValue = i.iter.SeekGE(lowerBound)
 	} else {
@@ -297,6 +349,7 @@ func (i *Iterator) Last() bool {
 		return false
 	}
 
+	i.prefix = nil
 	if upperBound := i.opts.GetUpperBound(); upperBound != nil {
 		i.iterKey, i.iterValue = i.iter.SeekLT(upperBound)
 	} else {
diff --git a/level_iter.go b/level_iter.go
index 2799994..cf8944e 100644
--- a/level_iter.go
+++ b/level_iter.go
@@ -206,6 +206,18 @@ func (l *levelIter) SeekGE(key []byte) (*db.InternalKey, []byte) {
 	return l.skipEmptyFileForward()
 }
 
+func (l *levelIter) SeekPrefixGE(prefix, key []byte) (*db.InternalKey, []byte) {
+	// NB: the top-level Iterator has already adjusted key based on
+	// IterOptions.LowerBound.
+	if !l.loadFile(l.findFileGE(key), 1) {
+		return nil, nil
+	}
+	if key, val := l.iter.SeekPrefixGE(prefix, key); key != nil {
+		return key, val
+	}
+	return l.skipEmptyFileForward()
+}
+
 func (l *levelIter) SeekLT(key []byte) (*db.InternalKey, []byte) {
 	// NB: the top-level Iterator has already adjusted key based on
 	// IterOptions.UpperBound.
diff --git a/merging_iter.go b/merging_iter.go
index 1bb6f41..5b72f77 100644
--- a/merging_iter.go
+++ b/merging_iter.go
@@ -183,6 +183,7 @@ type mergingIter struct {
 	largestUserKeys [][]byte
 	heap            mergingIterHeap
 	err             error
+	prefix          []byte
 }
 
 // mergingIter implements the internalIterator interface.
@@ -512,7 +513,11 @@ func (m *mergingIter) seekGE(key []byte, level int) {
 
 	for ; level < len(m.iters); level++ {
 		iter := m.iters[level]
-		iter.SeekGE(key)
+		if m.prefix != nil {
+			iter.SeekPrefixGE(m.prefix, key)
+		} else {
+			iter.SeekGE(key)
+		}
 
 		if m.rangeDelIters != nil {
 			if rangeDelIter := m.rangeDelIters[level]; rangeDelIter != nil {
@@ -535,6 +540,13 @@ func (m *mergingIter) seekGE(key []byte, level int) {
 }
 
 func (m *mergingIter) SeekGE(key []byte) (*db.InternalKey, []byte) {
+	m.prefix = nil
+	m.seekGE(key, 0 /* start level */)
+	return m.findNextEntry()
+}
+
+func (m *mergingIter) SeekPrefixGE(prefix, key []byte) (*db.InternalKey, []byte) {
+	m.prefix = prefix
 	m.seekGE(key, 0 /* start level */)
 	return m.findNextEntry()
 }
@@ -542,7 +554,7 @@ func (m *mergingIter) SeekGE(key []byte) (*db.InternalKey, []byte) {
 func (m *mergingIter) seekLT(key []byte, level int) {
 	// See the comment in seekLT regarding using tombstones to adjust the seek
 	// target per level.
-
+	m.prefix = nil
 	for ; level < len(m.iters); level++ {
 		m.iters[level].SeekLT(key)
 
@@ -562,11 +574,13 @@ func (m *mergingIter) seekLT(key []byte, level int) {
 }
 
 func (m *mergingIter) SeekLT(key []byte) (*db.InternalKey, []byte) {
+	m.prefix = nil
 	m.seekLT(key, 0 /* start level */)
 	return m.findPrevEntry()
 }
 
 func (m *mergingIter) First() (*db.InternalKey, []byte) {
+	m.prefix = nil
 	m.heap.items = m.heap.items[:0]
 	for _, t := range m.iters {
 		// TODO(peter): save key and value so we don't have to access t.Key() and
@@ -578,6 +592,7 @@ func (m *mergingIter) First() (*db.InternalKey, []byte) {
 }
 
 func (m *mergingIter) Last() (*db.InternalKey, []byte) {
+	m.prefix = nil
 	for _, t := range m.iters {
 		// TODO(peter): save key and value so we don't have to access t.Key() and
 		// t.Value() in initHeap().
diff --git a/open.go b/open.go
index dab675f..5bca547 100644
--- a/open.go
+++ b/open.go
@@ -63,6 +63,7 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 		cmp:            opts.Comparer.Compare,
 		equal:          opts.Comparer.Equal,
 		merge:          opts.Merger.Merge,
+		split:          opts.Comparer.Split,
 		abbreviatedKey: opts.Comparer.AbbreviatedKey,
 		logRecycler:    logRecycler{limit: opts.MemTableStopWritesThreshold + 1},
 	}
@@ -311,20 +312,16 @@ func (d *DB) replayWAL(
 	}
 
 	if mem != nil && !mem.empty() {
-		c := newFlush(d.opts, d.mu.versions.currentVersion(),
-			1 /* base level */, []flushable{mem})
-		newVE, pendingOutputs, err := d.runCompaction(c)
+		meta, err := d.writeLevel0Table(mem.newIter(nil),
+			true /* allowRangeTombstoneElision */)
 		if err != nil {
 			return 0, err
 		}
-		ve.newFiles = append(ve.newFiles, newVE.newFiles...)
-		// Strictly speaking, it's too early to delete from d.pendingOutputs, but
-		// we are replaying the log file, which happens before Open returns, so
-		// there is no possibility of deleteObsoleteFiles being called concurrently
-		// here.
-		for _, fileNum := range pendingOutputs {
-			delete(d.mu.compact.pendingOutputs, fileNum)
-		}
+		ve.newFiles = append(ve.newFiles, newFileEntry{level: 0, meta: meta})
+		// Strictly speaking, it's too early to delete meta.fileNum from d.pendingOutputs,
+		// but we are replaying the log file, which happens before Open returns, so there
+		// is no possibility of deleteObsoleteFiles being called concurrently here.
+		delete(d.mu.compact.pendingOutputs, meta.fileNum)
 	}
 
 	return maxSeqNum, nil
diff --git a/sstable/block.go b/sstable/block.go
index f1dc04f..37b52f7 100644
--- a/sstable/block.go
+++ b/sstable/block.go
@@ -354,6 +354,13 @@ func (i *blockIter) SeekGE(key []byte) (*db.InternalKey, []byte) {
 	return nil, nil
 }
 
+// SeekPrefixGE implements internalIterator.SeekPrefixGE, as documented in the
+// pebble package.
+func (i *blockIter) SeekPrefixGE(prefix, key []byte) (*db.InternalKey, []byte) {
+	// This should never be called as prefix iteration is handled by sstable.Iterator.
+	panic("pebble: SeekPrefixGE unimplemented")
+}
+
 // SeekLT implements internalIterator.SeekLT, as documented in the pebble
 // package.
 func (i *blockIter) SeekLT(key []byte) (*db.InternalKey, []byte) {
diff --git a/sstable/raw_block.go b/sstable/raw_block.go
index 8e1b21c..db05bc6 100644
--- a/sstable/raw_block.go
+++ b/sstable/raw_block.go
@@ -142,6 +142,13 @@ func (i *rawBlockIter) SeekGE(key []byte) bool {
 	return i.Valid()
 }
 
+// SeekPrefixGE implements internalIterator.SeekPrefixGE, as documented in the
+// pebble package.
+func (i *rawBlockIter) SeekPrefixGE(key []byte) bool {
+	// This should never be called as prefix iteration is never used with raw blocks.
+	panic("pebble: SeekPrefixGE unimplemented")
+}
+
 // SeekLT implements internalIterator.SeekLT, as documented in the pebble
 // package.
 func (i *rawBlockIter) SeekLT(key []byte) bool {
diff --git a/sstable/reader.go b/sstable/reader.go
index 7bd315a..41b8e1a 100644
--- a/sstable/reader.go
+++ b/sstable/reader.go
@@ -200,7 +200,44 @@ func (i *Iterator) SeekGE(key []byte) (*db.InternalKey, []byte) {
 		return nil, nil
 	}
 	if i.blockUpper != nil && i.cmp(ikey.UserKey, i.blockUpper) >= 0 {
-		i.data.offset = i.data.restarts
+		i.data.invalidateUpper() // force i.data.Valid() to return false
+		return nil, nil
+	}
+	return ikey, val
+}
+
+// SeekPrefixGE implements internalIterator.SeekPrefixGE, as documented in the
+// pebble package. Note that SeekPrefixGE only checks the upper bound. It is up
+// to the caller to ensure that key is greater than or equal to the lower bound.
+func (i *Iterator) SeekPrefixGE(prefix, key []byte) (*db.InternalKey, []byte) {
+	if i.err != nil {
+		return nil, nil
+	}
+
+	// Check prefix bloom filter.
+	if i.reader.tableFilter != nil {
+		data, err := i.reader.readFilter()
+		if err != nil {
+			return nil, nil
+		}
+		if !i.reader.tableFilter.mayContain(data, prefix) {
+			i.data.invalidateUpper() // force i.data.Valid() to return false
+			return nil, nil
+		}
+	}
+
+	if ikey, _ := i.index.SeekGE(key); ikey == nil {
+		return nil, nil
+	}
+	if !i.loadBlock() {
+		return nil, nil
+	}
+	ikey, val := i.data.SeekGE(key)
+	if ikey == nil {
+		return nil, nil
+	}
+	if i.blockUpper != nil && i.cmp(ikey.UserKey, i.blockUpper) >= 0 {
+		i.data.invalidateUpper() // force i.data.Valid() to return false
 		return nil, nil
 	}
 	return ikey, val
@@ -295,7 +332,7 @@ func (i *Iterator) Last() (*db.InternalKey, []byte) {
 		return nil, nil
 	}
 	if i.blockLower != nil && i.cmp(i.data.ikey.UserKey, i.blockLower) < 0 {
-		i.data.offset = -1
+		i.data.invalidateLower()
 		return nil, nil
 	}
 	return &i.data.ikey, i.data.val
@@ -309,7 +346,7 @@ func (i *Iterator) Next() (*db.InternalKey, []byte) {
 	}
 	if key, val := i.data.Next(); key != nil {
 		if i.blockUpper != nil && i.cmp(key.UserKey, i.blockUpper) >= 0 {
-			i.data.offset = i.data.restarts
+			i.data.invalidateUpper()
 			return nil, nil
 		}
 		return key, val
@@ -328,7 +365,7 @@ func (i *Iterator) Next() (*db.InternalKey, []byte) {
 				return nil, nil
 			}
 			if i.blockUpper != nil && i.cmp(key.UserKey, i.blockUpper) >= 0 {
-				i.data.offset = i.data.restarts
+				i.data.invalidateUpper()
 				return nil, nil
 			}
 			return key, val
@@ -345,7 +382,7 @@ func (i *Iterator) Prev() (*db.InternalKey, []byte) {
 	}
 	if key, val := i.data.Prev(); key != nil {
 		if i.blockLower != nil && i.cmp(key.UserKey, i.blockLower) < 0 {
-			i.data.offset = -1
+			i.data.invalidateLower()
 			return nil, nil
 		}
 		return key, val
@@ -364,7 +401,7 @@ func (i *Iterator) Prev() (*db.InternalKey, []byte) {
 				return nil, nil
 			}
 			if i.blockLower != nil && i.cmp(key.UserKey, i.blockLower) < 0 {
-				i.data.offset = -1
+				i.data.invalidateLower()
 				return nil, nil
 			}
 			return key, val
diff --git a/sstable/writer.go b/sstable/writer.go
index e46f918..013126e 100644
--- a/sstable/writer.go
+++ b/sstable/writer.go
@@ -16,6 +16,7 @@ import (
 	"github.com/petermattis/pebble/db"
 	"github.com/petermattis/pebble/internal/crc"
 	"github.com/petermattis/pebble/internal/rangedel"
+	"github.com/petermattis/pebble/vfs"
 )
 
 // WriterMetadata holds info about a finished sstable.
@@ -72,21 +73,12 @@ func (m *WriterMetadata) Largest(cmp db.Compare) db.InternalKey {
 	return m.LargestRange
 }
 
-type flusher interface {
-	Flush() error
-}
-
-type writeCloseSyncer interface {
-	io.WriteCloser
-	Sync() error
-}
-
 // Writer is a table writer. It implements the DB interface, as documented
 // in the pebble/db package.
 type Writer struct {
 	writer    io.Writer
 	bufWriter *bufio.Writer
-	syncer    writeCloseSyncer
+	file      vfs.File
 	meta      WriterMetadata
 	err       error
 	// The following fields are copied from db.Options.
@@ -359,16 +351,12 @@ func (w *Writer) writeRawBlock(b []byte, compression db.Compression) (blockHandl
 	binary.LittleEndian.PutUint32(w.tmp[1:5], checksum)
 
 	// Write the bytes to the file.
-	n, err := w.writer.Write(b)
-	if err != nil {
+	if _, err := w.writer.Write(b); err != nil {
 		return blockHandle{}, err
 	}
-	w.meta.Size += uint64(n)
-	n, err = w.writer.Write(w.tmp[:5])
-	if err != nil {
+	if _, err := w.writer.Write(w.tmp[:5]); err != nil {
 		return blockHandle{}, err
 	}
-	w.meta.Size += uint64(n)
 	bh := blockHandle{w.offset, uint64(len(b))}
 	w.offset += uint64(len(b)) + blockTrailerLen
 
@@ -379,14 +367,14 @@ func (w *Writer) writeRawBlock(b []byte, compression db.Compression) (blockHandl
 // table was written to.
 func (w *Writer) Close() (err error) {
 	defer func() {
-		if w.syncer == nil {
+		if w.file == nil {
 			return
 		}
-		err1 := w.syncer.Close()
+		err1 := w.file.Close()
 		if err == nil {
 			err = err1
 		}
-		w.syncer = nil
+		w.file = nil
 	}()
 	if w.err != nil {
 		return w.err
@@ -491,12 +479,10 @@ func (w *Writer) Close() (err error) {
 		metaindexBH: metaindexBH,
 		indexBH:     indexBH,
 	}
-	var n int
-	if n, err = w.writer.Write(footer.encode(w.tmp[:])); err != nil {
+	if _, err := w.writer.Write(footer.encode(w.tmp[:])); err != nil {
 		w.err = err
 		return w.err
 	}
-	w.meta.Size += uint64(n)
 
 	// Flush the buffer.
 	if w.bufWriter != nil {
@@ -506,11 +492,24 @@ func (w *Writer) Close() (err error) {
 		}
 	}
 
-	if err := w.syncer.Sync(); err != nil {
+	if err := w.file.Sync(); err != nil {
 		w.err = err
 		return err
 	}
 
+	stat, err := w.file.Stat()
+	if err != nil {
+		w.err = err
+		return err
+	}
+
+	size := stat.Size()
+	if size < 0 {
+		w.err = fmt.Errorf("pebble: file has negative size %d", size)
+		return err
+	}
+	w.meta.Size = uint64(size)
+
 	// Make any future calls to Set or Close return an error.
 	w.err = errors.New("pebble: writer is closed")
 	return nil
@@ -525,7 +524,7 @@ func (w *Writer) EstimatedSize() uint64 {
 // Metadata returns the metadata for the finished sstable. Only valid to call
 // after the sstable has been finished.
 func (w *Writer) Metadata() (*WriterMetadata, error) {
-	if w.syncer != nil {
+	if w.file != nil {
 		return nil, errors.New("pebble: writer is not closed")
 	}
 	return &w.meta, nil
@@ -533,12 +532,18 @@ func (w *Writer) Metadata() (*WriterMetadata, error) {
 
 // NewWriter returns a new table writer for the file. Closing the writer will
 // close the file.
-func NewWriter(f writeCloseSyncer, o *db.Options, lo db.LevelOptions) *Writer {
+func NewWriter(f vfs.File, o *db.Options, lo db.LevelOptions) *Writer {
 	o = o.EnsureDefaults()
 	lo = *lo.EnsureDefaults()
 
+	if f != nil {
+		f = vfs.NewSyncingFile(f, vfs.SyncingFileOptions{
+			BytesPerSync: o.BytesPerSync,
+		})
+	}
+
 	w := &Writer{
-		syncer: f,
+		file: f,
 		meta: WriterMetadata{
 			SmallestSeqNum: math.MaxUint64,
 		},
@@ -589,6 +594,9 @@ func NewWriter(f writeCloseSyncer, o *db.Options, lo db.LevelOptions) *Writer {
 	w.props.Version = 2 // TODO(peter): what is this?
 
 	// If f does not have a Flush method, do our own buffering.
+	type flusher interface {
+		Flush() error
+	}
 	if _, ok := f.(flusher); ok {
 		w.writer = f
 	} else {


Test patch:
diff --git a/compaction_test.go b/compaction_test.go
index 189fe85..10901db 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -1070,7 +1070,7 @@ func TestCompactionAllowZeroSeqNum(t *testing.T) {
 						c.inputs[i] = append(c.inputs[i], meta)
 					}
 					c.outputLevel = c.startLevel + 1
-					fmt.Fprintf(&buf, "%t\n", c.allowZeroSeqNum(nil))
+					fmt.Fprintf(&buf, "%t\n", c.allowZeroSeqNum())
 				}
 				return buf.String()
 
diff --git a/data_test.go b/data_test.go
index b97b3d0..ae5731e 100644
--- a/data_test.go
+++ b/data_test.go
@@ -21,6 +21,10 @@ const (
 	iterCmdVerboseKey iterCmdOpt = iota
 )
 
+func checkValidPrefix(prefix, key []byte) bool {
+	return prefix == nil || bytes.HasPrefix(key, prefix)
+}
+
 func runIterCmd(d *datadriven.TestData, iter *Iterator) string {
 	var b bytes.Buffer
 	for _, line := range strings.Split(d.Input, "\n") {
@@ -35,6 +39,11 @@ func runIterCmd(d *datadriven.TestData, iter *Iterator) string {
 				return fmt.Sprintf("seek-ge <key>\n")
 			}
 			valid = iter.SeekGE([]byte(strings.TrimSpace(parts[1])))
+		case "seek-prefix-ge":
+			if len(parts) != 2 {
+				return fmt.Sprintf("seek-prefix-ge <key>\n")
+			}
+			valid = iter.SeekPrefixGE([]byte(strings.TrimSpace(parts[1])))
 		case "seek-lt":
 			if len(parts) != 2 {
 				return fmt.Sprintf("seek-lt <key>\n")
@@ -51,12 +60,12 @@ func runIterCmd(d *datadriven.TestData, iter *Iterator) string {
 		default:
 			return fmt.Sprintf("unknown op: %s", parts[0])
 		}
-		if valid != iter.Valid() {
+		if err := iter.Error(); err != nil {
+			fmt.Fprintf(&b, "err=%v\n", err)
+		} else if valid != iter.Valid() {
 			fmt.Fprintf(&b, "mismatched valid states: %t vs %t\n", valid, iter.Valid())
 		} else if valid {
 			fmt.Fprintf(&b, "%s:%s\n", iter.Key(), iter.Value())
-		} else if err := iter.Error(); err != nil {
-			fmt.Fprintf(&b, "err=%v\n", err)
 		} else {
 			fmt.Fprintf(&b, ".\n")
 		}
@@ -73,6 +82,7 @@ func runInternalIterCmd(d *datadriven.TestData, iter internalIterator, opts ...i
 	}
 
 	var b bytes.Buffer
+	var prefix []byte
 	for _, line := range strings.Split(d.Input, "\n") {
 		parts := strings.Fields(line)
 		if len(parts) == 0 {
@@ -83,15 +93,25 @@ func runInternalIterCmd(d *datadriven.TestData, iter internalIterator, opts ...i
 			if len(parts) != 2 {
 				return fmt.Sprintf("seek-ge <key>\n")
 			}
+			prefix = nil
 			iter.SeekGE([]byte(strings.TrimSpace(parts[1])))
+		case "seek-prefix-ge":
+			if len(parts) != 2 {
+				return fmt.Sprintf("seek-prefix-ge <key>\n")
+			}
+			prefix = []byte(strings.TrimSpace(parts[1]))
+			iter.SeekPrefixGE(prefix, prefix /* key */)
 		case "seek-lt":
 			if len(parts) != 2 {
 				return fmt.Sprintf("seek-lt <key>\n")
 			}
+			prefix = nil
 			iter.SeekLT([]byte(strings.TrimSpace(parts[1])))
 		case "first":
+			prefix = nil
 			iter.First()
 		case "last":
+			prefix = nil
 			iter.Last()
 		case "next":
 			iter.Next()
@@ -100,7 +120,7 @@ func runInternalIterCmd(d *datadriven.TestData, iter internalIterator, opts ...i
 		default:
 			return fmt.Sprintf("unknown op: %s", parts[0])
 		}
-		if iter.Valid() {
+		if iter.Valid() && checkValidPrefix(prefix, iter.Key().UserKey) {
 			if verboseKey {
 				fmt.Fprintf(&b, "%s:%s\n", iter.Key(), iter.Value())
 			} else {
@@ -243,19 +263,18 @@ func runDBDefineCmd(td *datadriven.TestData) (*DB, error) {
 			return nil
 		}
 
-		c := newFlush(d.opts, d.mu.versions.currentVersion(),
-			d.mu.versions.picker.baseLevel, []flushable{mem})
-		c.disableRangeTombstoneElision = true
-		newVE, _, err := d.runCompaction(c)
+		iter := mem.newIter(nil)
+		if rangeDelIter := mem.newRangeDelIter(nil); rangeDelIter != nil {
+			iter = newMergingIter(d.cmp, iter, rangeDelIter)
+		}
+		meta, err := d.writeLevel0Table(iter, false /* allowRangeTombstoneElision */)
 		if err != nil {
 			return nil
 		}
-		for _, f := range newVE.newFiles {
-			ve.newFiles = append(ve.newFiles, newFileEntry{
-				level: level,
-				meta:  f.meta,
-			})
-		}
+		ve.newFiles = append(ve.newFiles, newFileEntry{
+			level: level,
+			meta:  meta,
+		})
 		level = -1
 		return nil
 	}
diff --git a/internal/arenaskl/skl_test.go b/internal/arenaskl/skl_test.go
index b403251..83c662b 100644
--- a/internal/arenaskl/skl_test.go
+++ b/internal/arenaskl/skl_test.go
@@ -61,6 +61,10 @@ func (i *iterAdapter) SeekGE(key []byte) bool {
 	return i.verify(i.Iterator.SeekGE(key))
 }
 
+func (i *iterAdapter) SeekPrefixGE(prefix, key []byte) bool {
+	return i.verify(i.Iterator.SeekPrefixGE(prefix, key))
+}
+
 func (i *iterAdapter) SeekLT(key []byte) bool {
 	return i.verify(i.Iterator.SeekLT(key))
 }
diff --git a/internal_test.go b/internal_test.go
index 3fa29a9..47773de 100644
--- a/internal_test.go
+++ b/internal_test.go
@@ -39,6 +39,10 @@ func (i *internalIterAdapter) SeekGE(key []byte) bool {
 	return i.verify(i.internalIterator.SeekGE(key))
 }
 
+func (i *internalIterAdapter) SeekPrefixGE(prefix, key []byte) bool {
+	return i.verify(i.internalIterator.SeekPrefixGE(prefix, key))
+}
+
 func (i *internalIterAdapter) SeekLT(key []byte) bool {
 	return i.verify(i.internalIterator.SeekLT(key))
 }
diff --git a/iterator_test.go b/iterator_test.go
index 7dd03b2..21c2d81 100644
--- a/iterator_test.go
+++ b/iterator_test.go
@@ -77,6 +77,10 @@ func (f *fakeIter) SeekGE(key []byte) (*db.InternalKey, []byte) {
 	return nil, nil
 }
 
+func (f *fakeIter) SeekPrefixGE(prefix, key []byte) (*db.InternalKey, []byte) {
+	return f.SeekGE(key)
+}
+
 func (f *fakeIter) SeekLT(key []byte) (*db.InternalKey, []byte) {
 	f.valid = false
 	for f.index = len(f.keys) - 1; f.index >= 0; f.index-- {
@@ -295,6 +299,7 @@ func TestIterator(t *testing.T) {
 	newIter := func(seqNum uint64, opts *db.IterOptions) *Iterator {
 		cmp := db.DefaultComparer.Compare
 		equal := db.DefaultComparer.Equal
+		split := func(a []byte) int { return len(a) }
 		// NB: Use a mergingIter to filter entries newer than seqNum.
 		iter := newMergingIter(cmp, &fakeIter{
 			lower: opts.GetLowerBound(),
@@ -307,6 +312,7 @@ func TestIterator(t *testing.T) {
 			opts:  opts,
 			cmp:   cmp,
 			equal: equal,
+			split: split,
 			merge: db.DefaultMerger.Merge,
 			iter:  iter,
 		}
diff --git a/sstable/reader_test.go b/sstable/reader_test.go
index 2f5326e..fbc258f 100644
--- a/sstable/reader_test.go
+++ b/sstable/reader_test.go
@@ -49,6 +49,10 @@ func (i *iterAdapter) SeekGE(key []byte) bool {
 	return i.verify(i.Iterator.SeekGE(key))
 }
 
+func (i *iterAdapter) SeekPrefixGE(prefix, key []byte) bool {
+	return i.verify(i.Iterator.SeekPrefixGE(prefix, key))
+}
+
 func (i *iterAdapter) SeekLT(key []byte) bool {
 	return i.verify(i.Iterator.SeekLT(key))
 }
@@ -101,6 +105,11 @@ func TestReader(t *testing.T) {
 		},
 	}
 
+	testDirs := map[string]string {
+		"default": "testdata/reader",
+		"prefixFilter": "testdata/prefixreader",
+	}
+
 	for lName, levelOpt := range levelOpts {
 		for oName, opt := range opts {
 			levelOpt.EnsureDefaults()
@@ -109,13 +118,13 @@ func TestReader(t *testing.T) {
 			o.EnsureDefaults()
 
 			t.Run(fmt.Sprintf("opts=%s,levelOpts=%s", oName, lName), func(t *testing.T) {
-				runTestReader(t, o)
+				runTestReader(t, o, testDirs[oName])
 			})
 		}
 	}
 }
 
-func runTestReader(t *testing.T, o db.Options) {
+func runTestReader(t *testing.T, o db.Options, dir string) {
 	makeIkeyValue := func(s string) (db.InternalKey, []byte) {
 		j := strings.Index(s, ":")
 		k := strings.Index(s, "=")
@@ -129,7 +138,7 @@ func runTestReader(t *testing.T, o db.Options) {
 	mem := vfs.NewMem()
 	var r *Reader
 
-	datadriven.Walk(t, "testdata/reader", func(t *testing.T, path string) {
+	datadriven.Walk(t, dir, func(t *testing.T, path string) {
 		datadriven.RunTest(t, path, func(d *datadriven.TestData) string {
 			switch d.Cmd {
 			case "build":
@@ -179,6 +188,7 @@ func runTestReader(t *testing.T, o db.Options) {
 				}
 
 				var b bytes.Buffer
+				var prefix []byte
 				for _, line := range strings.Split(d.Input, "\n") {
 					parts := strings.Fields(line)
 					if len(parts) == 0 {
@@ -189,22 +199,32 @@ func runTestReader(t *testing.T, o db.Options) {
 						if len(parts) != 2 {
 							return fmt.Sprintf("seek-ge <key>\n")
 						}
+						prefix = nil
 						iter.SeekGE([]byte(strings.TrimSpace(parts[1])))
+					case "seek-prefix-ge":
+						if len(parts) != 2 {
+							return fmt.Sprintf("seek-prefix-ge <key>\n")
+						}
+						prefix = []byte(strings.TrimSpace(parts[1]))
+						iter.SeekPrefixGE(prefix, prefix /* key */)
 					case "seek-lt":
 						if len(parts) != 2 {
 							return fmt.Sprintf("seek-lt <key>\n")
 						}
+						prefix = nil
 						iter.SeekLT([]byte(strings.TrimSpace(parts[1])))
 					case "first":
+						prefix = nil
 						iter.First()
 					case "last":
+						prefix = nil
 						iter.Last()
 					case "next":
 						iter.Next()
 					case "prev":
 						iter.Prev()
 					}
-					if iter.Valid() {
+					if iter.Valid() && checkValidPrefix(prefix, iter.Key().UserKey){
 						fmt.Fprintf(&b, "<%s:%d>", iter.Key().UserKey, iter.Key().SeqNum())
 					} else if err := iter.Error(); err != nil {
 						fmt.Fprintf(&b, "<err=%v>", err)
@@ -233,6 +253,10 @@ func runTestReader(t *testing.T, o db.Options) {
 	})
 }
 
+func checkValidPrefix(prefix, key []byte) bool {
+	return prefix == nil || bytes.HasPrefix(key, prefix)
+}
+
 func buildBenchmarkTable(b *testing.B, blockSize, restartInterval int) (*Reader, [][]byte) {
 	mem := vfs.NewMem()
 	f0, err := mem.Create("bench")
diff --git a/sstable/testdata/prefixreader/bloom b/sstable/testdata/prefixreader/bloom
new file mode 100644
index 0000000..acd00e1
--- /dev/null
+++ b/sstable/testdata/prefixreader/bloom
@@ -0,0 +1,18 @@
+build
+a:1=A,aa:2=AA,aaa:3=AAA,aaaa:4=AAAA,b:5=B
+----
+
+get
+b
+ab
+a
+aaa
+aa
+aaaa
+----
+B
+<err: pebble: not found>
+A
+AAA
+AA
+AAAA
diff --git a/sstable/testdata/prefixreader/iter b/sstable/testdata/prefixreader/iter
new file mode 100644
index 0000000..952cfb6
--- /dev/null
+++ b/sstable/testdata/prefixreader/iter
@@ -0,0 +1,196 @@
+build
+a:1=A,aa:2=AA,c:3=C,d:4=D
+----
+
+iter
+seek-prefix-ge a
+next
+next
+----
+<a:1><aa:2>.
+
+iter
+seek-prefix-ge aa
+next
+----
+<aa:2>.
+
+iter
+seek-prefix-ge aa
+prev
+----
+<aa:2>.
+
+iter
+seek-prefix-ge c
+prev
+----
+<c:3>.
+
+iter
+seek-prefix-ge b
+----
+.
+
+iter
+seek-prefix-ge c
+next
+----
+<c:3>.
+
+iter
+seek-prefix-ge d
+next
+----
+<d:4>.
+
+iter
+seek-prefix-ge e
+----
+.
+
+iter
+seek-prefix-ge c
+seek-prefix-ge d
+seek-prefix-ge e
+----
+<c:3><d:4>.
+
+iter
+seek-prefix-ge c
+next
+seek-prefix-ge a
+next
+next
+----
+<c:3>.<a:1><aa:2>.
+
+iter
+seek-prefix-ge aa
+next
+seek-prefix-ge a
+next
+next
+seek-prefix-ge c
+next
+----
+<aa:2>.<a:1><aa:2>.<c:3>.
+
+iter
+seek-prefix-ge c
+next
+seek-prefix-ge aa
+next
+seek-prefix-ge a
+next
+next
+----
+<c:3>.<aa:2>.<a:1><aa:2>.
+
+iter
+seek-prefix-ge a
+next
+next
+----
+<a:1><aa:2>.
+
+iter
+seek-prefix-ge a
+next
+prev
+prev
+----
+<a:1><aa:2><a:1>.
+
+iter
+seek-prefix-ge a
+prev
+----
+<a:1>.
+
+iter
+seek-prefix-ge a
+seek-ge a
+next
+next
+next
+next
+----
+<a:1><a:1><aa:2><c:3><d:4>.
+
+iter
+seek-prefix-ge a
+seek-ge aa
+next
+next
+next
+----
+<a:1><aa:2><c:3><d:4>.
+
+iter
+seek-prefix-ge aa
+seek-ge c
+next
+next
+----
+<aa:2><c:3><d:4>.
+
+iter
+seek-prefix-ge aa
+seek-lt c
+next
+next
+next
+----
+<aa:2><aa:2><c:3><d:4>.
+
+iter
+seek-prefix-ge aa
+seek-lt c
+prev
+prev
+----
+<aa:2><aa:2><a:1>.
+
+iter
+seek-lt c
+seek-prefix-ge aa
+prev
+----
+<aa:2><aa:2>.
+
+iter
+seek-lt c
+seek-prefix-ge a
+next
+next
+
+----
+<aa:2><a:1><aa:2>.
+
+iter
+seek-ge aa
+seek-prefix-ge a
+next
+next
+
+----
+<aa:2><a:1><aa:2>.
+
+iter
+seek-prefix-ge 1
+----
+.
+
+get
+a
+aa
+f
+d
+c
+----
+A
+AA
+<err: pebble: not found>
+D
+C
diff --git a/testdata/event_listener b/testdata/event_listener
index cb768a3..ec4ade8 100644
--- a/testdata/event_listener
+++ b/testdata/event_listener
@@ -31,7 +31,7 @@ sync: wal/000002.log
 create: db/000006.sst
 sync: db/000006.sst
 sync: db
-[JOB 3] flushed to L0: 1 (959 B)
+[JOB 3] flushed to L0 (959 B)
 create: db/MANIFEST-000007
 sync: db/MANIFEST-000007
 create: db/CURRENT.000007.dbtmp
@@ -53,7 +53,7 @@ sync: wal/000005.log
 create: db/000009.sst
 sync: db/000009.sst
 sync: db
-[JOB 5] flushed to L0: 1 (959 B)
+[JOB 5] flushed to L0 (959 B)
 create: db/MANIFEST-000010
 sync: db/MANIFEST-000010
 create: db/CURRENT.000010.dbtmp
diff --git a/testdata/internal_iter_next b/testdata/internal_iter_next
index 46caf70..a66b8ed 100644
--- a/testdata/internal_iter_next
+++ b/testdata/internal_iter_next
@@ -16,6 +16,10 @@ seek-lt a
 seek-lt b
 seek-lt c
 seek-lt d
+seek-prefix-ge a
+seek-prefix-ge b
+seek-prefix-ge c
+seek-prefix-ge d
 ----
 a:2
 b:2
@@ -25,6 +29,10 @@ c:2
 a:1
 b:1
 c:1
+a:2
+b:2
+c:2
+.
 
 iter
 first
diff --git a/testdata/iterator b/testdata/iterator
index 42c07ff..367ad05 100644
--- a/testdata/iterator
+++ b/testdata/iterator
@@ -45,6 +45,24 @@ a:c
 .
 a:c
 
+iter seq=2
+seek-prefix-ge a
+next
+prev
+----
+a:b
+.
+a:b
+
+iter seq=3
+seek-prefix-ge a
+next
+prev
+----
+a:c
+.
+a:c
+
 
 define
 a.DEL.2:
@@ -77,6 +95,15 @@ a:b
 .
 a:b
 
+iter seq=3
+seek-prefix-ge a
+----
+.
+
+iter seq=2
+seek-prefix-ge 1
+----
+.
 
 define
 a.DEL.2:
@@ -101,6 +128,21 @@ seek-ge a
 ----
 a:b
 
+iter seq=4
+seek-prefix-ge a
+----
+.
+
+iter seq=3
+seek-prefix-ge a
+----
+.
+
+iter seq=2
+seek-prefix-ge a
+----
+a:b
+
 
 define
 a.SET.1:a
@@ -170,6 +212,35 @@ a:a
 .
 a:a
 
+iter seq=4
+seek-prefix-ge a
+next
+----
+a:a
+.
+
+iter seq=4
+seek-prefix-ge b
+next
+----
+b:b
+.
+
+
+iter seq=4
+seek-prefix-ge c
+next
+----
+c:c
+.
+
+
+iter seq=4
+seek-prefix-ge d
+----
+.
+
+
 define
 a.SET.b2:b
 b.SET.2:c
@@ -212,6 +283,187 @@ a:b
 .
 a:b
 
+iter seq=2
+seek-prefix-ge a
+next
+prev
+----
+a:b
+.
+a:b
+
+iter seq=2
+seek-prefix-ge b
+----
+.
+
+
+define
+a.SET.1:a
+aa.SET.2:aa
+aaa.SET.3:aaa
+b.SET.4:b
+----
+
+iter seq=5
+seek-prefix-ge a
+next
+next
+next
+----
+a:a
+aa:aa
+aaa:aaa
+.
+
+iter seq=5
+seek-prefix-ge a
+next
+next
+prev
+next
+next
+----
+a:a
+aa:aa
+aaa:aaa
+aa:aa
+aaa:aaa
+.
+
+iter seq=5
+seek-prefix-ge aa
+prev
+----
+aa:aa
+.
+
+iter seq=5
+seek-prefix-ge aa
+next
+prev
+next
+next
+----
+aa:aa
+aaa:aaa
+aa:aa
+aaa:aaa
+.
+
+iter seq=5
+seek-prefix-ge aa
+next
+prev
+prev
+----
+aa:aa
+aaa:aaa
+aa:aa
+.
+
+iter seq=5
+seek-prefix-ge aaa
+next
+----
+aaa:aaa
+.
+
+iter seq=5
+seek-prefix-ge aaa
+prev
+----
+aaa:aaa
+.
+
+iter seq=5
+seek-prefix-ge b
+next
+----
+b:b
+.
+
+iter seq=5
+seek-prefix-ge aa
+last
+prev
+prev
+prev
+prev
+----
+aa:aa
+b:b
+aaa:aaa
+aa:aa
+a:a
+.
+
+iter seq=5
+seek-prefix-ge aa
+first
+next
+next
+next
+next
+----
+aa:aa
+a:a
+aa:aa
+aaa:aaa
+b:b
+.
+
+iter seq=5
+seek-prefix-ge aaa
+seek-ge aa
+next
+next
+next
+----
+aaa:aaa
+aa:aa
+aaa:aaa
+b:b
+.
+
+iter seq=5
+seek-prefix-ge aaa
+seek-ge aaa
+next
+next
+----
+aaa:aaa
+aaa:aaa
+b:b
+.
+
+iter seq=5
+seek-prefix-ge aaa
+seek-lt aa
+next
+next
+next
+next
+----
+aaa:aaa
+a:a
+aa:aa
+aaa:aaa
+b:b
+.
+
+
+iter seq=5
+seek-prefix-ge aaa
+seek-lt b
+next
+next
+----
+aaa:aaa
+aaa:aaa
+b:b
+.
+
 
 # NB: RANGEDEL entries are ignored.
 define
@@ -343,6 +595,147 @@ a:d
 b:b
 a:d
 
+iter seq=3
+seek-prefix-ge a
+next
+----
+a:cd
+.
+
+iter seq=2
+seek-prefix-ge a
+next
+----
+a:d
+.
+
+iter seq=4
+seek-prefix-ge a
+next
+prev
+next
+----
+a:bcd
+.
+a:bcd
+.
+
+iter seq=2
+seek-prefix-ge a
+next
+prev
+next
+----
+a:d
+.
+a:d
+.
+
+iter seq=3
+seek-prefix-ge a
+next
+prev
+next
+----
+a:cd
+.
+a:cd
+.
+
+iter seq=3
+seek-prefix-ge c
+----
+.
+
+iter seq=3
+seek-prefix-ge 1
+----
+.
+
+iter seq=3
+seek-prefix-ge a
+prev
+----
+a:cd
+.
+
+
+# NB: RANGEDEL entries are ignored.
+define
+a.RANGEDEL.4:c
+a.MERGE.3:b
+a.RANGEDEL.2:c
+a.MERGE.2:c
+a.RANGEDEL.1:c
+a.MERGE.1:d
+aa.RANGEDEL.3:c
+aa.MERGE.2:a
+aa.RANGEDEL.1:c
+aa.MERGE.1:b
+b.RANGEDEL.3:c
+b.MERGE.2:a
+b.RANGEDEL.1:c
+b.MERGE.1:b
+----
+
+iter seq=3
+seek-prefix-ge a
+next
+next
+----
+a:cd
+aa:ab
+.
+
+iter seq=2
+seek-prefix-ge a
+next
+next
+----
+a:d
+aa:b
+.
+
+iter seq=4
+seek-prefix-ge a
+next
+prev
+next
+----
+a:bcd
+aa:ab
+a:bcd
+aa:ab
+
+iter seq=2
+seek-prefix-ge a
+next
+prev
+next
+----
+a:d
+aa:b
+a:d
+aa:b
+
+iter seq=3
+seek-prefix-ge aa
+next
+prev
+next
+----
+aa:ab
+.
+aa:ab
+.
+
+iter seq=4
+seek-prefix-ge aa
+prev
+----
+aa:ab
+.
+
 define
 a.SET.1:a
 b.SET.1:b
@@ -434,6 +827,145 @@ next
 b:b
 .
 
+
+define
+a.SET.1:a
+aa.SET.1:aa
+aaa.SET.1:aaa
+b.SET.1:b
+----
+
+iter seq=2 lower=a
+seek-prefix-ge a
+first
+prev
+----
+a:a
+a:a
+.
+
+
+iter seq=2 lower=aa
+seek-prefix-ge a
+next
+prev
+prev
+----
+aa:aa
+aaa:aaa
+aa:aa
+.
+
+iter seq=2 lower=b
+seek-prefix-ge a
+----
+err=pebble: SeekPrefixGE supplied with key outside of lower bound
+
+iter seq=2 lower=aaa
+seek-prefix-ge a
+----
+aaa:aaa
+
+iter seq=2 lower=aa
+seek-prefix-ge a
+next
+next
+----
+aa:aa
+aaa:aaa
+.
+
+iter seq=2 lower=aa
+seek-prefix-ge a
+first
+next
+next
+next
+----
+aa:aa
+aa:aa
+aaa:aaa
+b:b
+.
+
+iter seq=2 lower=a upper=aa
+seek-prefix-ge a
+next
+----
+a:a
+.
+
+iter seq=2 lower=a upper=aaa
+seek-prefix-ge a
+next
+next
+----
+a:a
+aa:aa
+.
+
+iter seq=2 lower=a upper=b
+seek-prefix-ge a
+next
+next
+next
+----
+a:a
+aa:aa
+aaa:aaa
+.
+
+iter seq=2 lower=a upper=c
+seek-prefix-ge a
+next
+next
+next
+----
+a:a
+aa:aa
+aaa:aaa
+.
+
+iter seq=2 lower=aa upper=aaa
+seek-prefix-ge a
+next
+----
+aa:aa
+.
+
+iter seq=2 lower=aa upper=b
+seek-prefix-ge a
+next
+next
+----
+aa:aa
+aaa:aaa
+.
+
+iter seq=2 lower=aa upper=aa
+seek-prefix-ge a
+----
+.
+
+iter seq=2 lower=aa upper=aa
+seek-prefix-ge a
+----
+.
+
+iter seq=2 lower=a upper=aaa
+seek-prefix-ge aa
+prev
+----
+aa:aa
+.
+
+iter seq=2 lower=a upper=aaa
+seek-prefix-ge aa
+next
+----
+aa:aa
+.
+
 # NB: RANGEDEL entries are ignored.
 define
 a.RANGEDEL.2:c
diff --git a/testdata/level_iter b/testdata/level_iter
index 8212b99..3c17c13 100644
--- a/testdata/level_iter
+++ b/testdata/level_iter
@@ -1,6 +1,7 @@
 define
 a.SET.1:1 b.SET.2:2
 c.SET.3:3 d.SET.4:4
+dd.SET.5:5
 ----
 
 iter
@@ -9,11 +10,13 @@ next
 next
 next
 next
+next
 ----
 a:1
 b:2
 c:3
 d:4
+dd:5
 .
 
 iter
@@ -21,26 +24,39 @@ seek-ge b
 next
 next
 next
+next
 ----
 b:2
 c:3
 d:4
+dd:5
 .
 
 iter
 seek-ge c
 next
 next
+next
 ----
 c:3
 d:4
+dd:5
 .
 
 iter
 seek-ge d
 next
+next
 ----
 d:4
+dd:5
+.
+
+iter
+seek-ge dd
+next
+----
+dd:5
 .
 
 iter
@@ -86,13 +102,63 @@ prev
 prev
 prev
 prev
+prev
 ----
+dd:5
 d:4
 c:3
 b:2
 a:1
 .
 
+iter
+seek-prefix-ge a
+next
+----
+a:1
+.
+
+iter
+seek-prefix-ge d
+next
+next
+----
+d:4
+dd:5
+.
+
+iter
+seek-prefix-ge dd
+next
+----
+dd:5
+.
+
+iter
+seek-prefix-ge d
+next
+prev
+prev
+----
+d:4
+dd:5
+d:4
+.
+
+iter
+seek-prefix-ge d
+prev
+----
+d:4
+.
+
+iter
+seek-prefix-ge dd
+prev
+----
+dd:5
+.
+
 iter lower=a
 seek-ge a
 first
@@ -155,8 +221,8 @@ iter upper=e
 seek-lt e
 last
 ----
-d:4
-d:4
+dd:5
+dd:5
 
 # levelIter only checks upper bound when loading sstables.
 iter upper=d
@@ -187,3 +253,40 @@ last
 ----
 .
 .
+
+iter upper=dd
+seek-prefix-ge d
+next
+----
+d:4
+.
+
+iter upper=e
+seek-prefix-ge d
+next
+next
+----
+d:4
+dd:5
+.
+
+iter lower=dd
+seek-prefix-ge d
+next
+----
+dd:5
+.
+
+iter lower=d
+seek-prefix-ge dd
+prev
+----
+dd:5
+.
+
+iter lower=c
+seek-prefix-ge dd
+prev
+----
+dd:5
+.
diff --git a/testdata/level_iter_boundaries b/testdata/level_iter_boundaries
index 43f04dc..b980f92 100644
--- a/testdata/level_iter_boundaries
+++ b/testdata/level_iter_boundaries
@@ -26,6 +26,17 @@ d#72057594037927935,15:
 a#1,15:
 .
 
+iter
+seek-prefix-ge c
+seek-prefix-ge d
+seek-lt b
+prev
+----
+.
+.
+a#1,15:
+.
+
 iter
 seek-ge e
 seek-lt a
@@ -33,6 +44,13 @@ seek-lt a
 .
 .
 
+iter
+seek-prefix-ge e
+seek-lt a
+----
+.
+.
+
 clear
 ----
 
diff --git a/testdata/merging_iter_seek b/testdata/merging_iter_seek
index a1098a0..d6d7bdb 100644
--- a/testdata/merging_iter_seek
+++ b/testdata/merging_iter_seek
@@ -68,6 +68,88 @@ a1:1
 a0:0
 .
 
+iter
+seek-prefix-ge a0
+next
+----
+a0:0
+.
+
+iter
+seek-prefix-ge a0
+prev
+----
+a0:0
+.
+
+iter
+seek-prefix-ge a0
+first
+next
+next
+next
+----
+a0:0
+a0:0
+a1:1
+a2:2
+.
+
+iter
+seek-prefix-ge a0
+last
+next
+----
+a0:0
+a2:2
+.
+
+iter
+seek-prefix-ge a0
+seek-ge a0
+next
+next
+next
+----
+a0:0
+a0:0
+a1:1
+a2:2
+.
+
+iter
+seek-prefix-ge a0
+seek-lt a2
+next
+next
+----
+a0:0
+a1:1
+a2:2
+.
+
+iter
+seek-prefix-ge a1
+last
+prev
+prev
+prev
+----
+a1:1
+a2:2
+a1:1
+a0:0
+.
+
+iter
+seek-prefix-ge a1
+first
+prev
+----
+a1:1
+a0:0
+.
+
 define
 a0.SET.0:0 b3.SET.3:3
 a1.SET.1:1
@@ -91,3 +173,137 @@ prev
 a1:1
 a0:0
 .
+
+define
+a.SET.0:0 b.SET.3:3
+aa.SET.1:1
+aaa.SET.2:2
+----
+
+iter
+seek-prefix-ge a
+next
+next
+next
+----
+a:0
+aa:1
+aaa:2
+.
+
+iter
+seek-prefix-ge aa
+prev
+----
+aa:1
+.
+
+iter
+seek-prefix-ge aa
+next
+prev
+prev
+----
+aa:1
+aaa:2
+aa:1
+.
+
+iter
+seek-prefix-ge aa
+next
+prev
+next
+next
+----
+aa:1
+aaa:2
+aa:1
+aaa:2
+.
+
+iter
+seek-prefix-ge aaa
+next
+----
+aaa:2
+.
+
+iter
+seek-prefix-ge aaa
+prev
+----
+aaa:2
+.
+
+iter
+seek-prefix-ge b
+prev
+----
+b:3
+.
+
+iter
+seek-prefix-ge b
+next
+----
+b:3
+.
+
+iter
+seek-prefix-ge aa
+last
+prev
+prev
+prev
+prev
+----
+aa:1
+b:3
+aaa:2
+aa:1
+a:0
+.
+
+iter
+seek-prefix-ge aa
+first
+next
+next
+next
+next
+----
+aa:1
+a:0
+aa:1
+aaa:2
+b:3
+.
+
+iter
+seek-prefix-ge aa
+seek-ge a
+next
+next
+next
+next
+----
+aa:1
+a:0
+aa:1
+aaa:2
+b:3
+.
+
+iter
+seek-prefix-ge aa
+seek-lt aaa
+next
+next
+next
+----
+aa:1
+aa:1
+aaa:2
+b:3
+.