var sizes [numLevels]int64

opts := &Options{}

opts.EnsureDefaults()

if len(d.CmdArgs) != 1 {

return nil, nil, sizes, 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, sizes, err.Error()

}

var files [numLevels][]*fileMetadata

if len(d.Input) > 0 {

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, sizes, fmt.Sprintf("malformed test:\n%s", d.Input)

}

level, err := strconv.Atoi(parts[0])

if err != nil {

return nil, nil, sizes, err.Error()

}

if files[level] != nil {

return nil, nil, sizes, fmt.Sprintf("level %d already filled", level)

}

size, err := strconv.ParseUint(strings.TrimSpace(parts[1]), 10, 64)

if err != nil {

return nil, nil, sizes, err.Error()

}

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{

SmallestSeqNum: key.SeqNum(),

LargestSeqNum: key.SeqNum(),

Size: 1,

}).ExtendPointKeyBounds(opts.Comparer.Compare, key, key)

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

}

files[level] = append(files[level], m)

sizes[level] += int64(m.Size)

}

}

}

vers := newVersion(opts, files)

return vers, opts, sizes, ""

datadriven.RunTest(t, "testdata/compaction_picker_level_max_bytes",

func(d *datadriven.TestData) string {

switch d.Cmd {

case "init":

vers, opts, sizes, errMsg := loadVersion(d)

if errMsg != "" {

return errMsg

}

p, ok := newCompactionPicker(vers, opts, nil, sizes, diskAvailBytesInf).(*compactionPickerByScore)

require.True(t, ok)

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)

}

})

var vers *version

var opts *Options

var sizes [numLevels]int64

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(d *datadriven.TestData) string {

switch d.Cmd {

case "init":

var errMsg string

vers, opts, sizes, errMsg = loadVersion(d)

if errMsg != "" {

return errMsg

}

return ""

case "init_cp":

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()

}

}

p := newCompactionPicker(vers, opts, inProgress, sizes, diskAvailBytesInf)

var ok bool

pickerByScore, ok = p.(*compactionPickerByScore)

require.True(t, ok)

return fmt.Sprintf("base: %d", pickerByScore.baseLevel)

case "queue":

var b strings.Builder

var inProgress []compactionInfo

for {

env := compactionEnv{

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,

})

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

})

}

}

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]

iter := vers.Levels[c.inputs[0].level].Iter()

for f := iter.First(); f != nil; f = iter.Next() {

if !f.IsCompacting() {

f.CompactionState = manifest.CompactionStateCompacting

c.inputs[0].files = iter.Take().Slice()

break

}

}

switch {

case 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

})

}

case c.inputs[0].level != c.outputLevel:

// Ln->Ln+1: mark 1 file in Ln+1 as compacting.

iter := vers.Levels[c.outputLevel].Iter()

for f := iter.First(); f != nil; f = iter.Next() {

if !f.IsCompacting() {

f.CompactionState = manifest.CompactionStateCompacting

c.inputs[1].files = iter.Take().Slice()

break

}

}

}

}

pc := pickerByScore.pickAuto(compactionEnv{

earliestUnflushedSeqNum: InternalKeySeqNumMax,

inProgressCompactions: inProgress,

})

if pc == nil {

return "no compaction"

}

return fmt.Sprintf("L%d->L%d: %0.1f", pc.startLevel.level, pc.outputLevel.level, pc.score)

case "pick_manual":

startLevel := 0

start := ""

end := ""

if len(d.CmdArgs) > 0 {

for _, arg := range d.CmdArgs {

switch arg.Key {

case "level":

startLevel64, err := strconv.ParseInt(arg.Vals[0], 10, 64)

if err != nil {

return err.Error()

}

startLevel = int(startLevel64)

case "start":

start = arg.Vals[0]

case "end":

end = arg.Vals[0]

default:

return "unknown arg: " + arg.Key

}

}

}

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 := pickerByScore.pickManual(compactionEnv{

earliestUnflushedSeqNum: InternalKeySeqNumMax,

}, 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)

}

})

datadriven.RunTest(t, "testdata/compaction_picker_estimated_debt",

func(d *datadriven.TestData) string {

switch d.Cmd {

case "init":

vers, opts, sizes, errMsg := loadVersion(d)

if errMsg != "" {

return errMsg

}

opts.MemTableSize = 1000

p := newCompactionPicker(vers, opts, nil, sizes, diskAvailBytesInf)

return fmt.Sprintf("%d\n", p.estimatedCompactionDebt(0))

default:

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

}

})

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()

return m, nil

}

var picker *compactionPickerByScore

var inProgressCompactions []compactionInfo

var pc *pickedCompaction

datadriven.RunTest(t, "testdata/compaction_picker_L0", func(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.Compare,

cmpName: DefaultComparer.Name,

}

vs.versions.Init(nil)

vs.append(version)

picker = &compactionPickerByScore{

opts: opts,

vers: version,

baseLevel: baseLevel,

diskAvailBytes: diskAvailBytesInf,

}

vs.picker = picker

for l := 0; l < len(picker.levelSizes); l++ {

version.Levels[l].Slice().Each(func(m *fileMetadata) {

picker.levelSizes[l] += int64(m.Size)

})

}

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":

for _, arg := range td.CmdArgs {

var err error

switch arg.Key {

case "l0_compaction_threshold":

opts.L0CompactionThreshold, err = strconv.Atoi(arg.Vals[0])

if err != nil {

return err.Error()

}

case "l0_compaction_file_threshold":

opts.L0CompactionFileThreshold, err = strconv.Atoi(arg.Vals[0])

if err != nil {

return err.Error()

}

}

}

pc = picker.pickAuto(compactionEnv{

earliestUnflushedSeqNum: math.MaxUint64,

inProgressCompactions: inProgressCompactions,

})

var result strings.Builder

if pc != nil {

c := newCompaction(pc, opts)

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)

})

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])),

)

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()

return m, nil

}

var picker *compactionPickerByScore

var inProgressCompactions []compactionInfo

datadriven.RunTest(t, "testdata/compaction_picker_concurrency", func(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.Compare,

cmpName: DefaultComparer.Name,

}

vs.versions.Init(nil)

vs.append(version)

var sizes [numLevels]int64

for l := 0; l < len(sizes); l++ {

version.Levels[l].Slice().Each(func(m *fileMetadata) {

sizes[l] += int64(m.Size)

})

}

picker = newCompactionPicker(version, opts, inProgressCompactions, sizes, diskAvailBytesInf).(*compactionPickerByScore)

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":

for _, arg := range td.CmdArgs {

var err error

switch arg.Key {

case "l0_compaction_threshold":

opts.L0CompactionThreshold, err = strconv.Atoi(arg.Vals[0])

if err != nil {

return err.Error()

}

case "l0_compaction_concurrency":

opts.Experimental.L0CompactionConcurrency, err = strconv.Atoi(arg.Vals[0])

if err != nil {

return err.Error()

}

case "compaction_debt_concurrency":

opts.Experimental.CompactionDebtConcurrency, err = strconv.Atoi(arg.Vals[0])

if err != nil {

return err.Error()

}

}

}

pc := picker.pickAuto(compactionEnv{

earliestUnflushedSeqNum: math.MaxUint64,

inProgressCompactions: inProgressCompactions,

})

var result strings.Builder

if pc != nil {

c := newCompaction(pc, opts)

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)

})

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])),

)

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()

return m, nil

}

datadriven.RunTest(t, "testdata/compaction_picker_read_triggered", func(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.Compare,

cmpName: DefaultComparer.Name,

}

vs.versions.Init(nil)

vs.append(vers)

var sizes [numLevels]int64

for l := 0; l < len(sizes); l++ {

vers.Levels[l].Slice().Each(func(m *fileMetadata) {

sizes[l] += int64(m.Size)

})

}

var inProgressCompactions []compactionInfo

picker = newCompactionPicker(vers, opts, inProgressCompactions, sizes, diskAvailBytesInf).(*compactionPickerByScore)

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)

})

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()

return m

}

datadriven.RunTest(t, "testdata/compaction_setup_inputs",

func(d *datadriven.TestData) string {

switch d.Cmd {

case "setup-inputs":

var availBytes uint64 = math.MaxUint64

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