Repository: petermattis/pebble (go)

Issue:
DeleteRange truncation issues
There are a couple issues with range tombstone truncation exposed in these test cases: https://gist.github.com/ajkr/9619663ad7dc5586ac79b29bbc06921f

I believe the current approach (implicit truncation) works nicely for point lookups and iterator next/prev. But the above test cases shows it doesn't handle all cases of iterator seek or writing compaction outputs.

For iterator seek, it should be enough to limit the seek over shadowed data to the sstable boundaries.

For writing compaction outputs it is a bit more difficult. We need to truncate tombstones before writing them but can't simply use the input sstables' boundaries due to cases like https://github.com/facebook/rocksdb/pull/4432#discussion_r221072219. In RocksDB we solved it by truncating to the compaction unit boundary. I believe the analogous thing should work here, even though compaction units are computed differently.

Pull request description:
Prevent sstable boundaries from expanding during compaction
Enforce the invariant that the boundaries of the output of a compaction
can be larger than the boundaries of the input. This fixes a problem
where a compaction can cause a range tombstone to expand to cover a
newer key at a lower level in the LSM.

Fixes #26
Closes #27

Patch:
diff --git a/compaction.go b/compaction.go
index 8083835..13ee4cd 100644
--- a/compaction.go
+++ b/compaction.go
@@ -14,6 +14,7 @@ import (
 	"unsafe"
 
 	"github.com/petermattis/pebble/db"
+	"github.com/petermattis/pebble/internal/rangedel"
 	"github.com/petermattis/pebble/sstable"
 	"github.com/petermattis/pebble/storage"
 )
@@ -121,7 +122,7 @@ func (c *compaction) expandInputs(inputs []fileMetadata) []fileMetadata {
 	end := start + len(inputs)
 	for ; end < len(files); end++ {
 		cur := &files[end-1]
-		next := files[end]
+		next := &files[end]
 		if c.cmp(cur.largest.UserKey, next.smallest.UserKey) < 0 {
 			break
 		}
@@ -232,6 +233,57 @@ func (c *compaction) elideRangeTombstone(start, end []byte) bool {
 	return true
 }
 
+// atomicUnitBounds returns the bounds of the atomic compaction unit containing
+// the specified sstable (identified by a pointer to its fileMetadata).
+func (c *compaction) atomicUnitBounds(f *fileMetadata) (lower, upper []byte) {
+	for i := range c.inputs {
+		files := c.inputs[i]
+		for j := range files {
+			if f == &files[j] {
+				lowerBound := f.smallest.UserKey
+				for k := j; k > 0; k-- {
+					cur := &files[k]
+					prev := &files[k-1]
+					if c.cmp(prev.largest.UserKey, cur.smallest.UserKey) < 0 {
+						break
+					}
+					if prev.largest.Trailer == db.InternalKeyRangeDeleteSentinel {
+						// The range deletion sentinel key is set for the largest key in a
+						// table when a range deletion tombstone straddles a table. It
+						// isn't necessary to include the next table in the atomic
+						// compaction unit as cur.largest.UserKey does not actually exist
+						// in the table.
+						break
+					}
+					lowerBound = prev.smallest.UserKey
+				}
+
+				upperBound := f.largest.UserKey
+				for k := j + 1; k < len(files); k++ {
+					cur := &files[k-1]
+					next := &files[k]
+					if c.cmp(cur.largest.UserKey, next.smallest.UserKey) < 0 {
+						break
+					}
+					if cur.largest.Trailer == db.InternalKeyRangeDeleteSentinel {
+						// The range deletion sentinel key is set for the largest key in a
+						// table when a range deletion tombstone straddles a table. It
+						// isn't necessary to include the next table in the atomic
+						// compaction unit as cur.largest.UserKey does not actually exist
+						// in the table.
+						break
+					}
+					// cur.largest.UserKey == next.largest.UserKey, so next is part of
+					// the atomic compaction unit.
+					upperBound = next.largest.UserKey
+				}
+				return lowerBound, upperBound
+			}
+		}
+	}
+	return nil, nil
+}
+
 // newInputIter returns an iterator over all the input tables in a compaction.
 func (c *compaction) newInputIter(
 	newIters tableNewIters,
@@ -267,11 +319,17 @@ func (c *compaction) newInputIter(
 				rangeDelIter = nil
 			}
 		}
+		if rangeDelIter != nil {
+			// Truncate the range tombstones returned by the iterator to the upper
+			// bound of the atomic compaction unit.
+			lowerBound, upperBound := c.atomicUnitBounds(f)
+			if lowerBound != nil || upperBound != nil {
+				rangeDelIter = rangedel.Truncate(c.cmp, rangeDelIter, lowerBound, upperBound)
+			}
+		}
 		return rangeDelIter, nil, err
 	}
 
-	// TODO(peter,rangedel): test that range tombstones are properly included in
-	// the output sstable.
 	if c.level != 0 {
 		iters = append(iters, newLevelIter(nil, c.cmp, newIters, c.inputs[0]))
 		iters = append(iters, newLevelIter(nil, c.cmp, newRangeDelIter, c.inputs[0]))
diff --git a/internal/rangedel/get.go b/internal/rangedel/get.go
index 3096817..2b214a9 100644
--- a/internal/rangedel/get.go
+++ b/internal/rangedel/get.go
@@ -15,6 +15,9 @@ type iterator interface {
 	// than the given key.
 	SeekLT(key []byte) bool
 
+	// First moves the iterator the the first key/value pair.
+	First() bool
+
 	// Last moves the iterator the the last key/value pair.
 	Last() bool
 
diff --git a/internal/rangedel/truncate.go b/internal/rangedel/truncate.go
new file mode 100644
index 0000000..540f121
--- /dev/null
+++ b/internal/rangedel/truncate.go
@@ -0,0 +1,31 @@
+// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package rangedel
+
+import (
+	"github.com/petermattis/pebble/db"
+)
+
+// Truncate creates a new iterator where every tombstone in the supplied
+// iterator is truncated to be contained within the range [lower, upper).
+func Truncate(cmp db.Compare, iter iterator, lower, upper []byte) *Iter {
+	var tombstones []Tombstone
+	for valid := iter.First(); valid; valid = iter.Next() {
+		t := Tombstone{
+			Start: iter.Key(),
+			End:   iter.Value(),
+		}
+		if cmp(t.Start.UserKey, lower) < 0 {
+			t.Start.UserKey = lower
+		}
+		if cmp(t.End, upper) > 0 {
+			t.End = upper
+		}
+		if cmp(t.Start.UserKey, t.End) < 0 {
+			tombstones = append(tombstones, t)
+		}
+	}
+	return NewIter(cmp, tombstones)
+}


Test patch:
diff --git a/compaction_test.go b/compaction_test.go
index f703f3e..333619a 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -900,3 +900,56 @@ func TestCompactionExpandInputs(t *testing.T) {
 			}
 		})
 }
+
+func TestCompactionAtomicUnitBounds(t *testing.T) {
+	cmp := db.DefaultComparer.Compare
+	var files []fileMetadata
+
+	parseMeta := func(s string) fileMetadata {
+		parts := strings.Split(s, "-")
+		if len(parts) != 2 {
+			t.Fatalf("malformed table spec: %s", s)
+		}
+		return fileMetadata{
+			smallest: db.ParseInternalKey(parts[0]),
+			largest:  db.ParseInternalKey(parts[1]),
+		}
+	}
+
+	datadriven.RunTest(t, "testdata/compaction_atomic_unit_bounds",
+		func(d *datadriven.TestData) string {
+			switch d.Cmd {
+			case "define":
+				files = nil
+				if len(d.Input) == 0 {
+					return ""
+				}
+				for _, data := range strings.Split(d.Input, "\n") {
+					meta := parseMeta(data)
+					meta.fileNum = uint64(len(files))
+					files = append(files, meta)
+				}
+				sort.Sort(bySmallest{files, cmp})
+				return ""
+
+			case "atomic-unit-bounds":
+				c := &compaction{
+					cmp: cmp,
+				}
+				c.inputs[0] = files
+				if len(d.CmdArgs) != 1 {
+					return fmt.Sprintf("%s expects 1 argument", d.Cmd)
+				}
+				index, err := strconv.ParseInt(d.CmdArgs[0].String(), 10, 64)
+				if err != nil {
+					return err.Error()
+				}
+
+				lower, upper := c.atomicUnitBounds(&files[index])
+				return fmt.Sprintf("%s-%s\n", lower, upper)
+
+			default:
+				return fmt.Sprintf("unknown command: %s", d.Cmd)
+			}
+		})
+}
diff --git a/internal/rangedel/testdata/truncate b/internal/rangedel/testdata/truncate
new file mode 100644
index 0000000..a531ba6
--- /dev/null
+++ b/internal/rangedel/testdata/truncate
@@ -0,0 +1,166 @@
+build
+1:  b-d
+2:  d-f
+3:  f-h
+----
+1:  b-d
+2:    d-f
+3:      f-h
+
+
+truncate a-b
+----
+
+truncate a-c
+----
+1:  bc
+
+truncate a-d
+----
+1:  b-d
+
+truncate a-e
+----
+1:  b-d
+2:    de
+
+truncate a-f
+----
+1:  b-d
+2:    d-f
+
+truncate a-g
+----
+1:  b-d
+2:    d-f
+3:      fg
+
+truncate a-h
+----
+1:  b-d
+2:    d-f
+3:      f-h
+
+
+truncate b-b
+----
+
+truncate b-c
+----
+1:  bc
+
+truncate b-d
+----
+1:  b-d
+
+truncate b-e
+----
+1:  b-d
+2:    de
+
+truncate b-f
+----
+1:  b-d
+2:    d-f
+
+truncate b-g
+----
+1:  b-d
+2:    d-f
+3:      fg
+
+truncate b-h
+----
+1:  b-d
+2:    d-f
+3:      f-h
+
+
+truncate c-c
+----
+
+truncate c-d
+----
+1:   cd
+
+truncate c-e
+----
+1:   cd
+2:    de
+
+truncate c-f
+----
+1:   cd
+2:    d-f
+
+truncate c-g
+----
+1:   cd
+2:    d-f
+3:      fg
+
+truncate c-h
+----
+1:   cd
+2:    d-f
+3:      f-h
+
+
+truncate d-d
+----
+
+truncate d-e
+----
+2:    de
+
+truncate d-f
+----
+2:    d-f
+
+truncate d-g
+----
+2:    d-f
+3:      fg
+
+truncate d-h
+----
+2:    d-f
+3:      f-h
+
+
+truncate e-e
+----
+
+truncate e-f
+----
+2:     ef
+
+truncate e-g
+----
+2:     ef
+3:      fg
+
+truncate e-h
+----
+2:     ef
+3:      f-h
+
+
+truncate f-f
+----
+
+truncate f-g
+----
+3:      fg
+
+truncate f-h
+----
+3:      f-h
+
+
+truncate g-g
+----
+
+truncate g-h
+----
+3:       gh
diff --git a/internal/rangedel/truncate_test.go b/internal/rangedel/truncate_test.go
new file mode 100644
index 0000000..09fe78a
--- /dev/null
+++ b/internal/rangedel/truncate_test.go
@@ -0,0 +1,48 @@
+// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package rangedel
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/petermattis/pebble/db"
+	"github.com/petermattis/pebble/internal/datadriven"
+)
+
+func TestTruncate(t *testing.T) {
+	cmp := db.DefaultComparer.Compare
+	var iter iterator
+
+	datadriven.RunTest(t, "testdata/truncate", func(d *datadriven.TestData) string {
+		switch d.Cmd {
+		case "build":
+			tombstones := buildTombstones(t, cmp, d.Input)
+			iter = NewIter(cmp, tombstones)
+			return formatTombstones(tombstones)
+
+		case "truncate":
+			if len(d.Input) > 0 {
+				t.Fatalf("unexpected input: %s", d.Input)
+			}
+			if len(d.CmdArgs) != 1 {
+				t.Fatalf("expected 1 argument: %s", d.CmdArgs)
+			}
+			parts := strings.Split(d.CmdArgs[0].String(), "-")
+			if len(parts) != 2 {
+				t.Fatalf("malformed arg: %s", d.CmdArgs[0])
+			}
+			lower := []byte(parts[0])
+			upper := []byte(parts[1])
+
+			truncated := Truncate(cmp, iter, lower, upper)
+			return formatTombstones(truncated.tombstones)
+
+		default:
+			return fmt.Sprintf("unknown command: %s", d.Cmd)
+		}
+	})
+}
diff --git a/testdata/compaction_atomic_unit_bounds b/testdata/compaction_atomic_unit_bounds
new file mode 100644
index 0000000..15aa456
--- /dev/null
+++ b/testdata/compaction_atomic_unit_bounds
@@ -0,0 +1,62 @@
+define
+a.SET.1-b.SET.2
+----
+
+atomic-unit-bounds 0
+----
+a-b
+
+define
+a.SET.1-b.SET.2
+c.SET.3-d.SET.4
+e.SET.5-f.SET.6
+----
+
+atomic-unit-bounds 0
+----
+a-b
+
+atomic-unit-bounds 1
+----
+c-d
+
+atomic-unit-bounds 2
+----
+e-f
+
+define
+a.SET.1-b.RANGEDEL.3
+b.SET.2-c.RANGEDEL.5
+c.SET.4-d.SET.6
+----
+
+atomic-unit-bounds 0
+----
+a-d
+
+atomic-unit-bounds 1
+----
+a-d
+
+atomic-unit-bounds 2
+----
+a-d
+
+define
+a.SET.1-b.RANGEDEL.72057594037927935
+b.SET.2-c.RANGEDEL.5
+c.SET.4-d.SET.6
+----
+
+atomic-unit-bounds 0
+----
+a-b
+
+atomic-unit-bounds 1
+----
+b-d
+
+atomic-unit-bounds 2
+----
+b-d
+
diff --git a/testdata/range_del b/testdata/range_del
index 4db52bf..2d7a1ad 100644
--- a/testdata/range_del
+++ b/testdata/range_del
@@ -1192,3 +1192,100 @@ prev
 c:v
 a:v
 .
+
+# A range tombstone straddles two sstables. One is compacted two
+# levels lower. The other is compacted one level lower. The one that
+# is compacted one level lower should not see its boundaries expand
+# causing it to delete more keys. A snapshot is used 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
+L2
+  d.SET.0:v
+----
+mem: 1
+0: a-e a-a c-c
+2: d-d
+
+compact a-b
+----
+1: a-c c-e
+2: d-d
+
+compact d-e
+----
+1: a-c
+3: c-d d-e
+
+get seq=4
+c
+----
+v
+
+compact a-b L1
+----
+2: a-c
+3: c-d d-e
+
+get seq=4
+c
+----
+v
+
+# A slight variation on the scenario above where a range tombstone is
+# expanded past the boundaries of its "atomic compaction unit".
+
+define target-file-sizes=(100, 1) snapshots=(1)
+L0
+  a.RANGEDEL.1:e
+L0
+  a.SET.2:v
+L0
+  c.SET.3:v
+L0
+  f.SET.4:v
+L2
+  d.SET.0:v
+----
+mem: 1
+0: a-e a-a c-c f-f
+2: d-d
+
+compact a-b
+----
+0: f-f
+1: a-c c-e
+2: d-d
+
+compact d-e
+----
+0: f-f
+1: a-c
+3: c-d d-e
+
+get seq=4
+c
+----
+v
+
+compact f-f L0
+----
+1: a-c f-f
+3: c-d d-e
+
+compact a-f L1
+----
+2: a-c c-f
+3: c-d d-e
+
+get seq=4
+c
+----
+v