From b5c5df0302e39c9d933c9c100cf62043beaf2472 Mon Sep 17 00:00:00 2001
From: Bilal Akhtar <bilal@cockroachlabs.com>
Date: Mon, 28 Sep 2020 15:51:44 -0400
Subject: [PATCH] *: Add a simple test for forced compactions

This change adds a test to test compaction picking logic where
MarkedForCompaction = true. Previously, this boolean was completely
untested, resulting in panics when Pebble was used on a previously-rocksdb
store directory, as RocksDB often set that boolean to true in the manifest.

Fixes #868.
---
 compaction_picker_test.go               | 25 ++++++++----
 testdata/compaction_picker_target_level | 52 +++++++++++++++++++++++++
 2 files changed, 70 insertions(+), 7 deletions(-)

diff --git a/compaction_picker_test.go b/compaction_picker_test.go
index ababd4be1d..1a3026d8b9 100644
--- a/compaction_picker_test.go
+++ b/compaction_picker_test.go
@@ -37,8 +37,9 @@ func loadVersion(d *datadriven.TestData) (*version, *Options, [numLevels]int64,
 	var files [numLevels][]*fileMetadata
 	if len(d.Input) > 0 {
 		for _, data := range strings.Split(d.Input, "\n") {
-			parts := strings.Split(data, ":")
-			if len(parts) != 2 {
+			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])
@@ -49,17 +50,27 @@ func loadVersion(d *datadriven.TestData) (*version, *Options, [numLevels]int64,
 				return nil, nil, sizes, fmt.Sprintf("level %d already filled", level)
 			}
 			size, err := strconv.ParseUint(strings.TrimSpace(parts[1]), 10, 64)
+			markedForCompaction := false
+			if len(parts) > 2 {
+				for _, field := range parts[2:] {
+					switch field {
+					case "marked_for_compaction":
+						markedForCompaction = true
+					}
+				}
+			}
 			if err != nil {
 				return nil, nil, sizes, err.Error()
 			}
 			for i := uint64(1); sizes[level] < int64(size); i++ {
 				key := base.MakeInternalKey([]byte(fmt.Sprintf("%04d", i)), i, InternalKeyKindSet)
 				m := &fileMetadata{
-					Smallest:       key,
-					Largest:        key,
-					SmallestSeqNum: key.SeqNum(),
-					LargestSeqNum:  key.SeqNum(),
-					Size:           1,
+					Smallest:            key,
+					Largest:             key,
+					SmallestSeqNum:      key.SeqNum(),
+					LargestSeqNum:       key.SeqNum(),
+					Size:                1,
+					MarkedForCompaction: markedForCompaction,
 				}
 				if size >= 100 {
 					// If the requested size of the level is very large only add a single
diff --git a/testdata/compaction_picker_target_level b/testdata/compaction_picker_target_level
index 60e5eb8c01..a3b5b57cb8 100644
--- a/testdata/compaction_picker_target_level
+++ b/testdata/compaction_picker_target_level
@@ -342,3 +342,55 @@ base: 5
 pick_manual level=5 start=0 end=12
 ----
 nil, retryLater = false
+
+# Ensure that files marked for compaction in the manifest are lower priority
+# than score-based compactions.
+
+init 5
+0: 20
+5: 1   marked_for_compaction
+6: 10
+----
+
+init_cp
+----
+base: 5
+
+queue
+----
+L0->L5: 25.0
+L0->L0: 11.9
+
+pick ongoing=(0,5)
+----
+L0->L0: 11.9
+
+# Ensure that files marked for compaction in the manifest are chosen at some
+# point, when there are no higher priority compactions.
+
+init 10
+0: 1
+5: 10
+6: 110
+----
+
+init_cp
+----
+base: 5
+
+queue
+----
+
+init 10
+0: 1
+5: 10  marked_for_compaction
+6: 110
+----
+
+init_cp
+----
+base: 5
+
+queue
+----
+L5->L6: 1.0