Repository: petermattis/pebble
Pull request #81 (2019-04-19 19:10:32)

Trim {Lower,Upper}Bound for table iteration
NB: this change is in anticipation of pushing lower/upper bound checks down into `sstable.Reader`.

Trim `IterOptions.{Lower,Upper}Bound` when loading a table to avoid requiring lower/upper bounds checks if all of the keys within the table pass the checks.

See #28

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/81)
<!-- Reviewable:end -->

petermattis: I added another commit which finishes off the work to push handling of {lower,upper}-bound to the leaf iterators. This reduces average latency for `pebble scan --rows=1000` by 6-7%. The approach is ready to take a look, but I need to add a lot of tests.
petermattis: TFTR! I'll ping this PR again once I get tests written for the last commit (probably not today).
petermattis: I fleshed out the testing for the bounds checking in `arenaskl.Iterator`, `batchskl.Iterator` and `sstable.Iterator`. This is ready for another look and should be good to go modulo code review comments.
petermattis: @ajkr Ping. Can you give this another look?

Patch:
diff --git a/batch.go b/batch.go
index 0847242..be77e28 100644
--- a/batch.go
+++ b/batch.go
@@ -468,7 +468,7 @@ func (b *Batch) newInternalIter(o *db.IterOptions) internalIterator {
 	return &batchIter{
 		cmp:   b.storage.cmp,
 		batch: b,
-		iter:  b.index.NewIter(),
+		iter:  b.index.NewIter(o.GetLowerBound(), o.GetUpperBound()),
 	}
 }
 
@@ -493,7 +493,7 @@ func (b *Batch) newRangeDelIter(o *db.IterOptions) internalIterator {
 		it := &batchIter{
 			cmp:   b.storage.cmp,
 			batch: b,
-			iter:  b.rangeDelIndex.NewIter(),
+			iter:  b.rangeDelIndex.NewIter(nil, nil),
 		}
 		for it.Next() {
 			frag.Add(it.Key(), it.Value())
diff --git a/compaction.go b/compaction.go
index 17256b0..9664b61 100644
--- a/compaction.go
+++ b/compaction.go
@@ -306,8 +306,10 @@ func (c *compaction) newInputIter(
 	// levelIters per level, one which iterates over the point operations, and
 	// one which iterates over the range deletions. These two iterators are
 	// combined with a mergingIter.
-	newRangeDelIter := func(f *fileMetadata) (internalIterator, internalIterator, error) {
-		iter, rangeDelIter, err := newIters(f)
+	newRangeDelIter := func(
+		f *fileMetadata, _ *db.IterOptions,
+	) (internalIterator, internalIterator, error) {
+		iter, rangeDelIter, err := newIters(f, nil /* iter options */)
 		if err == nil {
 			// TODO(peter): It is mildly wasteful to open the point iterator only to
 			// immediately close it. One way to solve this would be to add new
@@ -336,7 +338,7 @@ func (c *compaction) newInputIter(
 	} else {
 		for i := range c.inputs[0] {
 			f := &c.inputs[0][i]
-			iter, rangeDelIter, err := newIters(f)
+			iter, rangeDelIter, err := newIters(f, nil /* iter options */)
 			if err != nil {
 				return nil, fmt.Errorf("pebble: could not open table %d: %v", f.fileNum, err)
 			}
diff --git a/db.go b/db.go
index 8777e25..07fc00a 100644
--- a/db.go
+++ b/db.go
@@ -438,7 +438,7 @@ func (d *DB) newIterInternal(
 	current := readState.current
 	for i := len(current.files[0]) - 1; i >= 0; i-- {
 		f := &current.files[0][i]
-		iter, rangeDelIter, err := d.newIters(f)
+		iter, rangeDelIter, err := d.newIters(f, o)
 		if err != nil {
 			dbi.err = err
 			return dbi
diff --git a/get_iter.go b/get_iter.go
index 3461f52..d9a7815 100644
--- a/get_iter.go
+++ b/get_iter.go
@@ -129,7 +129,7 @@ func (g *getIter) Next() bool {
 			// Create iterators from L0 from newest to oldest.
 			if n := len(g.l0); n > 0 {
 				l := &g.l0[n-1]
-				g.iter, g.rangeDelIter, g.err = g.newIters(l)
+				g.iter, g.rangeDelIter, g.err = g.newIters(l, nil /* iter options */)
 				if g.err != nil {
 					return false
 				}
diff --git a/ingest.go b/ingest.go
index 48dbab4..2ea634d 100644
--- a/ingest.go
+++ b/ingest.go
@@ -35,7 +35,7 @@ func ingestLoad1(opts *db.Options, path string, fileNum uint64) (*fileMetadata,
 	smallestSet, largestSet := false, false
 
 	{
-		iter := r.NewIter(nil)
+		iter := r.NewIter(nil /* lower */, nil /* upper */)
 		defer iter.Close()
 		if iter.First() {
 			meta.smallest = iter.Key().Clone()
@@ -50,7 +50,7 @@ func ingestLoad1(opts *db.Options, path string, fileNum uint64) (*fileMetadata,
 		}
 	}
 
-	if iter := r.NewRangeDelIter(nil); iter != nil {
+	if iter := r.NewRangeDelIter(); iter != nil {
 		defer iter.Close()
 		if iter.First() {
 			key := iter.Key()
diff --git a/internal/arenaskl/iterator.go b/internal/arenaskl/iterator.go
index 2720918..40ba1b0 100644
--- a/internal/arenaskl/iterator.go
+++ b/internal/arenaskl/iterator.go
@@ -37,8 +37,10 @@ func (s *splice) init(prev, next *node) {
 // to construct an iterator. The current state of the iterator can be cloned by
 // simply value copying the struct. All iterator methods are thread-safe.
 type Iterator struct {
-	list *Skiplist
-	nd   *node
+	list  *Skiplist
+	nd    *node
+	lower []byte
+	upper []byte
 }
 
 // Close resets the iterator.
@@ -54,47 +56,98 @@ func (it *Iterator) Error() error {
 }
 
 // SeekGE moves the iterator to the first entry whose key is greater than or
-// equal to the given key. Returns true if the iterator is pointing at a
-// valid entry and false otherwise.
+// equal to the given key. Returns true if the iterator is pointing at a valid
+// entry and false otherwise. Note that SeekGE 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 (it *Iterator) SeekGE(key []byte) bool {
 	_, it.nd, _ = it.seekForBaseSplice(key)
-	return it.nd != it.list.tail
+	if it.nd == it.list.tail {
+		return false
+	}
+	if it.upper != nil && it.list.cmp(it.upper, it.Key().UserKey) <= 0 {
+		it.nd = it.list.tail
+		return false
+	}
+	return true
 }
 
 // SeekLT moves the iterator to the last entry whose key is less than the given
 // key. Returns true if the iterator is pointing at a valid entry and false
-// otherwise.
+// otherwise. Note that SeekLT only checks the lower bound. It is up to the
+// caller to ensure that key is less than the upper bound.
 func (it *Iterator) SeekLT(key []byte) bool {
+	// NB: the top-level Iterator has already adjusted key based on
+	// the upper-bound.
 	it.nd, _, _ = it.seekForBaseSplice(key)
-	return it.nd != it.list.head
+	if it.nd == it.list.head {
+		return false
+	}
+	if it.lower != nil && it.list.cmp(it.lower, it.Key().UserKey) > 0 {
+		it.nd = it.list.head
+		return false
+	}
+	return true
 }
 
 // First seeks position at the first entry in list. Final state of iterator is
-// Valid() iff list is not empty.
+// Valid() iff list is not empty. Note that First only checks the upper
+// bound. It is up to the caller to ensure that key is greater than or equal to
+// the lower bound (e.g. via a call to SeekGE(lower)).
 func (it *Iterator) First() bool {
 	it.nd = it.list.getNext(it.list.head, 0)
-	return it.nd != it.list.tail
+	if it.nd == it.list.tail {
+		return false
+	}
+	if it.upper != nil && it.list.cmp(it.upper, it.Key().UserKey) <= 0 {
+		it.nd = it.list.tail
+		return false
+	}
+	return true
 }
 
 // Last seeks position at the last entry in list. Final state of iterator is
-// Valid() iff list is not empty.
+// Valid() iff list is not empty. Note that Last only checks the lower
+// bound. It is up to the caller to ensure that key is less than the upper
+// bound (e.g. via a call to SeekLT(upper)).
 func (it *Iterator) Last() bool {
 	it.nd = it.list.getPrev(it.list.tail, 0)
-	return it.nd != it.list.head
+	if it.nd == it.list.head {
+		return false
+	}
+	if it.lower != nil && it.list.cmp(it.lower, it.Key().UserKey) > 0 {
+		it.nd = it.list.head
+		return false
+	}
+	return true
 }
 
 // Next advances to the next position. If there are no following nodes, then
 // Valid() will be false after this call.
 func (it *Iterator) Next() bool {
 	it.nd = it.list.getNext(it.nd, 0)
-	return it.nd != it.list.tail
+	if it.nd == it.list.tail {
+		return false
+	}
+	if it.upper != nil && it.list.cmp(it.upper, it.Key().UserKey) <= 0 {
+		it.nd = it.list.tail
+		return false
+	}
+	return true
 }
 
 // Prev moves to the previous position. If there are no previous nodes, then
 // Valid() will be false after this call.
 func (it *Iterator) Prev() bool {
 	it.nd = it.list.getPrev(it.nd, 0)
-	return it.nd != it.list.head
+	if it.nd == it.list.head {
+		return false
+	}
+	if it.lower != nil && it.list.cmp(it.lower, it.Key().UserKey) > 0 {
+		it.nd = it.list.head
+		return false
+	}
+	return true
 }
 
 // Key returns the key at the current position.
@@ -145,6 +198,11 @@ func (it *Iterator) seekForBaseSplice(key []byte) (prev, next *node, found bool)
 		prev, next, found = it.list.findSpliceForLevel(ikey, level, prev)
 
 		if found {
+			if level != 0 {
+				// next is pointing at the target node, but we need to find previous on
+				// the bottom level.
+				prev = it.list.getPrev(next, 0)
+			}
 			break
 		}
 
diff --git a/internal/arenaskl/skl.go b/internal/arenaskl/skl.go
index 3bc778a..81aa775 100644
--- a/internal/arenaskl/skl.go
+++ b/internal/arenaskl/skl.go
@@ -301,10 +301,14 @@ func (s *Skiplist) addInternal(key db.InternalKey, value []byte, ins *Inserter)
 	return nil
 }
 
-// NewIter returns a new Iterator object. Note that it is safe for an
-// iterator to be copied by value.
-func (s *Skiplist) NewIter() Iterator {
-	return Iterator{list: s, nd: s.head}
+// NewIter returns a new Iterator object. The lower and upper bound parameters
+// control the range of keys the iterator will return. Specifying for nil for
+// lower or upper bound disables the check for that boundary. Note that lower
+// bound is not checked on {SeekGE,First} and upper bound is not check on
+// {SeekLT,Last}. The user is expected to perform that check. Note that it is
+// safe for an iterator to be copied by value.
+func (s *Skiplist) NewIter(lower, upper []byte) Iterator {
+	return Iterator{list: s, nd: s.head, lower: lower, upper: upper}
 }
 
 func (s *Skiplist) newNode(
diff --git a/internal/batchskl/iterator.go b/internal/batchskl/iterator.go
index 7d6c79e..1c5b0a2 100644
--- a/internal/batchskl/iterator.go
+++ b/internal/batchskl/iterator.go
@@ -17,7 +17,9 @@
 
 package batchskl
 
-import "github.com/petermattis/pebble/db"
+import (
+	"github.com/petermattis/pebble/db"
+)
 
 type splice struct {
 	prev uint32
@@ -33,8 +35,10 @@ func (s *splice) init(prev, next uint32) {
 // to construct an iterator. The current state of the iterator can be cloned
 // by simply value copying the struct.
 type Iterator struct {
-	list *Skiplist
-	nd   uint32
+	list  *Skiplist
+	nd    uint32
+	lower []byte
+	upper []byte
 }
 
 // Close resets the iterator.
@@ -45,47 +49,96 @@ func (it *Iterator) Close() error {
 }
 
 // SeekGE moves the iterator to the first entry whose key is greater than or
-// equal to the given key. Returns true if the iterator is pointing at a
-// valid entry and false otherwise.
+// equal to the given key. Returns true if the iterator is pointing at a valid
+// entry and false otherwise. Note that SeekGE 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 (it *Iterator) SeekGE(key []byte) bool {
 	_, it.nd, _ = it.seekForBaseSplice(key, it.list.storage.AbbreviatedKey(key))
-	return it.nd != it.list.tail
+	if it.nd == it.list.tail {
+		return false
+	}
+	if it.upper != nil && it.list.storage.Compare(it.upper, it.list.getKey(it.nd)) <= 0 {
+		it.nd = it.list.tail
+		return false
+	}
+	return true
 }
 
 // SeekLT moves the iterator to the last entry whose key is less the given
 // key. Returns true if the iterator is pointing at a valid entry and false
-// otherwise.
+// otherwise. Note that SeekLT only checks the lower bound. It is up to the
+// caller to ensure that key is less than the upper bound.
 func (it *Iterator) SeekLT(key []byte) (found bool) {
 	it.nd, _, _ = it.seekForBaseSplice(key, it.list.storage.AbbreviatedKey(key))
-	return it.nd != it.list.head
+	if it.nd == it.list.head {
+		return false
+	}
+	if it.lower != nil && it.list.storage.Compare(it.lower, it.list.getKey(it.nd)) > 0 {
+		it.nd = it.list.head
+		return false
+	}
+	return true
 }
 
-// First seeks position at the first entry in list.
-// Final state of iterator is Valid() iff list is not empty.
+// First seeks position at the first entry in list. Final state of iterator is
+// Valid() iff list is not empty. Note that First only checks the upper
+// bound. It is up to the caller to ensure that key is greater than or equal to
+// the lower bound (e.g. via a call to SeekGE(lower)).
 func (it *Iterator) First() bool {
 	it.nd = it.list.getNext(it.list.head, 0)
-	return it.nd != it.list.tail
+	if it.nd == it.list.tail {
+		return false
+	}
+	if it.upper != nil && it.list.storage.Compare(it.upper, it.list.getKey(it.nd)) <= 0 {
+		it.nd = it.list.tail
+		return false
+	}
+	return true
 }
 
-// Last seeks position at the last entry in list.
-// Final state of iterator is Valid() iff list is not empty.
+// Last seeks position at the last entry in list. Final state of iterator is
+// Valid() iff list is not empty. Note that Last only checks the lower
+// bound. It is up to the caller to ensure that key is less than the upper
+// bound (e.g. via a call to SeekLT(upper)).
 func (it *Iterator) Last() bool {
 	it.nd = it.list.getPrev(it.list.tail, 0)
-	return it.nd != it.list.head
+	if it.nd == it.list.head {
+		return false
+	}
+	if it.lower != nil && it.list.storage.Compare(it.lower, it.list.getKey(it.nd)) > 0 {
+		it.nd = it.list.head
+		return false
+	}
+	return true
 }
 
 // Next advances to the next position. If there are no following nodes, then
 // Valid() will be false after this call.
 func (it *Iterator) Next() bool {
 	it.nd = it.list.getNext(it.nd, 0)
-	return it.nd != it.list.tail
+	if it.nd == it.list.tail {
+		return false
+	}
+	if it.upper != nil && it.list.storage.Compare(it.upper, it.list.getKey(it.nd)) <= 0 {
+		it.nd = it.list.tail
+		return false
+	}
+	return true
 }
 
 // Prev moves to the previous position. If there are no previous nodes, then
 // Valid() will be false after this call.
 func (it *Iterator) Prev() bool {
 	it.nd = it.list.getPrev(it.nd, 0)
-	return it.nd != it.list.head
+	if it.nd == it.list.head {
+		return false
+	}
+	if it.lower != nil && it.list.storage.Compare(it.lower, it.list.getKey(it.nd)) > 0 {
+		it.nd = it.list.head
+		return false
+	}
+	return true
 }
 
 // Key returns the key at the current position.
@@ -120,6 +173,11 @@ func (it *Iterator) seekForBaseSplice(
 	for level := it.list.height - 1; ; level-- {
 		prev, next, found = it.list.findSpliceForLevel(key, abbreviatedKey, level, prev)
 		if found {
+			if level != 0 {
+				// next is pointing at the target node, but we need to find previous on
+				// the bottom level.
+				prev = it.list.getPrev(next, 0)
+			}
 			break
 		}
 		if level == 0 {
diff --git a/internal/batchskl/skl.go b/internal/batchskl/skl.go
index 4af1a46..6b66dfc 100644
--- a/internal/batchskl/skl.go
+++ b/internal/batchskl/skl.go
@@ -222,10 +222,14 @@ func (s *Skiplist) Add(keyOffset uint32) error {
 	return nil
 }
 
-// NewIter returns a new Iterator object. Note that it is safe for an iterator
-// to be copied by value.
-func (s *Skiplist) NewIter() Iterator {
-	return Iterator{list: s}
+// NewIter returns a new Iterator object. The lower and upper bound parameters
+// control the range of keys the iterator will return. Specifying for nil for
+// lower or upper bound disables the check for that boundary. Note that lower
+// bound is not checked on {SeekGE,First} and upper bound is not check on
+// {SeekLT,Last}. The user is expected to perform that check. Note that it is
+// safe for an iterator to be copied by value.
+func (s *Skiplist) NewIter(lower, upper []byte) Iterator {
+	return Iterator{list: s, lower: lower, upper: upper}
 }
 
 func (s *Skiplist) newNode(height, key uint32, abbreviatedKey uint64) uint32 {
diff --git a/iterator.go b/iterator.go
index 391c3a0..b8cf3e9 100644
--- a/iterator.go
+++ b/iterator.go
@@ -49,16 +49,11 @@ type Iterator struct {
 }
 
 func (i *Iterator) findNextEntry() bool {
-	upperBound := i.opts.GetUpperBound()
 	i.valid = false
 	i.pos = iterPosCur
 
 	for i.iterValid {
 		key := i.iter.Key()
-		if upperBound != nil && i.cmp(key.UserKey, upperBound) >= 0 {
-			break
-		}
-
 		switch key.Kind() {
 		case db.InternalKeyKindDelete:
 			i.nextUserKey()
@@ -107,16 +102,11 @@ func (i *Iterator) nextUserKey() {
 }
 
 func (i *Iterator) findPrevEntry() bool {
-	lowerBound := i.opts.GetLowerBound()
 	i.valid = false
 	i.pos = iterPosCur
 
 	for i.iterValid {
 		key := i.iter.Key()
-		if lowerBound != nil && i.cmp(key.UserKey, lowerBound) < 0 {
-			break
-		}
-
 		if i.valid {
 			if !i.equal(key.UserKey, i.key) {
 				// We've iterated to the previous user key.
@@ -287,10 +277,10 @@ func (i *Iterator) First() bool {
 	}
 
 	if lowerBound := i.opts.GetLowerBound(); lowerBound != nil {
-		return i.SeekGE(lowerBound)
+		i.iterValid = i.iter.SeekGE(lowerBound)
+	} else {
+		i.iterValid = i.iter.First()
 	}
-
-	i.iterValid = i.iter.First()
 	return i.findNextEntry()
 }
 
@@ -302,10 +292,10 @@ func (i *Iterator) Last() bool {
 	}
 
 	if upperBound := i.opts.GetUpperBound(); upperBound != nil {
-		return i.SeekLT(upperBound)
+		i.iterValid = i.iter.SeekLT(upperBound)
+	} else {
+		i.iterValid = i.iter.Last()
 	}
-
-	i.iterValid = i.iter.Last()
 	return i.findPrevEntry()
 }
 
@@ -319,6 +309,14 @@ func (i *Iterator) Next() bool {
 	case iterPosCur:
 		i.nextUserKey()
 	case iterPosPrev:
+		// The underlying iterator is pointed to the previous key (this can only
+		// happen when switching iteration directions). We set i.valid to false
+		// here to force the calls to nextUserKey to save the current key i.iter is
+		// pointing at in order to determine when the next user-key is reached.
+		//
+		// TODO(peter): This is subtle. Perhaps we should provide a special version
+		// of nextUserKey which advances until a larger user key is found.
+		i.valid = false
 		i.nextUserKey()
 		i.nextUserKey()
 	case iterPosNext:
@@ -336,6 +334,14 @@ func (i *Iterator) Prev() bool {
 	case iterPosCur:
 		i.prevUserKey()
 	case iterPosNext:
+		// The underlying iterator is pointed to the next key (this can only happen
+		// when switching iteration directions). We set i.valid to false here to
+		// force the calls to prevUserKey to save the current key i.iter is
+		// pointing at in order to determine when the next user-key is reached.
+		//
+		// TODO(peter): This is subtle. Perhaps we should provide a special version
+		// of prevUserKey which advances until a smaller user key is found.
+		i.valid = false
 		i.prevUserKey()
 		i.prevUserKey()
 	case iterPosPrev:
diff --git a/level_iter.go b/level_iter.go
index 916e04c..06e36cd 100644
--- a/level_iter.go
+++ b/level_iter.go
@@ -12,7 +12,9 @@ import (
 
 // tableNewIters creates a new point and range-del iterator for the given file
 // number.
-type tableNewIters func(meta *fileMetadata) (internalIterator, internalIterator, error)
+type tableNewIters func(
+	meta *fileMetadata, opts *db.IterOptions,
+) (internalIterator, internalIterator, error)
 
 // levelIter provides a merged view of the sstables in a level.
 //
@@ -28,9 +30,10 @@ type tableNewIters func(meta *fileMetadata) (internalIterator, internalIterator,
 // heap. Note that Iterator treat a range deletion tombstone as a no-op and
 // processes range deletions via mergingIter.
 type levelIter struct {
-	opts  *db.IterOptions
-	cmp   db.Compare
-	index int
+	opts      *db.IterOptions
+	tableOpts *db.IterOptions
+	cmp       db.Compare
+	index     int
 	// The key to return when iterating past an sstable boundary and that
 	// boundary is a range deletion tombstone. Note that if boundary != nil, then
 	// iter == nil, and if iter != nil, then boundary == nil.
@@ -86,7 +89,7 @@ func (l *levelIter) initLargestUserKey(largestUserKey *[]byte) {
 func (l *levelIter) findFileGE(key []byte) int {
 	// Find the earliest file whose largest key is >= ikey. Note that the range
 	// deletion sentinel key is handled specially and a search for K will not
-	// find a table for K<range-del-sentinel> is the largest key. This prevents
+	// find a table where K<range-del-sentinel> is the largest key. This prevents
 	// loading untruncated range deletions from a table which can't possibly
 	// contain the target key and is required for correctness by DB.Get.
 	//
@@ -132,7 +135,8 @@ func (l *levelIter) loadFile(index, dir int) bool {
 		}
 
 		f := &l.files[l.index]
-		if lowerBound := l.opts.GetLowerBound(); lowerBound != nil {
+		lowerBound := l.opts.GetLowerBound()
+		if lowerBound != nil {
 			if l.cmp(f.largest.UserKey, lowerBound) < 0 {
 				// The largest key in the sstable is smaller than the lower bound.
 				if dir < 0 {
@@ -140,8 +144,15 @@ func (l *levelIter) loadFile(index, dir int) bool {
 				}
 				continue
 			}
+			if l.cmp(lowerBound, f.smallest.UserKey) < 0 {
+				// The lower bound is smaller than the smallest key in the
+				// table. Iteration within the table does not need to check the lower
+				// bound.
+				lowerBound = nil
+			}
 		}
-		if upperBound := l.opts.GetUpperBound(); upperBound != nil {
+		upperBound := l.opts.GetUpperBound()
+		if upperBound != nil {
 			if l.cmp(f.smallest.UserKey, upperBound) >= 0 {
 				// The smallest key in the sstable is greater than or equal to the
 				// lower bound.
@@ -150,10 +161,26 @@ func (l *levelIter) loadFile(index, dir int) bool {
 				}
 				continue
 			}
+			if l.cmp(upperBound, f.largest.UserKey) > 0 {
+				// The upper bound is greater than the largest key in the
+				// table. Iteration within the table does not need to check the upper
+				// bound.
+				upperBound = nil
+			}
+		}
+
+		var opts *db.IterOptions
+		if lowerBound != nil || upperBound != nil {
+			if l.tableOpts == nil {
+				l.tableOpts = &db.IterOptions{}
+			}
+			l.tableOpts.LowerBound = lowerBound
+			l.tableOpts.UpperBound = upperBound
+			opts = l.tableOpts
 		}
 
 		var rangeDelIter internalIterator
-		l.iter, rangeDelIter, l.err = l.newIters(f)
+		l.iter, rangeDelIter, l.err = l.newIters(f, opts)
 		if l.err != nil || l.iter == nil {
 			return false
 		}
diff --git a/mem_table.go b/mem_table.go
index 116d3f1..e3a9b5e 100644
--- a/mem_table.go
+++ b/mem_table.go
@@ -97,7 +97,7 @@ func (m *memTable) readyForFlush() bool {
 // Get gets the value for the given key. It returns ErrNotFound if the DB does
 // not contain the key.
 func (m *memTable) get(key []byte) (value []byte, err error) {
-	it := m.skl.NewIter()
+	it := m.skl.NewIter(nil, nil)
 	if !it.SeekGE(key) {
 		return nil, db.ErrNotFound
 	}
@@ -167,8 +167,8 @@ func (m *memTable) apply(batch *Batch, seqNum uint64) error {
 // 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.
-func (m *memTable) newIter(*db.IterOptions) internalIterator {
-	it := m.skl.NewIter()
+func (m *memTable) newIter(o *db.IterOptions) internalIterator {
+	it := m.skl.NewIter(o.GetLowerBound(), o.GetUpperBound())
 	return &it
 }
 
@@ -216,7 +216,7 @@ func (f *rangeTombstoneFrags) get(m *memTable) []rangedel.Tombstone {
 				f.tombstones = append(f.tombstones, fragmented...)
 			},
 		}
-		for it := m.rangeDelSkl.NewIter(); it.Next(); {
+		for it := m.rangeDelSkl.NewIter(nil, nil); it.Next(); {
 			frag.Add(it.Key(), it.Value())
 		}
 		frag.Finish()
diff --git a/sstable/reader.go b/sstable/reader.go
index 855adb0..de59554 100644
--- a/sstable/reader.go
+++ b/sstable/reader.go
@@ -49,11 +49,19 @@ type block []byte
 // to seek for a given key, it first looks in the index for the block that
 // contains that key, and then looks inside that block.
 type Iterator struct {
-	reader    *Reader
-	index     blockIter
-	data      blockIter
-	err       error
-	closeHook func() error
+	cmp db.Compare
+	// Global lower/upper bound for the iterator.
+	lower []byte
+	upper []byte
+	// Per-block lower/upper bound. Nil if the bound does not apply to the block
+	// because we determined the block lies completely within the bound.
+	blockLower []byte
+	blockUpper []byte
+	reader     *Reader
+	index      blockIter
+	data       blockIter
+	err        error
+	closeHook  func() error
 }
 
 func (i *Iterator) init(r *Reader) error {
@@ -63,10 +71,35 @@ func (i *Iterator) init(r *Reader) error {
 	if i.err != nil {
 		return i.err
 	}
-	i.err = i.index.init(r.compare, index, r.Properties.GlobalSeqNum)
+	i.cmp = r.compare
+	i.err = i.index.init(i.cmp, index, r.Properties.GlobalSeqNum)
 	return i.err
 }
 
+func (i *Iterator) initBounds() {
+	if i.lower == nil && i.upper == nil {
+		return
+	}
+
+	// Trim the iteration bounds for the current block. We don't have to check
+	// the bounds on each iteration if the block is entirely contained within the
+	// iteration bounds.
+	i.blockLower = i.lower
+	if i.blockLower != nil && i.data.First() &&
+		i.cmp(i.blockLower, i.data.Key().UserKey) < 0 {
+		// The lower-bound is less than the first key in the block. No need
+		// to check the lower-bound again for this block.
+		i.blockLower = nil
+	}
+	i.blockUpper = i.upper
+	if i.blockUpper != nil && i.cmp(i.blockUpper, i.index.Key().UserKey) >= 0 {
+		// The upper-bound is greater than or equal to the index key which
+		// itself is greater than every key in the block. No need to check the
+		// upper-bound again for this block.
+		i.blockUpper = nil
+	}
+}
+
 // loadBlock loads the block at the current index position and leaves i.data
 // unpositioned. If unsuccessful, it sets i.err to any error encountered, which
 // may be nil if we have simply exhausted the entire table.
@@ -91,8 +124,12 @@ func (i *Iterator) loadBlock() bool {
 		i.err = err
 		return false
 	}
-	i.err = i.data.init(i.reader.compare, block, i.reader.Properties.GlobalSeqNum)
-	return i.err == nil
+	i.err = i.data.init(i.cmp, block, i.reader.Properties.GlobalSeqNum)
+	if i.err != nil {
+		return false
+	}
+	i.initBounds()
+	return true
 }
 
 // seekBlock loads the block at the current index position and positions i.data
@@ -116,109 +153,127 @@ func (i *Iterator) seekBlock(key []byte) bool {
 		i.err = err
 		return false
 	}
-	i.err = i.data.init(i.reader.compare, block, i.reader.Properties.GlobalSeqNum)
+	i.err = i.data.init(i.cmp, block, i.reader.Properties.GlobalSeqNum)
 	if i.err != nil {
 		return false
 	}
 	// Look for the key inside that block.
+	i.initBounds()
 	i.data.SeekGE(key)
 	return true
 }
 
 // SeekGE implements internalIterator.SeekGE, as documented in the pebble
-// package.
+// package. Note that SeekGE 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) SeekGE(key []byte) bool {
 	if i.err != nil {
 		return false
 	}
 
-	// NB: the top-level Iterator has already adjusted key based on
-	// IterOptions.LowerBound.
-
 	if !i.index.SeekGE(key) {
 		return false
 	}
 	if !i.loadBlock() {
 		return false
 	}
-	return i.data.SeekGE(key)
+	if !i.data.SeekGE(key) {
+		return false
+	}
+	if i.blockUpper != nil && i.cmp(i.data.Key().UserKey, i.blockUpper) >= 0 {
+		return false
+	}
+	return true
 }
 
 // SeekLT implements internalIterator.SeekLT, as documented in the pebble
-// package.
+// package. Note that SeekLT only checks the lower bound. It is up to the
+// caller to ensure that key is less than the upper bound.
 func (i *Iterator) SeekLT(key []byte) bool {
 	if i.err != nil {
 		return false
 	}
 
-	// NB: the top-level Iterator has already adjusted key based on
-	// IterOptions.UpperBound.
-
 	if !i.index.SeekGE(key) {
 		i.index.Last()
 	}
 	if !i.loadBlock() {
 		return false
 	}
-	if i.data.SeekLT(key) {
-		return true
-	}
-	// The index contains separator keys which may lie between
-	// user-keys. Consider the user-keys:
-	//
-	//   complete
-	// ---- new block ---
-	//   complexion
-	//
-	// If these two keys end one block and start the next, the index key may
-	// be chosen as "compleu". The SeekGE in the index block will then point
-	// us to the block containing "complexion". If this happens, we want the
-	// last key from the previous data block.
-	if !i.index.Prev() {
-		return false
+	if !i.data.SeekLT(key) {
+		// The index contains separator keys which may lie between
+		// user-keys. Consider the user-keys:
+		//
+		//   complete
+		// ---- new block ---
+		//   complexion
+		//
+		// If these two keys end one block and start the next, the index key may
+		// be chosen as "compleu". The SeekGE in the index block will then point
+		// us to the block containing "complexion". If this happens, we want the
+		// last key from the previous data block.
+		if !i.index.Prev() {
+			return false
+		}
+		if !i.loadBlock() {
+			return false
+		}
+		if !i.data.Last() {
+			return false
+		}
 	}
-	if !i.loadBlock() {
+	if i.blockLower != nil && i.cmp(i.data.Key().UserKey, i.blockLower) < 0 {
 		return false
 	}
-	return i.data.Last()
+	return true
 }
 
 // First implements internalIterator.First, as documented in the pebble
-// package.
+// package. Note that First only checks the upper bound. It is up to the caller
+// to ensure that key is greater than or equal to the lower bound (e.g. via a
+// call to SeekGE(lower)).
 func (i *Iterator) First() bool {
 	if i.err != nil {
 		return false
 	}
 
-	// NB: the top-level Iterator will call SeekGE if IterOptions.LowerBound is
-	// set.
-
 	if !i.index.First() {
 		return false
 	}
 	if !i.loadBlock() {
 		return false
 	}
-	return i.data.First()
+	if !i.data.First() {
+		return false
+	}
+	if i.blockUpper != nil && i.cmp(i.data.Key().UserKey, i.blockUpper) >= 0 {
+		return false
+	}
+	return true
 }
 
 // Last implements internalIterator.Last, as documented in the pebble
-// package.
+// package. Note that Last only checks the lower bound. It is up to the caller
+// to ensure that key is less than the upper bound (e.g. via a call to
+// SeekLT(upper))
 func (i *Iterator) Last() bool {
 	if i.err != nil {
 		return false
 	}
 
-	// NB: the top-level Iterator will call SeekLT if IterOptions.UpperBound is
-	// set.
-
 	if !i.index.Last() {
 		return false
 	}
 	if !i.loadBlock() {
 		return false
 	}
-	return i.data.Last()
+	if !i.data.Last() {
+		return false
+	}
+	if i.blockLower != nil && i.cmp(i.data.Key().UserKey, i.blockLower) < 0 {
+		return false
+	}
+	return true
 }
 
 // Next implements internalIterator.Next, as documented in the pebble
@@ -228,6 +283,9 @@ func (i *Iterator) Next() bool {
 		return false
 	}
 	if i.data.Next() {
+		if i.blockUpper != nil && i.cmp(i.data.Key().UserKey, i.blockUpper) >= 0 {
+			return false
+		}
 		return true
 	}
 	for {
@@ -239,7 +297,13 @@ func (i *Iterator) Next() bool {
 			break
 		}
 		if i.loadBlock() {
-			return i.data.First()
+			if !i.data.First() {
+				return false
+			}
+			if i.blockUpper != nil && i.cmp(i.data.Key().UserKey, i.blockUpper) >= 0 {
+				return false
+			}
+			return true
 		}
 	}
 	return false
@@ -252,6 +316,9 @@ func (i *Iterator) Prev() bool {
 		return false
 	}
 	if i.data.Prev() {
+		if i.blockLower != nil && i.cmp(i.data.Key().UserKey, i.blockLower) < 0 {
+			return false
+		}
 		return true
 	}
 	for {
@@ -263,7 +330,13 @@ func (i *Iterator) Prev() bool {
 			break
 		}
 		if i.loadBlock() {
-			return i.data.Last()
+			if !i.data.Last() {
+				return false
+			}
+			if i.blockLower != nil && i.cmp(i.data.Key().UserKey, i.blockLower) < 0 {
+				return false
+			}
+			return true
 		}
 	}
 	return false
@@ -361,7 +434,7 @@ func (r *Reader) Close() error {
 
 // get is a testing helper that simulates a read and helps verify bloom filters
 // until they are available through iterators.
-func (r *Reader) get(key []byte, o *db.IterOptions) (value []byte, err error) {
+func (r *Reader) get(key []byte) (value []byte, err error) {
 	if r.err != nil {
 		return nil, r.err
 	}
@@ -399,14 +472,14 @@ func (r *Reader) get(key []byte, o *db.IterOptions) (value []byte, err error) {
 }
 
 // NewIter returns an internal iterator for the contents of the table.
-func (r *Reader) NewIter(o *db.IterOptions) *Iterator {
+func (r *Reader) NewIter(lower, upper []byte) *Iterator {
 	// NB: pebble.tableCache wraps the returned iterator with one which performs
 	// reference counting on the Reader, preventing the Reader from being closed
 	// until the final iterator closes.
 	if r.err != nil {
 		return &Iterator{err: r.err}
 	}
-	i := &Iterator{}
+	i := &Iterator{lower: lower, upper: upper}
 	_ = i.init(r)
 	return i
 }
@@ -414,7 +487,7 @@ func (r *Reader) NewIter(o *db.IterOptions) *Iterator {
 // NewRangeDelIter returns an internal iterator for the contents of the
 // range-del block for the table. Returns nil if the table does not contain any
 // range deletions.
-func (r *Reader) NewRangeDelIter(o *db.IterOptions) *blockIter {
+func (r *Reader) NewRangeDelIter() *blockIter {
 	if r.rangeDel.bh.length == 0 {
 		return nil
 	}
diff --git a/table_cache.go b/table_cache.go
index ce1c055..1e182fd 100644
--- a/table_cache.go
+++ b/table_cache.go
@@ -49,7 +49,9 @@ func (c *tableCache) init(dirname string, fs storage.Storage, opts *db.Options,
 	}
 }
 
-func (c *tableCache) newIters(meta *fileMetadata) (internalIterator, internalIterator, error) {
+func (c *tableCache) newIters(
+	meta *fileMetadata, opts *db.IterOptions,
+) (internalIterator, internalIterator, error) {
 	// Calling findNode gives us the responsibility of decrementing n's
 	// refCount. If opening the underlying table resulted in error, then we
 	// decrement this straight away. Otherwise, we pass that responsibility to
@@ -67,7 +69,7 @@ func (c *tableCache) newIters(meta *fileMetadata) (internalIterator, internalIte
 	}
 	n.result <- x
 
-	iter := x.reader.NewIter(nil)
+	iter := x.reader.NewIter(opts.GetLowerBound(), opts.GetUpperBound())
 	atomic.AddInt32(&c.mu.iterCount, 1)
 	if raceEnabled {
 		c.mu.Lock()
@@ -92,7 +94,7 @@ func (c *tableCache) newIters(meta *fileMetadata) (internalIterator, internalIte
 
 	// NB: range-del iterator does not maintain a reference to the table, nor
 	// does it need to read from it after creation.
-	if rangeDelIter := x.reader.NewRangeDelIter(nil); rangeDelIter != nil {
+	if rangeDelIter := x.reader.NewRangeDelIter(); rangeDelIter != nil {
 		return iter, rangeDelIter, nil
 	}
 	// NB: Translate a nil range-del iterator into a nil interface.


Test patch:
diff --git a/compaction_test.go b/compaction_test.go
index 333619a..471b124 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -615,7 +615,7 @@ func TestCompaction(t *testing.T) {
 				defer f.Close()
 				r := sstable.NewReader(f, meta.fileNum, nil)
 				defer r.Close()
-				ss = append(ss, get1(r.NewIter(nil))+".")
+				ss = append(ss, get1(r.NewIter(nil /* lower */, nil /* upper */))+".")
 			}
 		}
 		sort.Strings(ss)
diff --git a/get_iter_test.go b/get_iter_test.go
index c1e6f53..5890c52 100644
--- a/get_iter_test.go
+++ b/get_iter_test.go
@@ -440,7 +440,9 @@ func TestGetIter(t *testing.T) {
 
 		// m is a map from file numbers to DBs.
 		m := map[uint64]*memTable{}
-		newIter := func(meta *fileMetadata) (internalIterator, internalIterator, error) {
+		newIter := func(
+			meta *fileMetadata, _ *db.IterOptions,
+		) (internalIterator, internalIterator, error) {
 			d, ok := m[meta.fileNum]
 			if !ok {
 				return nil, nil, errors.New("no such file")
diff --git a/internal/arenaskl/skl_test.go b/internal/arenaskl/skl_test.go
index 62257f2..96c177d 100644
--- a/internal/arenaskl/skl_test.go
+++ b/internal/arenaskl/skl_test.go
@@ -61,7 +61,7 @@ func makeInserterAdd(s *Skiplist) func(key db.InternalKey, value []byte) error {
 func length(s *Skiplist) int {
 	count := 0
 
-	it := s.NewIter()
+	it := s.NewIter(nil, nil)
 	for valid := it.First(); valid; valid = it.Next() {
 		count++
 	}
@@ -73,7 +73,7 @@ func length(s *Skiplist) int {
 func lengthRev(s *Skiplist) int {
 	count := 0
 
-	it := s.NewIter()
+	it := s.NewIter(nil, nil)
 	for valid := it.Last(); valid; valid = it.Prev() {
 		count++
 	}
@@ -84,7 +84,7 @@ func lengthRev(s *Skiplist) int {
 func TestEmpty(t *testing.T) {
 	key := makeKey("aaa")
 	l := NewSkiplist(NewArena(arenaSize, 0), bytes.Compare)
-	it := l.NewIter()
+	it := l.NewIter(nil, nil)
 
 	require.False(t, it.Valid())
 
@@ -121,7 +121,7 @@ func TestBasic(t *testing.T) {
 	for _, inserter := range []bool{false, true} {
 		t.Run(fmt.Sprintf("inserter=%t", inserter), func(t *testing.T) {
 			l := NewSkiplist(NewArena(arenaSize, 0), bytes.Compare)
-			it := l.NewIter()
+			it := l.NewIter(nil, nil)
 
 			add := l.Add
 			if inserter {
@@ -216,7 +216,7 @@ func TestConcurrentBasic(t *testing.T) {
 				go func(i int) {
 					defer wg.Done()
 
-					it := l.NewIter()
+					it := l.NewIter(nil, nil)
 					require.True(t, it.SeekGE(makeKey(fmt.Sprintf("%05d", i))))
 					require.EqualValues(t, fmt.Sprintf("%05d", i), it.Key().UserKey)
 				}(i)
@@ -269,7 +269,7 @@ func TestConcurrentOneKey(t *testing.T) {
 				go func() {
 					defer wg.Done()
 
-					it := l.NewIter()
+					it := l.NewIter(nil, nil)
 					it.SeekGE(key)
 					require.True(t, it.Valid())
 					require.True(t, bytes.Equal(key, it.Key().UserKey))
@@ -292,7 +292,7 @@ func TestSkiplistAdd(t *testing.T) {
 	for _, inserter := range []bool{false, true} {
 		t.Run(fmt.Sprintf("inserter=%t", inserter), func(t *testing.T) {
 			l := NewSkiplist(NewArena(arenaSize, 0), bytes.Compare)
-			it := l.NewIter()
+			it := l.NewIter(nil, nil)
 
 			add := l.Add
 			if inserter {
@@ -307,7 +307,7 @@ func TestSkiplistAdd(t *testing.T) {
 			require.EqualValues(t, []byte{}, it.Value())
 
 			l = NewSkiplist(NewArena(arenaSize, 0), bytes.Compare)
-			it = l.NewIter()
+			it = l.NewIter(nil, nil)
 
 			add = l.Add
 			if inserter {
@@ -381,7 +381,7 @@ func TestConcurrentAdd(t *testing.T) {
 
 			for f := 0; f < 2; f++ {
 				go func() {
-					it := l.NewIter()
+					it := l.NewIter(nil, nil)
 					add := l.Add
 					if inserter {
 						add = makeInserterAdd(l)
@@ -416,7 +416,7 @@ func TestConcurrentAdd(t *testing.T) {
 func TestIteratorNext(t *testing.T) {
 	const n = 100
 	l := NewSkiplist(NewArena(arenaSize, 0), bytes.Compare)
-	it := l.NewIter()
+	it := l.NewIter(nil, nil)
 
 	require.False(t, it.Valid())
 
@@ -441,7 +441,7 @@ func TestIteratorNext(t *testing.T) {
 func TestIteratorPrev(t *testing.T) {
 	const n = 100
 	l := NewSkiplist(NewArena(arenaSize, 0), bytes.Compare)
-	it := l.NewIter()
+	it := l.NewIter(nil, nil)
 
 	require.False(t, it.Valid())
 
@@ -466,7 +466,7 @@ func TestIteratorPrev(t *testing.T) {
 func TestIteratorSeekGE(t *testing.T) {
 	const n = 100
 	l := NewSkiplist(NewArena(arenaSize, 0), bytes.Compare)
-	it := l.NewIter()
+	it := l.NewIter(nil, nil)
 
 	require.False(t, it.Valid())
 	it.First()
@@ -516,7 +516,7 @@ func TestIteratorSeekGE(t *testing.T) {
 func TestIteratorSeekLT(t *testing.T) {
 	const n = 100
 	l := NewSkiplist(NewArena(arenaSize, 0), bytes.Compare)
-	it := l.NewIter()
+	it := l.NewIter(nil, nil)
 
 	require.False(t, it.Valid())
 	it.First()
@@ -564,6 +564,75 @@ func TestIteratorSeekLT(t *testing.T) {
 	require.EqualValues(t, "", it.Key().UserKey)
 }
 
+// TODO(peter): test First and Last.
+func TestIteratorBounds(t *testing.T) {
+	l := NewSkiplist(NewArena(arenaSize, 0), bytes.Compare)
+	for i := 1; i < 10; i++ {
+		err := l.Add(makeIntKey(i), makeValue(i))
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	key := func(i int) []byte {
+		return makeIntKey(i).UserKey
+	}
+
+	it := l.NewIter(key(3), key(7))
+
+	// SeekGE within the lower and upper bound succeeds.
+	for i := 3; i <= 6; i++ {
+		k := key(i)
+		require.True(t, it.SeekGE(k))
+		require.EqualValues(t, string(k), string(it.Key().UserKey))
+	}
+
+	// SeekGE before the lower bound still succeeds (only the upper bound is
+	// checked).
+	for i := 1; i < 3; i++ {
+		k := key(i)
+		require.True(t, it.SeekGE(k))
+		require.EqualValues(t, string(k), string(it.Key().UserKey))
+	}
+
+	// SeekGE beyond the upper bound fails.
+	for i := 7; i < 10; i++ {
+		require.False(t, it.SeekGE(key(i)))
+	}
+
+	require.True(t, it.SeekGE(key(6)))
+	require.EqualValues(t, "00006", it.Key().UserKey)
+	require.EqualValues(t, "v00006", it.Value())
+
+	// Next into the upper bound fails.
+	require.False(t, it.Next())
+
+	// SeekLT within the lower and upper bound succeeds.
+	for i := 4; i <= 7; i++ {
+		require.True(t, it.SeekLT(key(i)))
+		require.EqualValues(t, string(key(i-1)), string(it.Key().UserKey))
+	}
+
+	// SeekLT beyond the upper bound still succeeds (only the lower bound is
+	// checked).
+	for i := 8; i < 9; i++ {
+		require.True(t, it.SeekLT(key(8)))
+		require.EqualValues(t, string(key(i-1)), string(it.Key().UserKey))
+	}
+
+	// SeekLT before the lower bound fails.
+	for i := 1; i < 4; i++ {
+		require.False(t, it.SeekLT(key(i)))
+	}
+
+	require.True(t, it.SeekLT(key(4)))
+	require.EqualValues(t, "00003", it.Key().UserKey)
+	require.EqualValues(t, "v00003", it.Value())
+
+	// Prev into the lower bound fails.
+	require.False(t, it.Prev())
+}
+
 func randomKey(rng *rand.Rand, b []byte) db.InternalKey {
 	key := rng.Uint32()
 	key2 := rng.Uint32()
@@ -582,7 +651,7 @@ func BenchmarkReadWrite(b *testing.B) {
 			b.ResetTimer()
 			var count int
 			b.RunParallel(func(pb *testing.PB) {
-				it := l.NewIter()
+				it := l.NewIter(nil, nil)
 				rng := rand.New(rand.NewSource(time.Now().UnixNano()))
 				buf := make([]byte, 8)
 
@@ -629,7 +698,7 @@ func BenchmarkIterNext(b *testing.B) {
 		}
 	}
 
-	it := l.NewIter()
+	it := l.NewIter(nil, nil)
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		if !it.Valid() {
@@ -649,7 +718,7 @@ func BenchmarkIterPrev(b *testing.B) {
 		}
 	}
 
-	it := l.NewIter()
+	it := l.NewIter(nil, nil)
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		if !it.Valid() {
diff --git a/internal/batchskl/skl_test.go b/internal/batchskl/skl_test.go
index e7572a3..24e9103 100644
--- a/internal/batchskl/skl_test.go
+++ b/internal/batchskl/skl_test.go
@@ -33,7 +33,7 @@ import (
 func length(s *Skiplist) int {
 	count := 0
 
-	it := s.NewIter()
+	it := s.NewIter(nil, nil)
 	for valid := it.First(); valid; valid = it.Next() {
 		count++
 	}
@@ -45,7 +45,7 @@ func length(s *Skiplist) int {
 func lengthRev(s *Skiplist) int {
 	count := 0
 
-	it := s.NewIter()
+	it := s.NewIter(nil, nil)
 	for valid := it.Last(); valid; valid = it.Prev() {
 		count++
 	}
@@ -82,7 +82,7 @@ func (d *testStorage) Compare(a []byte, b uint32) int {
 func TestEmpty(t *testing.T) {
 	key := makeKey("aaa")
 	l := NewSkiplist(&testStorage{}, 0)
-	it := l.NewIter()
+	it := l.NewIter(nil, nil)
 
 	require.False(t, it.Valid())
 
@@ -100,7 +100,7 @@ func TestEmpty(t *testing.T) {
 func TestBasic(t *testing.T) {
 	d := &testStorage{}
 	l := NewSkiplist(d, 0)
-	it := l.NewIter()
+	it := l.NewIter(nil, nil)
 
 	// Try adding values.
 	require.Nil(t, l.Add(d.add("key1")))
@@ -126,7 +126,7 @@ func TestBasic(t *testing.T) {
 func TestSkiplistAdd(t *testing.T) {
 	d := &testStorage{}
 	l := NewSkiplist(d, 0)
-	it := l.NewIter()
+	it := l.NewIter(nil, nil)
 
 	// Add empty key.
 	require.Nil(t, l.Add(d.add("")))
@@ -163,7 +163,7 @@ func TestIteratorNext(t *testing.T) {
 	const n = 100
 	d := &testStorage{}
 	l := NewSkiplist(d, 0)
-	it := l.NewIter()
+	it := l.NewIter(nil, nil)
 
 	require.False(t, it.Valid())
 
@@ -188,7 +188,7 @@ func TestIteratorPrev(t *testing.T) {
 	const n = 100
 	d := &testStorage{}
 	l := NewSkiplist(d, 0)
-	it := l.NewIter()
+	it := l.NewIter(nil, nil)
 
 	require.False(t, it.Valid())
 
@@ -212,7 +212,7 @@ func TestIteratorSeekGE(t *testing.T) {
 	const n = 1000
 	d := &testStorage{}
 	l := NewSkiplist(d, 0)
-	it := l.NewIter()
+	it := l.NewIter(nil, nil)
 
 	require.False(t, it.Valid())
 	it.First()
@@ -254,7 +254,7 @@ func TestIteratorSeekLT(t *testing.T) {
 	const n = 100
 	d := &testStorage{}
 	l := NewSkiplist(d, 0)
-	it := l.NewIter()
+	it := l.NewIter(nil, nil)
 
 	require.False(t, it.Valid())
 	it.First()
@@ -295,6 +295,74 @@ func TestIteratorSeekLT(t *testing.T) {
 	require.EqualValues(t, "", it.Key().UserKey)
 }
 
+// TODO(peter): test First and Last.
+func TestIteratorBounds(t *testing.T) {
+	d := &testStorage{}
+	l := NewSkiplist(d, 0)
+	for i := 1; i < 10; i++ {
+		err := l.Add(d.add(fmt.Sprintf("%05d", i)))
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	it := l.NewIter(makeKey("00003"), makeKey("00007"))
+
+	// SeekGE within the lower and upper bound succeeds.
+	for i := 3; i <= 6; i++ {
+		key := makeKey(fmt.Sprintf("%05d", i))
+		require.True(t, it.SeekGE(key))
+		require.EqualValues(t, string(key), string(it.Key().UserKey))
+	}
+
+	// SeekGE before the lower bound still succeeds (only the upper bound is
+	// checked).
+	for i := 1; i < 3; i++ {
+		key := makeKey(fmt.Sprintf("%05d", i))
+		require.True(t, it.SeekGE(key))
+		require.EqualValues(t, string(key), string(it.Key().UserKey))
+	}
+
+	// SeekGE beyond the upper bound fails.
+	for i := 7; i < 10; i++ {
+		key := makeKey(fmt.Sprintf("%05d", i))
+		require.False(t, it.SeekGE(key))
+	}
+
+	require.True(t, it.SeekGE(makeKey("00006")))
+	require.EqualValues(t, "00006", it.Key().UserKey)
+
+	// Next into the upper bound fails.
+	require.False(t, it.Next())
+
+	// SeekLT within the lower and upper bound succeeds.
+	for i := 4; i <= 7; i++ {
+		key := makeKey(fmt.Sprintf("%05d", i))
+		require.True(t, it.SeekLT(key))
+		require.EqualValues(t, fmt.Sprintf("%05d", i-1), string(it.Key().UserKey))
+	}
+
+	// SeekLT beyond the upper bound still succeeds (only the lower bound is
+	// checked).
+	for i := 8; i < 9; i++ {
+		key := makeKey(fmt.Sprintf("%05d", i))
+		require.True(t, it.SeekLT(key))
+		require.EqualValues(t, fmt.Sprintf("%05d", i-1), string(it.Key().UserKey))
+	}
+
+	// SeekLT before the lower bound fails.
+	for i := 1; i < 4; i++ {
+		key := makeKey(fmt.Sprintf("%05d", i))
+		require.False(t, it.SeekLT(key))
+	}
+
+	require.True(t, it.SeekLT(makeKey("00004")))
+	require.EqualValues(t, "00003", it.Key().UserKey)
+
+	// Prev into the lower bound fails.
+	require.False(t, it.Prev())
+}
+
 func randomKey(rng *rand.Rand, b []byte) []byte {
 	key := rng.Uint32()
 	key2 := rng.Uint32()
@@ -314,7 +382,7 @@ func BenchmarkReadWrite(b *testing.B) {
 				keys: make([][]byte, 0, b.N),
 			}
 			l := NewSkiplist(d, 0)
-			it := l.NewIter()
+			it := l.NewIter(nil, nil)
 			rng := rand.New(rand.NewSource(time.Now().UnixNano()))
 
 			b.ResetTimer()
@@ -348,7 +416,7 @@ func BenchmarkIterNext(b *testing.B) {
 		_ = l.Add(offset)
 	}
 
-	it := l.NewIter()
+	it := l.NewIter(nil, nil)
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		if !it.Valid() {
@@ -373,7 +441,7 @@ func BenchmarkIterPrev(b *testing.B) {
 		_ = l.Add(offset)
 	}
 
-	it := l.NewIter()
+	it := l.NewIter(nil, nil)
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		if !it.Valid() {
diff --git a/iterator_test.go b/iterator_test.go
index 74572ec..f2571da 100644
--- a/iterator_test.go
+++ b/iterator_test.go
@@ -32,9 +32,12 @@ var testKeyValuePairs = []string{
 }
 
 type fakeIter struct {
+	lower    []byte
+	upper    []byte
 	keys     []db.InternalKey
 	vals     [][]byte
 	index    int
+	valid    bool
 	closeErr error
 }
 
@@ -55,13 +58,19 @@ func newFakeIterator(closeErr error, keys ...string) *fakeIter {
 	return &fakeIter{
 		keys:     ikeys,
 		index:    0,
+		valid:    len(ikeys) > 0,
 		closeErr: closeErr,
 	}
 }
 
 func (f *fakeIter) SeekGE(key []byte) bool {
+	f.valid = false
 	for f.index = 0; f.index < len(f.keys); f.index++ {
 		if db.DefaultComparer.Compare(key, f.Key().UserKey) <= 0 {
+			if f.upper != nil && db.DefaultComparer.Compare(f.upper, f.Key().UserKey) <= 0 {
+				return false
+			}
+			f.valid = true
 			return true
 		}
 	}
@@ -69,8 +78,13 @@ func (f *fakeIter) SeekGE(key []byte) bool {
 }
 
 func (f *fakeIter) SeekLT(key []byte) bool {
+	f.valid = false
 	for f.index = len(f.keys) - 1; f.index >= 0; f.index-- {
 		if db.DefaultComparer.Compare(key, f.Key().UserKey) > 0 {
+			if f.lower != nil && db.DefaultComparer.Compare(f.lower, f.Key().UserKey) > 0 {
+				return false
+			}
+			f.valid = true
 			return true
 		}
 	}
@@ -78,29 +92,61 @@ func (f *fakeIter) SeekLT(key []byte) bool {
 }
 
 func (f *fakeIter) First() bool {
+	f.valid = false
 	f.index = -1
-	return f.Next()
+	if !f.Next() {
+		return false
+	}
+	if f.upper != nil && db.DefaultComparer.Compare(f.upper, f.Key().UserKey) <= 0 {
+		return false
+	}
+	f.valid = true
+	return true
 }
 
 func (f *fakeIter) Last() bool {
+	f.valid = false
 	f.index = len(f.keys)
-	return f.Prev()
+	if !f.Prev() {
+		return false
+	}
+	if f.lower != nil && db.DefaultComparer.Compare(f.lower, f.Key().UserKey) > 0 {
+		return false
+	}
+	f.valid = true
+	return true
 }
 
 func (f *fakeIter) Next() bool {
+	f.valid = false
 	if f.index == len(f.keys) {
 		return false
 	}
 	f.index++
-	return f.index < len(f.keys)
+	if f.index == len(f.keys) {
+		return false
+	}
+	if f.upper != nil && db.DefaultComparer.Compare(f.upper, f.Key().UserKey) <= 0 {
+		return false
+	}
+	f.valid = true
+	return true
 }
 
 func (f *fakeIter) Prev() bool {
+	f.valid = false
 	if f.index < 0 {
 		return false
 	}
 	f.index--
-	return f.index >= 0
+	if f.index < 0 {
+		return false
+	}
+	if f.lower != nil && db.DefaultComparer.Compare(f.lower, f.Key().UserKey) > 0 {
+		return false
+	}
+	f.valid = true
+	return true
 }
 
 func (f *fakeIter) Key() db.InternalKey {
@@ -115,7 +161,7 @@ func (f *fakeIter) Value() []byte {
 }
 
 func (f *fakeIter) Valid() bool {
-	return f.index >= 0 && f.index < len(f.keys)
+	return f.index >= 0 && f.index < len(f.keys) && f.valid
 }
 
 func (f *fakeIter) Error() error {
@@ -247,7 +293,12 @@ func TestIterator(t *testing.T) {
 		cmp := db.DefaultComparer.Compare
 		equal := db.DefaultComparer.Equal
 		// NB: Use a mergingIter to filter entries newer than seqNum.
-		iter := newMergingIter(cmp, &fakeIter{keys: keys, vals: vals})
+		iter := newMergingIter(cmp, &fakeIter{
+			lower: opts.GetLowerBound(),
+			upper: opts.GetUpperBound(),
+			keys:  keys,
+			vals:  vals,
+		})
 		iter.snapshot = seqNum
 		return &Iterator{
 			opts:  opts,
diff --git a/level_iter_test.go b/level_iter_test.go
index f723e57..0b1c3c5 100644
--- a/level_iter_test.go
+++ b/level_iter_test.go
@@ -24,7 +24,9 @@ func TestLevelIter(t *testing.T) {
 	var iters []*fakeIter
 	var files []fileMetadata
 
-	newIters := func(meta *fileMetadata) (internalIterator, internalIterator, error) {
+	newIters := func(
+		meta *fileMetadata, opts *db.IterOptions,
+	) (internalIterator, internalIterator, error) {
 		f := *iters[meta.fileNum]
 		return &f, nil, nil
 	}
@@ -74,6 +76,44 @@ func TestLevelIter(t *testing.T) {
 			defer iter.Close()
 			return runInternalIterCmd(d, iter)
 
+		case "load":
+			// The "load" command allows testing the iterator options passed to load
+			// sstables.
+			//
+			// load <key> [lower=<key>] [upper=<key>]
+			var opts db.IterOptions
+			var key string
+			for _, arg := range d.CmdArgs {
+				if len(arg.Vals) == 0 {
+					key = arg.Key
+					continue
+				}
+				if len(arg.Vals) != 1 {
+					return fmt.Sprintf("%s: %s=<value>", d.Cmd, arg.Key)
+				}
+				switch arg.Key {
+				case "lower":
+					opts.LowerBound = []byte(arg.Vals[0])
+				case "upper":
+					opts.UpperBound = []byte(arg.Vals[0])
+				default:
+					return fmt.Sprintf("%s: unknown arg: %s", d.Cmd, arg.Key)
+				}
+			}
+
+			var tableOpts *db.IterOptions
+			newIters2 := func(
+				meta *fileMetadata, opts *db.IterOptions,
+			) (internalIterator, internalIterator, error) {
+				tableOpts = opts
+				return newIters(meta, opts)
+			}
+
+			iter := newLevelIter(&opts, db.DefaultComparer.Compare, newIters2, files)
+			iter.SeekGE([]byte(key))
+			lower, upper := tableOpts.GetLowerBound(), tableOpts.GetUpperBound()
+			return fmt.Sprintf("[%s,%s]\n", lower, upper)
+
 		default:
 			return fmt.Sprintf("unknown command: %s", d.Cmd)
 		}
@@ -86,8 +126,10 @@ func TestLevelIterBoundaries(t *testing.T) {
 	var readers []*sstable.Reader
 	var files []fileMetadata
 
-	newIters := func(meta *fileMetadata) (internalIterator, internalIterator, error) {
-		return readers[meta.fileNum].NewIter(nil), nil, nil
+	newIters := func(
+		meta *fileMetadata, _ *db.IterOptions,
+	) (internalIterator, internalIterator, error) {
+		return readers[meta.fileNum].NewIter(nil /* lower */, nil /* upper */), nil, nil
 	}
 
 	datadriven.RunTest(t, "testdata/level_iter_boundaries", func(d *datadriven.TestData) string {
@@ -223,7 +265,7 @@ func buildLevelIterTables(
 
 	meta := make([]fileMetadata, len(readers))
 	for i := range readers {
-		iter := readers[i].NewIter(nil)
+		iter := readers[i].NewIter(nil /* lower */, nil /* upper */)
 		iter.First()
 		meta[i].fileNum = uint64(i)
 		meta[i].smallest = iter.Key()
@@ -243,8 +285,10 @@ func BenchmarkLevelIterSeekGE(b *testing.B) {
 					b.Run(fmt.Sprintf("count=%d", count),
 						func(b *testing.B) {
 							readers, files, keys := buildLevelIterTables(b, blockSize, restartInterval, count)
-							newIters := func(meta *fileMetadata) (internalIterator, internalIterator, error) {
-								return readers[meta.fileNum].NewIter(nil), nil, nil
+							newIters := func(
+								meta *fileMetadata, _ *db.IterOptions,
+							) (internalIterator, internalIterator, error) {
+								return readers[meta.fileNum].NewIter(nil /* lower */, nil /* upper */), nil, nil
 							}
 							l := newLevelIter(nil, db.DefaultComparer.Compare, newIters, files)
 							rng := rand.New(rand.NewSource(time.Now().UnixNano()))
@@ -269,8 +313,10 @@ func BenchmarkLevelIterNext(b *testing.B) {
 					b.Run(fmt.Sprintf("count=%d", count),
 						func(b *testing.B) {
 							readers, files, _ := buildLevelIterTables(b, blockSize, restartInterval, count)
-							newIters := func(meta *fileMetadata) (internalIterator, internalIterator, error) {
-								return readers[meta.fileNum].NewIter(nil), nil, nil
+							newIters := func(
+								meta *fileMetadata, _ *db.IterOptions,
+							) (internalIterator, internalIterator, error) {
+								return readers[meta.fileNum].NewIter(nil /* lower */, nil /* upper */), nil, nil
 							}
 							l := newLevelIter(nil, db.DefaultComparer.Compare, newIters, files)
 
@@ -297,8 +343,10 @@ func BenchmarkLevelIterPrev(b *testing.B) {
 					b.Run(fmt.Sprintf("count=%d", count),
 						func(b *testing.B) {
 							readers, files, _ := buildLevelIterTables(b, blockSize, restartInterval, count)
-							newIters := func(meta *fileMetadata) (internalIterator, internalIterator, error) {
-								return readers[meta.fileNum].NewIter(nil), nil, nil
+							newIters := func(
+								meta *fileMetadata, _ *db.IterOptions,
+							) (internalIterator, internalIterator, error) {
+								return readers[meta.fileNum].NewIter(nil /* lower */, nil /* upper */), nil, nil
 							}
 							l := newLevelIter(nil, db.DefaultComparer.Compare, newIters, files)
 
diff --git a/merging_iter_test.go b/merging_iter_test.go
index 2d319f4..268b048 100644
--- a/merging_iter_test.go
+++ b/merging_iter_test.go
@@ -199,7 +199,7 @@ func BenchmarkMergingIterSeekGE(b *testing.B) {
 							readers, keys := buildMergingIterTables(b, blockSize, restartInterval, count)
 							iters := make([]internalIterator, len(readers))
 							for i := range readers {
-								iters[i] = readers[i].NewIter(nil)
+								iters[i] = readers[i].NewIter(nil /* lower */, nil /* upper */)
 							}
 							m := newMergingIter(db.DefaultComparer.Compare, iters...)
 							rng := rand.New(rand.NewSource(time.Now().UnixNano()))
@@ -226,7 +226,7 @@ func BenchmarkMergingIterNext(b *testing.B) {
 							readers, _ := buildMergingIterTables(b, blockSize, restartInterval, count)
 							iters := make([]internalIterator, len(readers))
 							for i := range readers {
-								iters[i] = readers[i].NewIter(nil)
+								iters[i] = readers[i].NewIter(nil /* lower */, nil /* upper */)
 							}
 							m := newMergingIter(db.DefaultComparer.Compare, iters...)
 
@@ -255,7 +255,7 @@ func BenchmarkMergingIterPrev(b *testing.B) {
 							readers, _ := buildMergingIterTables(b, blockSize, restartInterval, count)
 							iters := make([]internalIterator, len(readers))
 							for i := range readers {
-								iters[i] = readers[i].NewIter(nil)
+								iters[i] = readers[i].NewIter(nil /* lower */, nil /* upper */)
 							}
 							m := newMergingIter(db.DefaultComparer.Compare, iters...)
 
diff --git a/sstable/reader_test.go b/sstable/reader_test.go
index f45809d..08a986a 100644
--- a/sstable/reader_test.go
+++ b/sstable/reader_test.go
@@ -122,7 +122,7 @@ func runTestReader(t *testing.T, o db.Options) {
 					}
 				}
 
-				iter := r.NewIter(nil)
+				iter := r.NewIter(nil /* lower */, nil /* upper */)
 				if err := iter.Error(); err != nil {
 					t.Fatal(err)
 				}
@@ -167,7 +167,7 @@ func runTestReader(t *testing.T, o db.Options) {
 			case "get":
 				var b bytes.Buffer
 				for _, k := range strings.Split(d.Input, "\n") {
-					v, err := r.get([]byte(k), nil)
+					v, err := r.get([]byte(k))
 					if err != nil {
 						fmt.Fprintf(&b, "<err: %s>\n", err)
 					} else {
@@ -227,7 +227,7 @@ func BenchmarkTableIterSeekGE(b *testing.B) {
 		b.Run(fmt.Sprintf("restart=%d", restartInterval),
 			func(b *testing.B) {
 				r, keys := buildBenchmarkTable(b, blockSize, restartInterval)
-				it := r.NewIter(nil)
+				it := r.NewIter(nil /* lower */, nil /* upper */)
 				rng := rand.New(rand.NewSource(time.Now().UnixNano()))
 
 				b.ResetTimer()
@@ -245,7 +245,7 @@ func BenchmarkTableIterSeekLT(b *testing.B) {
 		b.Run(fmt.Sprintf("restart=%d", restartInterval),
 			func(b *testing.B) {
 				r, keys := buildBenchmarkTable(b, blockSize, restartInterval)
-				it := r.NewIter(nil)
+				it := r.NewIter(nil /* lower */, nil /* upper */)
 				rng := rand.New(rand.NewSource(time.Now().UnixNano()))
 
 				b.ResetTimer()
@@ -263,7 +263,7 @@ func BenchmarkTableIterNext(b *testing.B) {
 		b.Run(fmt.Sprintf("restart=%d", restartInterval),
 			func(b *testing.B) {
 				r, _ := buildBenchmarkTable(b, blockSize, restartInterval)
-				it := r.NewIter(nil)
+				it := r.NewIter(nil /* lower */, nil /* upper */)
 
 				b.ResetTimer()
 				var sum int64
@@ -288,7 +288,7 @@ func BenchmarkTableIterPrev(b *testing.B) {
 		b.Run(fmt.Sprintf("restart=%d", restartInterval),
 			func(b *testing.B) {
 				r, _ := buildBenchmarkTable(b, blockSize, restartInterval)
-				it := r.NewIter(nil)
+				it := r.NewIter(nil /* lower */, nil /* upper */)
 
 				b.ResetTimer()
 				var sum int64
diff --git a/sstable/table_test.go b/sstable/table_test.go
index 588e20a..19467ac 100644
--- a/sstable/table_test.go
+++ b/sstable/table_test.go
@@ -10,11 +10,13 @@ import (
 	"encoding/binary"
 	"fmt"
 	"io"
+	"math/rand"
 	"os"
 	"path/filepath"
 	"sort"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/kr/pretty"
 	"github.com/petermattis/pebble/bloom"
@@ -119,16 +121,18 @@ func check(f storage.File, comparer *db.Comparer, fp db.FilterPolicy) error {
 	})
 
 	// Check that each key/value pair in wordCount is also in the table.
+	words := make([]string, 0, len(wordCount))
 	for k, v := range wordCount {
+		words = append(words, k)
 		// Check using Get.
-		if v1, err := r.get([]byte(k), nil); string(v1) != string(v) || err != nil {
+		if v1, err := r.get([]byte(k)); string(v1) != string(v) || err != nil {
 			return fmt.Errorf("Get %q: got (%q, %v), want (%q, %v)", k, v1, err, v, error(nil))
 		} else if len(v1) != cap(v1) {
 			return fmt.Errorf("Get %q: len(v1)=%d, cap(v1)=%d", k, len(v1), cap(v1))
 		}
 
 		// Check using SeekGE.
-		i := r.NewIter(nil)
+		i := r.NewIter(nil /* lower */, nil /* upper */)
 		if !i.SeekGE([]byte(k)) || string(i.Key().UserKey) != k {
 			return fmt.Errorf("Find %q: key was not in the table", k)
 		}
@@ -169,12 +173,12 @@ func check(f storage.File, comparer *db.Comparer, fp db.FilterPolicy) error {
 	// Check that nonsense words are not in the table.
 	for _, s := range nonsenseWords {
 		// Check using Get.
-		if _, err := r.get([]byte(s), nil); err != db.ErrNotFound {
+		if _, err := r.get([]byte(s)); err != db.ErrNotFound {
 			return fmt.Errorf("Get %q: got %v, want ErrNotFound", s, err)
 		}
 
 		// Check using Find.
-		i := r.NewIter(nil)
+		i := r.NewIter(nil /* lower */, nil /* upper */)
 		if i.SeekGE([]byte(s)) && s == string(i.Key().UserKey) {
 			return fmt.Errorf("Find %q: unexpectedly found key in the table", s)
 		}
@@ -200,7 +204,7 @@ func check(f storage.File, comparer *db.Comparer, fp db.FilterPolicy) error {
 		{0, "~"},
 	}
 	for _, ct := range countTests {
-		n, i := 0, r.NewIter(nil)
+		n, i := 0, r.NewIter(nil /* lower */, nil /* upper */)
 		for valid := i.SeekGE([]byte(ct.start)); valid; valid = i.Next() {
 			n++
 		}
@@ -222,6 +226,84 @@ func check(f storage.File, comparer *db.Comparer, fp db.FilterPolicy) error {
 		}
 	}
 
+	// Check lower/upper bounds behavior. Randomly choose a lower and upper bound
+	// and then guarantee that iteration finds the expected number if entries.
+	rng := rand.New(rand.NewSource(time.Now().UnixNano()))
+	sort.Strings(words)
+	for i := 0; i < 10; i++ {
+		lowerIdx := -1
+		upperIdx := len(words)
+		if rng.Intn(5) != 0 {
+			lowerIdx = rng.Intn(len(words))
+		}
+		if rng.Intn(5) != 0 {
+			upperIdx = rng.Intn(len(words))
+		}
+		if lowerIdx > upperIdx {
+			lowerIdx, upperIdx = upperIdx, lowerIdx
+		}
+
+		var lower, upper []byte
+		if lowerIdx >= 0 {
+			lower = []byte(words[lowerIdx])
+		} else {
+			lowerIdx = 0
+		}
+		if upperIdx < len(words) {
+			upper = []byte(words[upperIdx])
+		}
+
+		i := r.NewIter(lower, upper)
+
+		{
+			// NB: the semantics of First are that it starts iteration from the
+			// beginning, not respecting the lower bound.
+			n := 0
+			for valid := i.First(); valid; valid = i.Next() {
+				n++
+			}
+			if expected := upperIdx; expected != n {
+				return fmt.Errorf("expected %d, but found %d", expected, n)
+			}
+		}
+
+		{
+			// NB: the semantics of Last are that it starts iteration from the end, not
+			// respecting the upper bound.
+			n := 0
+			for valid := i.Last(); valid; valid = i.Prev() {
+				n++
+			}
+			if expected := len(words) - lowerIdx; expected != n {
+				return fmt.Errorf("expected %d, but found %d", expected, n)
+			}
+		}
+
+		if lower != nil {
+			n := 0
+			for valid := i.SeekGE(lower); valid; valid = i.Next() {
+				n++
+			}
+			if expected := upperIdx - lowerIdx; expected != n {
+				return fmt.Errorf("expected %d, but found %d", expected, n)
+			}
+		}
+
+		if upper != nil {
+			n := 0
+			for valid := i.SeekLT(upper); valid; valid = i.Prev() {
+				n++
+			}
+			if expected := upperIdx - lowerIdx; expected != n {
+				return fmt.Errorf("expected %d, but found %d", expected, n)
+			}
+		}
+
+		if err := i.Close(); err != nil {
+			return err
+		}
+	}
+
 	return r.Close()
 }
 
@@ -386,7 +468,7 @@ func TestBloomFilterFalsePositiveRate(t *testing.T) {
 	key := []byte("m!....")
 	for i := 0; i < n; i++ {
 		binary.LittleEndian.PutUint32(key[2:6], uint32(i))
-		r.get(key, nil)
+		r.get(key)
 	}
 
 	if c.truePositives != 0 {
@@ -500,7 +582,7 @@ func TestFinalBlockIsWritten(t *testing.T) {
 				continue
 			}
 			r := NewReader(rf, 0, nil)
-			i := r.NewIter(nil)
+			i := r.NewIter(nil /* lower */, nil /* upper */)
 			for valid := i.First(); valid; valid = i.Next() {
 				got++
 			}
@@ -532,7 +614,7 @@ func TestReaderGlobalSeqNum(t *testing.T) {
 	const globalSeqNum = 42
 	r.Properties.GlobalSeqNum = globalSeqNum
 
-	i := r.NewIter(nil)
+	i := r.NewIter(nil /* lower */, nil /* upper */)
 	for valid := i.First(); valid; valid = i.Next() {
 		if globalSeqNum != i.Key().SeqNum() {
 			t.Fatalf("expected %d, but found %d", globalSeqNum, i.Key().SeqNum())
diff --git a/sstable/writer_test.go b/sstable/writer_test.go
index 3583813..7f49a3a 100644
--- a/sstable/writer_test.go
+++ b/sstable/writer_test.go
@@ -129,7 +129,7 @@ func TestWriter(t *testing.T) {
 				meta.SmallestSeqNum, meta.LargestSeqNum)
 
 		case "scan":
-			iter := r.NewIter(nil)
+			iter := r.NewIter(nil /* lower */, nil /* upper */)
 			defer iter.Close()
 
 			var buf bytes.Buffer
@@ -139,7 +139,7 @@ func TestWriter(t *testing.T) {
 			return buf.String()
 
 		case "scan-range-del":
-			iter := r.NewRangeDelIter(nil)
+			iter := r.NewRangeDelIter()
 			if iter == nil {
 				return ""
 			}
diff --git a/table_cache_test.go b/table_cache_test.go
index 7ca3a4d..23be6fe 100644
--- a/table_cache_test.go
+++ b/table_cache_test.go
@@ -169,7 +169,7 @@ func testTableCacheRandomAccess(t *testing.T, concurrent bool) {
 			rngMu.Lock()
 			fileNum, sleepTime := rng.Intn(tableCacheTestNumTables), rng.Intn(1000)
 			rngMu.Unlock()
-			iter, _, err := c.newIters(&fileMetadata{fileNum: uint64(fileNum)})
+			iter, _, err := c.newIters(&fileMetadata{fileNum: uint64(fileNum)}, nil /* iter options */)
 			if err != nil {
 				errc <- fmt.Errorf("i=%d, fileNum=%d: find: %v", i, fileNum, err)
 				return
@@ -228,7 +228,7 @@ func TestTableCacheFrequentlyUsed(t *testing.T) {
 
 	for i := 0; i < N; i++ {
 		for _, j := range [...]int{pinned0, i % tableCacheTestNumTables, pinned1} {
-			iter, _, err := c.newIters(&fileMetadata{fileNum: uint64(j)})
+			iter, _, err := c.newIters(&fileMetadata{fileNum: uint64(j)}, nil /* iter options */)
 			if err != nil {
 				t.Fatalf("i=%d, j=%d: find: %v", i, j, err)
 			}
@@ -263,7 +263,7 @@ func TestTableCacheEvictions(t *testing.T) {
 	rng := rand.New(rand.NewSource(2))
 	for i := 0; i < N; i++ {
 		j := rng.Intn(tableCacheTestNumTables)
-		iter, _, err := c.newIters(&fileMetadata{fileNum: uint64(j)})
+		iter, _, err := c.newIters(&fileMetadata{fileNum: uint64(j)}, nil /* iter options */)
 		if err != nil {
 			t.Fatalf("i=%d, j=%d: find: %v", i, j, err)
 		}
@@ -303,7 +303,7 @@ func TestTableCacheIterLeak(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if _, _, err := c.newIters(&fileMetadata{fileNum: 0}); err != nil {
+	if _, _, err := c.newIters(&fileMetadata{fileNum: 0}, nil /* iter options */); err != nil {
 		t.Fatal(err)
 	}
 	if err := c.Close(); err == nil {
diff --git a/testdata/level_iter b/testdata/level_iter
index 8f3ac49..8212b99 100644
--- a/testdata/level_iter
+++ b/testdata/level_iter
@@ -100,6 +100,27 @@ first
 a:1
 a:1
 
+# levelIter trims lower/upper bound in the options passed to sstables.
+load a
+----
+[,]
+
+load a lower=a upper=b
+----
+[a,b]
+
+load a lower=a upper=c
+----
+[a,]
+
+load c lower=b upper=d
+----
+[,d]
+
+load c lower=b upper=e
+----
+[,]
+
 # levelIter only checks lower bound when loading sstables.
 iter lower=b
 seek-ge a