Repository: petermattis/pebble (go)

Issue:
db: check merger name on startup
The `db.Merger.Name` field has a comment that is current unimplemented:

	// Pebble stores the merger name on disk, and opening a database with a
	// different merger from the one it was created with will result in an error.

The merger name is stored in the sstable `rocksdb.merge.operator` property, but we don't seem to check it anywhere currently. Need to check how RocksDB handles this and do something similar.

Pull request description:
Verify comparer and merger name in OPTIONS file
Fixes #84

Patch:
diff --git a/db/options.go b/db/options.go
index 49c344f..a40ca42 100644
--- a/db/options.go
+++ b/db/options.go
@@ -7,6 +7,7 @@ package db
 import (
 	"bytes"
 	"fmt"
+	"strings"
 
 	"github.com/petermattis/pebble/cache"
 	"github.com/petermattis/pebble/storage"
@@ -387,6 +388,55 @@ func (o *Options) String() string {
 	return buf.String()
 }
 
+// Check verifies the options are compatible with the previous options
+// serialized by Options.String(). For example, the Comparer and Merger must be
+// the same, or data will not be able to be properly read from the DB.
+func (o *Options) Check(s string) error {
+	var section string
+	for _, line := range strings.Split(s, "\n") {
+		line = strings.TrimSpace(line)
+		if len(line) == 0 {
+			// Skip blank lines.
+			continue
+		}
+		if line[0] == ';' || line[0] == '#' {
+			// Skip comments.
+			continue
+		}
+		n := len(line)
+		if line[0] == '[' && line[n-1] == ']' {
+			// Parse section.
+			section = line[1 : n-1]
+			continue
+		}
+
+		pos := strings.Index(line, "=")
+		if pos < 0 {
+			return fmt.Errorf("pebble: invalid key=value syntax: %s", line)
+		}
+
+		key := strings.TrimSpace(line[:pos])
+		value := strings.TrimSpace(line[pos+1:])
+
+		switch section {
+		case "Options":
+			switch key {
+			case "comparer":
+				if value != o.Comparer.Name {
+					return fmt.Errorf("pebble: comparer name from file %q != comparer name from options %q",
+						value, o.Comparer.Name)
+				}
+			case "merger":
+				if value != o.Merger.Name {
+					return fmt.Errorf("pebble: merger name from file %q != meger name from options %q",
+						value, o.Merger.Name)
+				}
+			}
+		}
+	}
+	return nil
+}
+
 // IterOptions hold the optional per-query parameters for NewIter.
 //
 // Like Options, a nil *IterOptions is valid and means to use the default
diff --git a/open.go b/open.go
index 50d7d9c..bace5d4 100644
--- a/open.go
+++ b/open.go
@@ -8,6 +8,7 @@ import (
 	"bytes"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"os"
 	"path/filepath"
 	"sort"
@@ -124,13 +125,12 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 		return nil, err
 	}
 
-	// Replay any newer log files than the ones named in the manifest.
-	var ve versionEdit
 	ls, err := fs.List(dirname)
 	if err != nil {
 		return nil, err
 	}
 
+	// Replay any newer log files than the ones named in the manifest.
 	type fileNumAndName struct {
 		num  uint64
 		name string
@@ -138,13 +138,24 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 	var logFiles []fileNumAndName
 	for _, filename := range ls {
 		ft, fn, ok := parseDBFilename(filename)
-		if ok && ft == fileTypeLog && (fn >= d.mu.versions.logNumber || fn == d.mu.versions.prevLogNumber) {
-			logFiles = append(logFiles, fileNumAndName{fn, filename})
+		if !ok {
+			continue
+		}
+		switch ft {
+		case fileTypeLog:
+			if fn >= d.mu.versions.logNumber || fn == d.mu.versions.prevLogNumber {
+				logFiles = append(logFiles, fileNumAndName{fn, filename})
+			}
+		case fileTypeOptions:
+			if err := checkOptions(opts, filepath.Join(dirname, filename)); err != nil {
+				return nil, err
+			}
 		}
 	}
 	sort.Slice(logFiles, func(i, j int) bool {
 		return logFiles[i].num < logFiles[j].num
 	})
+	var ve versionEdit
 	for _, lf := range logFiles {
 		maxSeqNum, err := d.replayWAL(&ve, fs, filepath.Join(dirname, lf.name), lf.num)
 		if err != nil {
@@ -283,3 +294,17 @@ func (d *DB) replayWAL(
 
 	return maxSeqNum, nil
 }
+
+func checkOptions(opts *db.Options, path string) error {
+	f, err := opts.Storage.Open(path)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	data, err := ioutil.ReadAll(f)
+	if err != nil {
+		return err
+	}
+	return opts.Check(string(data))
+}


Test patch:
diff --git a/db/options_test.go b/db/options_test.go
index 8bbcf3d..c08282f 100644
--- a/db/options_test.go
+++ b/db/options_test.go
@@ -6,6 +6,8 @@ package db
 
 import (
 	"testing"
+
+	"github.com/stretchr/testify/require"
 )
 
 func TestLevelOptions(t *testing.T) {
@@ -66,3 +68,19 @@ func TestOptionsString(t *testing.T) {
 		t.Fatalf("expected\n%s\nbut found\n%s", expected, v)
 	}
 }
+
+func TestOptionsCheck(t *testing.T) {
+	var opts *Options
+	opts = opts.EnsureDefaults()
+	s := opts.String()
+	require.NoError(t, opts.Check(s))
+	require.Regexp(t, `invalid key=value syntax`, opts.Check("foo\n"))
+
+	tmp := *opts
+	tmp.Comparer = &Comparer{Name: "foo"}
+	require.Regexp(t, `comparer name from file.*!=.*`, tmp.Check(s))
+
+	tmp = *opts
+	tmp.Merger = &Merger{Name: "foo"}
+	require.Regexp(t, `merger name from file.*!=.*`, tmp.Check(s))
+}
diff --git a/open_test.go b/open_test.go
index 860c7df..4e4aad9 100644
--- a/open_test.go
+++ b/open_test.go
@@ -14,6 +14,7 @@ import (
 
 	"github.com/petermattis/pebble/db"
 	"github.com/petermattis/pebble/storage"
+	"github.com/stretchr/testify/require"
 )
 
 func TestErrorIfDBExists(t *testing.T) {
@@ -163,3 +164,30 @@ func TestOpenCloseOpenClose(t *testing.T) {
 		}
 	}
 }
+
+func TestOpenOptionsCheck(t *testing.T) {
+	mem := storage.NewMem()
+	opts := &db.Options{Storage: mem}
+
+	d, err := Open("", opts)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := d.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	opts = &db.Options{
+		Comparer: &db.Comparer{Name: "foo"},
+		Storage:  mem,
+	}
+	_, err = Open("", opts)
+	require.Regexp(t, `comparer name from file.*!=.*`, err)
+
+	opts = &db.Options{
+		Merger:  &db.Merger{Name: "bar"},
+		Storage: mem,
+	}
+	_, err = Open("", opts)
+	require.Regexp(t, `merger name from file.*!=.*`, err)
+}