From beb36d9c1e977af6da033fd6b95e0a937eb52ed3 Mon Sep 17 00:00:00 2001
From: Mike Kolupaev <kolmike@fb.com>
Date: Mon, 5 Dec 2016 15:07:01 -0800
Subject: [PATCH] Fixed CompactionFilter::Decision::kRemoveAndSkipUntil

Summary:
Embarassingly enough, the first time I tried to use my new feature in logdevice it crashed with this assertion failure:

  db/pinned_iterators_manager.h:30: void rocksdb::PinnedIteratorsManager::StartPinning(): Assertion `pinning_enabled == false' failed

The issue was that `pinned_iters_mgr_.StartPinning()` was called but `pinned_iters_mgr_.ReleasePinnedData()` wasn't.
Closes https://github.com/facebook/rocksdb/pull/1611

Differential Revision: D4265622

Pulled By: al13n321

fbshipit-source-id: 747b10f
---
 db/compaction_iterator.cc      | 8 +++++---
 db/compaction_iterator_test.cc | 8 ++++----
 2 files changed, 9 insertions(+), 7 deletions(-)

diff --git a/db/compaction_iterator.cc b/db/compaction_iterator.cc
index e46c679bf..152a1de54 100644
--- a/db/compaction_iterator.cc
+++ b/db/compaction_iterator.cc
@@ -437,9 +437,7 @@ void CompactionIterator::NextFromInput() {
                                 bottommost_level_);
       merge_out_iter_.SeekToFirst();
 
-      if (merge_helper_->FilteredUntil(&skip_until)) {
-        need_skip = true;
-      } else if (merge_out_iter_.Valid()) {
+      if (merge_out_iter_.Valid()) {
         // NOTE: key, value, and ikey_ refer to old entries.
         //       These will be correctly set below.
         key_ = merge_out_iter_.key();
@@ -460,6 +458,10 @@ void CompactionIterator::NextFromInput() {
         // coming after the merges
         has_current_user_key_ = false;
         pinned_iters_mgr_.ReleasePinnedData();
+
+        if (merge_helper_->FilteredUntil(&skip_until)) {
+          need_skip = true;
+        }
       }
     } else {
       // 1. new user key -OR-
diff --git a/db/compaction_iterator_test.cc b/db/compaction_iterator_test.cc
index de5bdee34..dd398696d 100644
--- a/db/compaction_iterator_test.cc
+++ b/db/compaction_iterator_test.cc
@@ -272,8 +272,8 @@ TEST_F(CompactionIteratorTest, CompactionFilterSkipUntil) {
         return Decision::kKeep;
       }
       if (k == "i") {
-        EXPECT_EQ(ValueType::kValue, t);
-        EXPECT_EQ("iv95", v);
+        EXPECT_EQ(ValueType::kMergeOperand, t);
+        EXPECT_EQ("im95", v);
         *skip_until = "z";
         return Decision::kRemoveAndSkipUntil;
       }
@@ -299,10 +299,10 @@ TEST_F(CompactionIteratorTest, CompactionFilterSkipUntil) {
        test::KeyStr("f", 30, kTypeMerge),  // skip to "g+"
        test::KeyStr("f", 25, kTypeValue), test::KeyStr("g", 90, kTypeValue),
        test::KeyStr("h", 91, kTypeValue),  // keep
-       test::KeyStr("i", 95, kTypeValue),  // skip to "z"
+       test::KeyStr("i", 95, kTypeMerge),  // skip to "z"
        test::KeyStr("j", 99, kTypeValue)},
       {"av50", "am45", "bv60", "bv40", "cv35", "dm70", "em71", "fm65", "fm30",
-       "fv25", "gv90", "hv91", "iv95", "jv99"},
+       "fv25", "gv90", "hv91", "im95", "jv99"},
       {}, {}, kMaxSequenceNumber, &merge_op, &filter);
 
   // Compaction should output just "a", "e" and "h" keys.