Repository: cockroachdb/pebble
Pull request #593 (2020-03-28 00:18:56)

internal/metamorphic: a quartet of changes
* internal/metamorphic: fix ingest_using_apply implementation
* internal/metamorphic: randomize MaxConcurrentCompactions
* internal/metamorphic: enable run comparison by default
* internal/metamorphic: temporarily disable ingest_using_apply

petermattis: 

<!-- Reviewable:start -->
This change is [<img src="https://reviewable.io/review_button.png" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/cockroachdb/pebble/593)
<!-- Reviewable:end -->

petermattis: I've add a 4th commit here which fixes `ingest_using_apply` and reenables it.

Patch:
diff --git a/compaction.go b/compaction.go
index 74ba8990..bcf2caa9 100644
--- a/compaction.go
+++ b/compaction.go
@@ -725,7 +725,7 @@ func (c *compaction) newInputIter(newIters tableNewIters) (_ internalIterator, r
 			f := c.inputs[0][i]
 			iter, rangeDelIter, err := newIters(f, nil /* iter options */, &c.bytesIterated)
 			if err != nil {
-				return nil, fmt.Errorf("pebble: could not open table %d: %w", f.FileNum, err)
+				return nil, fmt.Errorf("pebble: could not open table %d: %v", f.FileNum, err)
 			}
 			iters = append(iters, iter)
 			if rangeDelIter != nil {
diff --git a/internal/manifest/version.go b/internal/manifest/version.go
index 7e10eba2..5af898bd 100644
--- a/internal/manifest/version.go
+++ b/internal/manifest/version.go
@@ -370,7 +370,7 @@ func (v *Version) Overlaps(level int, cmp Compare, start, end []byte) (ret []*Fi
 func (v *Version) CheckOrdering(cmp Compare, format base.Formatter) error {
 	for level, files := range v.Files {
 		if err := CheckOrdering(cmp, format, level, files); err != nil {
-			return fmt.Errorf("%w\n%s", err, v.DebugString(format))
+			return fmt.Errorf("%s\n%s", err, v.DebugString(format))
 		}
 	}
 	return nil
diff --git a/internal/manifest/version_edit.go b/internal/manifest/version_edit.go
index a5d985bc..345168c3 100644
--- a/internal/manifest/version_edit.go
+++ b/internal/manifest/version_edit.go
@@ -539,7 +539,7 @@ func (b *BulkVersionEdit) Apply(
 			}
 			SortBySeqNum(v.Files[level])
 			if err := CheckOrdering(cmp, format, 0, v.Files[level]); err != nil {
-				return nil, nil, fmt.Errorf("pebble: internal error: %w", err)
+				return nil, nil, fmt.Errorf("pebble: internal error: %v", err)
 			}
 			continue
 		}
diff --git a/internal/metamorphic/ops.go b/internal/metamorphic/ops.go
index 7ac69b21..e41370bf 100644
--- a/internal/metamorphic/ops.go
+++ b/internal/metamorphic/ops.go
@@ -200,9 +200,13 @@ func (o *ingestOp) run(t *test, h *history) {
 	if t.testOpts.ingestUsingApply && len(o.batchIDs) == 1 {
 		id := o.batchIDs[0]
 		b := t.getBatch(id)
-		w := t.getWriter(makeObjID(dbTag, 0))
-		err := w.Apply(b, t.writeOpts)
+		c, err := o.collapseBatch(t, b)
+		if err == nil {
+			w := t.getWriter(makeObjID(dbTag, 0))
+			err = w.Apply(c, t.writeOpts)
+		}
 		_ = b.Close()
+		_ = c.Close()
 		t.clearObj(id)
 		h.Recordf("%s // %v", o, err)
 		return
@@ -299,6 +303,88 @@ func (o *ingestOp) build(t *test, h *history, b *pebble.Batch, i int) (string, e
 	return path, nil
 }
 
+// collapseBatch collapses the mutations in a batch to be equivalent to an
+// sstable ingesting those mutations. Duplicate updates to a key are collapsed
+// so that only the latest update is performed. All range deletions are
+// performed first in the batch to match the semantics of ingestion where a
+// range deletion does not delete a point record contained in the sstable.
+func (o *ingestOp) collapseBatch(t *test, b *pebble.Batch) (*pebble.Batch, error) {
+	iter, rangeDelIter := private.BatchSort(b)
+	defer func() {
+		if iter != nil {
+			iter.Close()
+		}
+		if rangeDelIter != nil {
+			rangeDelIter.Close()
+		}
+	}()
+
+	equal := t.opts.Comparer.Equal
+	collapsed := t.db.NewBatch()
+
+	if rangeDelIter != nil {
+		// NB: The range tombstones have already been fragmented by the Batch.
+		var lastUserKey []byte
+		var lastValue []byte
+		for key, value := rangeDelIter.First(); key != nil; key, value = rangeDelIter.Next() {
+			// Ignore duplicate tombstones.
+			if equal(lastUserKey, key.UserKey) && equal(lastValue, value) {
+				continue
+			}
+			// NB: We don't have to copy the key or value since we're reading from a
+			// batch which doesn't do prefix compression.
+			lastUserKey = key.UserKey
+			lastValue = value
+
+			if err := collapsed.DeleteRange(key.UserKey, value, nil); err != nil {
+				return nil, err
+			}
+		}
+		if err := rangeDelIter.Close(); err != nil {
+			return nil, err
+		}
+		rangeDelIter = nil
+	}
+
+	if iter != nil {
+		var lastUserKey []byte
+		for key, value := iter.First(); key != nil; key, value = iter.Next() {
+			// Ignore duplicate keys.
+			if equal(lastUserKey, key.UserKey) {
+				continue
+			}
+			// NB: We don't have to copy the key or value since we're reading from a
+			// batch which doesn't do prefix compression.
+			lastUserKey = key.UserKey
+
+			var err error
+			switch key.Kind() {
+			case pebble.InternalKeyKindDelete:
+				err = collapsed.Delete(key.UserKey, nil)
+			case pebble.InternalKeyKindSingleDelete:
+				err = collapsed.SingleDelete(key.UserKey, nil)
+			case pebble.InternalKeyKindSet:
+				err = collapsed.Set(key.UserKey, value, nil)
+			case pebble.InternalKeyKindMerge:
+				err = collapsed.Merge(key.UserKey, value, nil)
+			case pebble.InternalKeyKindLogData:
+				err = collapsed.LogData(key.UserKey, nil)
+			default:
+				err = fmt.Errorf("unknown batch record kind: %d", key.Kind())
+			}
+			if err != nil {
+				return nil, err
+			}
+		}
+		if err := iter.Close(); err != nil {
+			return nil, err
+		}
+		iter = nil
+	}
+
+	return collapsed, nil
+}
+
 func (o *ingestOp) String() string {
 	var buf strings.Builder
 	buf.WriteString("db.Ingest(")
diff --git a/internal/metamorphic/options.go b/internal/metamorphic/options.go
index 34cc73d8..d44b3322 100644
--- a/internal/metamorphic/options.go
+++ b/internal/metamorphic/options.go
@@ -158,6 +158,10 @@ func standardOptions() []*testOptions {
   strictfs=true
 `,
 		18: `
+[Options]
+  max_concurrent_compactions=2
+`,
+		19: `
 [TestOptions]
   ingest_using_apply=true
 `,
@@ -185,6 +189,7 @@ func randomOptions(rng *rand.Rand) *testOptions {
 		opts.L0StopWritesThreshold = opts.L0CompactionThreshold
 	}
 	opts.LBaseMaxBytes = 1 << uint(rng.Intn(30))       // 1B - 1GB
+	opts.MaxConcurrentCompactions = rng.Intn(4)        // 0-3
 	opts.MaxManifestFileSize = 1 << uint(rng.Intn(30)) // 1B  - 1GB
 	opts.MemTableSize = 1 << (10 + uint(rng.Intn(17))) // 1KB - 256MB
 	opts.MemTableStopWritesThreshold = 2 + rng.Intn(5) // 2 - 5
diff --git a/internal/record/log_writer.go b/internal/record/log_writer.go
index 682d9841..9e4118bd 100644
--- a/internal/record/log_writer.go
+++ b/internal/record/log_writer.go
@@ -446,7 +446,7 @@ func (w *LogWriter) flushPending(
 		// the stack that created the panic if panic'ing itself hits a panic
 		// (e.g. unlock of unlocked mutex).
 		if r := recover(); r != nil {
-			err = fmt.Errorf("%w\n%s", err, debug.Stack())
+			err = fmt.Errorf("%v\n%s", err, debug.Stack())
 		}
 	}()
 
diff --git a/level_checker.go b/level_checker.go
index 11cc7c0a..f17a78cf 100644
--- a/level_checker.go
+++ b/level_checker.go
@@ -175,7 +175,7 @@ func (m *simpleMergingIter) step() bool {
 			m.valueMerger, m.err = m.merge(item.key.UserKey, item.value)
 		}
 		if m.err != nil {
-			m.err = fmt.Errorf("merge processing error on key %s in %s: %w",
+			m.err = fmt.Errorf("merge processing error on key %s in %s: %v",
 				item.key.Pretty(m.format), l.iter, m.err)
 			return false
 		}
@@ -238,7 +238,7 @@ func (m *simpleMergingIter) step() bool {
 		if m.valueMerger != nil {
 			_, m.err = m.valueMerger.Finish()
 			if m.err != nil {
-				m.err = fmt.Errorf("merge processing error on key %s in %s: %w",
+				m.err = fmt.Errorf("merge processing error on key %s in %s: %v",
 					item.key.Pretty(m.format), lastRecordMsg, m.err)
 			}
 			m.valueMerger = nil
diff --git a/open.go b/open.go
index 94f86a4e..a3146b40 100644
--- a/open.go
+++ b/open.go
@@ -172,7 +172,7 @@ func Open(dirname string, opts *Options) (db *DB, _ error) {
 			return nil, err
 		}
 	} else if err != nil {
-		return nil, fmt.Errorf("pebble: database %q: %w", dirname, err)
+		return nil, fmt.Errorf("pebble: database %q: %v", dirname, err)
 	} else if opts.ErrorIfExists {
 		return nil, fmt.Errorf("pebble: database %q already exists", dirname)
 	} else {
diff --git a/sstable/table.go b/sstable/table.go
index 4b578f33..4b1bcf24 100644
--- a/sstable/table.go
+++ b/sstable/table.go
@@ -199,7 +199,7 @@ func readFooter(f vfs.File) (footer, error) {
 	var footer footer
 	stat, err := f.Stat()
 	if err != nil {
-		return footer, fmt.Errorf("pebble/table: invalid table (could not stat file): %w", err)
+		return footer, fmt.Errorf("pebble/table: invalid table (could not stat file): %v", err)
 	}
 	if stat.Size() < minFooterLen {
 		return footer, errors.New("pebble/table: invalid table (file size is too small)")
@@ -212,7 +212,7 @@ func readFooter(f vfs.File) (footer, error) {
 	}
 	n, err := f.ReadAt(buf, off)
 	if err != nil && err != io.EOF {
-		return footer, fmt.Errorf("pebble/table: invalid table (could not read footer): %w", err)
+		return footer, fmt.Errorf("pebble/table: invalid table (could not read footer): %v", err)
 	}
 	buf = buf[:n]
 
diff --git a/version_set.go b/version_set.go
index 3bc865c5..499d462a 100644
--- a/version_set.go
+++ b/version_set.go
@@ -155,7 +155,7 @@ func (vs *versionSet) load(dirname string, opts *Options, mu *sync.Mutex) error
 	// Read the CURRENT file to find the current manifest file.
 	current, err := vs.fs.Open(base.MakeFilename(vs.fs, dirname, fileTypeCurrent, 0))
 	if err != nil {
-		return fmt.Errorf("pebble: could not open CURRENT file for DB %q: %w", dirname, err)
+		return fmt.Errorf("pebble: could not open CURRENT file for DB %q: %v", dirname, err)
 	}
 	defer current.Close()
 	stat, err := current.Stat()
@@ -188,7 +188,7 @@ func (vs *versionSet) load(dirname string, opts *Options, mu *sync.Mutex) error
 	var bve bulkVersionEdit
 	manifest, err := vs.fs.Open(vs.fs.PathJoin(dirname, string(b)))
 	if err != nil {
-		return fmt.Errorf("pebble: could not open manifest file %q for DB %q: %w", b, dirname, err)
+		return fmt.Errorf("pebble: could not open manifest file %q for DB %q: %v", b, dirname, err)
 	}
 	defer manifest.Close()
 	rr := record.NewReader(manifest, 0 /* logNum */)


Test patch:
diff --git a/internal/metamorphic/meta_test.go b/internal/metamorphic/meta_test.go
index ff1a9cad..3ea0c28a 100644
--- a/internal/metamorphic/meta_test.go
+++ b/internal/metamorphic/meta_test.go
@@ -37,17 +37,17 @@ import (
 // - Add support for Writer.LogData
 
 var (
-	// TODO(peter): Enable comparing of output by default. Currently disabled as
-	// comparing output shows differences between runs that appear to be due to
-	// bugs in Pebble.
-	compare = flag.Bool("compare", false, "")
-	dir     = flag.String("dir", "_meta", "")
-	disk    = flag.Bool("disk", false, "")
-	failRE  = flag.String("fail", "",
+	dir = flag.String("dir", "_meta",
+		"the directory storing test state")
+	disk = flag.Bool("disk", false,
+		"whether to use an in-mem DB or on-disk (in-mem is significantly faster)")
+	failRE = flag.String("fail", "",
 		"fail the test if the supplied regular expression matches the output")
-	keep   = flag.Bool("keep", false, "")
+	keep = flag.Bool("keep", false,
+		"keep the DB directory even on successful runs")
 	ops    = randvar.NewFlag("uniform:5000-10000")
-	runDir = flag.String("run-dir", "", "")
+	runDir = flag.String("run-dir", "",
+		"the specific configuration to (re-)run (used for post-mortem debugging)")
 )
 
 func init() {
@@ -222,40 +222,38 @@ func TestMeta(t *testing.T) {
 		return
 	}
 
-	if *compare {
-		// Read a history file, stripping out lines that begin with a comment.
-		readHistory := func(name string) []string {
-			historyPath := filepath.Join(metaDir, name, "history")
-			data, err := ioutil.ReadFile(historyPath)
-			require.NoError(t, err)
-			lines := difflib.SplitLines(string(data))
-			newLines := make([]string, 0, len(lines))
-			for _, line := range lines {
-				if strings.HasPrefix(line, "// ") {
-					continue
-				}
-				newLines = append(newLines, line)
+	// Read a history file, stripping out lines that begin with a comment.
+	readHistory := func(name string) []string {
+		historyPath := filepath.Join(metaDir, name, "history")
+		data, err := ioutil.ReadFile(historyPath)
+		require.NoError(t, err)
+		lines := difflib.SplitLines(string(data))
+		newLines := make([]string, 0, len(lines))
+		for _, line := range lines {
+			if strings.HasPrefix(line, "// ") {
+				continue
 			}
-			return newLines
+			newLines = append(newLines, line)
 		}
+		return newLines
+	}
 
-		base := readHistory(names[0])
-		for i := 1; i < len(names); i++ {
-			lines := readHistory(names[i])
-			diff := difflib.UnifiedDiff{
-				A:       base,
-				B:       lines,
-				Context: 5,
-			}
-			text, err := difflib.GetUnifiedDiffString(diff)
-			require.NoError(t, err)
-			if text != "" {
-				// NB: We force an exit rather than using t.Fatal because the later
-				// will run another instance of the test if -count is specified, while
-				// we're happy to exit on the first failure.
-				fmt.Printf("diff %s/{%s,%s}\n%s\n", metaDir, names[0], names[i], text)
-				os.Exit(1)
-			}
+	base := readHistory(names[0])
+	for i := 1; i < len(names); i++ {
+		lines := readHistory(names[i])
+		diff := difflib.UnifiedDiff{
+			A:       base,
+			B:       lines,
+			Context: 5,
+		}
+		text, err := difflib.GetUnifiedDiffString(diff)
+		require.NoError(t, err)
+		if text != "" {
+			// NB: We force an exit rather than using t.Fatal because the latter
+			// will run another instance of the test if -count is specified, while
+			// we're happy to exit on the first failure.
+			fmt.Printf("diff %s/{%s,%s}\n%s\n", metaDir, names[0], names[i], text)
+			os.Exit(1)
 		}
 	}
 }