opt(DropPrefix): check if there exist some data to drop before dropping prefixes (#1621)

We currently do not check if there exist some data to drop for prefixes in DropPrefix(). This can make drop prefix to be expensive because it has to do a bunch of compactions, etc.

This PR optimizes that behavior by not doing any of that in case no data exist for prefixes.

db.go

@@ -1638,6 +1638,16 @@ func (db *DB) DropPrefix(prefixes ...[]byte) error {
 		return err
 	}
 	defer f()
+
+	var filtered [][]byte
+	if filtered, err = db.filterPrefixesToDrop(prefixes); err != nil {
+		return err
+	}
+	// If there is no prefix for which the data already exist, do not do anything.
+	if len(filtered) == 0 {
+		db.opt.Infof("No prefixes to drop")
+		return nil
+	}
 	// Block all foreign interactions with memory tables.
 	db.Lock()
 	defer db.Unlock()
@@ -1651,7 +1661,7 @@ func (db *DB) DropPrefix(prefixes ...[]byte) error {
 		task := flushTask{
 			mt: memtable,
 			// Ensure that the head of value log gets persisted to disk.
-			dropPrefixes: prefixes,
+			dropPrefixes: filtered,
 		}
 		db.opt.Debugf("Flushing memtable")
 		if err := db.handleFlushTask(task); err != nil {
@@ -1669,13 +1679,35 @@ func (db *DB) DropPrefix(prefixes ...[]byte) error {
 	}
 
 	// Drop prefixes from the levels.
-	if err := db.lc.dropPrefixes(prefixes); err != nil {
+	if err := db.lc.dropPrefixes(filtered); err != nil {
 		return err
 	}
 	db.opt.Infof("DropPrefix done")
 	return nil
 }
 
+func (db *DB) filterPrefixesToDrop(prefixes [][]byte) ([][]byte, error) {
+	var filtered [][]byte
+	for _, prefix := range prefixes {
+		err := db.View(func(txn *Txn) error {
+			iopts := DefaultIteratorOptions
+			iopts.Prefix = prefix
+			iopts.PrefetchValues = false
+			itr := txn.NewIterator(iopts)
+			defer itr.Close()
+			itr.Rewind()
+			if itr.ValidForPrefix(prefix) {
+				filtered = append(filtered, prefix)
+			}
+			return nil
+		})
+		if err != nil {
+			return filtered, err
+		}
+	}
+	return filtered, nil
+}
+
 // KVList contains a list of key-value pairs.
 type KVList = pb.KVList
 

db2_test.go

@@ -739,6 +739,30 @@ func TestWindowsDataLoss(t *testing.T) {
 	require.ElementsMatch(t, keyList, result)
 }
 
+func TestDropPrefixWithNoData(t *testing.T) {
+	runBadgerTest(t, nil, func(t *testing.T, db *DB) {
+		val := []byte("value")
+		require.NoError(t, db.Update(func(txn *Txn) error {
+			require.NoError(t, txn.Set([]byte("aaa"), val))
+			require.NoError(t, txn.Set([]byte("aab"), val))
+			require.NoError(t, txn.Set([]byte("aba"), val))
+			require.NoError(t, txn.Set([]byte("aca"), val))
+			return nil
+		}))
+
+		// If we drop prefix, we flush the memtables and create a new mutable memtable. Hence, the
+		// nextMemFid increases by 1. But if there does not exist any data for the prefixes, we
+		// don't do that.
+		memFid := db.nextMemFid
+		prefixes := [][]byte{[]byte("bbb")}
+		require.NoError(t, db.DropPrefix(prefixes...))
+		require.Equal(t, memFid, db.nextMemFid)
+		prefixes = [][]byte{[]byte("aba"), []byte("bbb")}
+		require.NoError(t, db.DropPrefix(prefixes...))
+		require.Equal(t, memFid+1, db.nextMemFid)
+	})
+}
+
 func TestDropAllDropPrefix(t *testing.T) {
 	key := func(i int) []byte {
 		return []byte(fmt.Sprintf("%10d", i))