Repository: petermattis/pebble (go)

Issue:
perf: add support for storing the WAL in a separate directory
RocksDB supports storing the WAL in a separate directory. That directory can be on a different disk than sstables, and that disk might be a different media that is significantly faster. For example, the WAL could be stored on SSD or Optane, while sstables are stored in spinning disk. This feature is relatively straightforward to implement, mainly just plumbing a configuration option.

Pull request:
Support storing WALs in a separate directory
Add `Options.WALDir` which specifies the directory to store write-ahead
logs in. This allows write-ahead logs to be stored on faster disk than
sstables. For example, write-ahead logs could be stored on
battery-backed SSD, while sstables are stored on spinning disk. Or
write-ahead logs could be stored on an Optane drive, while sstables are
stored on SSDs. If `Options.WALDir` is empty, write-ahead logs are
stored in the primary data directory.

Fixes #65

Patch:
diff --git a/compaction.go b/compaction.go
index 1af02b4..1e42c2a 100644
--- a/compaction.go
+++ b/compaction.go
@@ -534,7 +534,7 @@ func (d *DB) flush1() error {
 		}
 	}
 
-	err = d.mu.versions.logAndApply(ve, d.dir)
+	err = d.mu.versions.logAndApply(ve, d.dataDir)
 	for i := range ve.newFiles {
 		f := &ve.newFiles[i]
 		if _, ok := d.mu.compact.pendingOutputs[f.meta.fileNum]; !ok {
@@ -686,7 +686,7 @@ func (d *DB) writeLevel0Table(
 		return fileMetadata{}, errEmptyTable
 	}
 
-	if err := d.dir.Sync(); err != nil {
+	if err := d.dataDir.Sync(); err != nil {
 		return fileMetadata{}, err
 	}
 
@@ -800,7 +800,7 @@ func (d *DB) compact1() (err error) {
 	if err != nil {
 		return err
 	}
-	err = d.mu.versions.logAndApply(ve, d.dir)
+	err = d.mu.versions.logAndApply(ve, d.dataDir)
 	for _, fileNum := range pendingOutputs {
 		if _, ok := d.mu.compact.pendingOutputs[fileNum]; !ok {
 			panic("pebble: expected pending output not present")
@@ -1011,7 +1011,7 @@ func (d *DB) compactDiskTables(c *compaction) (ve *versionEdit, pendingOutputs [
 		}
 	}
 
-	if err := d.dir.Sync(); err != nil {
+	if err := d.dataDir.Sync(); err != nil {
 		return nil, pendingOutputs, err
 	}
 	return ve, pendingOutputs, nil
@@ -1037,6 +1037,15 @@ func (d *DB) scanObsoleteFiles() {
 		return
 	}
 
+	if d.dirname != d.walDirname {
+		list2, err := fs.List(d.walDirname)
+		if err != nil {
+			// Ignore any filesystem errors.
+		} else {
+			list = append(list, list2...)
+		}
+	}
+
 	// Grab d.mu again in order to get a snapshot of the live state. Note that we
 	// need to this after the directory list because after releasing the lock
 	// again new files can be created.
diff --git a/db.go b/db.go
index 32ee038..cceb9f2 100644
--- a/db.go
+++ b/db.go
@@ -130,13 +130,16 @@ type Writer interface {
 //	})
 type DB struct {
 	dirname        string
-	dir            vfs.File
+	walDirname     string
 	opts           *db.Options
 	cmp            db.Compare
 	equal          db.Equal
 	merge          db.Merge
 	abbreviatedKey db.AbbreviatedKey
 
+	dataDir vfs.File
+	walDir  vfs.File
+
 	tableCache tableCache
 	newIters   tableNewIters
 
@@ -600,7 +603,7 @@ func (d *DB) Close() error {
 	d.commit.Close()
 	d.mu.closed = true
 
-	err = firstError(err, d.dir.Close())
+	err = firstError(err, d.dataDir.Close())
 
 	if err == nil {
 		d.readState.val.unrefLocked()
@@ -789,7 +792,7 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 			d.mu.mem.switching = true
 			d.mu.Unlock()
 
-			newLogName := dbFilename(d.dirname, fileTypeLog, newLogNumber)
+			newLogName := dbFilename(d.walDirname, fileTypeLog, newLogNumber)
 
 			// Try to use a recycled log file. Recycling log files is an important
 			// performance optimization as it is faster to sync a file that has
@@ -799,7 +802,7 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 			// preallocation is performed (e.g. fallocate).
 			recycleLogNumber := d.logRecycler.peek()
 			if recycleLogNumber > 0 {
-				recycleLogName := dbFilename(d.dirname, fileTypeLog, recycleLogNumber)
+				recycleLogName := dbFilename(d.walDirname, fileTypeLog, recycleLogNumber)
 				err = d.opts.FS.Rename(recycleLogName, newLogName)
 			}
 
@@ -810,7 +813,7 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 			if err == nil {
 				// TODO(peter): RocksDB delays sync of the parent directory until the
 				// first time the log is synced. Is that worthwhile?
-				err = d.dir.Sync()
+				err = d.walDir.Sync()
 			}
 
 			if err == nil {
diff --git a/db/options.go b/db/options.go
index 398d66d..662be85 100644
--- a/db/options.go
+++ b/db/options.go
@@ -286,6 +286,11 @@ type Options struct {
 	// of TableFormatRocksDBv2. We should ensure it is only used when writing an
 	// sstable directly, and not used when opening a database.
 	TableFormat TableFormat
+
+	// WALDir specifies the directory to store write-ahead logs (WALs) in. If
+	// empty (the default), WALs will be stored in the same directory as sstables
+	// (i.e. the directory passed to pebble.Open).
+	WALDir string
 }
 
 // EnsureDefaults ensures that the default values for all options are set if a
@@ -385,6 +390,7 @@ func (o *Options) String() string {
 	fmt.Fprintf(&buf, "  mem_table_size=%d\n", o.MemTableSize)
 	fmt.Fprintf(&buf, "  mem_table_stop_writes_threshold=%d\n", o.MemTableStopWritesThreshold)
 	fmt.Fprintf(&buf, "  merger=%s\n", o.Merger.Name)
+	fmt.Fprintf(&buf, "  wal_dir=%s\n", o.WALDir)
 
 	for i := range o.Levels {
 		l := &o.Levels[i]
diff --git a/ingest.go b/ingest.go
index 4ae0521..f60a32d 100644
--- a/ingest.go
+++ b/ingest.go
@@ -289,7 +289,7 @@ func (d *DB) Ingest(paths []string) error {
 	// point before we update the MANIFEST (via logAndApply), otherwise a crash
 	// can have the tables referenced in the MANIFEST, but not present in the
 	// directory.
-	if err := d.dir.Sync(); err != nil {
+	if err := d.dataDir.Sync(); err != nil {
 		return err
 	}
 
@@ -389,7 +389,7 @@ func (d *DB) ingestApply(meta []*fileMetadata) (*versionEdit, error) {
 		ve.newFiles[i].level = ingestTargetLevel(d.cmp, current, m)
 		ve.newFiles[i].meta = *m
 	}
-	if err := d.mu.versions.logAndApply(ve, d.dir); err != nil {
+	if err := d.mu.versions.logAndApply(ve, d.dataDir); err != nil {
 		return nil, err
 	}
 	d.updateReadStateLocked()
diff --git a/open.go b/open.go
index b69a800..fbc6ff8 100644
--- a/open.go
+++ b/open.go
@@ -58,6 +58,7 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 	opts = opts.EnsureDefaults()
 	d := &DB{
 		dirname:        dirname,
+		walDirname:     opts.WALDir,
 		opts:           opts,
 		cmp:            opts.Comparer.Compare,
 		equal:          opts.Comparer.Equal,
@@ -99,10 +100,6 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 	if err != nil {
 		return nil, err
 	}
-	d.dir, err = opts.FS.OpenDir(dirname)
-	if err != nil {
-		return nil, err
-	}
 	fileLock, err := opts.FS.Lock(dbFilename(dirname, fileTypeLock, 0))
 	if err != nil {
 		return nil, err
@@ -113,12 +110,29 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 		}
 	}()
 
+	d.dataDir, err = opts.FS.OpenDir(dirname)
+	if err != nil {
+		return nil, err
+	}
+	if d.walDirname == "" {
+		d.walDirname = d.dirname
+	}
+	if d.walDirname == d.dirname {
+		d.walDir = d.dataDir
+	} else {
+		err := opts.FS.MkdirAll(d.walDirname, 0755)
+		if err != nil {
+			return nil, err
+		}
+		d.walDir, err = opts.FS.OpenDir(d.walDirname)
+	}
+
 	if _, err := opts.FS.Stat(dbFilename(dirname, fileTypeCurrent, 0)); os.IsNotExist(err) {
 		// Create the DB if it did not already exist.
 		if err := createDB(dirname, opts); err != nil {
 			return nil, err
 		}
-		if err := d.dir.Sync(); err != nil {
+		if err := d.dataDir.Sync(); err != nil {
 			return nil, err
 		}
 	} else if err != nil {
@@ -133,7 +147,7 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 		return nil, err
 	}
 
-	ls, err := opts.FS.List(dirname)
+	ls, err := opts.FS.List(d.walDirname)
 	if err != nil {
 		return nil, err
 	}
@@ -165,7 +179,7 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 	})
 	var ve versionEdit
 	for _, lf := range logFiles {
-		maxSeqNum, err := d.replayWAL(&ve, opts.FS, filepath.Join(dirname, lf.name), lf.num)
+		maxSeqNum, err := d.replayWAL(&ve, opts.FS, filepath.Join(d.walDirname, lf.name), lf.num)
 		if err != nil {
 			return nil, err
 		}
@@ -179,11 +193,11 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 	// Create an empty .log file.
 	ve.logNumber = d.mu.versions.nextFileNum()
 	d.mu.log.queue = append(d.mu.log.queue, ve.logNumber)
-	logFile, err := opts.FS.Create(dbFilename(dirname, fileTypeLog, ve.logNumber))
+	logFile, err := opts.FS.Create(dbFilename(d.walDirname, fileTypeLog, ve.logNumber))
 	if err != nil {
 		return nil, err
 	}
-	if err := d.dir.Sync(); err != nil {
+	if err := d.walDir.Sync(); err != nil {
 		return nil, err
 	}
 	logFile = vfs.NewSyncingFile(logFile, vfs.SyncingFileOptions{
@@ -193,7 +207,7 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 	d.mu.log.LogWriter = record.NewLogWriter(logFile, ve.logNumber)
 
 	// Write a new manifest to disk.
-	if err := d.mu.versions.logAndApply(&ve, d.dir); err != nil {
+	if err := d.mu.versions.logAndApply(&ve, d.dataDir); err != nil {
 		return nil, err
 	}
 	d.updateReadStateLocked()
@@ -208,7 +222,7 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 		return nil, err
 	}
 	optionsFile.Close()
-	if err := d.dir.Sync(); err != nil {
+	if err := d.dataDir.Sync(); err != nil {
 		return nil, err
 	}
 


Test patch:
diff --git a/data_test.go b/data_test.go
index 19ed850..e02bf9d 100644
--- a/data_test.go
+++ b/data_test.go
@@ -303,7 +303,7 @@ func runDBDefineCmd(td *datadriven.TestData) (*DB, error) {
 	}
 
 	if len(ve.newFiles) > 0 {
-		if err := d.mu.versions.logAndApply(ve, d.dir); err != nil {
+		if err := d.mu.versions.logAndApply(ve, d.dataDir); err != nil {
 			return nil, err
 		}
 		d.updateReadStateLocked()
diff --git a/db/options_test.go b/db/options_test.go
index 4d88322..5bf0f81 100644
--- a/db/options_test.go
+++ b/db/options_test.go
@@ -53,6 +53,7 @@ func TestOptionsString(t *testing.T) {
   mem_table_size=4194304
   mem_table_stop_writes_threshold=2
   merger=pebble.concatenate
+  wal_dir=
 
 [Level "0"]
   block_restart_interval=16
diff --git a/event_listener_test.go b/event_listener_test.go
index 13c8368..b9e0991 100644
--- a/event_listener_test.go
+++ b/event_listener_test.go
@@ -135,6 +135,7 @@ func TestEventListener(t *testing.T) {
 				FS:                  loggingFS{mem, &buf},
 				EventListener:       newLoggingEventListener(&buf),
 				MaxManifestFileSize: 1,
+				WALDir:              "wal",
 			})
 			if err != nil {
 				return err.Error()
diff --git a/open_test.go b/open_test.go
index 904e225..73d0504 100644
--- a/open_test.go
+++ b/open_test.go
@@ -82,79 +82,86 @@ func testOpenCloseOpenClose(t *testing.T, fs vfs.FS, root string) {
 	}
 
 	for _, startFromEmpty := range []bool{false, true} {
-		for _, length := range []int{-1, 0, 1, 1000, 10000, 100000} {
-			dirname := "sharedDatabase"
-			if startFromEmpty {
-				dirname = "startFromEmpty" + strconv.Itoa(length)
-			}
-			dirname = filepath.Join(root, dirname)
+		for _, walDirname := range []string{"", "wal"} {
+			for _, length := range []int{-1, 0, 1, 1000, 10000, 100000} {
+				dirname := "sharedDatabase"
+				if startFromEmpty {
+					dirname = "startFromEmpty" + strconv.Itoa(length)
+				}
+				dirname = filepath.Join(root, dirname)
+				if walDirname == "" {
+					opts.WALDir = ""
+				} else {
+					opts.WALDir = filepath.Join(dirname, walDirname)
+				}
 
-			got, xxx := []byte(nil), ""
-			if length >= 0 {
-				xxx = strings.Repeat("x", length)
-			}
+				got, xxx := []byte(nil), ""
+				if length >= 0 {
+					xxx = strings.Repeat("x", length)
+				}
 
-			d0, err := Open(dirname, opts)
-			if err != nil {
-				t.Errorf("sfe=%t, length=%d: Open #0: %v",
-					startFromEmpty, length, err)
-				continue
-			}
-			if length >= 0 {
-				err = d0.Set([]byte("key"), []byte(xxx), nil)
+				d0, err := Open(dirname, opts)
 				if err != nil {
-					t.Errorf("sfe=%t, length=%d: Set: %v",
+					t.Errorf("sfe=%t, length=%d: Open #0: %v",
 						startFromEmpty, length, err)
 					continue
 				}
-			}
-			err = d0.Close()
-			if err != nil {
-				t.Errorf("sfe=%t, length=%d: Close #0: %v",
-					startFromEmpty, length, err)
-				continue
-			}
-
-			d1, err := Open(dirname, opts)
-			if err != nil {
-				t.Errorf("sfe=%t, length=%d: Open #1: %v",
-					startFromEmpty, length, err)
-				continue
-			}
-			if length >= 0 {
-				got, err = d1.Get([]byte("key"))
+				if length >= 0 {
+					err = d0.Set([]byte("key"), []byte(xxx), nil)
+					if err != nil {
+						t.Errorf("sfe=%t, length=%d: Set: %v",
+							startFromEmpty, length, err)
+						continue
+					}
+				}
+				err = d0.Close()
 				if err != nil {
-					t.Errorf("sfe=%t, length=%d: Get: %v",
+					t.Errorf("sfe=%t, length=%d: Close #0: %v",
 						startFromEmpty, length, err)
 					continue
 				}
-			}
-			err = d1.Close()
-			if err != nil {
-				t.Errorf("sfe=%t, length=%d: Close #1: %v",
-					startFromEmpty, length, err)
-				continue
-			}
 
-			if length >= 0 && string(got) != xxx {
-				t.Errorf("sfe=%t, length=%d: got value differs from set value",
-					startFromEmpty, length)
-				continue
-			}
-
-			{
-				got, err := opts.FS.List(dirname)
+				d1, err := Open(dirname, opts)
 				if err != nil {
-					t.Fatalf("List: %v", err)
+					t.Errorf("sfe=%t, length=%d: Open #1: %v",
+						startFromEmpty, length, err)
+					continue
 				}
-				var optionsCount int
-				for _, s := range got {
-					if t, _, ok := parseDBFilename(s); ok && t == fileTypeOptions {
-						optionsCount++
+				if length >= 0 {
+					got, err = d1.Get([]byte("key"))
+					if err != nil {
+						t.Errorf("sfe=%t, length=%d: Get: %v",
+							startFromEmpty, length, err)
+						continue
 					}
 				}
-				if optionsCount != 1 {
-					t.Fatalf("expected 1 OPTIONS file, but found %d", optionsCount)
+				err = d1.Close()
+				if err != nil {
+					t.Errorf("sfe=%t, length=%d: Close #1: %v",
+						startFromEmpty, length, err)
+					continue
+				}
+
+				if length >= 0 && string(got) != xxx {
+					t.Errorf("sfe=%t, length=%d: got value differs from set value",
+						startFromEmpty, length)
+					continue
+				}
+
+				{
+					got, err := opts.FS.List(dirname)
+					if err != nil {
+						t.Fatalf("List: %v", err)
+					}
+					var optionsCount int
+					for _, s := range got {
+						if t, _, ok := parseDBFilename(s); ok && t == fileTypeOptions {
+							optionsCount++
+						}
+					}
+					if optionsCount != 1 {
+						t.Fatalf("expected 1 OPTIONS file, but found %d", optionsCount)
+					}
 				}
 			}
 		}
diff --git a/testdata/event_listener b/testdata/event_listener
index d893382..e0569ca 100644
--- a/testdata/event_listener
+++ b/testdata/event_listener
@@ -1,13 +1,14 @@
 open
 ----
 open-dir: db
+open-dir: wal
 create: db/MANIFEST-000001
 create: db/CURRENT.000001.dbtmp
 sync: db/CURRENT.000001.dbtmp
 rename: db/CURRENT.000001.dbtmp -> db/CURRENT
 sync: db
-create: db/000002.log
-sync: db
+create: wal/000002.log
+sync: wal
 create: db/MANIFEST-000003
 sync: db/MANIFEST-000003
 create: db/CURRENT.000003.dbtmp
@@ -19,10 +20,10 @@ sync: db
 
 flush
 ----
-sync: db/000002.log
-create: db/000005.log
-sync: db
-sync: db/000002.log
+sync: wal/000002.log
+create: wal/000005.log
+sync: wal
+sync: wal/000002.log
 #2: WAL created: 5 recycled=0
 #3: flush begin
 create: db/000006.sst
@@ -38,11 +39,11 @@ sync: db
 
 compact
 ----
-sync: db/000005.log
-rename: db/000002.log -> db/000008.log
-create: db/000008.log
-sync: db
-sync: db/000005.log
+sync: wal/000005.log
+rename: wal/000002.log -> wal/000008.log
+create: wal/000008.log
+sync: wal
+sync: wal/000005.log
 #4: WAL created: 8 recycled=2
 #5: flush begin
 create: db/000009.sst