Repository: petermattis/pebble
Pull request #62 (2019-03-18 04:56:56)

Fix iterator bug when range tombstone ends after its sstable
A range tombstone's end key may extend past its containing sstable's
upper bound. In that case, the portion of the range tombstone outside
the sstable's boundaries does not necessarily cover keys in lower levels.
The iterator optimization for seeking past keys covered by a range
tombstone in lower levels should take this into account and limit its
seek to the sstable's upper bound.

The approach taken here is to maintain the sstable upper bounds in
`mergingIter` for the partitioned levels. `levelIter` holds a pointer to
the upper bound for its level and updates it when the current sstable
changes. `mergingIter` then uses this info to prevent seeking too far in
lower levels upon encountering a range tombstone.

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

petermattis: Apologies for the delay in reviewing this. The approach generally looks good, though I need a bit more time to scrutinize the details. The CI failure is something I've seen before and seems to be go1.12 specific, though I couldn't reproduce it. Perhaps you can.
ajkr: Got it. This failed three times:

```
$ go version
go version go1.11.6 linux/amd64
$ go test ./internal/arenaskl/ -run=TestConcurrentOneKey/inserter=true -count=20000
...
```

This failed 11 times:
```
$ go version
go version go1.12.1 linux/amd64
$ go test ./internal/arenaskl/ -run=TestConcurrentOneKey/inserter=true -count=20000
...
```
So I guess it's more frequent with go1.12, but seems to be an issue in go1.11 too.
ajkr: Fixed in https://github.com/petermattis/pebble/pull/64.
ajkr: 



___
*[level_iter.go, line 43 at r1](https://reviewable.io/reviews/petermattis/pebble/62#-LbEClA_BPuvP1DnKp9n:-LbEKIiYDm750o83czdD:b-8gmeul) ([raw file](https://github.com/petermattis/pebble/blob/2dcfed983054070fe4bc4624ba3ffd13a3cc3b18/level_iter.go#L43)):*
> struggling a bit with the name partitionUpperBound. Do we use the term partition to refer to tables elsewhere?  When non-nil, we initialize this field to point to the largest user-key in the table. Perhaps largestUserKey would be a more explicit name. I'm open to hearing arguments about why partitionUpperBound is more appropriate.

I think some papers call the strategy we use in L1+ "partitioned leveling" to distinguish from the strategy in the classic level compaction paper, which did Ln->Ln+1 compaction by compacting all the data in Ln with all the data in Ln+1. The name `partitionUpperBound` is kind of convenient here because we don't need it populated for L0 files, and since we don't partition L0, it makes sense that `partitionUpperBound` would be null for them.


<!-- Sent from Reviewable.io -->

ajkr: 



___
*[level_iter.go, line 43 at r1](https://reviewable.io/reviews/petermattis/pebble/62#-LbEClA_BPuvP1DnKp9n:-LbERH2537r93fF-cBmV:bshj231) ([raw file](https://github.com/petermattis/pebble/blob/2dcfed983054070fe4bc4624ba3ffd13a3cc3b18/level_iter.go#L43)):*
<details><summary><i>Previously, ajkr (Andrew Kryczka) wrote…</i></summary><blockquote>

> struggling a bit with the name partitionUpperBound. Do we use the term partition to refer to tables elsewhere?  When non-nil, we initialize this field to point to the largest user-key in the table. Perhaps largestUserKey would be a more explicit name. I'm open to hearing arguments about why partitionUpperBound is more appropriate.

I think some papers call the strategy we use in L1+ "partitioned leveling" to distinguish from the strategy in the classic level compaction paper, which did Ln->Ln+1 compaction by compacting all the data in Ln with all the data in Ln+1. The name `partitionUpperBound` is kind of convenient here because we don't need it populated for L0 files, and since we don't partition L0, it makes sense that `partitionUpperBound` would be null for them.
</blockquote></details>

I mean the merging iterator's `partitionUpperBounds[i]` would be null for `i` referring to unpartitioned "levels" (memtable or L0 files). Within level iterator, the only downside with renaming to `largestUserKey` I can think of is it looks different from what we call it in merging iterator. Did you prefer changing it in both places or just in level iterator?


<!-- Sent from Reviewable.io -->


Patch:
diff --git a/db.go b/db.go
index 0e456e4..8b42e69 100644
--- a/db.go
+++ b/db.go
@@ -402,11 +402,12 @@ func (d *DB) newIterInternal(
 	// Bundle various structures under a single umbrella in order to allocate
 	// them together.
 	var buf struct {
-		dbi           Iterator
-		merging       mergingIter
-		iters         [3 + numLevels]internalIterator
-		rangeDelIters [3 + numLevels]internalIterator
-		levels        [numLevels]levelIter
+		dbi             Iterator
+		merging         mergingIter
+		iters           [3 + numLevels]internalIterator
+		rangeDelIters   [3 + numLevels]internalIterator
+		largestUserKeys [3 + numLevels][]byte
+		levels          [numLevels]levelIter
 	}
 
 	dbi := &buf.dbi
@@ -418,9 +419,11 @@ func (d *DB) newIterInternal(
 
 	iters := buf.iters[:0]
 	rangeDelIters := buf.rangeDelIters[:0]
+	largestUserKeys := buf.largestUserKeys[:0]
 	if batchIter != nil {
 		iters = append(iters, batchIter)
 		rangeDelIters = append(rangeDelIters, batchRangeDelIter)
+		largestUserKeys = append(largestUserKeys, nil)
 	}
 
 	// TODO(peter): We only need to add memtables which contain sequence numbers
@@ -430,6 +433,7 @@ func (d *DB) newIterInternal(
 		mem := memtables[i]
 		iters = append(iters, mem.newIter(o))
 		rangeDelIters = append(rangeDelIters, mem.newRangeDelIter(o))
+		largestUserKeys = append(largestUserKeys, nil)
 	}
 
 	// The level 0 files need to be added from newest to oldest.
@@ -442,6 +446,7 @@ func (d *DB) newIterInternal(
 		}
 		iters = append(iters, iter)
 		rangeDelIters = append(rangeDelIters, rangeDelIter)
+		largestUserKeys = append(largestUserKeys, nil)
 	}
 
 	start := len(rangeDelIters)
@@ -450,9 +455,12 @@ func (d *DB) newIterInternal(
 			continue
 		}
 		rangeDelIters = append(rangeDelIters, nil)
+		largestUserKeys = append(largestUserKeys, nil)
 	}
 	buf.merging.rangeDelIters = rangeDelIters
+	buf.merging.largestUserKeys = largestUserKeys
 	rangeDelIters = rangeDelIters[start:]
+	largestUserKeys = largestUserKeys[start:]
 
 	// Add level iterators for the remaining files.
 	levels := buf.levels[:]
@@ -471,8 +479,10 @@ func (d *DB) newIterInternal(
 
 		li.init(o, d.cmp, d.newIters, current.files[level])
 		li.initRangeDel(&rangeDelIters[0])
+		li.initLargestUserKey(&largestUserKeys[0])
 		iters = append(iters, li)
 		rangeDelIters = rangeDelIters[1:]
+		largestUserKeys = largestUserKeys[1:]
 	}
 
 	buf.merging.init(d.cmp, iters...)
diff --git a/level_iter.go b/level_iter.go
index c1e5268..916e04c 100644
--- a/level_iter.go
+++ b/level_iter.go
@@ -40,6 +40,18 @@ type levelIter struct {
 	rangeDelIter *internalIterator
 	files        []fileMetadata
 	err          error
+	// Pointer into this level's entry in `mergingIter::largestUserKeys`. We populate it
+	// with the largest user key for the currently opened file. It is used to limit the optimization
+	// that seeks lower-level iterators past keys shadowed by a range tombstone. Limiting this
+	// seek to the file upper-bound is necessary since range tombstones are stored untruncated,
+	// while they only apply to keys within their containing file's boundaries. For a detailed
+	// example, see comment above `mergingIter`.
+	//
+	// This field differs from the `boundary` field in a few ways:
+	// - `boundary` is only populated when the iterator is positioned exactly on the sentinel key.
+	// - `boundary` can hold either the lower- or upper-bound, depending on the iterator direction.
+	// - `boundary` is not exposed to the next higher-level iterator, i.e., `mergingIter`.
+	largestUserKey *[]byte
 }
 
 // levelIter implements the internalIterator interface.
@@ -67,6 +79,10 @@ func (l *levelIter) initRangeDel(rangeDelIter *internalIterator) {
 	l.rangeDelIter = rangeDelIter
 }
 
+func (l *levelIter) initLargestUserKey(largestUserKey *[]byte) {
+	l.largestUserKey = largestUserKey
+}
+
 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
@@ -144,6 +160,9 @@ func (l *levelIter) loadFile(index, dir int) bool {
 		if l.rangeDelIter != nil {
 			*l.rangeDelIter = rangeDelIter
 		}
+		if l.largestUserKey != nil {
+			*l.largestUserKey = f.largest.UserKey
+		}
 		return true
 	}
 }
diff --git a/merging_iter.go b/merging_iter.go
index 9c61e65..8314bcc 100644
--- a/merging_iter.go
+++ b/merging_iter.go
@@ -104,6 +104,18 @@ import (
 // "e" and find the range tombstone [d,h) and similar logic holds. By the time
 // we get to level 4 we're seeking to "n".
 //
+// One consequence of not truncating tombstone end keys to sstable boundaries is
+// the seeking process described above cannot always seek to the tombstone end
+// key in the older level. For example, imagine in the above example r3 is a
+// partitioned level (i.e., L1+ in our LSM), and the sstable containing [j, n)
+// has "k" as its upper boundary. In this situation, compactions involving keys
+// at or after "k" can output those keys to r4+, even if they're newer than our
+// tombstone [j, n). So instead of seeking to "n" in r4 we can only seek to "k".
+// To achieve this, the instance variable `largestUserKeys` maintains the
+// upper bounds of the current sstables in the partitioned levels. In this
+// example, `largestUserKeys[3]` holds "k", telling us to limit the seek
+// triggered by a tombstone in r3 to "k".
+//
 // During actual iteration levels can contain both point operations and range
 // deletions. Within a level, when a range deletion contains a point operation
 // the sequence numbers must must be checked to determine if the point
@@ -164,12 +176,13 @@ import (
 // scenarios and have each step display the current state (i.e. the current
 // heap and range-del iterator positioning).
 type mergingIter struct {
-	dir           int
-	snapshot      uint64
-	iters         []internalIterator
-	rangeDelIters []internalIterator
-	heap          mergingIterHeap
-	err           error
+	dir             int
+	snapshot        uint64
+	iters           []internalIterator
+	rangeDelIters   []internalIterator
+	largestUserKeys [][]byte
+	heap            mergingIterHeap
+	err             error
 }
 
 // mergingIter implements the internalIterator interface.
@@ -507,7 +520,12 @@ func (m *mergingIter) seekGE(key []byte, level int) {
 				// the search key.
 				tombstone := rangedel.SeekGE(m.heap.cmp, rangeDelIter, key, m.snapshot)
 				if !tombstone.Empty() && tombstone.Contains(m.heap.cmp, key) {
-					key = tombstone.End
+					if m.largestUserKeys[level] != nil &&
+						m.heap.cmp(m.largestUserKeys[level], tombstone.End) < 0 {
+						key = m.largestUserKeys[level]
+					} else {
+						key = tombstone.End
+					}
 				}
 			}
 		}


Test patch:
diff --git a/compaction_test.go b/compaction_test.go
index e1e4107..f703f3e 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -747,14 +747,7 @@ func TestManualCompaction(t *testing.T) {
 			return b.String()
 
 		case "compact":
-			if len(td.CmdArgs) != 1 {
-				return fmt.Sprintf("%s expects 1 argument", td.Cmd)
-			}
-			parts := strings.Split(td.CmdArgs[0].Key, "-")
-			if len(parts) != 2 {
-				return fmt.Sprintf("malformed test case: %s", td.Input)
-			}
-			if err := d.Compact([]byte(parts[0]), []byte(parts[1])); err != nil {
+			if err := runCompactCommand(td, d); err != nil {
 				return err.Error()
 			}
 
diff --git a/data_test.go b/data_test.go
index 0a19709..dac8627 100644
--- a/data_test.go
+++ b/data_test.go
@@ -156,19 +156,83 @@ func runBatchDefineCmd(d *datadriven.TestData, b *Batch) error {
 	return nil
 }
 
+func runCompactCommand(td *datadriven.TestData, d *DB) error {
+	if len(td.CmdArgs) > 2 {
+		return fmt.Errorf("%s expects at most two arguments", td.Cmd)
+	}
+	parts := strings.Split(td.CmdArgs[0].Key, "-")
+	if len(parts) != 2 {
+		return fmt.Errorf("expected <begin>-<end>: %s", td.Input)
+	}
+	if len(td.CmdArgs) == 2 {
+		levelString := td.CmdArgs[1].String()
+		iStart := db.MakeInternalKey([]byte(parts[0]), db.InternalKeySeqNumMax, db.InternalKeyKindMax)
+		iEnd := db.MakeInternalKey([]byte(parts[1]), 0, 0)
+		if levelString[0] != 'L' {
+			return fmt.Errorf("expected L<n>: %s", levelString)
+		}
+		level, err := strconv.Atoi(levelString[1:])
+		if err != nil {
+			return err
+		}
+		return d.manualCompact(&manualCompaction{
+			done:  make(chan error, 1),
+			level: level,
+			start: iStart,
+			end:   iEnd,
+		})
+	} else {
+		return d.Compact([]byte(parts[0]), []byte(parts[1]))
+	}
+}
+
 func runDBDefineCmd(td *datadriven.TestData) (*DB, error) {
 	if td.Input == "" {
 		return nil, fmt.Errorf("empty test input")
 	}
 
 	fs := storage.NewMem()
-	d, err := Open("", &db.Options{
+	opts := db.Options{
 		Storage: fs,
-	})
+	}
+	var snapshots []uint64
+	for _, arg := range td.CmdArgs {
+		switch arg.Key {
+		case "target-file-sizes":
+			opts.Levels = make([]db.LevelOptions, len(arg.Vals))
+			for i := range arg.Vals {
+				size, err := strconv.ParseInt(arg.Vals[i], 10, 64)
+				if err != nil {
+					return nil, err
+				}
+				opts.Levels[i].TargetFileSize = size
+			}
+		case "snapshots":
+			snapshots = make([]uint64, len(arg.Vals))
+			for i := range arg.Vals {
+				seqNum, err := strconv.ParseUint(arg.Vals[i], 10, 64)
+				if err != nil {
+					return nil, err
+				}
+				snapshots[i] = seqNum
+				if i > 0 && snapshots[i] < snapshots[i-1] {
+					return nil, fmt.Errorf("Snapshots must be in ascending order")
+				}
+			}
+		default:
+			return nil, fmt.Errorf("%s: unknown arg: %s", td.Cmd, arg.Key)
+		}
+	}
+	d, err := Open("", &opts)
 	if err != nil {
 		return nil, err
 	}
 	d.mu.Lock()
+	for i := range snapshots {
+		s := &Snapshot{db: d}
+		s.seqNum = snapshots[i]
+		d.mu.snapshots.pushBack(s)
+	}
 	defer d.mu.Unlock()
 
 	var mem *memTable
diff --git a/range_del_test.go b/range_del_test.go
index 05970ac..d86ce61 100644
--- a/range_del_test.go
+++ b/range_del_test.go
@@ -34,6 +34,15 @@ func TestRangeDel(t *testing.T) {
 			d.mu.Unlock()
 			return s
 
+		case "compact":
+			if err := runCompactCommand(td, d); err != nil {
+				return err.Error()
+			}
+			d.mu.Lock()
+			s := d.mu.versions.currentVersion().String()
+			d.mu.Unlock()
+			return s
+
 		case "get":
 			snap := Snapshot{
 				db:     d,
diff --git a/testdata/range_del b/testdata/range_del
index 757cf86..4db52bf 100644
--- a/testdata/range_del
+++ b/testdata/range_del
@@ -1127,3 +1127,68 @@ get seq=3
 a
 ----
 pebble: not found
+
+# A range tombstone straddles two SSTs. One is compacted to a lower level. Its
+# keys that are newer than the range tombstone should not disappear.
+#
+# Uses a snapshot to prevent range tombstone from being elided when it gets
+# compacted to the bottommost level.
+
+define target-file-sizes=(100, 1) snapshots=(1)
+L0
+  a.RANGEDEL.1:e
+L0
+  a.SET.2:v
+L0
+  c.SET.3:v
+----
+mem: 1
+0: a-e a-a c-c
+
+compact a-e
+----
+1: a-c c-e
+
+compact d-e
+----
+1: a-c
+2: c-e
+
+iter seq=4
+seek-ge b
+next
+----
+c:v
+.
+
+# Reverse the above test: compact the left file containing the split range
+# tombstone downwards, and iterate from right to left.
+
+define target-file-sizes=(100, 1) snapshots=(1)
+L0
+  a.RANGEDEL.1:e
+L0
+  a.SET.2:v
+L0
+  c.SET.3:v
+----
+mem: 1
+0: a-e a-a c-c
+
+compact a-e
+----
+1: a-c c-e
+
+compact a-b
+----
+1: c-e
+2: a-c
+
+iter seq=4
+seek-lt d
+prev
+prev
+----
+c:v
+a:v
+.