Repository: petermattis/pebble (go)

Issue:
db: target file size for level needs to be adjusted based on dynamic base level
Currently, the target file size for a level is constant (specified in `db.LevelOptions.TargetFileSize`), and does not account for how the number of levels in the LSM adjusts based on the amount of data stored in it. The "dynamic level bytes" heuristic allows Pebble to compact into L6 when the amount of data in the database is small, but it uses a target file size that is too large resulting in overly large compactions. RocksDB had this same issue, and fixed it in https://github.com/facebook/rocksdb/commit/a703432808c578768df96769da16097f39ea9741. 

Pull request description:
Fix dynamic-level-bytes heuristic
The implementation of dynamic-level-bytes was incomplete: we were
dynamically determining the base level and the size for each level, but the
base level wasn't being utilized to compute the compaction output level
when compacting from L0. The end result is that we'd determine the LSM only
needed 2 levels, but then compact from L0 -> L1, thus invalidating the
dyanmic-level-bytes heuristic (which stops at the first non-empty level).

Also fixed the target file size calculation to account for the dynamic base
level.

Fixes #108

Patch:
diff --git a/compaction.go b/compaction.go
index e3f8806..4cd6bff 100644
--- a/compaction.go
+++ b/compaction.go
@@ -39,9 +39,13 @@ type compaction struct {
 	cmp     db.Compare
 	version *version
 
-	// level is the level that is being compacted. Inputs from level and
-	// level+1 will be merged to produce a set of level+1 files.
-	level int
+	// startLevel is the level that is being compacted. Inputs from startLevel
+	// and outputLevel will be merged to produce a set of outputLevel files.
+	startLevel int
+	// outputLevel is the level that files are being produced in. outputLevel is
+	// equal to startLevel+1 except when startLevel is 0 in which case it is
+	// equal to compactionPicker.baseLevel.
+	outputLevel int
 
 	// maxOutputFileSize is the maximum size of an individual table created
 	// during compaction.
@@ -64,14 +68,32 @@ type compaction struct {
 	seenKey         bool   // some output key has been seen
 }
 
-func newCompaction(opts *db.Options, cur *version, level int) *compaction {
+func newCompaction(opts *db.Options, cur *version, startLevel, baseLevel int) *compaction {
+	if startLevel > 0 && startLevel < baseLevel {
+		panic(fmt.Sprintf("invalid compaction: start level %d should be empty (base level %d)",
+			startLevel, baseLevel))
+	}
+
+	outputLevel := startLevel + 1
+	if startLevel == 0 {
+		outputLevel = baseLevel
+	}
+	if outputLevel >= numLevels-1 {
+		outputLevel = numLevels - 1
+	}
+	// Output level is in the range [baseLevel,numLevels]. For the purpose of
+	// determining the target output file size, overlap bytes, and expanded
+	// bytes, we want to adjust the range to [1,numLevels].
+	adjustedOutputLevel := 1 + outputLevel - baseLevel
+
 	c := &compaction{
 		cmp:               opts.Comparer.Compare,
 		version:           cur,
-		level:             level,
-		maxOutputFileSize: uint64(opts.Level(level + 1).TargetFileSize),
-		maxOverlapBytes:   maxGrandparentOverlapBytes(opts, level+1),
-		maxExpandedBytes:  expandedCompactionByteSizeLimit(opts, level+1),
+		startLevel:        startLevel,
+		outputLevel:       outputLevel,
+		maxOutputFileSize: uint64(opts.Level(adjustedOutputLevel).TargetFileSize),
+		maxOverlapBytes:   maxGrandparentOverlapBytes(opts, adjustedOutputLevel),
+		maxExpandedBytes:  expandedCompactionByteSizeLimit(opts, adjustedOutputLevel),
 	}
 	return c
 }
@@ -81,7 +103,7 @@ func newCompaction(opts *db.Options, cur *version, level int) *compaction {
 func (c *compaction) setupOtherInputs() {
 	c.inputs[0] = c.expandInputs(c.inputs[0])
 	smallest0, largest0 := ikeyRange(c.cmp, c.inputs[0], nil)
-	c.inputs[1] = c.version.overlaps(c.level+1, c.cmp, smallest0.UserKey, largest0.UserKey)
+	c.inputs[1] = c.version.overlaps(c.outputLevel, c.cmp, smallest0.UserKey, largest0.UserKey)
 	smallest01, largest01 := ikeyRange(c.cmp, c.inputs[0], c.inputs[1])
 
 	// Grow the inputs if it doesn't affect the number of level+1 files.
@@ -89,9 +111,9 @@ func (c *compaction) setupOtherInputs() {
 		smallest01, largest01 = ikeyRange(c.cmp, c.inputs[0], c.inputs[1])
 	}
 
-	// Compute the set of level+2 files that overlap this compaction.
-	if c.level+2 < numLevels {
-		c.grandparents = c.version.overlaps(c.level+2, c.cmp, smallest01.UserKey, largest01.UserKey)
+	// Compute the set of outputLevel+1 files that overlap this compaction.
+	if c.outputLevel+1 < numLevels {
+		c.grandparents = c.version.overlaps(c.outputLevel+1, c.cmp, smallest01.UserKey, largest01.UserKey)
 	}
 }
 
@@ -101,12 +123,12 @@ func (c *compaction) setupOtherInputs() {
 // current input tables such that the rightmost table has a "clean cut". A
 // clean cut is either a change in user keys, or
 func (c *compaction) expandInputs(inputs []fileMetadata) []fileMetadata {
-	if c.level == 0 {
+	if c.startLevel == 0 {
 		// We already call version.overlaps for L0 and that call guarantees that we
 		// get a "clean cut".
 		return inputs
 	}
-	files := c.version.files[c.level]
+	files := c.version.files[c.startLevel]
 	// Pointer arithmetic to figure out the index if inputs[0] with
 	// files[0]. This requires that the inputs slice is a sub-slice of
 	// files. This is true for non-L0 files returned from version.overlaps.
@@ -145,7 +167,7 @@ func (c *compaction) grow(sm, la db.InternalKey) bool {
 	if len(c.inputs[1]) == 0 {
 		return false
 	}
-	grow0 := c.version.overlaps(c.level, c.cmp, sm.UserKey, la.UserKey)
+	grow0 := c.version.overlaps(c.startLevel, c.cmp, sm.UserKey, la.UserKey)
 	grow0 = c.expandInputs(grow0)
 	if len(grow0) <= len(c.inputs[0]) {
 		return false
@@ -154,7 +176,7 @@ func (c *compaction) grow(sm, la db.InternalKey) bool {
 		return false
 	}
 	sm1, la1 := ikeyRange(c.cmp, grow0, nil)
-	grow1 := c.version.overlaps(c.level+1, c.cmp, sm1.UserKey, la1.UserKey)
+	grow1 := c.version.overlaps(c.outputLevel, c.cmp, sm1.UserKey, la1.UserKey)
 	if len(grow1) != len(c.inputs[1]) {
 		return false
 	}
@@ -227,7 +249,7 @@ func (c *compaction) elideTombstone(key []byte) bool {
 	// TODO(peter): this can be faster if ukey is always increasing between
 	// successive elideTombstones calls and we can keep some state in between
 	// calls.
-	for level := c.level + 2; level < numLevels; level++ {
+	for level := c.outputLevel + 1; level < numLevels; level++ {
 		for _, f := range c.version.files[level] {
 			if c.cmp(key, f.largest.UserKey) <= 0 {
 				if c.cmp(key, f.smallest.UserKey) >= 0 {
@@ -244,9 +266,10 @@ func (c *compaction) elideTombstone(key []byte) bool {
 
 // elideRangeTombstone returns true if it is ok to elide the specified range
 // tombstone. A return value of true guarantees that there are no key/value
-// pairs at c.level+2 or higher that possibly overlap the specified tombstone.
+// pairs at c.outputLevel+1 or higher that possibly overlap the specified
+// tombstone.
 func (c *compaction) elideRangeTombstone(start, end []byte) bool {
-	for level := c.level + 2; level < numLevels; level++ {
+	for level := c.outputLevel + 1; level < numLevels; level++ {
 		overlaps := c.version.overlaps(level, c.cmp, start, end)
 		if len(overlaps) > 0 {
 			return false
@@ -354,7 +377,7 @@ func (c *compaction) newInputIter(
 		return rangeDelIter, nil, err
 	}
 
-	if c.level != 0 {
+	if c.startLevel != 0 {
 		iters = append(iters, newLevelIter(nil, c.cmp, newIters, c.inputs[0]))
 		iters = append(iters, newLevelIter(nil, c.cmp, newRangeDelIter, c.inputs[0]))
 	} else {
@@ -379,7 +402,11 @@ func (c *compaction) newInputIter(
 func (c *compaction) String() string {
 	var buf bytes.Buffer
 	for i := range c.inputs {
-		fmt.Fprintf(&buf, "%d:", i+c.level)
+		level := c.startLevel
+		if i == 1 {
+			level = c.outputLevel
+		}
+		fmt.Fprintf(&buf, "%d:", level)
 		for _, f := range c.inputs[i] {
 			fmt.Fprintf(&buf, " %d:%s-%s", f.fileNum, f.smallest, f.largest)
 		}
@@ -389,10 +416,11 @@ func (c *compaction) String() string {
 }
 
 type manualCompaction struct {
-	level int
-	done  chan error
-	start db.InternalKey
-	end   db.InternalKey
+	level       int
+	outputLevel int
+	done        chan error
+	start       db.InternalKey
+	end         db.InternalKey
 }
 
 // maybeScheduleFlush schedules a flush if necessary.
@@ -743,8 +771,8 @@ func (d *DB) compact1() (err error) {
 			Err:   err,
 		}
 		if err != nil {
-			info.Input.Level = c.level
-			info.Output.Level = c.level + 1
+			info.Input.Level = c.startLevel
+			info.Output.Level = c.outputLevel
 			for i := range c.inputs {
 				for j := range c.inputs[i] {
 					m := &c.inputs[i][j]
@@ -788,14 +816,14 @@ func (d *DB) compactDiskTables(c *compaction) (ve *versionEdit, pendingOutputs [
 	// the move could create a parent file that will require a very expensive
 	// merge later on.
 	if len(c.inputs[0]) == 1 && len(c.inputs[1]) == 0 &&
-		totalSize(c.grandparents) <= maxGrandparentOverlapBytes(d.opts, c.level+1) {
+		totalSize(c.grandparents) <= maxGrandparentOverlapBytes(d.opts, c.outputLevel) {
 		meta := &c.inputs[0][0]
 		return &versionEdit{
 			deletedFiles: map[deletedFileEntry]bool{
-				deletedFileEntry{level: c.level, fileNum: meta.fileNum}: true,
+				deletedFileEntry{level: c.startLevel, fileNum: meta.fileNum}: true,
 			},
 			newFiles: []newFileEntry{
-				{level: c.level + 1, meta: *meta},
+				{level: c.outputLevel, meta: *meta},
 			},
 		}, nil, nil
 	}
@@ -859,10 +887,10 @@ func (d *DB) compactDiskTables(c *compaction) (ve *versionEdit, pendingOutputs [
 			return err
 		}
 		filenames = append(filenames, filename)
-		tw = sstable.NewWriter(file, d.opts, d.opts.Level(c.level+1))
+		tw = sstable.NewWriter(file, d.opts, d.opts.Level(c.outputLevel))
 
 		ve.newFiles = append(ve.newFiles, newFileEntry{
-			level: c.level + 1,
+			level: c.outputLevel,
 			meta: fileMetadata{
 				fileNum: fileNum,
 			},
@@ -961,9 +989,13 @@ func (d *DB) compactDiskTables(c *compaction) (ve *versionEdit, pendingOutputs [
 	}
 
 	for i := range c.inputs {
+		level := c.startLevel
+		if i == 1 {
+			level = c.outputLevel
+		}
 		for _, f := range c.inputs[i] {
 			ve.deletedFiles[deletedFileEntry{
-				level:   c.level + i,
+				level:   level,
 				fileNum: f.fileNum,
 			}] = true
 		}
diff --git a/compaction_picker.go b/compaction_picker.go
index bc7fadf..52de0e0 100644
--- a/compaction_picker.go
+++ b/compaction_picker.go
@@ -232,11 +232,11 @@ func (p *compactionPicker) pickAuto(opts *db.Options) (c *compaction) {
 	}
 
 	vers := p.vers
-	c = newCompaction(opts, vers, p.level)
-	c.inputs[0] = vers.files[c.level][p.file : p.file+1]
+	c = newCompaction(opts, vers, p.level, p.baseLevel)
+	c.inputs[0] = vers.files[c.startLevel][p.file : p.file+1]
 
 	// Files in level 0 may overlap each other, so pick up all overlapping ones.
-	if c.level == 0 {
+	if c.startLevel == 0 {
 		cmp := opts.Comparer.Compare
 		smallest, largest := ikeyRange(cmp, c.inputs[0], nil)
 		c.inputs[0] = vers.overlaps(0, cmp, smallest.UserKey, largest.UserKey)
@@ -256,7 +256,8 @@ func (p *compactionPicker) pickManual(opts *db.Options, manual *manualCompaction
 
 	// TODO(peter): The logic here is untested and possibly incomplete.
 	cur := p.vers
-	c = newCompaction(opts, cur, manual.level)
+	c = newCompaction(opts, cur, manual.level, p.baseLevel)
+	manual.outputLevel = c.outputLevel
 	cmp := opts.Comparer.Compare
 	c.inputs[0] = cur.overlaps(manual.level, cmp, manual.start.UserKey, manual.end.UserKey)
 	if len(c.inputs[0]) == 0 {
diff --git a/db.go b/db.go
index df8a3c7..2a3a36e 100644
--- a/db.go
+++ b/db.go
@@ -662,7 +662,7 @@ func (d *DB) Compact(start, end []byte /* CompactionOptions */) error {
 		<-mem.flushed()
 	}
 
-	for level := 0; level < maxLevelWithFiles; level++ {
+	for level := 0; level < maxLevelWithFiles; {
 		manual := &manualCompaction{
 			done:  make(chan error, 1),
 			level: level,
@@ -672,6 +672,12 @@ func (d *DB) Compact(start, end []byte /* CompactionOptions */) error {
 		if err := d.manualCompact(manual); err != nil {
 			return err
 		}
+		level = manual.outputLevel
+		if level == numLevels-1 {
+			// A manual compaction of the bottommost level occured. There is no next
+			// level to try and compact.
+			break
+		}
 	}
 	return nil
 }
diff --git a/internal/datadriven/datadriven.go b/internal/datadriven/datadriven.go
index 48bda64..e5ad838 100644
--- a/internal/datadriven/datadriven.go
+++ b/internal/datadriven/datadriven.go
@@ -18,9 +18,11 @@ import (
 	"bufio"
 	"flag"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"testing"
 )
@@ -65,7 +67,31 @@ var (
 // actual results of the case, which this function compares with the expected
 // results, and either succeeds or fails the test.
 func RunTest(t *testing.T, path string, f func(d *TestData) string) {
-	r := newTestDataReader(t, path)
+	t.Helper()
+	file, err := os.OpenFile(path, os.O_RDWR, 0644 /* irrelevant */)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		_ = file.Close()
+	}()
+
+	runTestInternal(t, path, file, f, *rewriteTestFiles)
+}
+
+// RunTestFromString is a version of RunTest which takes the contents of a test
+// directly.
+func RunTestFromString(t *testing.T, input string, f func(d *TestData) string) {
+	t.Helper()
+	runTestInternal(t, "<string>" /* optionalPath */, strings.NewReader(input), f, *rewriteTestFiles)
+}
+
+func runTestInternal(
+	t *testing.T, sourceName string, reader io.Reader, f func(d *TestData) string, rewrite bool,
+) {
+	t.Helper()
+
+	r := newTestDataReader(t, sourceName, reader, rewrite)
 	for r.Next(t) {
 		d := &r.data
 		actual := func() string {
@@ -95,7 +121,12 @@ func RunTest(t *testing.T, path string, f func(d *TestData) string) {
 		} else if d.Expected != actual {
 			t.Fatalf("\n%s: %s\nexpected:\n%s\nfound:\n%s", d.Pos, d.Input, d.Expected, actual)
 		} else if testing.Verbose() {
-			fmt.Printf("\n%s:\n%s\n----\n%s", d.Pos, d.Input, actual)
+			input := d.Input
+			if input == "" {
+				input = "<no input to command>"
+			}
+			// TODO(tbg): it's awkward to reproduce the args, but it would be helpful.
+			fmt.Printf("\n%s:\n%s [%d args]\n%s\n----\n%s", d.Pos, d.Cmd, len(d.CmdArgs), input, actual)
 		}
 	}
 
@@ -104,9 +135,18 @@ func RunTest(t *testing.T, path string, f func(d *TestData) string) {
 		if l := len(data); l > 2 && data[l-1] == '\n' && data[l-2] == '\n' {
 			data = data[:l-1]
 		}
-		err := ioutil.WriteFile(path, data, 0644)
-		if err != nil {
-			t.Fatal(err)
+		if dest, ok := reader.(*os.File); ok {
+			if _, err := dest.WriteAt(data, 0); err != nil {
+				t.Fatal(err)
+			}
+			if err := dest.Truncate(int64(len(data))); err != nil {
+				t.Fatal(err)
+			}
+			if err := dest.Sync(); err != nil {
+				t.Fatal(err)
+			}
+		} else {
+			t.Logf("input is not a file; rewritten output is:\n%s", data)
 		}
 	}
 }
@@ -160,7 +200,7 @@ func Walk(t *testing.T, path string, f func(t *testing.T, path string)) {
 // TestData contains information about one data-driven test case that was
 // parsed from the test file.
 type TestData struct {
-	Pos string // file and line number
+	Pos string // reader and line number
 
 	// Cmd is the first string on the directive line (up to the first whitespace).
 	Cmd string
@@ -171,6 +211,67 @@ type TestData struct {
 	Expected string
 }
 
+// ScanArgs looks up the first CmdArg matching the given key and scans it into
+// the given destinations in order. If the arg does not exist, the number of
+// destinations does not match that of the arguments, or a destination can not
+// be populated from its matching value, a fatal error results.
+//
+// For example, for a TestData originating from
+//
+// cmd arg1=50 arg2=yoruba arg3=(50, 50, 50)
+//
+// the following would be valid:
+//
+// var i1, i2, i3, i4 int
+// var s string
+// td.ScanArgs(t, "arg1", &i1)
+// td.ScanArgs(t, "arg2", &s)
+// td.ScanArgs(t, "arg3", &i2, &i3, &i4)
+func (td *TestData) ScanArgs(t *testing.T, key string, dests ...interface{}) {
+	t.Helper()
+	var arg CmdArg
+	for i := range td.CmdArgs {
+		if td.CmdArgs[i].Key == key {
+			arg = td.CmdArgs[i]
+			break
+		}
+	}
+	if arg.Key == "" {
+		t.Fatalf("missing argument: %s", key)
+	}
+	if len(dests) != len(arg.Vals) {
+		t.Fatalf("%s: got %d destinations, but %d values", arg.Key, len(dests), len(arg.Vals))
+	}
+
+	for i := range dests {
+		val := arg.Vals[i]
+		switch dest := dests[i].(type) {
+		case *string:
+			*dest = val
+		case *int:
+			n, err := strconv.ParseInt(val, 10, 64)
+			if err != nil {
+				t.Fatal(err)
+			}
+			*dest = int(n) // assume 64bit ints
+		case *uint64:
+			n, err := strconv.ParseUint(val, 10, 64)
+			if err != nil {
+				t.Fatal(err)
+			}
+			*dest = n
+		case *bool:
+			b, err := strconv.ParseBool(val)
+			if err != nil {
+				t.Fatal(err)
+			}
+			*dest = b
+		default:
+			t.Fatalf("unsupported type %T for destination #%d (might be easy to add it)", dest, i+1)
+		}
+	}
+}
+
 // CmdArg contains information about an argument on the directive line. An
 // argument is specified in one of the following forms:
 //  - argument
@@ -181,7 +282,7 @@ type CmdArg struct {
 	Vals []string
 }
 
-func (arg *CmdArg) String() string {
+func (arg CmdArg) String() string {
 	switch len(arg.Vals) {
 	case 0:
 		return arg.Key
diff --git a/version_set.go b/version_set.go
index 7c77d80..f417173 100644
--- a/version_set.go
+++ b/version_set.go
@@ -29,6 +29,9 @@ type versionSet struct {
 	fs      vfs.FS
 	cmp     db.Compare
 	cmpName string
+	// Dynamic base level allows the dynamic base level computation to be
+	// disabled. Used by tests which want to create specific LSM structures.
+	dynamicBaseLevel bool
 
 	// Mutable fields.
 	versions versionList
@@ -62,6 +65,7 @@ func (vs *versionSet) load(dirname string, opts *db.Options, mu *sync.Mutex) err
 	vs.fs = opts.VFS
 	vs.cmp = opts.Comparer.Compare
 	vs.cmpName = opts.Comparer.Name
+	vs.dynamicBaseLevel = true
 	vs.versions.init()
 	// For historical reasons, the next file number is initialized to 2.
 	vs.nextFileNumber = 2
@@ -225,6 +229,9 @@ func (vs *versionSet) logAndApply(ve *versionEdit) error {
 			}
 		}
 		picker = newCompactionPicker(newVersion, vs.opts)
+		if !vs.dynamicBaseLevel {
+			picker.baseLevel = 1
+		}
 		return nil
 	}(); err != nil {
 		return err


Test patch:
diff --git a/compaction_test.go b/compaction_test.go
index 3c9858a..0f2a69a 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -69,8 +69,9 @@ func TestPickCompaction(t *testing.T) {
 				},
 			},
 			picker: compactionPicker{
-				score: 99,
-				level: 0,
+				score:     99,
+				level:     0,
+				baseLevel: 1,
 			},
 			want: "100  ",
 		},
@@ -96,8 +97,9 @@ func TestPickCompaction(t *testing.T) {
 				},
 			},
 			picker: compactionPicker{
-				score: 99,
-				level: 0,
+				score:     99,
+				level:     0,
+				baseLevel: 1,
 			},
 			want: "100  ",
 		},
@@ -123,8 +125,9 @@ func TestPickCompaction(t *testing.T) {
 				},
 			},
 			picker: compactionPicker{
-				score: 99,
-				level: 0,
+				score:     99,
+				level:     0,
+				baseLevel: 1,
 			},
 			want: "100,110  ",
 		},
@@ -150,8 +153,9 @@ func TestPickCompaction(t *testing.T) {
 				},
 			},
 			picker: compactionPicker{
-				score: 99,
-				level: 0,
+				score:     99,
+				level:     0,
+				baseLevel: 1,
 			},
 			want: "100,110  ",
 		},
@@ -185,8 +189,9 @@ func TestPickCompaction(t *testing.T) {
 				},
 			},
 			picker: compactionPicker{
-				score: 99,
-				level: 0,
+				score:     99,
+				level:     0,
+				baseLevel: 1,
 			},
 			want: "100  ",
 		},
@@ -246,8 +251,9 @@ func TestPickCompaction(t *testing.T) {
 				},
 			},
 			picker: compactionPicker{
-				score: 99,
-				level: 0,
+				score:     99,
+				level:     0,
+				baseLevel: 1,
 			},
 			want: "100 210 310,320,330",
 		},
@@ -299,8 +305,9 @@ func TestPickCompaction(t *testing.T) {
 				},
 			},
 			picker: compactionPicker{
-				score: 99,
-				level: 1,
+				score:     99,
+				level:     1,
+				baseLevel: 1,
 			},
 			want: "200,210,220 300 ",
 		},
@@ -352,8 +359,9 @@ func TestPickCompaction(t *testing.T) {
 				},
 			},
 			picker: compactionPicker{
-				score: 99,
-				level: 1,
+				score:     99,
+				level:     1,
+				baseLevel: 1,
 			},
 			want: "200 300 ",
 		},
@@ -405,8 +413,9 @@ func TestPickCompaction(t *testing.T) {
 				},
 			},
 			picker: compactionPicker{
-				score: 99,
-				level: 1,
+				score:     99,
+				level:     1,
+				baseLevel: 1,
 			},
 			want: "200 300 ",
 		},
@@ -436,7 +445,7 @@ func TestPickCompaction(t *testing.T) {
 	}
 }
 
-func TestIsBaseLevelForUkey(t *testing.T) {
+func TestElideTombstone(t *testing.T) {
 	testCases := []struct {
 		desc    string
 		level   int
@@ -560,9 +569,10 @@ func TestIsBaseLevelForUkey(t *testing.T) {
 
 	for _, tc := range testCases {
 		c := compaction{
-			cmp:     db.DefaultComparer.Compare,
-			version: &tc.version,
-			level:   tc.level,
+			cmp:         db.DefaultComparer.Compare,
+			version:     &tc.version,
+			startLevel:  tc.level,
+			outputLevel: tc.level + 1,
 		}
 		for ukey, want := range tc.wants {
 			if got := c.elideTombstone([]byte(ukey)); got != want {
@@ -852,6 +862,33 @@ func TestCompactionShouldStopBefore(t *testing.T) {
 		})
 }
 
+func TestCompactionOutputLevel(t *testing.T) {
+	opts := (*db.Options)(nil).EnsureDefaults()
+	version := &version{}
+
+	datadriven.RunTest(t, "testdata/compaction_output_level",
+		func(d *datadriven.TestData) (res string) {
+			defer func() {
+				if r := recover(); r != nil {
+					res = fmt.Sprintln(r)
+				}
+			}()
+
+			switch d.Cmd {
+			case "compact":
+				var start, base int
+				d.ScanArgs(t, "start", &start)
+				d.ScanArgs(t, "base", &base)
+				c := newCompaction(opts, version, start, base)
+				return fmt.Sprintf("output=%d\nmax-output-file-size=%d\n",
+					c.outputLevel, c.maxOutputFileSize)
+
+			default:
+				return fmt.Sprintf("unknown command: %s", d.Cmd)
+			}
+		})
+}
+
 func TestCompactionExpandInputs(t *testing.T) {
 	cmp := db.DefaultComparer.Compare
 	var files []fileMetadata
@@ -885,11 +922,11 @@ func TestCompactionExpandInputs(t *testing.T) {
 
 			case "expand-inputs":
 				c := &compaction{
-					cmp:     cmp,
-					version: &version{},
-					level:   1,
+					cmp:        cmp,
+					version:    &version{},
+					startLevel: 1,
 				}
-				c.version.files[c.level] = files
+				c.version.files[c.startLevel] = files
 				if len(d.CmdArgs) != 1 {
 					return fmt.Sprintf("%s expects 1 argument", d.Cmd)
 				}
@@ -1016,22 +1053,23 @@ func TestCompactionAllowZeroSeqNum(t *testing.T) {
 					}
 					c.inputs[0] = nil
 					c.inputs[1] = nil
-					c.level = -1
+					c.startLevel = -1
 
 					for _, p := range parts {
 						level, meta := parseMeta(p)
 						i := 0
 						switch {
-						case c.level == -1:
-							c.level = level
-						case c.level+1 == level:
+						case c.startLevel == -1:
+							c.startLevel = level
+						case c.startLevel+1 == level:
 							i = 1
-						case c.level != level:
+						case c.startLevel != level:
 							return fmt.Sprintf("invalid level %d: expected %d or %d",
-								level, c.level, c.level+1)
+								level, c.startLevel, c.startLevel+1)
 						}
 						c.inputs[i] = append(c.inputs[i], meta)
 					}
+					c.outputLevel = c.startLevel + 1
 					fmt.Fprintf(&buf, "%t\n", c.allowZeroSeqNum())
 				}
 				return buf.String()
diff --git a/data_test.go b/data_test.go
index c2500f3..3ce919c 100644
--- a/data_test.go
+++ b/data_test.go
@@ -181,9 +181,8 @@ func runCompactCommand(td *datadriven.TestData, d *DB) error {
 			start: iStart,
 			end:   iEnd,
 		})
-	} else {
-		return d.Compact([]byte(parts[0]), []byte(parts[1]))
 	}
+	return d.Compact([]byte(parts[0]), []byte(parts[1]))
 }
 
 func runDBDefineCmd(td *datadriven.TestData) (*DB, error) {
@@ -227,6 +226,7 @@ func runDBDefineCmd(td *datadriven.TestData) (*DB, error) {
 		return nil, err
 	}
 	d.mu.Lock()
+	d.mu.versions.dynamicBaseLevel = false
 	for i := range snapshots {
 		s := &Snapshot{db: d}
 		s.seqNum = snapshots[i]
diff --git a/internal/datadriven/datadriven_test.go b/internal/datadriven/datadriven_test.go
new file mode 100644
index 0000000..d2a964e
--- /dev/null
+++ b/internal/datadriven/datadriven_test.go
@@ -0,0 +1,50 @@
+// Copyright 2019 The Cockroach Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License.
+
+package datadriven
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+)
+
+func TestDataDriven(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	input := `
+# NB: we allow duplicate args. It's unclear at this time whether this is useful,
+# either way, ScanArgs simply picks the first occurrence.
+make argTuple=(1, 🍌) argInt=12 argString=greedily argString=totally_ignored
+sentence
+----
+Did the following: make sentence
+1 hungry monkey eats a 🍌
+while 12 other monkeys watch greedily
+`
+
+	RunTestFromString(t, input, func(d *TestData) string {
+		var one int
+		var twelve int
+		var banana string
+		var greedily string
+		d.ScanArgs(t, "argTuple", &one, &banana)
+		d.ScanArgs(t, "argInt", &twelve)
+		d.ScanArgs(t, "argString", &greedily)
+		return fmt.Sprintf("Did the following: %s %s\n%d hungry monkey eats a %s\nwhile %d other monkeys watch %s\n",
+			d.Cmd, d.Input, one, banana, twelve, greedily,
+		)
+	})
+}
diff --git a/internal/datadriven/test_data_reader.go b/internal/datadriven/test_data_reader.go
index 0b284d8..315fbf2 100644
--- a/internal/datadriven/test_data_reader.go
+++ b/internal/datadriven/test_data_reader.go
@@ -17,43 +17,37 @@ package datadriven
 import (
 	"bytes"
 	"fmt"
-	"os"
+	"io"
 	"regexp"
 	"strings"
 	"testing"
 )
 
 type testDataReader struct {
-	path    string
-	file    *os.File
-	scanner *lineScanner
-	data    TestData
-	rewrite *bytes.Buffer
+	sourceName string
+	reader     io.Reader
+	scanner    *lineScanner
+	data       TestData
+	rewrite    *bytes.Buffer
 }
 
-func newTestDataReader(t *testing.T, path string) *testDataReader {
+func newTestDataReader(
+	t *testing.T, sourceName string, file io.Reader, record bool,
+) *testDataReader {
 	t.Helper()
 
-	file, err := os.Open(path)
-	if err != nil {
-		t.Fatal(err)
-	}
 	var rewrite *bytes.Buffer
-	if *rewriteTestFiles {
+	if record {
 		rewrite = &bytes.Buffer{}
 	}
 	return &testDataReader{
-		path:    path,
-		file:    file,
-		scanner: newLineScanner(file),
-		rewrite: rewrite,
+		sourceName: sourceName,
+		reader:     file,
+		scanner:    newLineScanner(file),
+		rewrite:    rewrite,
 	}
 }
 
-func (r *testDataReader) Close() error {
-	return r.file.Close()
-}
-
 func (r *testDataReader) Next(t *testing.T) bool {
 	t.Helper()
 
@@ -81,7 +75,7 @@ func (r *testDataReader) Next(t *testing.T) bool {
 			continue
 		}
 		cmd := fields[0]
-		r.data.Pos = fmt.Sprintf("%s:%d", r.path, r.scanner.line)
+		r.data.Pos = fmt.Sprintf("%s:%d", r.sourceName, r.scanner.line)
 		r.data.Cmd = cmd
 
 		for _, arg := range fields[1:] {
@@ -185,7 +179,7 @@ func (r *testDataReader) emit(s string) {
 	}
 }
 
-var splitDirectivesRE = regexp.MustCompile(`^ *[a-zA-Z0-9_,-\.]+(|=[-a-zA-Z0-9_@:]+|=\([^)]*\))( |$)`)
+var splitDirectivesRE = regexp.MustCompile(`^ *[a-zA-Z0-9_,-\.]+(|=[-a-zA-Z0-9_@]+|=\([^)]*\))( |$)`)
 
 // splits a directive line into tokens, where each token is
 // either:
diff --git a/range_del_test.go b/range_del_test.go
index bc983c1..9d17728 100644
--- a/range_del_test.go
+++ b/range_del_test.go
@@ -30,6 +30,8 @@ func TestRangeDel(t *testing.T) {
 			}
 
 			d.mu.Lock()
+			// Disable the "dynamic base level" code for this test.
+			d.mu.versions.picker.baseLevel = 1
 			s := fmt.Sprintf("mem: %d\n%s", len(d.mu.mem.queue), d.mu.versions.currentVersion())
 			d.mu.Unlock()
 			return s
@@ -39,6 +41,8 @@ func TestRangeDel(t *testing.T) {
 				return err.Error()
 			}
 			d.mu.Lock()
+			// Disable the "dynamic base level" code for this test.
+			d.mu.versions.picker.baseLevel = 1
 			s := d.mu.versions.currentVersion().String()
 			d.mu.Unlock()
 			return s
@@ -127,6 +131,10 @@ func TestRangeDelCompactionTruncation(t *testing.T) {
 	}
 	defer d.Close()
 
+	d.mu.Lock()
+	d.mu.versions.dynamicBaseLevel = false
+	d.mu.Unlock()
+
 	lsm := func() string {
 		d.mu.Lock()
 		s := d.mu.versions.currentVersion().DebugString()
diff --git a/testdata/compaction_output_level b/testdata/compaction_output_level
new file mode 100644
index 0000000..13a0eb3
--- /dev/null
+++ b/testdata/compaction_output_level
@@ -0,0 +1,199 @@
+compact start=0 base=1
+----
+output=1
+max-output-file-size=4194304
+
+compact start=1 base=1
+----
+output=2
+max-output-file-size=8388608
+
+compact start=2 base=1
+----
+output=3
+max-output-file-size=16777216
+
+compact start=3 base=1
+----
+output=4
+max-output-file-size=33554432
+
+compact start=4 base=1
+----
+output=5
+max-output-file-size=67108864
+
+compact start=5 base=1
+----
+output=6
+max-output-file-size=134217728
+
+compact start=6 base=1
+----
+output=6
+max-output-file-size=134217728
+
+
+compact start=0 base=2
+----
+output=2
+max-output-file-size=4194304
+
+compact start=1 base=2
+----
+invalid compaction: start level 1 should be empty (base level 2)
+
+compact start=2 base=2
+----
+output=3
+max-output-file-size=8388608
+
+compact start=3 base=2
+----
+output=4
+max-output-file-size=16777216
+
+compact start=4 base=2
+----
+output=5
+max-output-file-size=33554432
+
+compact start=5 base=2
+----
+output=6
+max-output-file-size=67108864
+
+compact start=6 base=2
+----
+output=6
+max-output-file-size=67108864
+
+
+compact start=0 base=3
+----
+output=3
+max-output-file-size=4194304
+
+compact start=1 base=3
+----
+invalid compaction: start level 1 should be empty (base level 3)
+
+compact start=2 base=3
+----
+invalid compaction: start level 2 should be empty (base level 3)
+
+compact start=3 base=3
+----
+output=4
+max-output-file-size=8388608
+
+compact start=4 base=3
+----
+output=5
+max-output-file-size=16777216
+
+compact start=5 base=3
+----
+output=6
+max-output-file-size=33554432
+
+compact start=6 base=3
+----
+output=6
+max-output-file-size=33554432
+
+
+compact start=0 base=4
+----
+output=4
+max-output-file-size=4194304
+
+compact start=1 base=4
+----
+invalid compaction: start level 1 should be empty (base level 4)
+
+compact start=2 base=4
+----
+invalid compaction: start level 2 should be empty (base level 4)
+
+compact start=3 base=4
+----
+invalid compaction: start level 3 should be empty (base level 4)
+
+compact start=4 base=4
+----
+output=5
+max-output-file-size=8388608
+
+compact start=5 base=4
+----
+output=6
+max-output-file-size=16777216
+
+compact start=6 base=4
+----
+output=6
+max-output-file-size=16777216
+
+
+compact start=0 base=5
+----
+output=5
+max-output-file-size=4194304
+
+compact start=1 base=5
+----
+invalid compaction: start level 1 should be empty (base level 5)
+
+compact start=2 base=5
+----
+invalid compaction: start level 2 should be empty (base level 5)
+
+compact start=3 base=5
+----
+invalid compaction: start level 3 should be empty (base level 5)
+
+compact start=4 base=5
+----
+invalid compaction: start level 4 should be empty (base level 5)
+
+compact start=5 base=5
+----
+output=6
+max-output-file-size=8388608
+
+compact start=6 base=5
+----
+output=6
+max-output-file-size=8388608
+
+
+compact start=0 base=6
+----
+output=6
+max-output-file-size=4194304
+
+compact start=1 base=6
+----
+invalid compaction: start level 1 should be empty (base level 6)
+
+compact start=2 base=6
+----
+invalid compaction: start level 2 should be empty (base level 6)
+
+compact start=3 base=6
+----
+invalid compaction: start level 3 should be empty (base level 6)
+
+compact start=4 base=6
+----
+invalid compaction: start level 4 should be empty (base level 6)
+
+compact start=5 base=6
+----
+invalid compaction: start level 5 should be empty (base level 6)
+
+compact start=6 base=6
+----
+output=6
+max-output-file-size=4194304
diff --git a/testdata/manual_compaction b/testdata/manual_compaction
index e0278b5..e0eb336 100644
--- a/testdata/manual_compaction
+++ b/testdata/manual_compaction
@@ -5,7 +5,7 @@ set b 2
 
 compact a-b
 ----
-1: a-b
+6: a-b
 
 batch
 set c 3
@@ -14,7 +14,7 @@ set d 4
 
 compact c-d
 ----
-1: a-b c-d
+6: a-b c-d
 
 batch
 set b 5
@@ -23,7 +23,7 @@ set c 6
 
 compact a-d
 ----
-2: a-d
+6: a-d
 
 # This also tests flushing a memtable that only contains range
 # deletions.