Repository: petermattis/pebble (go)

Issue:
db: add max_manifest_file_size option
The RocksDB [max_manifest_file_size](https://github.com/facebook/rocksdb/blob/master/include/rocksdb/options.h#L604-L608) option limits how large the manifest can become:

>   // manifest file is rolled over on reaching this limit.
  // The older manifest file be deleted.
  // The default value is 1GB so that the manifest file can grow, but not
  // reach the limit of storage capacity.

CRDB sets this to 128MB as we found that 1GB can cause an excessive startup time as the manifest is "replayed". Currently, Pebble has no support for rolling the manifest when it becomes too large. There is a [TODO](https://github.com/petermattis/pebble/blob/master/version_set.go#L189) in the code that this needs to be addressed.

Pull request description:
Add support for capping the size of the MANIFEST
The MANIFEST is a log (in `internal/record` format) containing a
sequence of version edits. A version edit is created whenever a flush or
compaction occurs. The actual current version is obtained by replaying
the log at startup (the current version is also maintained in memory, of
course). In order to avoid the MANIFEST growing excessively large, it is
"rolled" whenever its size exceeds
`Options.MaxManifestFileSize` (default 128 MB). The new MANIFEST is
completely self-contained with the version entry being a snapshot of the
entire state (essentially a delta from the empty state).

Re-jiggered the way the MANIFEST is initially created which resulted in
slightly different file numbering at database creation: the first log
now gets file number 2, and the manifest file number 3.

Fixes #42

Patch:
diff --git a/compaction.go b/compaction.go
index e3f8806..de8a57c 100644
--- a/compaction.go
+++ b/compaction.go
@@ -910,14 +910,6 @@ func (d *DB) compactDiskTables(c *compaction) (ve *versionEdit, pendingOutputs [
 				// table's largest key. We need the tables to be key-space partitioned,
 				// so force the boundary to a key that we know is larger than the
 				// previous key.
-				//
-				// We use seqnum zero since seqnums are in descending order, and our
-				// goal is to ensure this forged key does not overlap with the previous
-				// file. `db.InternalKeyRangeDeleteSentinel` is actually the first key
-				// kind as key kinds are also in descending order. But, this is OK
-				// because choosing seqnum zero is already enough to prevent overlap
-				// (the previous file could not end with a key at seqnum zero if this
-				// file had a tombstone extending into it).
 				writerMeta.SmallestRange = db.MakeInternalKey(
 					prevMeta.largest.UserKey, 0, db.InternalKeyKindRangeDelete)
 			}
diff --git a/db.go b/db.go
index df8a3c7..8876d47 100644
--- a/db.go
+++ b/db.go
@@ -864,9 +864,8 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 		imm := d.mu.mem.mutable
 		d.mu.mem.mutable = newMemTable(d.opts)
 		// NB: When the immutable memtable is flushed to disk it will apply a
-		// versionEdit to the manifest telling it that log files < newLogNumber
-		// have been applied. newLogNumber corresponds to the WAL that contains
-		// mutations that are present in the new memtable.
+		// versionEdit to the manifest telling it that log files <= newLogNumber
+		// have been applied.
 		d.mu.mem.mutable.logNum = newLogNumber
 		d.mu.mem.queue = append(d.mu.mem.queue, d.mu.mem.mutable)
 		d.updateReadStateLocked()
diff --git a/db/options.go b/db/options.go
index f2ed1de..9a52f20 100644
--- a/db/options.go
+++ b/db/options.go
@@ -243,6 +243,11 @@ type Options struct {
 	// The default logger uses the Go standard library log package.
 	Logger Logger
 
+	// MaxManifestFileSize is the maximum size the MANIFEST file is allowed to
+	// become. When the MANIFEST exceeds this size it is rolled over and a new
+	// MANIFEST is created.
+	MaxManifestFileSize int64
+
 	// MaxOpenFiles is a soft limit on the number of open files that can be
 	// used by the DB.
 	//
@@ -322,6 +327,9 @@ func (o *Options) EnsureDefaults() *Options {
 	if o.Logger == nil {
 		o.Logger = defaultLogger{}
 	}
+	if o.MaxManifestFileSize == 0 {
+		o.MaxManifestFileSize = 128 << 20 // 128 MB
+	}
 	if o.MaxOpenFiles == 0 {
 		o.MaxOpenFiles = 1000
 	}
@@ -368,6 +376,7 @@ func (o *Options) String() string {
 	fmt.Fprintf(&buf, "  l0_slowdown_writes_threshold=%d\n", o.L0SlowdownWritesThreshold)
 	fmt.Fprintf(&buf, "  l0_stop_writes_threshold=%d\n", o.L0StopWritesThreshold)
 	fmt.Fprintf(&buf, "  l1_max_bytes=%d\n", o.L1MaxBytes)
+	fmt.Fprintf(&buf, "  max_manifest_file_size=%d\n", o.MaxManifestFileSize)
 	fmt.Fprintf(&buf, "  max_open_files=%d\n", o.MaxOpenFiles)
 	fmt.Fprintf(&buf, "  mem_table_size=%d\n", o.MemTableSize)
 	fmt.Fprintf(&buf, "  mem_table_stop_writes_threshold=%d\n", o.MemTableStopWritesThreshold)
diff --git a/filenames.go b/filenames.go
index e3bba61..7467bc9 100644
--- a/filenames.go
+++ b/filenames.go
@@ -95,6 +95,9 @@ func setCurrentFile(dirname string, fs vfs.FS, fileNum uint64) error {
 	if _, err := fmt.Fprintf(f, "MANIFEST-%06d\n", fileNum); err != nil {
 		return err
 	}
+	if err := f.Sync(); err != nil {
+		return err
+	}
 	if err := f.Close(); err != nil {
 		return err
 	}
diff --git a/internal/record/record.go b/internal/record/record.go
index 71c4fdb..e0068dc 100644
--- a/internal/record/record.go
+++ b/internal/record/record.go
@@ -543,6 +543,11 @@ func (w *Writer) WriteRecord(p []byte) (int64, error) {
 	return offset, w.err
 }
 
+// Size returns the current size of the file.
+func (w *Writer) Size() int64 {
+	return w.blockNumber*blockSize + int64(w.j)
+}
+
 // LastRecordOffset returns the offset in the underlying io.Writer of the last
 // record so far - the one created by the most recent Next call. It is the
 // offset of the first chunk header, suitable to pass to Reader.SeekRecord.
diff --git a/version_set.go b/version_set.go
index ef246cb..7c77d80 100644
--- a/version_set.go
+++ b/version_set.go
@@ -144,7 +144,6 @@ func (vs *versionSet) load(dirname string, opts *db.Options, mu *sync.Mutex) err
 	}
 	vs.markFileNumUsed(vs.logNumber)
 	vs.markFileNumUsed(vs.prevLogNumber)
-	vs.manifestFileNumber = vs.nextFileNum()
 
 	newVersion, err := bve.apply(opts, nil, vs.cmp)
 	if err != nil {
@@ -180,6 +179,13 @@ func (vs *versionSet) logAndApply(ve *versionEdit) error {
 	currentVersion := vs.currentVersion()
 	var newVersion *version
 
+	// Generate a new manifest if we don't currently have one, or the current one
+	// is too large.
+	var newManifestFileNumber uint64
+	if vs.manifest == nil || vs.manifest.Size() >= vs.opts.MaxManifestFileSize {
+		newManifestFileNumber = vs.nextFileNum()
+	}
+
 	var picker *compactionPicker
 	if err := func() error {
 		vs.mu.Unlock()
@@ -194,10 +200,8 @@ func (vs *versionSet) logAndApply(ve *versionEdit) error {
 			return err
 		}
 
-		// TODO(peter): if vs.manifest becomes too large, create a new one. Be sure
-		// to add the old manifest to obsoleteManifests.
-		if vs.manifest == nil {
-			if err := vs.createManifest(vs.dirname); err != nil {
+		if newManifestFileNumber != 0 {
+			if err := vs.createManifest(vs.dirname, newManifestFileNumber); err != nil {
 				return err
 			}
 		}
@@ -215,8 +219,10 @@ func (vs *versionSet) logAndApply(ve *versionEdit) error {
 		if err := vs.manifestFile.Sync(); err != nil {
 			return err
 		}
-		if err := setCurrentFile(vs.dirname, vs.fs, vs.manifestFileNumber); err != nil {
-			return err
+		if newManifestFileNumber != 0 {
+			if err := setCurrentFile(vs.dirname, vs.fs, newManifestFileNumber); err != nil {
+				return err
+			}
 		}
 		picker = newCompactionPicker(newVersion, vs.opts)
 		return nil
@@ -232,14 +238,20 @@ func (vs *versionSet) logAndApply(ve *versionEdit) error {
 	if ve.prevLogNumber != 0 {
 		vs.prevLogNumber = ve.prevLogNumber
 	}
+	if newManifestFileNumber != 0 {
+		if vs.manifestFileNumber != 0 {
+			vs.obsoleteManifests = append(vs.obsoleteManifests, vs.manifestFileNumber)
+		}
+		vs.manifestFileNumber = newManifestFileNumber
+	}
 	vs.picker = picker
 	return nil
 }
 
 // createManifest creates a manifest file that contains a snapshot of vs.
-func (vs *versionSet) createManifest(dirname string) (err error) {
+func (vs *versionSet) createManifest(dirname string, fileNum uint64) (err error) {
 	var (
-		filename     = dbFilename(dirname, fileTypeManifest, vs.manifestFileNumber)
+		filename     = dbFilename(dirname, fileTypeManifest, fileNum)
 		manifestFile vfs.File
 		manifest     *record.Writer
 	)


Test patch:
diff --git a/db/options_test.go b/db/options_test.go
index 81a2e7d..4d88322 100644
--- a/db/options_test.go
+++ b/db/options_test.go
@@ -48,6 +48,7 @@ func TestOptionsString(t *testing.T) {
   l0_slowdown_writes_threshold=8
   l0_stop_writes_threshold=12
   l1_max_bytes=67108864
+  max_manifest_file_size=134217728
   max_open_files=1000
   mem_table_size=4194304
   mem_table_stop_writes_threshold=2
diff --git a/db_test.go b/db_test.go
index 9d24f39..23b6e3b 100644
--- a/db_test.go
+++ b/db_test.go
@@ -19,6 +19,7 @@ import (
 	"github.com/petermattis/pebble/cache"
 	"github.com/petermattis/pebble/db"
 	"github.com/petermattis/pebble/vfs"
+	"github.com/stretchr/testify/require"
 	"golang.org/x/exp/rand"
 )
 
@@ -614,3 +615,78 @@ func TestFlushEmpty(t *testing.T) {
 		t.Fatal(err)
 	}
 }
+
+func TestRollManifest(t *testing.T) {
+	d, err := Open("", &db.Options{
+		MaxManifestFileSize: 1,
+		VFS:                 vfs.NewMem(),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	manifestFileNumber := func() uint64 {
+		d.mu.Lock()
+		defer d.mu.Unlock()
+		return d.mu.versions.manifestFileNumber
+	}
+
+	current := func() string {
+		f, err := d.opts.VFS.Open(dbFilename(d.dirname, fileTypeCurrent, 0))
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer f.Close()
+		stat, err := f.Stat()
+		if err != nil {
+			t.Fatal(err)
+		}
+		n := stat.Size()
+		b := make([]byte, n)
+		if _, err = f.ReadAt(b, 0); err != nil {
+			t.Fatal(err)
+		}
+		return string(b)
+	}
+
+	lastManifestNum := manifestFileNumber()
+	for i := 0; i < 5; i++ {
+		if err := d.Set([]byte("a"), nil, nil); err != nil {
+			t.Fatal(err)
+		}
+		if err := d.Flush(); err != nil {
+			t.Fatal(err)
+		}
+		num := manifestFileNumber()
+		if lastManifestNum == num {
+			t.Fatalf("manifest failed to roll: %d == %d", lastManifestNum, num)
+		}
+		lastManifestNum = num
+
+		expectedCurrent := fmt.Sprintf("MANIFEST-%06d\n", lastManifestNum)
+		if v := current(); expectedCurrent != v {
+			t.Fatalf("expected %s, but found %s", expectedCurrent, v)
+		}
+	}
+
+	files, err := d.opts.VFS.List("")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var manifests []string
+	for _, filename := range files {
+		fileType, _, ok := parseDBFilename(filename)
+		if !ok {
+			continue
+		}
+		if fileType == fileTypeManifest {
+			manifests = append(manifests, filename)
+		}
+	}
+	expected := []string{fmt.Sprintf("MANIFEST-%06d", lastManifestNum)}
+	require.EqualValues(t, expected, manifests)
+
+	if err := d.Close(); err != nil {
+		t.Fatal(err)
+	}
+}
diff --git a/event_listener_test.go b/event_listener_test.go
index 9017d80..4fb56be 100644
--- a/event_listener_test.go
+++ b/event_listener_test.go
@@ -94,7 +94,7 @@ func TestEventListener(t *testing.T) {
 #3: flush end: 6
 #4: compaction begin: L0 -> L1
 #4: compaction end: L0 -> L1
-#5: WAL created: 7 recycled=3
+#5: WAL created: 7 recycled=2
 #6: flush begin
 #6: flush end: 8
 #7: compaction begin: L0 -> L1
diff --git a/internal/record/record_test.go b/internal/record/record_test.go
index 16021a5..869b1e1 100644
--- a/internal/record/record_test.go
+++ b/internal/record/record_test.go
@@ -818,6 +818,27 @@ func TestInvalidLogNum(t *testing.T) {
 	}
 }
 
+func TestSize(t *testing.T) {
+	var buf bytes.Buffer
+	zeroes := make([]byte, 8<<10)
+	w := NewWriter(&buf)
+	for i := 0; i < 100; i++ {
+		n := rand.Intn(len(zeroes))
+		if _, err := w.WriteRecord(zeroes[:n]); err != nil {
+			t.Fatal(err)
+		}
+		if err := w.Flush(); err != nil {
+			t.Fatal(err)
+		}
+		if buf.Len() != int(w.Size()) {
+			t.Fatalf("expected %d, but found %d", buf.Len(), w.Size())
+		}
+	}
+	if err := w.Close(); err != nil {
+		t.Fatal(err)
+	}
+}
+
 func BenchmarkRecordWrite(b *testing.B) {
 	for _, size := range []int{8, 16, 32, 64, 128} {
 		b.Run(fmt.Sprintf("size=%d", size), func(b *testing.B) {
diff --git a/open_test.go b/open_test.go
index 06246cc..82e1587 100644
--- a/open_test.go
+++ b/open_test.go
@@ -67,9 +67,9 @@ func TestNewDBFilenames(t *testing.T) {
 	sort.Strings(got)
 	// TODO(peter): should there be a LOCK file here?
 	want := []string{
-		"000003.log",
+		"000002.log",
 		"CURRENT",
-		"MANIFEST-000002",
+		"MANIFEST-000003",
 		"OPTIONS-000004",
 	}
 	if !reflect.DeepEqual(got, want) {