ceph/src/rocksdb/db/merge_helper_test.cc

   1 //  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
   2 //  This source code is licensed under both the GPLv2 (found in the
   3 //  COPYING file in the root directory) and Apache 2.0 License
   4 //  (found in the LICENSE.Apache file in the root directory).
   5
   6 #include <algorithm>
   7 #include <string>
   8 #include <vector>
   9
  10 #include "db/merge_helper.h"
  11 #include "rocksdb/comparator.h"
  12 #include "util/coding.h"
  13 #include "util/testharness.h"
  14 #include "util/testutil.h"
  15 #include "utilities/merge_operators.h"
  16
  17 namespace rocksdb {
  18
  19 class MergeHelperTest : public testing::Test {
  20  public:
  21   MergeHelperTest() { env_ = Env::Default(); }
  22
  23   ~MergeHelperTest() = default;
  24
  25   Status Run(SequenceNumber stop_before, bool at_bottom,
  26              SequenceNumber latest_snapshot = 0) {
  27     iter_.reset(new test::VectorIterator(ks_, vs_));
  28     iter_->SeekToFirst();
  29     merge_helper_.reset(new MergeHelper(env_, BytewiseComparator(),
  30                                         merge_op_.get(), filter_.get(), nullptr,
  31                                         false, latest_snapshot));
  32     return merge_helper_->MergeUntil(iter_.get(), nullptr /* range_del_agg */,
  33                                      stop_before, at_bottom);
  34   }
  35
  36   void AddKeyVal(const std::string& user_key, const SequenceNumber& seq,
  37                  const ValueType& t, const std::string& val,
  38                  bool corrupt = false) {
  39     InternalKey ikey(user_key, seq, t);
  40     if (corrupt) {
  41       test::CorruptKeyType(&ikey);
  42     }
  43     ks_.push_back(ikey.Encode().ToString());
  44     vs_.push_back(val);
  45   }
  46
  47   Env* env_;
  48   std::unique_ptr<test::VectorIterator> iter_;
  49   std::shared_ptr<MergeOperator> merge_op_;
  50   std::unique_ptr<MergeHelper> merge_helper_;
  51   std::vector<std::string> ks_;
  52   std::vector<std::string> vs_;
  53   std::unique_ptr<test::FilterNumber> filter_;
  54 };
  55
  56 // If MergeHelper encounters a new key on the last level, we know that
  57 // the key has no more history and it can merge keys.
  58 TEST_F(MergeHelperTest, MergeAtBottomSuccess) {
  59   merge_op_ = MergeOperators::CreateUInt64AddOperator();
  60
  61   AddKeyVal("a", 20, kTypeMerge, test::EncodeInt(1U));
  62   AddKeyVal("a", 10, kTypeMerge, test::EncodeInt(3U));
  63   AddKeyVal("b", 10, kTypeMerge, test::EncodeInt(4U));  // <- iter_ after merge
  64
  65   ASSERT_TRUE(Run(0, true).ok());
  66   ASSERT_EQ(ks_[2], iter_->key());
  67   ASSERT_EQ(test::KeyStr("a", 20, kTypeValue), merge_helper_->keys()[0]);
  68   ASSERT_EQ(test::EncodeInt(4U), merge_helper_->values()[0]);
  69   ASSERT_EQ(1U, merge_helper_->keys().size());
  70   ASSERT_EQ(1U, merge_helper_->values().size());
  71 }
  72
  73 // Merging with a value results in a successful merge.
  74 TEST_F(MergeHelperTest, MergeValue) {
  75   merge_op_ = MergeOperators::CreateUInt64AddOperator();
  76
  77   AddKeyVal("a", 40, kTypeMerge, test::EncodeInt(1U));
  78   AddKeyVal("a", 30, kTypeMerge, test::EncodeInt(3U));
  79   AddKeyVal("a", 20, kTypeValue, test::EncodeInt(4U));  // <- iter_ after merge
  80   AddKeyVal("a", 10, kTypeMerge, test::EncodeInt(1U));
  81
  82   ASSERT_TRUE(Run(0, false).ok());
  83   ASSERT_EQ(ks_[3], iter_->key());
  84   ASSERT_EQ(test::KeyStr("a", 40, kTypeValue), merge_helper_->keys()[0]);
  85   ASSERT_EQ(test::EncodeInt(8U), merge_helper_->values()[0]);
  86   ASSERT_EQ(1U, merge_helper_->keys().size());
  87   ASSERT_EQ(1U, merge_helper_->values().size());
  88 }
  89
  90 // Merging stops before a snapshot.
  91 TEST_F(MergeHelperTest, SnapshotBeforeValue) {
  92   merge_op_ = MergeOperators::CreateUInt64AddOperator();
  93
  94   AddKeyVal("a", 50, kTypeMerge, test::EncodeInt(1U));
  95   AddKeyVal("a", 40, kTypeMerge, test::EncodeInt(3U));  // <- iter_ after merge
  96   AddKeyVal("a", 30, kTypeMerge, test::EncodeInt(1U));
  97   AddKeyVal("a", 20, kTypeValue, test::EncodeInt(4U));
  98   AddKeyVal("a", 10, kTypeMerge, test::EncodeInt(1U));
  99
 100   ASSERT_TRUE(Run(31, true).IsMergeInProgress());
 101   ASSERT_EQ(ks_[2], iter_->key());
 102   ASSERT_EQ(test::KeyStr("a", 50, kTypeMerge), merge_helper_->keys()[0]);
 103   ASSERT_EQ(test::EncodeInt(4U), merge_helper_->values()[0]);
 104   ASSERT_EQ(1U, merge_helper_->keys().size());
 105   ASSERT_EQ(1U, merge_helper_->values().size());
 106 }
 107
 108 // MergeHelper preserves the operand stack for merge operators that
 109 // cannot do a partial merge.
 110 TEST_F(MergeHelperTest, NoPartialMerge) {
 111   merge_op_ = MergeOperators::CreateStringAppendTESTOperator();
 112
 113   AddKeyVal("a", 50, kTypeMerge, "v2");
 114   AddKeyVal("a", 40, kTypeMerge, "v");  // <- iter_ after merge
 115   AddKeyVal("a", 30, kTypeMerge, "v");
 116
 117   ASSERT_TRUE(Run(31, true).IsMergeInProgress());
 118   ASSERT_EQ(ks_[2], iter_->key());
 119   ASSERT_EQ(test::KeyStr("a", 40, kTypeMerge), merge_helper_->keys()[0]);
 120   ASSERT_EQ("v", merge_helper_->values()[0]);
 121   ASSERT_EQ(test::KeyStr("a", 50, kTypeMerge), merge_helper_->keys()[1]);
 122   ASSERT_EQ("v2", merge_helper_->values()[1]);
 123   ASSERT_EQ(2U, merge_helper_->keys().size());
 124   ASSERT_EQ(2U, merge_helper_->values().size());
 125 }
 126
 127 // A single operand can not be merged.
 128 TEST_F(MergeHelperTest, SingleOperand) {
 129   merge_op_ = MergeOperators::CreateUInt64AddOperator();
 130
 131   AddKeyVal("a", 50, kTypeMerge, test::EncodeInt(1U));
 132
 133   ASSERT_TRUE(Run(31, true).IsMergeInProgress());
 134   ASSERT_FALSE(iter_->Valid());
 135   ASSERT_EQ(test::KeyStr("a", 50, kTypeMerge), merge_helper_->keys()[0]);
 136   ASSERT_EQ(test::EncodeInt(1U), merge_helper_->values()[0]);
 137   ASSERT_EQ(1U, merge_helper_->keys().size());
 138   ASSERT_EQ(1U, merge_helper_->values().size());
 139 }
 140
 141 // Merging with a deletion turns the deletion into a value
 142 TEST_F(MergeHelperTest, MergeDeletion) {
 143   merge_op_ = MergeOperators::CreateUInt64AddOperator();
 144
 145   AddKeyVal("a", 30, kTypeMerge, test::EncodeInt(3U));
 146   AddKeyVal("a", 20, kTypeDeletion, "");
 147
 148   ASSERT_TRUE(Run(15, false).ok());
 149   ASSERT_FALSE(iter_->Valid());
 150   ASSERT_EQ(test::KeyStr("a", 30, kTypeValue), merge_helper_->keys()[0]);
 151   ASSERT_EQ(test::EncodeInt(3U), merge_helper_->values()[0]);
 152   ASSERT_EQ(1U, merge_helper_->keys().size());
 153   ASSERT_EQ(1U, merge_helper_->values().size());
 154 }
 155
 156 // The merge helper stops upon encountering a corrupt key
 157 TEST_F(MergeHelperTest, CorruptKey) {
 158   merge_op_ = MergeOperators::CreateUInt64AddOperator();
 159
 160   AddKeyVal("a", 30, kTypeMerge, test::EncodeInt(3U));
 161   AddKeyVal("a", 25, kTypeMerge, test::EncodeInt(1U));
 162   // Corrupt key
 163   AddKeyVal("a", 20, kTypeDeletion, "", true);  // <- iter_ after merge
 164
 165   ASSERT_TRUE(Run(15, false).IsMergeInProgress());
 166   ASSERT_EQ(ks_[2], iter_->key());
 167   ASSERT_EQ(test::KeyStr("a", 30, kTypeMerge), merge_helper_->keys()[0]);
 168   ASSERT_EQ(test::EncodeInt(4U), merge_helper_->values()[0]);
 169   ASSERT_EQ(1U, merge_helper_->keys().size());
 170   ASSERT_EQ(1U, merge_helper_->values().size());
 171 }
 172
 173 // The compaction filter is called on every merge operand
 174 TEST_F(MergeHelperTest, FilterMergeOperands) {
 175   merge_op_ = MergeOperators::CreateUInt64AddOperator();
 176   filter_.reset(new test::FilterNumber(5U));
 177
 178   AddKeyVal("a", 30, kTypeMerge, test::EncodeInt(3U));
 179   AddKeyVal("a", 29, kTypeMerge, test::EncodeInt(5U));  // Filtered
 180   AddKeyVal("a", 28, kTypeMerge, test::EncodeInt(3U));
 181   AddKeyVal("a", 27, kTypeMerge, test::EncodeInt(1U));
 182   AddKeyVal("a", 26, kTypeMerge, test::EncodeInt(5U));  // Filtered
 183   AddKeyVal("a", 25, kTypeValue, test::EncodeInt(1U));
 184
 185   ASSERT_TRUE(Run(15, false).ok());
 186   ASSERT_FALSE(iter_->Valid());
 187   MergeOutputIterator merge_output_iter(merge_helper_.get());
 188   merge_output_iter.SeekToFirst();
 189   ASSERT_EQ(test::KeyStr("a", 30, kTypeValue),
 190             merge_output_iter.key().ToString());
 191   ASSERT_EQ(test::EncodeInt(8U), merge_output_iter.value().ToString());
 192   merge_output_iter.Next();
 193   ASSERT_FALSE(merge_output_iter.Valid());
 194 }
 195
 196 TEST_F(MergeHelperTest, FilterAllMergeOperands) {
 197   merge_op_ = MergeOperators::CreateUInt64AddOperator();
 198   filter_.reset(new test::FilterNumber(5U));
 199
 200   AddKeyVal("a", 30, kTypeMerge, test::EncodeInt(5U));
 201   AddKeyVal("a", 29, kTypeMerge, test::EncodeInt(5U));
 202   AddKeyVal("a", 28, kTypeMerge, test::EncodeInt(5U));
 203   AddKeyVal("a", 27, kTypeMerge, test::EncodeInt(5U));
 204   AddKeyVal("a", 26, kTypeMerge, test::EncodeInt(5U));
 205   AddKeyVal("a", 25, kTypeMerge, test::EncodeInt(5U));
 206
 207   // filtered out all
 208   ASSERT_TRUE(Run(15, false).ok());
 209   ASSERT_FALSE(iter_->Valid());
 210   MergeOutputIterator merge_output_iter(merge_helper_.get());
 211   merge_output_iter.SeekToFirst();
 212   ASSERT_FALSE(merge_output_iter.Valid());
 213
 214   // we have one operand that will survive because it's a delete
 215   AddKeyVal("a", 24, kTypeDeletion, test::EncodeInt(5U));
 216   AddKeyVal("b", 23, kTypeValue, test::EncodeInt(5U));
 217   ASSERT_TRUE(Run(15, true).ok());
 218   merge_output_iter = MergeOutputIterator(merge_helper_.get());
 219   ASSERT_TRUE(iter_->Valid());
 220   merge_output_iter.SeekToFirst();
 221   ASSERT_FALSE(merge_output_iter.Valid());
 222
 223   // when all merge operands are filtered out, we leave the iterator pointing to
 224   // the Put/Delete that survived
 225   ASSERT_EQ(test::KeyStr("a", 24, kTypeDeletion), iter_->key().ToString());
 226   ASSERT_EQ(test::EncodeInt(5U), iter_->value().ToString());
 227 }
 228
 229 // Make sure that merge operands are filtered at the beginning
 230 TEST_F(MergeHelperTest, FilterFirstMergeOperand) {
 231   merge_op_ = MergeOperators::CreateUInt64AddOperator();
 232   filter_.reset(new test::FilterNumber(5U));
 233
 234   AddKeyVal("a", 31, kTypeMerge, test::EncodeInt(5U));  // Filtered
 235   AddKeyVal("a", 30, kTypeMerge, test::EncodeInt(5U));  // Filtered
 236   AddKeyVal("a", 29, kTypeMerge, test::EncodeInt(2U));
 237   AddKeyVal("a", 28, kTypeMerge, test::EncodeInt(1U));
 238   AddKeyVal("a", 27, kTypeMerge, test::EncodeInt(3U));
 239   AddKeyVal("a", 26, kTypeMerge, test::EncodeInt(5U));  // Filtered
 240   AddKeyVal("a", 25, kTypeMerge, test::EncodeInt(5U));  // Filtered
 241   AddKeyVal("b", 24, kTypeValue, test::EncodeInt(5U));  // next user key
 242
 243   ASSERT_OK(Run(15, true));
 244   ASSERT_TRUE(iter_->Valid());
 245   MergeOutputIterator merge_output_iter(merge_helper_.get());
 246   merge_output_iter.SeekToFirst();
 247   // sequence number is 29 here, because the first merge operand got filtered
 248   // out
 249   ASSERT_EQ(test::KeyStr("a", 29, kTypeValue),
 250             merge_output_iter.key().ToString());
 251   ASSERT_EQ(test::EncodeInt(6U), merge_output_iter.value().ToString());
 252   merge_output_iter.Next();
 253   ASSERT_FALSE(merge_output_iter.Valid());
 254
 255   // make sure that we're passing user keys into the filter
 256   ASSERT_EQ("a", filter_->last_merge_operand_key());
 257 }
 258
 259 // Make sure that merge operands are not filtered out if there's a snapshot
 260 // pointing at them
 261 TEST_F(MergeHelperTest, DontFilterMergeOperandsBeforeSnapshotTest) {
 262   merge_op_ = MergeOperators::CreateUInt64AddOperator();
 263   filter_.reset(new test::FilterNumber(5U));
 264
 265   AddKeyVal("a", 31, kTypeMerge, test::EncodeInt(5U));
 266   AddKeyVal("a", 30, kTypeMerge, test::EncodeInt(5U));
 267   AddKeyVal("a", 29, kTypeMerge, test::EncodeInt(2U));
 268   AddKeyVal("a", 28, kTypeMerge, test::EncodeInt(1U));
 269   AddKeyVal("a", 27, kTypeMerge, test::EncodeInt(3U));
 270   AddKeyVal("a", 26, kTypeMerge, test::EncodeInt(5U));
 271   AddKeyVal("a", 25, kTypeMerge, test::EncodeInt(5U));
 272   AddKeyVal("b", 24, kTypeValue, test::EncodeInt(5U));
 273
 274   ASSERT_OK(Run(15, true, 32));
 275   ASSERT_TRUE(iter_->Valid());
 276   MergeOutputIterator merge_output_iter(merge_helper_.get());
 277   merge_output_iter.SeekToFirst();
 278   ASSERT_EQ(test::KeyStr("a", 31, kTypeValue),
 279             merge_output_iter.key().ToString());
 280   ASSERT_EQ(test::EncodeInt(26U), merge_output_iter.value().ToString());
 281   merge_output_iter.Next();
 282   ASSERT_FALSE(merge_output_iter.Valid());
 283 }
 284
 285 }  // namespace rocksdb
 286
 287 int main(int argc, char** argv) {
 288   ::testing::InitGoogleTest(&argc, argv);
 289   return RUN_ALL_TESTS();
 290 }