Repository: petermattis/pebble
Pull request #235 (2019-08-20 13:22:28)

Add Options.ReadOnly
Allow a DB to be opened in read-only mode, which disables writes,
background flushes and compactions, and the flush which normally occurs
after replay the WAL at startup. This is useful for introspection tools
which wish to inspect the state of a DB without modifying it.


Patch:
diff --git a/compaction.go b/compaction.go
index 3f7e59f..5cf9229 100644
--- a/compaction.go
+++ b/compaction.go
@@ -601,7 +601,7 @@ func (d *DB) getCompactionPacerInfo() compactionPacerInfo {
 	d.mu.Lock()
 	estimatedMaxWAmp := d.mu.versions.picker.estimatedMaxWAmp
 	pacerInfo := compactionPacerInfo{
-		slowdownThreshold: uint64(estimatedMaxWAmp * float64(d.opts.MemTableSize)),
+		slowdownThreshold:   uint64(estimatedMaxWAmp * float64(d.opts.MemTableSize)),
 		totalCompactionDebt: d.mu.versions.picker.estimatedCompactionDebt(bytesFlushed),
 	}
 	d.mu.Unlock()
@@ -619,7 +619,7 @@ func (d *DB) getFlushPacerInfo() flushPacerInfo {
 //
 // d.mu must be held when calling this.
 func (d *DB) maybeScheduleFlush() {
-	if d.mu.compact.flushing || atomic.LoadInt32(&d.closed) != 0 {
+	if d.mu.compact.flushing || atomic.LoadInt32(&d.closed) != 0 || d.opts.ReadOnly {
 		return
 	}
 	if len(d.mu.mem.queue) <= 1 {
@@ -752,7 +752,7 @@ func (d *DB) flush1() error {
 //
 // d.mu must be held when calling this.
 func (d *DB) maybeScheduleCompaction() {
-	if d.mu.compact.compacting || atomic.LoadInt32(&d.closed) != 0 {
+	if d.mu.compact.compacting || atomic.LoadInt32(&d.closed) != 0 || d.opts.ReadOnly {
 		return
 	}
 
diff --git a/db.go b/db.go
index 9dc9f42..8228fd2 100644
--- a/db.go
+++ b/db.go
@@ -34,6 +34,9 @@ var (
 	// ErrClosed is returned when an operation is performed on a closed snapshot
 	// or DB.
 	ErrClosed = errors.New("pebble: closed")
+	// ErrReadOnly is returned when a write operation is performed on a read-only
+	// database.
+	ErrReadOnly = errors.New("pebble: read-only")
 )
 
 type flushable interface {
@@ -373,6 +376,9 @@ func (d *DB) Apply(batch *Batch, opts *WriteOptions) error {
 	if atomic.LoadInt32(&d.closed) != 0 {
 		panic(ErrClosed)
 	}
+	if d.opts.ReadOnly {
+		return ErrReadOnly
+	}
 
 	sync := opts.GetSync()
 	if sync && d.opts.DisableWAL {
@@ -645,7 +651,11 @@ func (d *DB) Close() error {
 		d.mu.compact.cond.Wait()
 	}
 	err := d.tableCache.Close()
-	err = firstError(err, d.mu.log.Close())
+	if !d.opts.ReadOnly {
+		err = firstError(err, d.mu.log.Close())
+	} else if d.mu.log.LogWriter != nil {
+		panic("pebble: log-writer should be nil in read-only mode")
+	}
 	err = firstError(err, d.fileLock.Close())
 	d.commit.Close()
 
@@ -676,6 +686,9 @@ func (d *DB) Compact(start, end []byte /* CompactionOptions */) error {
 	if atomic.LoadInt32(&d.closed) != 0 {
 		panic(ErrClosed)
 	}
+	if d.opts.ReadOnly {
+		return ErrReadOnly
+	}
 
 	iStart := base.MakeInternalKey(start, InternalKeySeqNumMax, InternalKeyKindMax)
 	iEnd := base.MakeInternalKey(end, 0, 0)
@@ -776,6 +789,9 @@ func (d *DB) AsyncFlush() (<-chan struct{}, error) {
 	if atomic.LoadInt32(&d.closed) != 0 {
 		panic(ErrClosed)
 	}
+	if d.opts.ReadOnly {
+		return nil, ErrReadOnly
+	}
 
 	d.commit.mu.Lock()
 	d.mu.Lock()
diff --git a/internal/base/options.go b/internal/base/options.go
index a23bac0..604776b 100644
--- a/internal/base/options.go
+++ b/internal/base/options.go
@@ -336,6 +336,12 @@ type Options struct {
 	// default is 1 MB/s.
 	MinFlushRate int
 
+	// ReadOnly indicates that the DB should be opened in read-only mode. Writes
+	// to the DB will return an error, background compactions are disabled, and
+	// the flush that normally occurs after replaying the WAL at startup is
+	// disabled.
+	ReadOnly bool
+
 	// TableFormat specifies the format version for sstables. The default is
 	// TableFormatRocksDBv2 which creates RocksDB compatible sstables. Use
 	// TableFormatLevelDB to create LevelDB compatible sstable which can be used
diff --git a/open.go b/open.go
index 1776680..74d6a7d 100644
--- a/open.go
+++ b/open.go
@@ -116,9 +116,11 @@ func Open(dirname string, opts *Options) (*DB, error) {
 	defer d.mu.Unlock()
 
 	// Lock the database directory.
-	err := opts.FS.MkdirAll(dirname, 0755)
-	if err != nil {
-		return nil, err
+	if !d.opts.ReadOnly {
+		err := opts.FS.MkdirAll(dirname, 0755)
+		if err != nil {
+			return nil, err
+		}
 	}
 	fileLock, err := opts.FS.Lock(dbFilename(dirname, fileTypeLock, 0))
 	if err != nil {
@@ -140,14 +142,16 @@ func Open(dirname string, opts *Options) (*DB, error) {
 	if d.walDirname == d.dirname {
 		d.walDir = d.dataDir
 	} else {
-		err := opts.FS.MkdirAll(d.walDirname, 0755)
-		if err != nil {
-			return nil, err
+		if !d.opts.ReadOnly {
+			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) {
+	if _, err := opts.FS.Stat(dbFilename(dirname, fileTypeCurrent, 0)); os.IsNotExist(err) && !d.opts.ReadOnly {
 		// Create the DB if it did not already exist.
 		if err := createDB(dirname, opts); err != nil {
 			return nil, err
@@ -217,41 +221,45 @@ func Open(dirname string, opts *Options) (*DB, error) {
 	}
 	d.mu.versions.visibleSeqNum = d.mu.versions.logSeqNum
 
-	// 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(d.walDirname, fileTypeLog, ve.logNumber))
-	if err != nil {
-		return nil, err
-	}
-	if err := d.walDir.Sync(); err != nil {
-		return nil, err
-	}
-	logFile = vfs.NewSyncingFile(logFile, vfs.SyncingFileOptions{
-		BytesPerSync:    d.opts.BytesPerSync,
-		PreallocateSize: d.walPreallocateSize(),
-	})
-	d.mu.log.LogWriter = record.NewLogWriter(logFile, ve.logNumber)
-	d.mu.versions.metrics.WAL.Files++
-
-	// Write a new manifest to disk.
-	if err := d.mu.versions.logAndApply(0, &ve, d.dataDir); err != nil {
-		return nil, err
+	if !d.opts.ReadOnly {
+		// 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(d.walDirname, fileTypeLog, ve.logNumber))
+		if err != nil {
+			return nil, err
+		}
+		if err := d.walDir.Sync(); err != nil {
+			return nil, err
+		}
+		logFile = vfs.NewSyncingFile(logFile, vfs.SyncingFileOptions{
+			BytesPerSync:    d.opts.BytesPerSync,
+			PreallocateSize: d.walPreallocateSize(),
+		})
+		d.mu.log.LogWriter = record.NewLogWriter(logFile, ve.logNumber)
+		d.mu.versions.metrics.WAL.Files++
+
+		// Write a new manifest to disk.
+		if err := d.mu.versions.logAndApply(0, &ve, d.dataDir); err != nil {
+			return nil, err
+		}
 	}
 	d.updateReadStateLocked()
 
-	// Write the current options to disk.
-	d.optionsFileNum = d.mu.versions.nextFileNum()
-	optionsFile, err := opts.FS.Create(dbFilename(dirname, fileTypeOptions, d.optionsFileNum))
-	if err != nil {
-		return nil, err
-	}
-	if _, err := optionsFile.Write([]byte(opts.String())); err != nil {
-		return nil, err
-	}
-	optionsFile.Close()
-	if err := d.dataDir.Sync(); err != nil {
-		return nil, err
+	if !d.opts.ReadOnly {
+		// Write the current options to disk.
+		d.optionsFileNum = d.mu.versions.nextFileNum()
+		optionsFile, err := opts.FS.Create(dbFilename(dirname, fileTypeOptions, d.optionsFileNum))
+		if err != nil {
+			return nil, err
+		}
+		if _, err := optionsFile.Write([]byte(opts.String())); err != nil {
+			return nil, err
+		}
+		optionsFile.Close()
+		if err := d.dataDir.Sync(); err != nil {
+			return nil, err
+		}
 	}
 
 	jobID := d.mu.nextJobID
@@ -287,6 +295,13 @@ func (d *DB) replayWAL(
 		mem *memTable
 		rr  = record.NewReader(file, logNum)
 	)
+
+	// In read-only mode, we replay directly into the mutable memtable which will
+	// never be flushed.
+	if d.opts.ReadOnly {
+		mem = d.mu.mem.mutable
+	}
+
 	for {
 		r, err := rr.Next()
 		if err == nil {
@@ -339,7 +354,7 @@ func (d *DB) replayWAL(
 		buf.Reset()
 	}
 
-	if mem != nil && !mem.empty() {
+	if mem != nil && !mem.empty() && !d.opts.ReadOnly {
 		c := newFlush(d.opts, d.mu.versions.currentVersion(),
 			1 /* base level */, []flushable{mem}, &d.bytesFlushed)
 		newVE, pendingOutputs, err := d.runCompaction(c, nilPacer)


Test patch:
diff --git a/event_listener_test.go b/event_listener_test.go
index cfe4a50..f1dca3c 100644
--- a/event_listener_test.go
+++ b/event_listener_test.go
@@ -8,6 +8,7 @@ import (
 	"bytes"
 	"fmt"
 	"io"
+	"os"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -90,6 +91,11 @@ func (fs loggingFS) Rename(oldname, newname string) error {
 	return fs.FS.Rename(oldname, newname)
 }
 
+func (fs loggingFS) MkdirAll(dir string, perm os.FileMode) error {
+	fmt.Fprintf(fs.w, "mkdir-all: %s %#o\n", dir, perm)
+	return fs.FS.MkdirAll(dir, perm)
+}
+
 type loggingFile struct {
 	vfs.File
 	name string
diff --git a/open_test.go b/open_test.go
index 34abe1e..60926d4 100644
--- a/open_test.go
+++ b/open_test.go
@@ -14,6 +14,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/kr/pretty"
 	"github.com/petermattis/pebble/vfs"
 	"github.com/stretchr/testify/require"
 )
@@ -218,3 +219,99 @@ func TestOpenOptionsCheck(t *testing.T) {
 	_, err = Open("", opts)
 	require.Regexp(t, `merger name from file.*!=.*`, err)
 }
+
+func TestOpenReadOnly(t *testing.T) {
+	mem := vfs.NewMem()
+
+	{
+		// Opening a non-existent DB in read-only mode should result in no mutable
+		// filesystem operations.
+		var buf syncedBuffer
+		_, err := Open("non-existent", &Options{
+			FS:       loggingFS{mem, &buf},
+			ReadOnly: true,
+			WALDir:   "non-existent-waldir",
+		})
+		if err == nil {
+			t.Fatalf("expected error, but found success")
+		}
+		const expected = `open-dir: non-existent`
+		if trimmed := strings.TrimSpace(buf.String()); expected != trimmed {
+			t.Fatalf("expected %s, but found %s", expected, trimmed)
+		}
+	}
+
+	var contents []string
+	{
+		// Create a new DB and populate it with a small amount of data.
+		d, err := Open("", &Options{
+			FS: mem,
+		})
+		require.NoError(t, err)
+		require.NoError(t, d.Set([]byte("test"), nil, nil))
+		require.NoError(t, d.Close())
+		contents, err = mem.List("")
+		require.NoError(t, err)
+		sort.Strings(contents)
+	}
+
+	{
+		// Re-open the DB read-only. The directory contents should be unchanged.
+		d, err := Open("", &Options{
+			FS:       mem,
+			ReadOnly: true,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		// Verify various write operations fail in read-only mode.
+		require.EqualValues(t, ErrReadOnly, d.Compact(nil, nil))
+		require.EqualValues(t, ErrReadOnly, d.Flush())
+		require.EqualValues(t, ErrReadOnly, func() error { _, err := d.AsyncFlush(); return err }())
+
+		require.EqualValues(t, ErrReadOnly, d.Delete(nil, nil))
+		require.EqualValues(t, ErrReadOnly, d.DeleteRange(nil, nil, nil))
+		require.EqualValues(t, ErrReadOnly, d.LogData(nil, nil))
+		require.EqualValues(t, ErrReadOnly, d.Merge(nil, nil, nil))
+		require.EqualValues(t, ErrReadOnly, d.Set(nil, nil, nil))
+
+		// Verify we can still read in read-only mode.
+		require.NoError(t, func() error { _, err := d.Get([]byte("test")); return err }())
+
+		checkIter := func(iter *Iterator) {
+			t.Helper()
+
+			var keys []string
+			for valid := iter.First(); valid; valid = iter.Next() {
+				keys = append(keys, string(iter.Key()))
+			}
+			require.NoError(t, iter.Close())
+			expectedKeys := []string{"test"}
+			if diff := pretty.Diff(keys, expectedKeys); diff != nil {
+				t.Fatalf("%s\n%s", strings.Join(diff, "\n"), keys)
+			}
+		}
+
+		checkIter(d.NewIter(nil))
+
+		b := d.NewIndexedBatch()
+		checkIter(b.NewIter(nil))
+		require.EqualValues(t, ErrReadOnly, b.Commit(nil))
+		require.EqualValues(t, ErrReadOnly, d.Apply(b, nil))
+
+		s := d.NewSnapshot()
+		checkIter(s.NewIter(nil))
+		require.NoError(t, s.Close())
+
+		require.NoError(t, d.Close())
+
+		newContents, err := mem.List("")
+		require.NoError(t, err)
+
+		sort.Strings(newContents)
+		if diff := pretty.Diff(contents, newContents); diff != nil {
+			t.Fatalf("%s", strings.Join(diff, "\n"))
+		}
+	}
+}
diff --git a/testdata/event_listener b/testdata/event_listener
index 55b36ef..47a6c3d 100644
--- a/testdata/event_listener
+++ b/testdata/event_listener
@@ -1,6 +1,8 @@
 open
 ----
+mkdir-all: db 0755
 open-dir: db
+mkdir-all: wal 0755
 open-dir: wal
 create: db/MANIFEST-000001
 create: db/CURRENT.000001.dbtmp