fix(compact): close vlog after the compaction at L0 has completed (#1752

)

Vlog needs to stay open while the compaction is being run. With the CompactL0OnClose option, it becomes necessary to close the vlog after all the compactions have been completed.

db.go

@@ -559,11 +559,6 @@ func (db *DB) close() (err error) {
 	db.closers.pub.SignalAndWait()
 	db.closers.cacheHealth.Signal()
 
-	// Now close the value log.
-	if vlogErr := db.vlog.Close(); vlogErr != nil {
-		err = y.Wrap(vlogErr, "DB.Close")
-	}
-
 	// Make sure that block writer is done pushing stuff into memtable!
 	// Otherwise, you will have a race condition: we are trying to flush memtables
 	// and remove them completely, while the block / memtable writer is still
@@ -619,6 +614,11 @@ func (db *DB) close() (err error) {
 		}
 	}
 
+	// Now close the value log.
+	if vlogErr := db.vlog.Close(); vlogErr != nil {
+		err = y.Wrap(vlogErr, "DB.Close")
+	}
+
 	db.opt.Infof(db.LevelsToString())
 	if lcErr := db.lc.close(); err == nil {
 		err = y.Wrap(lcErr, "DB.Close")
@@ -2113,7 +2113,7 @@ func (db *DB) Subscribe(ctx context.Context, cb func(kv *KVList) error, matches
 	drain := func() {
 		for {
 			select {
-			case <- s.sendCh:
+			case <-s.sendCh:
 			default:
 				return
 			}

db_test.go

@@ -2547,3 +2547,32 @@ func TestSeekTs(t *testing.T) {
 		}
 	})
 }
+
+func TestCompactL0OnClose(t *testing.T) {
+	opt := getTestOptions("")
+	opt.CompactL0OnClose = true
+	opt.ValueThreshold = 1 // Every value goes to value log
+	opt.NumVersionsToKeep = 1
+	runBadgerTest(t, &opt, func(t *testing.T, db *DB) {
+		var keys [][]byte
+		val := make([]byte, 1<<12)
+		for i := 0; i < 10; i++ {
+			key := make([]byte, 40)
+			_, err := rand.Read(key)
+			require.NoError(t, err)
+			keys = append(keys, key)
+
+			err = db.Update(func(txn *Txn) error {
+				return txn.SetEntry(NewEntry(key, val))
+			})
+			require.NoError(t, err)
+		}
+
+		for _, key := range keys {
+			err := db.Update(func(txn *Txn) error {
+				return txn.SetEntry(NewEntry(key, val))
+			})
+			require.NoError(t, err)
+		}
+	})
+}