compaction_picker_test.go

// Copyright 2018 The LevelDB-Go and Pebble Authors. All rights reserved. Use
// of this source code is governed by a BSD-style license that can be found in
// the LICENSE file.

package pebble

import (
	"bytes"
	"fmt"
	"math"
	"sort"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/cockroachdb/datadriven"
	"github.com/cockroachdb/errors"
	"github.com/cockroachdb/pebble/internal/base"
	"github.com/cockroachdb/pebble/internal/humanize"
	"github.com/cockroachdb/pebble/internal/manifest"
	"github.com/cockroachdb/pebble/internal/testkeys"
	"github.com/cockroachdb/pebble/vfs"
	"github.com/stretchr/testify/require"
)

func loadVersion(t *testing.T, d *datadriven.TestData) (*version, *Options, string) {
	var sizes [numLevels]int64
	opts := &Options{}
	opts.testingRandomized(t)
	opts.EnsureDefaults()

	if len(d.CmdArgs) != 1 {
		return nil, nil, fmt.Sprintf("%s expects 1 argument", d.Cmd)
	}
	var err error
	opts.LBaseMaxBytes, err = strconv.ParseInt(d.CmdArgs[0].Key, 10, 64)
	if err != nil {
		return nil, nil, err.Error()
	}

	var files [numLevels][]*fileMetadata
	if len(d.Input) > 0 {
		// Parse each line as
		//
		// <level>: <size> [compensation]
		//
		// Creating sstables within the level whose file sizes total to `size`
		// and whose compensated file sizes total to `size`+`compensation`. If
		// size is sufficiently large, only one single file is created. See
		// the TODO below.
		for _, data := range strings.Split(d.Input, "\n") {
			parts := strings.Split(data, " ")
			parts[0] = strings.TrimSuffix(strings.TrimSpace(parts[0]), ":")
			if len(parts) < 2 {
				return nil, nil, fmt.Sprintf("malformed test:\n%s", d.Input)
			}
			level, err := strconv.Atoi(parts[0])
			if err != nil {
				return nil, nil, err.Error()
			}
			if files[level] != nil {
				return nil, nil, fmt.Sprintf("level %d already filled", level)
			}
			size, err := strconv.ParseUint(strings.TrimSpace(parts[1]), 10, 64)
			if err != nil {
				return nil, nil, err.Error()
			}
			var compensation uint64
			if len(parts) == 3 {
				compensation, err = strconv.ParseUint(strings.TrimSpace(parts[2]), 10, 64)
				if err != nil {
					return nil, nil, err.Error()
				}
			}

			var lastFile *fileMetadata
			for i := uint64(1); sizes[level] < int64(size); i++ {
				var key InternalKey
				if level == 0 {
					// For L0, make `size` overlapping files.
					key = base.MakeInternalKey([]byte(fmt.Sprintf("%04d", 1)), i, InternalKeyKindSet)
				} else {
					key = base.MakeInternalKey([]byte(fmt.Sprintf("%04d", i)), i, InternalKeyKindSet)
				}
				m := (&fileMetadata{
					FileNum:               base.FileNum(uint64(level)*100_000 + i),
					SmallestSeqNum:        key.SeqNum(),
					LargestSeqNum:         key.SeqNum(),
					LargestSeqNumAbsolute: key.SeqNum(),
					Size:                  1,
					Stats: manifest.TableStats{
						RangeDeletionsBytesEstimate: 0,
					},
				}).ExtendPointKeyBounds(opts.Comparer.Compare, key, key)
				m.InitPhysicalBacking()
				m.StatsMarkValid()
				lastFile = m
				if size >= 100 {
					// If the requested size of the level is very large only add a single
					// file in order to avoid massive blow-up in the number of files in
					// the Version.
					//
					// TODO(peter): There is tension between the testing in
					// TestCompactionPickerLevelMaxBytes and
					// TestCompactionPickerTargetLevel. Clean this up somehow.
					m.Size = size
					if level != 0 {
						endKey := base.MakeInternalKey([]byte(fmt.Sprintf("%04d", size)), i, InternalKeyKindSet)
						m.ExtendPointKeyBounds(opts.Comparer.Compare, key, endKey)
					}
				}
				files[level] = append(files[level], m)
				sizes[level] += int64(m.Size)
			}
			// Let all the compensation be due to the last file.
			if lastFile != nil && compensation > 0 {
				lastFile.Stats.RangeDeletionsBytesEstimate = compensation
			}
		}
	}

	vers := newVersion(opts, files)
	return vers, opts, ""
}

func TestCompactionPickerByScoreLevelMaxBytes(t *testing.T) {
	datadriven.RunTest(t, "testdata/compaction_picker_level_max_bytes",
		func(t *testing.T, d *datadriven.TestData) string {
			switch d.Cmd {
			case "init":
				vers, opts, errMsg := loadVersion(t, d)
				if errMsg != "" {
					return errMsg
				}

				vb := manifest.MakeVirtualBackings()
				p := newCompactionPickerByScore(vers, &vb, opts, nil)
				var buf bytes.Buffer
				for level := p.getBaseLevel(); level < numLevels; level++ {
					fmt.Fprintf(&buf, "%d: %d\n", level, p.levelMaxBytes[level])
				}
				return buf.String()

			default:
				return fmt.Sprintf("unknown command: %s", d.Cmd)
			}
		})
}

func TestCompactionPickerTargetLevel(t *testing.T) {
	var vers *version
	var opts *Options
	var pickerByScore *compactionPickerByScore

	parseInProgress := func(vals []string) ([]compactionInfo, error) {
		var levels []int
		for _, s := range vals {
			l, err := strconv.ParseInt(s, 10, 8)
			if err != nil {
				return nil, err
			}
			levels = append(levels, int(l))
		}
		if len(levels)%2 != 0 {
			return nil, errors.New("odd number of levels with ongoing compactions")
		}
		var inProgress []compactionInfo
		for i := 0; i < len(levels); i += 2 {
			inProgress = append(inProgress, compactionInfo{
				inputs: []compactionLevel{
					{level: levels[i]},
					{level: levels[i+1]},
				},
				outputLevel: levels[i+1],
			})
		}
		return inProgress, nil
	}

	resetCompacting := func() {
		for _, files := range vers.Levels {
			files.Slice().Each(func(f *fileMetadata) {
				f.CompactionState = manifest.CompactionStateNotCompacting
			})
		}
	}

	datadriven.RunTest(t, "testdata/compaction_picker_target_level",
		func(t *testing.T, d *datadriven.TestData) string {
			switch d.Cmd {
			case "init":
				// loadVersion expects a single datadriven argument that it
				// sets as Options.LBaseMaxBytes. It parses the input as
				// newline-separated levels, specifying the level's file size
				// and optionally additional compensation to be added during
				// compensated file size calculations. Eg:
				//
				// init <LBaseMaxBytes>
				// <level>: <size> [compensation]
				// <level>: <size> [compensation]
				var errMsg string
				vers, opts, errMsg = loadVersion(t, d)
				if errMsg != "" {
					return errMsg
				}
				return runVersionFileSizes(vers)
			case "init_cp":
				resetCompacting()

				var inProgress []compactionInfo
				if arg, ok := d.Arg("ongoing"); ok {
					var err error
					inProgress, err = parseInProgress(arg.Vals)
					if err != nil {
						return err.Error()
					}
				}

				vb := manifest.MakeVirtualBackings()
				pickerByScore = newCompactionPickerByScore(vers, &vb, opts, inProgress)
				return fmt.Sprintf("base: %d", pickerByScore.baseLevel)
			case "queue":
				var b strings.Builder
				var inProgress []compactionInfo
				for {
					env := compactionEnv{
						diskAvailBytes:          math.MaxUint64,
						earliestUnflushedSeqNum: InternalKeySeqNumMax,
						inProgressCompactions:   inProgress,
					}
					pc := pickerByScore.pickAuto(env)
					if pc == nil {
						break
					}
					fmt.Fprintf(&b, "L%d->L%d: %.1f\n", pc.startLevel.level, pc.outputLevel.level, pc.score)
					inProgress = append(inProgress, compactionInfo{
						inputs:      pc.inputs,
						outputLevel: pc.outputLevel.level,
						smallest:    pc.smallest,
						largest:     pc.largest,
					})
					if pc.outputLevel.level == 0 {
						// Once we pick one L0->L0 compaction, we'll keep on doing so
						// because the test isn't marking files as Compacting.
						break
					}
					for _, cl := range pc.inputs {
						cl.files.Each(func(f *fileMetadata) {
							f.CompactionState = manifest.CompactionStateCompacting
							fmt.Fprintf(&b, "  %s marked as compacting\n", f)
						})
					}
				}

				resetCompacting()
				return b.String()
			case "pick":
				resetCompacting()

				var inProgress []compactionInfo
				if len(d.CmdArgs) == 1 {
					arg := d.CmdArgs[0]
					if arg.Key != "ongoing" {
						return "unknown arg: " + arg.Key
					}
					var err error
					inProgress, err = parseInProgress(arg.Vals)
					if err != nil {
						return err.Error()
					}
				}

				// Mark files as compacting for each in-progress compaction.
				for i := range inProgress {
					c := &inProgress[i]
					for j, cl := range c.inputs {
						iter := vers.Levels[cl.level].Iter()
						for f := iter.First(); f != nil; f = iter.Next() {
							if !f.IsCompacting() {
								f.CompactionState = manifest.CompactionStateCompacting
								c.inputs[j].files = iter.Take().Slice()
								break
							}
						}
					}
					if c.inputs[0].level == 0 && c.outputLevel != 0 {
						// L0->Lbase: mark all of Lbase as compacting.
						c.inputs[1].files = vers.Levels[c.outputLevel].Slice()
						for _, in := range c.inputs {
							in.files.Each(func(f *fileMetadata) {
								f.CompactionState = manifest.CompactionStateCompacting
							})
						}
					}
				}

				var b strings.Builder
				fmt.Fprintf(&b, "Initial state before pick:\n%s", runVersionFileSizes(vers))
				pc := pickerByScore.pickAuto(compactionEnv{
					earliestUnflushedSeqNum: InternalKeySeqNumMax,
					inProgressCompactions:   inProgress,
				})
				if pc != nil {
					fmt.Fprintf(&b, "Picked: L%d->L%d: %0.1f\n", pc.startLevel.level, pc.outputLevel.level, pc.score)
				}
				if pc == nil {
					fmt.Fprintln(&b, "Picked: no compaction")
				}
				return b.String()
			case "pick_manual":
				var startLevel int
				var start, end string
				d.MaybeScanArgs(t, "level", &startLevel)
				d.MaybeScanArgs(t, "start", &start)
				d.MaybeScanArgs(t, "end", &end)

				iStart := base.MakeInternalKey([]byte(start), InternalKeySeqNumMax, InternalKeyKindMax)
				iEnd := base.MakeInternalKey([]byte(end), 0, 0)
				manual := &manualCompaction{
					done:  make(chan error, 1),
					level: startLevel,
					start: iStart.UserKey,
					end:   iEnd.UserKey,
				}

				pc, retryLater := pickManualCompaction(
					pickerByScore.vers,
					opts,
					compactionEnv{
						earliestUnflushedSeqNum: InternalKeySeqNumMax,
					},
					pickerByScore.getBaseLevel(),
					manual)
				if pc == nil {
					return fmt.Sprintf("nil, retryLater = %v", retryLater)
				}

				return fmt.Sprintf("L%d->L%d, retryLater = %v", pc.startLevel.level, pc.outputLevel.level, retryLater)
			default:
				return fmt.Sprintf("unknown command: %s", d.Cmd)
			}
		})
}

func TestCompactionPickerEstimatedCompactionDebt(t *testing.T) {
	datadriven.RunTest(t, "testdata/compaction_picker_estimated_debt",
		func(t *testing.T, d *datadriven.TestData) string {
			switch d.Cmd {
			case "init":
				vers, opts, errMsg := loadVersion(t, d)
				if errMsg != "" {
					return errMsg
				}
				opts.MemTableSize = 1000

				vb := manifest.MakeVirtualBackings()
				p := newCompactionPickerByScore(vers, &vb, opts, nil)
				return fmt.Sprintf("%d\n", p.estimatedCompactionDebt(0))

			default:
				return fmt.Sprintf("unknown command: %s", d.Cmd)
			}
		})
}

func TestCompactionPickerL0(t *testing.T) {
	opts := (*Options)(nil).EnsureDefaults()
	opts.Experimental.L0CompactionConcurrency = 1

	parseMeta := func(s string) (*fileMetadata, error) {
		parts := strings.Split(s, ":")
		fileNum, err := strconv.Atoi(parts[0])
		if err != nil {
			return nil, err
		}
		fields := strings.Fields(parts[1])
		parts = strings.Split(fields[0], "-")
		if len(parts) != 2 {
			return nil, errors.Errorf("malformed table spec: %s", s)
		}
		m := (&fileMetadata{
			FileNum: base.FileNum(fileNum),
		}).ExtendPointKeyBounds(
			opts.Comparer.Compare,
			base.ParseInternalKey(strings.TrimSpace(parts[0])),
			base.ParseInternalKey(strings.TrimSpace(parts[1])),
		)
		m.SmallestSeqNum = m.Smallest.SeqNum()
		m.LargestSeqNum = m.Largest.SeqNum()
		if m.SmallestSeqNum > m.LargestSeqNum {
			m.SmallestSeqNum, m.LargestSeqNum = m.LargestSeqNum, m.SmallestSeqNum
		}
		m.LargestSeqNumAbsolute = m.LargestSeqNum
		m.InitPhysicalBacking()
		return m, nil
	}

	var picker *compactionPickerByScore
	var inProgressCompactions []compactionInfo
	var pc *pickedCompaction

	datadriven.RunTest(t, "testdata/compaction_picker_L0", func(t *testing.T, td *datadriven.TestData) string {
		switch td.Cmd {
		case "define":
			fileMetas := [manifest.NumLevels][]*fileMetadata{}
			baseLevel := manifest.NumLevels - 1
			level := 0
			var err error
			lines := strings.Split(td.Input, "\n")
			var compactionLines []string

			for len(lines) > 0 {
				data := strings.TrimSpace(lines[0])
				lines = lines[1:]
				switch data {
				case "L0", "L1", "L2", "L3", "L4", "L5", "L6":
					level, err = strconv.Atoi(data[1:])
					if err != nil {
						return err.Error()
					}
				case "compactions":
					compactionLines, lines = lines, nil
				default:
					meta, err := parseMeta(data)
					if err != nil {
						return err.Error()
					}
					if level != 0 && level < baseLevel {
						baseLevel = level
					}
					fileMetas[level] = append(fileMetas[level], meta)
				}
			}

			// Parse in-progress compactions in the form of:
			//   L0 000001 -> L2 000005
			inProgressCompactions = nil
			for len(compactionLines) > 0 {
				parts := strings.Fields(compactionLines[0])
				compactionLines = compactionLines[1:]

				var level int
				var info compactionInfo
				first := true
				compactionFiles := map[int][]*fileMetadata{}
				for _, p := range parts {
					switch p {
					case "L0", "L1", "L2", "L3", "L4", "L5", "L6":
						var err error
						level, err = strconv.Atoi(p[1:])
						if err != nil {
							return err.Error()
						}
						if len(info.inputs) > 0 && info.inputs[len(info.inputs)-1].level == level {
							// eg, L0 -> L0 compaction or L6 -> L6 compaction
							continue
						}
						if info.outputLevel < level {
							info.outputLevel = level
						}
						info.inputs = append(info.inputs, compactionLevel{level: level})
					case "->":
						continue
					default:
						fileNum, err := strconv.Atoi(p)
						if err != nil {
							return err.Error()
						}
						var compactFile *fileMetadata
						for _, m := range fileMetas[level] {
							if m.FileNum == FileNum(fileNum) {
								compactFile = m
							}
						}
						if compactFile == nil {
							return fmt.Sprintf("cannot find compaction file %s", FileNum(fileNum))
						}
						compactFile.CompactionState = manifest.CompactionStateCompacting
						if first || base.InternalCompare(DefaultComparer.Compare, info.largest, compactFile.Largest) < 0 {
							info.largest = compactFile.Largest
						}
						if first || base.InternalCompare(DefaultComparer.Compare, info.smallest, compactFile.Smallest) > 0 {
							info.smallest = compactFile.Smallest
						}
						first = false
						compactionFiles[level] = append(compactionFiles[level], compactFile)
					}
				}
				for i, cl := range info.inputs {
					files := compactionFiles[cl.level]
					info.inputs[i].files = manifest.NewLevelSliceSeqSorted(files)
					// Mark as intra-L0 compacting if the compaction is
					// L0 -> L0.
					if info.outputLevel == 0 {
						for _, f := range files {
							f.IsIntraL0Compacting = true
						}
					}
				}
				inProgressCompactions = append(inProgressCompactions, info)
			}

			version := newVersion(opts, fileMetas)
			version.L0Sublevels.InitCompactingFileInfo(inProgressL0Compactions(inProgressCompactions))
			vs := &versionSet{
				opts: opts,
				cmp:  DefaultComparer,
			}
			vs.versions.Init(nil)
			vs.append(version)
			picker = &compactionPickerByScore{
				opts:      opts,
				vers:      version,
				baseLevel: baseLevel,
			}
			vs.picker = picker
			picker.initLevelMaxBytes(inProgressCompactions)

			var buf bytes.Buffer
			fmt.Fprint(&buf, version.String())
			if len(inProgressCompactions) > 0 {
				fmt.Fprintln(&buf, "compactions")
				for _, c := range inProgressCompactions {
					fmt.Fprintf(&buf, "  %s\n", c.String())
				}
			}
			return buf.String()
		case "pick-auto":
			td.MaybeScanArgs(t, "l0_compaction_threshold", &opts.L0CompactionThreshold)
			td.MaybeScanArgs(t, "l0_compaction_file_threshold", &opts.L0CompactionFileThreshold)

			pc = picker.pickAuto(compactionEnv{
				diskAvailBytes:          math.MaxUint64,
				earliestUnflushedSeqNum: math.MaxUint64,
				inProgressCompactions:   inProgressCompactions,
			})
			var result strings.Builder
			if pc != nil {
				checkClone(t, pc)
				c := newCompaction(pc, opts, time.Now(), nil /* provider */)
				fmt.Fprintf(&result, "L%d -> L%d\n", pc.startLevel.level, pc.outputLevel.level)
				fmt.Fprintf(&result, "L%d: %s\n", pc.startLevel.level, fileNums(pc.startLevel.files))
				if !pc.outputLevel.files.Empty() {
					fmt.Fprintf(&result, "L%d: %s\n", pc.outputLevel.level, fileNums(pc.outputLevel.files))
				}
				if !c.grandparents.Empty() {
					fmt.Fprintf(&result, "grandparents: %s\n", fileNums(c.grandparents))
				}
			} else {
				return "nil"
			}
			return result.String()
		case "mark-for-compaction":
			var fileNum uint64
			td.ScanArgs(t, "file", &fileNum)
			for l, lm := range picker.vers.Levels {
				iter := lm.Iter()
				for f := iter.First(); f != nil; f = iter.Next() {
					if f.FileNum != base.FileNum(fileNum) {
						continue
					}
					f.MarkedForCompaction = true
					picker.vers.Stats.MarkedForCompaction++
					picker.vers.Levels[l].InvalidateAnnotation(markedForCompactionAnnotator{})
					return fmt.Sprintf("marked L%d.%s", l, f.FileNum)
				}
			}
			return "not-found"
		case "max-output-file-size":
			if pc == nil {
				return "no compaction"
			}
			return fmt.Sprintf("%d", pc.maxOutputFileSize)
		case "max-overlap-bytes":
			if pc == nil {
				return "no compaction"
			}
			return fmt.Sprintf("%d", pc.maxOverlapBytes)
		}
		return fmt.Sprintf("unrecognized command: %s", td.Cmd)
	})
}

func TestCompactionPickerConcurrency(t *testing.T) {
	opts := (*Options)(nil).EnsureDefaults()
	opts.Experimental.L0CompactionConcurrency = 1

	parseMeta := func(s string) (*fileMetadata, error) {
		parts := strings.Split(s, ":")
		fileNum, err := strconv.Atoi(parts[0])
		if err != nil {
			return nil, err
		}
		fields := strings.Fields(parts[1])
		parts = strings.Split(fields[0], "-")
		if len(parts) != 2 {
			return nil, errors.Errorf("malformed table spec: %s", s)
		}
		m := (&fileMetadata{
			FileNum: base.FileNum(fileNum),
			Size:    1028,
		}).ExtendPointKeyBounds(
			opts.Comparer.Compare,
			base.ParseInternalKey(strings.TrimSpace(parts[0])),
			base.ParseInternalKey(strings.TrimSpace(parts[1])),
		)
		m.InitPhysicalBacking()
		for _, p := range fields[1:] {
			if strings.HasPrefix(p, "size=") {
				v, err := strconv.Atoi(strings.TrimPrefix(p, "size="))
				if err != nil {
					return nil, err
				}
				m.Size = uint64(v)
			}
		}
		m.SmallestSeqNum = m.Smallest.SeqNum()
		m.LargestSeqNum = m.Largest.SeqNum()
		m.LargestSeqNumAbsolute = m.Largest.SeqNum()
		return m, nil
	}

	var picker *compactionPickerByScore
	var inProgressCompactions []compactionInfo

	datadriven.RunTest(t, "testdata/compaction_picker_concurrency", func(t *testing.T, td *datadriven.TestData) string {
		switch td.Cmd {
		case "define":
			fileMetas := [manifest.NumLevels][]*fileMetadata{}
			level := 0
			var err error
			lines := strings.Split(td.Input, "\n")
			var compactionLines []string

			for len(lines) > 0 {
				data := strings.TrimSpace(lines[0])
				lines = lines[1:]
				switch data {
				case "L0", "L1", "L2", "L3", "L4", "L5", "L6":
					level, err = strconv.Atoi(data[1:])
					if err != nil {
						return err.Error()
					}
				case "compactions":
					compactionLines, lines = lines, nil
				default:
					meta, err := parseMeta(data)
					if err != nil {
						return err.Error()
					}
					fileMetas[level] = append(fileMetas[level], meta)
				}
			}

			// Parse in-progress compactions in the form of:
			//   L0 000001 -> L2 000005
			inProgressCompactions = nil
			for len(compactionLines) > 0 {
				parts := strings.Fields(compactionLines[0])
				compactionLines = compactionLines[1:]

				var level int
				var info compactionInfo
				first := true
				compactionFiles := map[int][]*fileMetadata{}
				for _, p := range parts {
					switch p {
					case "L0", "L1", "L2", "L3", "L4", "L5", "L6":
						var err error
						level, err = strconv.Atoi(p[1:])
						if err != nil {
							return err.Error()
						}
						if len(info.inputs) > 0 && info.inputs[len(info.inputs)-1].level == level {
							// eg, L0 -> L0 compaction or L6 -> L6 compaction
							continue
						}
						if info.outputLevel < level {
							info.outputLevel = level
						}
						info.inputs = append(info.inputs, compactionLevel{level: level})
					case "->":
						continue
					default:
						fileNum, err := strconv.Atoi(p)
						if err != nil {
							return err.Error()
						}
						var compactFile *fileMetadata
						for _, m := range fileMetas[level] {
							if m.FileNum == FileNum(fileNum) {
								compactFile = m
							}
						}
						if compactFile == nil {
							return fmt.Sprintf("cannot find compaction file %s", FileNum(fileNum))
						}
						compactFile.CompactionState = manifest.CompactionStateCompacting
						if first || base.InternalCompare(DefaultComparer.Compare, info.largest, compactFile.Largest) < 0 {
							info.largest = compactFile.Largest
						}
						if first || base.InternalCompare(DefaultComparer.Compare, info.smallest, compactFile.Smallest) > 0 {
							info.smallest = compactFile.Smallest
						}
						first = false
						compactionFiles[level] = append(compactionFiles[level], compactFile)
					}
				}
				for i, cl := range info.inputs {
					files := compactionFiles[cl.level]
					if cl.level == 0 {
						info.inputs[i].files = manifest.NewLevelSliceSeqSorted(files)
					} else {
						info.inputs[i].files = manifest.NewLevelSliceKeySorted(DefaultComparer.Compare, files)
					}
					// Mark as intra-L0 compacting if the compaction is
					// L0 -> L0.
					if info.outputLevel == 0 {
						for _, f := range files {
							f.IsIntraL0Compacting = true
						}
					}
				}
				inProgressCompactions = append(inProgressCompactions, info)
			}

			version := newVersion(opts, fileMetas)
			version.L0Sublevels.InitCompactingFileInfo(inProgressL0Compactions(inProgressCompactions))
			vs := &versionSet{
				opts: opts,
				cmp:  DefaultComparer,
			}
			vs.versions.Init(nil)
			vs.append(version)

			vb := manifest.MakeVirtualBackings()
			picker = newCompactionPickerByScore(version, &vb, opts, inProgressCompactions)
			vs.picker = picker

			var buf bytes.Buffer
			fmt.Fprint(&buf, version.String())
			if len(inProgressCompactions) > 0 {
				fmt.Fprintln(&buf, "compactions")
				for _, c := range inProgressCompactions {
					fmt.Fprintf(&buf, "  %s\n", c.String())
				}
			}
			return buf.String()

		case "pick-auto":
			td.MaybeScanArgs(t, "l0_compaction_threshold", &opts.L0CompactionThreshold)
			td.MaybeScanArgs(t, "l0_compaction_concurrency", &opts.Experimental.L0CompactionConcurrency)
			td.MaybeScanArgs(t, "compaction_debt_concurrency", &opts.Experimental.CompactionDebtConcurrency)

			pc := picker.pickAuto(compactionEnv{
				earliestUnflushedSeqNum: math.MaxUint64,
				inProgressCompactions:   inProgressCompactions,
			})
			var result strings.Builder
			if pc != nil {
				c := newCompaction(pc, opts, time.Now(), nil /* provider */)
				fmt.Fprintf(&result, "L%d -> L%d\n", pc.startLevel.level, pc.outputLevel.level)
				fmt.Fprintf(&result, "L%d: %s\n", pc.startLevel.level, fileNums(pc.startLevel.files))
				if !pc.outputLevel.files.Empty() {
					fmt.Fprintf(&result, "L%d: %s\n", pc.outputLevel.level, fileNums(pc.outputLevel.files))
				}
				if !c.grandparents.Empty() {
					fmt.Fprintf(&result, "grandparents: %s\n", fileNums(c.grandparents))
				}
			} else {
				return "nil"
			}
			return result.String()
		}
		return fmt.Sprintf("unrecognized command: %s", td.Cmd)
	})
}

func TestCompactionPickerPickReadTriggered(t *testing.T) {
	opts := (*Options)(nil).EnsureDefaults()
	var picker *compactionPickerByScore
	var rcList readCompactionQueue
	var vers *version

	parseMeta := func(s string) (*fileMetadata, error) {
		parts := strings.Split(s, ":")
		fileNum, err := strconv.Atoi(parts[0])
		if err != nil {
			return nil, err
		}
		fields := strings.Fields(parts[1])
		parts = strings.Split(fields[0], "-")
		if len(parts) != 2 {
			return nil, errors.Errorf("malformed table spec: %s. usage: <file-num>:start.SET.1-end.SET.2", s)
		}
		m := (&fileMetadata{
			FileNum: base.FileNum(fileNum),
			Size:    1028,
		}).ExtendPointKeyBounds(
			opts.Comparer.Compare,
			base.ParseInternalKey(strings.TrimSpace(parts[0])),
			base.ParseInternalKey(strings.TrimSpace(parts[1])),
		)
		m.InitPhysicalBacking()
		for _, p := range fields[1:] {
			if strings.HasPrefix(p, "size=") {
				v, err := strconv.Atoi(strings.TrimPrefix(p, "size="))
				if err != nil {
					return nil, err
				}
				m.Size = uint64(v)
			}
		}
		m.SmallestSeqNum = m.Smallest.SeqNum()
		m.LargestSeqNum = m.Largest.SeqNum()
		m.LargestSeqNumAbsolute = m.Largest.SeqNum()
		return m, nil
	}

	datadriven.RunTest(t, "testdata/compaction_picker_read_triggered", func(t *testing.T, td *datadriven.TestData) string {
		switch td.Cmd {
		case "define":
			rcList = readCompactionQueue{}
			fileMetas := [manifest.NumLevels][]*fileMetadata{}
			level := 0
			var err error
			lines := strings.Split(td.Input, "\n")

			for len(lines) > 0 {
				data := strings.TrimSpace(lines[0])
				lines = lines[1:]
				switch data {
				case "L0", "L1", "L2", "L3", "L4", "L5", "L6":
					level, err = strconv.Atoi(data[1:])
					if err != nil {
						return err.Error()
					}
				default:
					meta, err := parseMeta(data)
					if err != nil {
						return err.Error()
					}
					fileMetas[level] = append(fileMetas[level], meta)
				}
			}

			vers = newVersion(opts, fileMetas)
			vs := &versionSet{
				opts: opts,
				cmp:  DefaultComparer,
			}
			vs.versions.Init(nil)
			vs.append(vers)
			var inProgressCompactions []compactionInfo
			vb := manifest.MakeVirtualBackings()
			picker = newCompactionPickerByScore(vers, &vb, opts, inProgressCompactions)
			vs.picker = picker

			var buf bytes.Buffer
			fmt.Fprint(&buf, vers.String())
			return buf.String()

		case "add-read-compaction":
			for _, line := range strings.Split(td.Input, "\n") {
				if line == "" {
					continue
				}
				parts := strings.Split(line, " ")
				if len(parts) != 3 {
					return "error: malformed data for add-read-compaction. usage: <level>: <start>-<end> <filenum>"
				}
				if l, err := strconv.Atoi(parts[0][:1]); err == nil {
					keys := strings.Split(parts[1], "-")
					fileNum, _ := strconv.Atoi(parts[2])

					rc := readCompaction{
						level:   l,
						start:   []byte(keys[0]),
						end:     []byte(keys[1]),
						fileNum: base.FileNum(fileNum),
					}
					rcList.add(&rc, DefaultComparer.Compare)
				} else {
					return err.Error()
				}
			}
			return ""

		case "show-read-compactions":
			var sb strings.Builder
			if rcList.size == 0 {
				sb.WriteString("(none)")
			}
			for i := 0; i < rcList.size; i++ {
				rc := rcList.at(i)
				sb.WriteString(fmt.Sprintf("(level: %d, start: %s, end: %s)\n", rc.level, string(rc.start), string(rc.end)))
			}
			return sb.String()

		case "pick-auto":
			pc := picker.pickAuto(compactionEnv{
				earliestUnflushedSeqNum: math.MaxUint64,
				readCompactionEnv: readCompactionEnv{
					readCompactions: &rcList,
					flushing:        false,
				},
			})
			var result strings.Builder
			if pc != nil {
				fmt.Fprintf(&result, "L%d -> L%d\n", pc.startLevel.level, pc.outputLevel.level)
				fmt.Fprintf(&result, "L%d: %s\n", pc.startLevel.level, fileNums(pc.startLevel.files))
				if !pc.outputLevel.files.Empty() {
					fmt.Fprintf(&result, "L%d: %s\n", pc.outputLevel.level, fileNums(pc.outputLevel.files))
				}
			} else {
				return "nil"
			}
			return result.String()
		}
		return fmt.Sprintf("unrecognized command: %s", td.Cmd)
	})
}

type alwaysMultiLevel struct{}

func (d alwaysMultiLevel) pick(
	pcOrig *pickedCompaction, opts *Options, diskAvailBytes uint64,
) *pickedCompaction {
	pcMulti := pcOrig.clone()
	if !pcMulti.setupMultiLevelCandidate(opts, diskAvailBytes) {
		return pcOrig
	}
	return pcMulti
}

func (d alwaysMultiLevel) allowL0() bool  { return false }
func (d alwaysMultiLevel) String() string { return "always" }

func TestPickedCompactionSetupInputs(t *testing.T) {
	opts := &Options{}
	opts.EnsureDefaults()

	parseMeta := func(s string) *fileMetadata {
		parts := strings.Split(strings.TrimSpace(s), " ")
		var fileSize uint64
		var compacting bool
		for _, part := range parts {
			switch {
			case part == "compacting":
				compacting = true
			case strings.HasPrefix(part, "size="):
				v, err := strconv.ParseUint(strings.TrimPrefix(part, "size="), 10, 64)
				require.NoError(t, err)
				fileSize = v
			}
		}
		tableParts := strings.Split(parts[0], "-")
		if len(tableParts) != 2 {
			t.Fatalf("malformed table spec: %s", s)
		}
		state := manifest.CompactionStateNotCompacting
		if compacting {
			state = manifest.CompactionStateCompacting
		}
		m := (&fileMetadata{
			CompactionState: state,
			Size:            fileSize,
		}).ExtendPointKeyBounds(
			opts.Comparer.Compare,
			base.ParseInternalKey(strings.TrimSpace(tableParts[0])),
			base.ParseInternalKey(strings.TrimSpace(tableParts[1])),
		)
		m.SmallestSeqNum = m.Smallest.SeqNum()
		m.LargestSeqNum = m.Largest.SeqNum()
		if m.SmallestSeqNum > m.LargestSeqNum {
			m.SmallestSeqNum, m.LargestSeqNum = m.LargestSeqNum, m.SmallestSeqNum
		}
		m.LargestSeqNumAbsolute = m.LargestSeqNum
		m.InitPhysicalBacking()
		return m
	}

	setupInputTest := func(t *testing.T, d *datadriven.TestData) string {
		switch d.Cmd {
		case "setup-inputs":
			var availBytes uint64 = math.MaxUint64
			var maxLevelBytes [7]int64
			args := d.CmdArgs

			if len(args) > 0 && args[0].Key == "avail-bytes" {
				require.Equal(t, 1, len(args[0].Vals))
				var err error
				availBytes, err = strconv.ParseUint(args[0].Vals[0], 10, 64)
				require.NoError(t, err)
				args = args[1:]
			}

			if len(args) != 2 {
				return "setup-inputs [avail-bytes=XXX] <start> <end>"
			}

			pc := &pickedCompaction{
				cmp:    DefaultComparer.Compare,
				inputs: []compactionLevel{{level: -1}, {level: -1}},
			}
			pc.startLevel, pc.outputLevel = &pc.inputs[0], &pc.inputs[1]
			var currentLevel int
			var files [numLevels][]*fileMetadata
			fileNum := FileNum(1)

			for _, data := range strings.Split(d.Input, "\n") {
				switch data[:2] {
				case "L0", "L1", "L2", "L3", "L4", "L5", "L6":
					levelArgs := strings.Fields(data)
					level, err := strconv.Atoi(levelArgs[0][1:])
					if err != nil {
						return err.Error()
					}
					currentLevel = level
					if len(levelArgs) > 1 {
						maxSizeArg := strings.Replace(levelArgs[1], "max-size=", "", 1)
						maxSize, err := strconv.ParseInt(maxSizeArg, 10, 64)
						if err != nil {
							return err.Error()
						}
						maxLevelBytes[level] = maxSize
					} else {
						maxLevelBytes[level] = math.MaxInt64
					}
					if pc.startLevel.level == -1 {
						pc.startLevel.level = level

					} else if pc.outputLevel.level == -1 {
						if pc.startLevel.level >= level {
							return fmt.Sprintf("startLevel=%d >= outputLevel=%d\n", pc.startLevel.level, level)
						}
						pc.outputLevel.level = level
					}
				default:
					meta := parseMeta(data)
					meta.FileNum = fileNum
					fileNum++
					files[currentLevel] = append(files[currentLevel], meta)
				}
			}

			for _, levelFiles := range files {
				for i := 1; i < len(levelFiles); i++ {
					if !checkTableBoundary(levelFiles[i-1], levelFiles[i], opts.Comparer.Compare) {
						d.Fatalf(t, "overlapping tables: %s and %s", levelFiles[i-1], levelFiles[i])
					}
				}
			}

			if pc.outputLevel.level == -1 {
				pc.outputLevel.level = pc.startLevel.level + 1
			}
			pc.version = newVersion(opts, files)
			pc.startLevel.files = pc.version.Overlaps(
				pc.startLevel.level,
				base.UserKeyBoundsInclusive([]byte(args[0].String()), []byte(args[1].String())),
			)

			var isCompacting bool
			if !pc.setupInputs(opts, availBytes, pc.startLevel) {
				isCompacting = true
			}
			origPC := pc
			pc = pc.maybeAddLevel(opts, availBytes)
			// If pc points to a new pickedCompaction, a new multi level compaction
			// was initialized.
			initMultiLevel := pc != origPC
			checkClone(t, pc)
			var buf bytes.Buffer
			for _, cl := range pc.inputs {
				if cl.files.Empty() {
					continue
				}

				fmt.Fprintf(&buf, "L%d\n", cl.level)
				cl.files.Each(func(f *fileMetadata) {
					fmt.Fprintf(&buf, "  %s\n", f)
				})
			}
			if isCompacting {
				fmt.Fprintf(&buf, "is-compacting\n")
			}

			if initMultiLevel {
				extraLevel := pc.extraLevels[0].level
				fmt.Fprintf(&buf, "init-multi-level(%d,%d,%d)\n", pc.startLevel.level, extraLevel,
					pc.outputLevel.level)
				fmt.Fprintf(&buf, "Original WriteAmp %.2f; ML WriteAmp %.2f\n", origPC.predictedWriteAmp(), pc.predictedWriteAmp())
				fmt.Fprintf(&buf, "Original OverlappingRatio %.2f; ML OverlappingRatio %.2f\n", origPC.overlappingRatio(), pc.overlappingRatio())
			}
			return buf.String()

		default:
			return fmt.Sprintf("unknown command: %s", d.Cmd)
		}
	}

	t.Logf("Test basic setup inputs behavior without multi level compactions")
	opts.Experimental.MultiLevelCompactionHeuristic = NoMultiLevel{}
	datadriven.RunTest(t, "testdata/compaction_setup_inputs",
		setupInputTest)

	t.Logf("Turning multi level compaction on")
	opts.Experimental.MultiLevelCompactionHeuristic = alwaysMultiLevel{}
	datadriven.RunTest(t, "testdata/compaction_setup_inputs_multilevel_dummy",
		setupInputTest)

	t.Logf("Try Write-Amp Heuristic")
	opts.Experimental.MultiLevelCompactionHeuristic = WriteAmpHeuristic{}
	datadriven.RunTest(t, "testdata/compaction_setup_inputs_multilevel_write_amp",
		setupInputTest)
}

func TestPickedCompactionExpandInputs(t *testing.T) {
	opts := &Options{}
	opts.EnsureDefaults()
	cmp := DefaultComparer.Compare
	var files []*fileMetadata

	parseMeta := func(s string) *fileMetadata {
		parts := strings.Split(s, "-")
		if len(parts) != 2 {
			t.Fatalf("malformed table spec: %s", s)
		}
		m := (&fileMetadata{}).ExtendPointKeyBounds(
			opts.Comparer.Compare,
			base.ParseInternalKey(parts[0]),
			base.ParseInternalKey(parts[1]),
		)
		m.InitPhysicalBacking()
		return m
	}

	datadriven.RunTest(t, "testdata/compaction_expand_inputs",
		func(t *testing.T, d *datadriven.TestData) string {
			switch d.Cmd {
			case "define":
				files = nil
				if len(d.Input) == 0 {
					return ""
				}
				for _, data := range strings.Split(d.Input, "\n") {
					meta := parseMeta(data)
					meta.FileNum = FileNum(len(files))
					files = append(files, meta)
				}
				manifest.SortBySmallest(files, cmp)
				// Verify that the tables have no user key overlap.
				for i := 1; i < len(files); i++ {
					if !checkTableBoundary(files[i-1], files[i], cmp) {
						d.Fatalf(t, "overlapping tables: %s and %s", files[i-1], files[i])
					}
				}
				return ""

			case "expand-inputs":
				pc := &pickedCompaction{
					cmp:    cmp,
					inputs: []compactionLevel{{level: 1}},
				}
				pc.startLevel = &pc.inputs[0]

				var filesLevelled [numLevels][]*fileMetadata
				filesLevelled[pc.startLevel.level] = files
				pc.version = newVersion(opts, filesLevelled)

				if len(d.CmdArgs) != 1 {
					return fmt.Sprintf("%s expects 1 argument", d.Cmd)
				}
				index, err := strconv.ParseInt(d.CmdArgs[0].String(), 10, 64)
				if err != nil {
					return err.Error()
				}

				// Advance the iterator to position `index`.
				iter := pc.version.Levels[pc.startLevel.level].Iter()
				_ = iter.First()
				for i := int64(0); i < index; i++ {
					_ = iter.Next()
				}

				var buf bytes.Buffer
				iter.Take().Slice().Each(func(f *fileMetadata) {
					fmt.Fprintf(&buf, "%d: %s-%s\n", f.FileNum, f.Smallest, f.Largest)
				})
				return buf.String()

			default:
				return fmt.Sprintf("unknown command: %s", d.Cmd)
			}
		})
}

func TestCompactionOutputFileSize(t *testing.T) {
	opts := (*Options)(nil).EnsureDefaults()
	var picker *compactionPickerByScore
	var vers *version

	parseMeta := func(s string) (*fileMetadata, error) {
		parts := strings.Split(s, ":")
		fileNum, err := strconv.Atoi(parts[0])
		if err != nil {
			return nil, err
		}
		fields := strings.Fields(parts[1])
		parts = strings.Split(fields[0], "-")
		if len(parts) != 2 {
			return nil, errors.Errorf("malformed table spec: %s. usage: <file-num>:start.SET.1-end.SET.2", s)
		}
		m := (&fileMetadata{
			FileNum: base.FileNum(fileNum),
			Size:    1028,
		}).ExtendPointKeyBounds(
			opts.Comparer.Compare,
			base.ParseInternalKey(strings.TrimSpace(parts[0])),
			base.ParseInternalKey(strings.TrimSpace(parts[1])),
		)
		m.InitPhysicalBacking()
		for _, p := range fields[1:] {
			if strings.HasPrefix(p, "size=") {
				v, err := strconv.Atoi(strings.TrimPrefix(p, "size="))
				if err != nil {
					return nil, err
				}
				m.Size = uint64(v)
			}
			if strings.HasPrefix(p, "range-deletions-bytes-estimate=") {
				v, err := strconv.Atoi(strings.TrimPrefix(p, "range-deletions-bytes-estimate="))
				if err != nil {
					return nil, err
				}
				m.Stats.RangeDeletionsBytesEstimate = uint64(v)
				m.Stats.NumDeletions = 1 // At least one range del responsible for the deletion bytes.
				m.StatsMarkValid()
			}
		}
		m.SmallestSeqNum = m.Smallest.SeqNum()
		m.LargestSeqNum = m.Largest.SeqNum()
		m.LargestSeqNumAbsolute = m.LargestSeqNum
		return m, nil
	}

	datadriven.RunTest(t, "testdata/compaction_output_file_size", func(t *testing.T, td *datadriven.TestData) string {
		switch td.Cmd {
		case "define":
			fileMetas := [manifest.NumLevels][]*fileMetadata{}
			level := 0
			var err error
			lines := strings.Split(td.Input, "\n")

			for len(lines) > 0 {
				data := strings.TrimSpace(lines[0])
				lines = lines[1:]
				switch data {
				case "L0", "L1", "L2", "L3", "L4", "L5", "L6":
					level, err = strconv.Atoi(data[1:])
					if err != nil {
						return err.Error()
					}
				default:
					meta, err := parseMeta(data)
					if err != nil {
						return err.Error()
					}
					fileMetas[level] = append(fileMetas[level], meta)
				}
			}

			vers = newVersion(opts, fileMetas)
			vs := &versionSet{
				opts: opts,
				cmp:  DefaultComparer,
			}
			vs.versions.Init(nil)
			vs.append(vers)
			var inProgressCompactions []compactionInfo
			vb := manifest.MakeVirtualBackings()
			picker = newCompactionPickerByScore(vers, &vb, opts, inProgressCompactions)
			vs.picker = picker

			var buf bytes.Buffer
			fmt.Fprint(&buf, vers.String())
			return buf.String()

		case "pick-auto":
			pc := picker.pickAuto(compactionEnv{
				earliestUnflushedSeqNum: math.MaxUint64,
				earliestSnapshotSeqNum:  math.MaxUint64,
			})
			var buf bytes.Buffer
			if pc != nil {
				fmt.Fprintf(&buf, "L%d -> L%d\n", pc.startLevel.level, pc.outputLevel.level)
				fmt.Fprintf(&buf, "L%d: %s\n", pc.startLevel.level, fileNums(pc.startLevel.files))
				fmt.Fprintf(&buf, "maxOutputFileSize: %d\n", pc.maxOutputFileSize)
			} else {
				return "nil"
			}
			return buf.String()

		default:
			return fmt.Sprintf("unrecognized command: %s", td.Cmd)
		}
	})
}

func TestCompactionPickerCompensatedSize(t *testing.T) {
	testCases := []struct {
		size                  uint64
		pointDelEstimateBytes uint64
		rangeDelEstimateBytes uint64
		wantBytes             uint64
	}{
		{
			size:                  100,
			pointDelEstimateBytes: 0,
			rangeDelEstimateBytes: 0,
			wantBytes:             100,
		},
		{
			size:                  100,
			pointDelEstimateBytes: 10,
			rangeDelEstimateBytes: 0,
			wantBytes:             100 + 10,
		},
		{
			size:                  100,
			pointDelEstimateBytes: 10,
			rangeDelEstimateBytes: 5,
			wantBytes:             100 + 10 + 5,
		},
	}

	for _, tc := range testCases {
		t.Run("", func(t *testing.T) {
			f := &fileMetadata{Size: tc.size}
			f.InitPhysicalBacking()
			f.Stats.PointDeletionsBytesEstimate = tc.pointDelEstimateBytes
			f.Stats.RangeDeletionsBytesEstimate = tc.rangeDelEstimateBytes
			gotBytes := compensatedSize(f)
			require.Equal(t, tc.wantBytes, gotBytes)
		})
	}
}

func TestCompactionPickerPickFile(t *testing.T) {
	fs := vfs.NewMem()
	opts := &Options{
		Comparer:           testkeys.Comparer,
		FormatMajorVersion: FormatNewest,
		FS:                 fs,
	}

	d, err := Open("", opts)
	require.NoError(t, err)
	defer func() {
		if d != nil {
			require.NoError(t, d.Close())
		}
	}()

	datadriven.RunTest(t, "testdata/compaction_picker_pick_file", func(t *testing.T, td *datadriven.TestData) string {
		switch td.Cmd {
		case "define":
			require.NoError(t, d.Close())

			d, err = runDBDefineCmd(td, opts)
			if err != nil {
				return err.Error()
			}
			d.mu.Lock()
			s := d.mu.versions.currentVersion().String()
			d.mu.Unlock()
			return s

		case "file-sizes":
			return runTableFileSizesCmd(td, d)

		case "build":
			if err := runBuildCmd(td, d, d.opts.FS); err != nil {
				return err.Error()
			}
			return ""

		case "ingest-and-excise":
			if err := runIngestAndExciseCmd(td, d, d.opts.FS); err != nil {
				return err.Error()
			}
			return ""

		case "lsm":
			d.mu.Lock()
			s := d.mu.versions.currentVersion().String()
			d.mu.Unlock()
			return s

		case "pick-file":
			s := strings.TrimPrefix(td.CmdArgs[0].String(), "L")
			level, err := strconv.Atoi(s)
			if err != nil {
				return fmt.Sprintf("unable to parse arg %q as level", td.CmdArgs[0].String())
			}
			if level == 0 {
				panic("L0 picking unimplemented")
			}
			d.mu.Lock()
			defer d.mu.Unlock()

			// Use maybeScheduleCompactionPicker to take care of all of the
			// initialization of the compaction-picking environment, but never
			// pick a compaction; just call pickFile using the user-provided
			// level.
			var lf manifest.LevelFile
			var ok bool
			d.maybeScheduleCompactionPicker(func(untypedPicker compactionPicker, env compactionEnv) *pickedCompaction {
				p := untypedPicker.(*compactionPickerByScore)
				lf, ok = pickCompactionSeedFile(p.vers, p.virtualBackings, opts, level, level+1, env.earliestSnapshotSeqNum)
				return nil
			})
			if !ok {
				return "(none)"
			}
			return lf.FileMetadata.String()

		default:
			return fmt.Sprintf("unknown command: %s", td.Cmd)
		}
	})
}

type pausableCleaner struct {
	mu      sync.Mutex
	cond    sync.Cond
	paused  bool
	cleaner Cleaner
}

func (c *pausableCleaner) Clean(fs vfs.FS, fileType base.FileType, path string) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	for c.paused {
		c.cond.Wait()
	}
	return c.cleaner.Clean(fs, fileType, path)
}

func (c *pausableCleaner) pause() {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.paused = true
}

func (c *pausableCleaner) resume() {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.paused = false
	c.cond.Broadcast()
}

func TestCompactionPickerScores(t *testing.T) {
	fs := vfs.NewMem()
	cleaner := pausableCleaner{cleaner: DeleteCleaner{}}
	cleaner.cond.L = &cleaner.mu
	opts := &Options{
		Cleaner:                     &cleaner,
		Comparer:                    testkeys.Comparer,
		DisableAutomaticCompactions: true,
		FormatMajorVersion:          FormatNewest,
		FS:                          fs,
	}

	d, err := Open("", opts)
	require.NoError(t, err)
	defer func() {
		if d != nil {
			cleaner.resume()
			require.NoError(t, closeAllSnapshots(d))
			require.NoError(t, d.Close())
		}
	}()

	var buf bytes.Buffer
	datadriven.RunTest(t, "testdata/compaction_picker_scores", func(t *testing.T, td *datadriven.TestData) string {
		switch td.Cmd {
		case "define":
			require.NoError(t, closeAllSnapshots(d))
			require.NoError(t, d.Close())

			if td.HasArg("pause-cleaning") {
				cleaner.pause()
			}

			d, err = runDBDefineCmd(td, opts)
			if err != nil {
				return err.Error()
			}
			d.mu.Lock()
			s := d.mu.versions.currentVersion().String()
			d.mu.Unlock()
			return s

		case "disable-table-stats":
			d.mu.Lock()
			d.opts.DisableTableStats = true
			d.mu.Unlock()
			return ""

		case "enable-table-stats":
			d.mu.Lock()
			d.opts.DisableTableStats = false
			d.maybeCollectTableStatsLocked()
			d.mu.Unlock()
			return ""

		case "resume-cleaning":
			cleaner.resume()
			return ""

		case "ingest":
			if err = runBuildCmd(td, d, d.opts.FS); err != nil {
				return err.Error()
			}
			if err = runIngestCmd(td, d, d.opts.FS); err != nil {
				return err.Error()
			}
			d.mu.Lock()
			s := d.mu.versions.currentVersion().String()
			d.mu.Unlock()
			return s

		case "lsm":
			return runLSMCmd(td, d)

		case "maybe-compact":
			buf.Reset()
			d.mu.Lock()
			d.opts.DisableAutomaticCompactions = false
			d.maybeScheduleCompaction()
			fmt.Fprintf(&buf, "%d compactions in progress:", d.mu.compact.compactingCount)
			for c := range d.mu.compact.inProgress {
				fmt.Fprintf(&buf, "\n%s", c)
			}
			d.opts.DisableAutomaticCompactions = true
			d.mu.Unlock()
			return buf.String()

		case "scores":
			waitFor := "completion"
			td.MaybeScanArgs(t, "wait-for-compaction", &waitFor)

			// Wait for any running compactions to complete before calculating
			// scores. Otherwise, the output of this command is
			// nondeterministic.
			switch waitFor {
			case "completion":
				d.mu.Lock()
				for d.mu.compact.compactingCount > 0 {
					d.mu.compact.cond.Wait()
				}
				d.mu.Unlock()
			case "version-edit":
				func() {
					for {
						d.mu.Lock()
						wait := len(d.mu.compact.inProgress) > 0
						for c := range d.mu.compact.inProgress {
							wait = wait && !c.versionEditApplied
						}
						d.mu.Unlock()
						if !wait {
							return
						}
						// d.mu.compact.cond isn't notified until the compaction
						// is removed from inProgress, so we need to just sleep
						// and check again soon.
						time.Sleep(10 * time.Millisecond)
					}
				}()
			default:
				panic(fmt.Sprintf("unrecognized `wait-for-compaction` value: %q", waitFor))
			}

			buf.Reset()
			fmt.Fprintf(&buf, "L       Size   Score\n")
			for l, lm := range d.Metrics().Levels {
				if l < numLevels-1 {
					fmt.Fprintf(&buf, "L%-3d\t%-7s%.1f\n", l, humanize.Bytes.Int64(lm.Size), lm.Score)
				} else {
					fmt.Fprintf(&buf, "L%-3d\t%-7s-\n", l, humanize.Bytes.Int64(lm.Size))
				}
			}
			return buf.String()

		case "wait-pending-table-stats":
			return runTableStatsCmd(td, d)

		default:
			return fmt.Sprintf("unknown command: %s", td.Cmd)
		}
	})
}

func fileNums(files manifest.LevelSlice) string {
	var ss []string
	files.Each(func(f *fileMetadata) {
		ss = append(ss, f.FileNum.String())
	})
	sort.Strings(ss)
	return strings.Join(ss, ",")
}

func checkClone(t *testing.T, pc *pickedCompaction) {
	pcClone := pc.clone()
	require.Equal(t, pc.String(), pcClone.String())

	// ensure all input files are in new address
	for i := range pc.inputs {
		// Len could be zero if setup inputs rejected a level
		if pc.inputs[i].files.Len() > 0 {
			require.NotEqual(t, &pc.inputs[i], &pcClone.inputs[i])
		}
	}
	for i := range pc.startLevel.l0SublevelInfo {
		if pc.startLevel.l0SublevelInfo[i].Len() > 0 {
			require.NotEqual(t, &pc.startLevel.l0SublevelInfo[i], &pcClone.startLevel.l0SublevelInfo[i])
		}
	}
}

func checkTableBoundary(a, b *fileMetadata, cmp base.Compare) (ok bool) {
	c := cmp(a.LargestPointKey.UserKey, b.SmallestPointKey.UserKey)
	return c < 0 || (c == 0 && a.LargestPointKey.IsExclusiveSentinel())
}