Enhance partial merge to support multiple arguments

Summary:
* PartialMerge api now takes a list of operands instead of two operands.
* Add min_pertial_merge_operands to Options, indicating the minimum
  number of operands to trigger partial merge.
* This diff is based on Schalk's previous diff (D14601), but it also
  includes necessary changes such as updating the pure C api for
  partial merge.

Test Plan:
* make check all
* develop tests for cases where partial merge takes more than two
  operands.

TODOs (from Schalk):
* Add test with min_partial_merge_operands > 2.
* Perform benchmarks to measure the performance improvements (can probably
  use results of task #2837810.)
* Add description of problem to doc/index.html.
* Change wiki pages to reflect the interface changes.

Reviewers: haobo, igor, vamsi

Reviewed By: haobo

CC: leveldb, dhruba

Differential Revision: https://reviews.facebook.net/D16815

Author: yhchiang

db/builder.cc

@@ -73,6 +73,7 @@ Status BuildTable(const std::string& dbname, Env* env, const Options& options,
 
     MergeHelper merge(internal_comparator.user_comparator(),
                       options.merge_operator.get(), options.info_log.get(),
+                      options.min_partial_merge_operands,
                       true /* internal key corruption is not ok */);
 
     if (purge) {

db/c.cc

@@ -159,12 +159,10 @@ struct rocksdb_mergeoperator_t : public MergeOperator {
       const char* const* operands_list, const size_t* operands_list_length,
       int num_operands,
       unsigned char* success, size_t* new_value_length);
-  char* (*partial_merge_)(
-      void*,
-      const char* key, size_t key_length,
-      const char* left_operand, size_t left_operand_length,
-      const char* right_operand, size_t right_operand_length,
-      unsigned char* success, size_t* new_value_length);
+  char* (*partial_merge_)(void*, const char* key, size_t key_length,
+                          const char* const* operands_list,
+                          const size_t* operands_list_length, int num_operands,
+                          unsigned char* success, size_t* new_value_length);
   void (*delete_value_)(
       void*,
       const char* value, size_t value_length);
@@ -219,21 +217,23 @@ struct rocksdb_mergeoperator_t : public MergeOperator {
     return success;
   }
 
-  virtual bool PartialMerge(
-      const Slice& key,
-      const Slice& left_operand,
-      const Slice& right_operand,
-      std::string* new_value,
-      Logger* logger) const {
+  virtual bool PartialMergeMulti(const Slice& key,
+                                 const std::deque<Slice>& operand_list,
+                                 std::string* new_value, Logger* logger) const {
+    size_t operand_count = operand_list.size();
+    std::vector<const char*> operand_pointers(operand_count);
+    std::vector<size_t> operand_sizes(operand_count);
+    for (size_t i = 0; i < operand_count; ++i) {
+      Slice operand(operand_list[i]);
+      operand_pointers[i] = operand.data();
+      operand_sizes[i] = operand.size();
+    }
 
     unsigned char success;
     size_t new_value_len;
     char* tmp_new_value = (*partial_merge_)(
-        state_,
-        key.data(), key.size(),
-        left_operand.data(), left_operand.size(),
-        right_operand.data(), right_operand.size(),
-        &success, &new_value_len);
+        state_, key.data(), key.size(), &operand_pointers[0], &operand_sizes[0],
+        operand_count, &success, &new_value_len);
     new_value->assign(tmp_new_value, new_value_len);
 
     if (delete_value_ != nullptr) {
@@ -1094,24 +1094,18 @@ rocksdb_filterpolicy_t* rocksdb_filterpolicy_create_bloom(int bits_per_key) {
 }
 
 rocksdb_mergeoperator_t* rocksdb_mergeoperator_create(
-    void* state,
-    void (*destructor)(void*),
-    char* (*full_merge)(
-        void*,
-        const char* key, size_t key_length,
-        const char* existing_value, size_t existing_value_length,
-        const char* const* operands_list, const size_t* operands_list_length,
-        int num_operands,
-        unsigned char* success, size_t* new_value_length),
-    char* (*partial_merge)(
-        void*,
-        const char* key, size_t key_length,
-        const char* left_operand, size_t left_operand_length,
-        const char* right_operand, size_t right_operand_length,
-        unsigned char* success, size_t* new_value_length),
-    void (*delete_value)(
-        void*,
-        const char* value, size_t value_length),
+    void* state, void (*destructor)(void*),
+    char* (*full_merge)(void*, const char* key, size_t key_length,
+                        const char* existing_value,
+                        size_t existing_value_length,
+                        const char* const* operands_list,
+                        const size_t* operands_list_length, int num_operands,
+                        unsigned char* success, size_t* new_value_length),
+    char* (*partial_merge)(void*, const char* key, size_t key_length,
+                           const char* const* operands_list,
+                           const size_t* operands_list_length, int num_operands,
+                           unsigned char* success, size_t* new_value_length),
+    void (*delete_value)(void*, const char* value, size_t value_length),
     const char* (*name)(void*)) {
   rocksdb_mergeoperator_t* result = new rocksdb_mergeoperator_t;
   result->state_ = state;

db/c_test.c

@@ -175,8 +175,8 @@ static char* MergeOperatorFullMerge(
 static char* MergeOperatorPartialMerge(
     void* arg,
     const char* key, size_t key_length,
-    const char* left_operand, size_t left_operand_length,
-    const char* right_operand, size_t right_operand_length,
+    const char* const* operands_list, const size_t* operands_list_length,
+    int num_operands,
     unsigned char* success, size_t* new_value_length) {
   *new_value_length = 4;
   *success = 1;

db/db_impl.cc

@@ -2463,6 +2463,7 @@ Status DBImpl::DoCompactionWork(CompactionState* compact,
   std::vector<char> delete_key; // for compaction filter
   MergeHelper merge(user_comparator(), options_.merge_operator.get(),
                     options_.info_log.get(),
+                    options_.min_partial_merge_operands,
                     false /* internal key corruption is expected */);
   auto compaction_filter = options_.compaction_filter;
   std::unique_ptr<CompactionFilter> compaction_filter_from_factory = nullptr;

db/db_iter.cc

@@ -313,21 +313,6 @@ void DBIter::MergeValuesNewToOld() {
       // when complete, add result to operands and continue.
       const Slice& value = iter_->value();
       operands.push_front(value.ToString());
-      while(operands.size() >= 2) {
-        // Call user associative-merge until it returns false
-        if (user_merge_operator_->PartialMerge(ikey.user_key,
-                                               Slice(operands[0]),
-                                               Slice(operands[1]),
-                                               &merge_result,
-                                               logger_)) {
-          operands.pop_front();
-          swap(operands.front(), merge_result);
-        } else {
-          // Associative merge returns false ==> stack the operands
-          break;
-        }
-      }
-
     }
   }
 

db/memtable.cc

@@ -36,7 +36,7 @@ MemTable::MemTable(const InternalKeyComparator& cmp, const Options& options)
       kWriteBufferSize(options.write_buffer_size),
       arena_(options.arena_block_size),
       table_(options.memtable_factory->CreateMemTableRep(
-             comparator_, &arena_, options.prefix_extractor.get())),
+          comparator_, &arena_, options.prefix_extractor.get())),
       flush_in_progress_(false),
       flush_completed_(false),
       file_number_(0),
@@ -353,17 +353,6 @@ static bool SaveValue(void* arg, const char* entry) {
         Slice v = GetLengthPrefixedSlice(key_ptr + key_length);
         *(s->merge_in_progress) = true;
         merge_context->PushOperand(v);
-        while (merge_context->GetNumOperands() >= 2) {
-          // Attempt to associative merge. (Returns true if successful)
-          if (merge_operator->PartialMerge(
-                  s->key->user_key(), merge_context->GetOperand(0),
-                  merge_context->GetOperand(1), &merge_result, s->logger)) {
-            merge_context->PushPartialMergeResult(merge_result);
-          } else {
-            // Stack them because user can't associative merge
-            break;
-          }
-        }
         return true;
       }
       default:

db/merge_context.h

@@ -25,12 +25,12 @@ class MergeContext {
       operand_list->clear();
     }
   }
-  // Replace the first two operands of merge_result, which are expected be the
-  // merge results of them.
+  // Replace all operands with merge_result, which are expected to be the
+  // merge result of them.
   void PushPartialMergeResult(std::string& merge_result) {
     assert (operand_list);
-    operand_list->pop_front();
-    swap(operand_list->front(), merge_result);
+    operand_list->clear();
+    operand_list->push_front(std::move(merge_result));
   }
   // Push a merge operand
   void PushOperand(const Slice& operand_slice) {

db/merge_helper.cc

@@ -129,28 +129,10 @@ void MergeHelper::MergeUntil(Iterator* iter, SequenceNumber stop_before,
       //   => then continue because we haven't yet seen a Put/Delete.
       assert(!operands_.empty()); // Should have at least one element in it
 
+      // keep queuing keys and operands until we either meet a put / delete
+      // request or later did a partial merge.
       keys_.push_front(iter->key().ToString());
       operands_.push_front(iter->value().ToString());
-      while (operands_.size() >= 2) {
-        // Returns false when the merge_operator can no longer process it
-        if (user_merge_operator_->PartialMerge(ikey.user_key,
-                                               Slice(operands_[0]),
-                                               Slice(operands_[1]),
-                                               &merge_result,
-                                               logger_)) {
-          // Merging of operands (associative merge) was successful.
-          // Replace these frontmost two operands with the merge result
-          keys_.pop_front();
-          operands_.pop_front();
-          swap(operands_.front(), merge_result);
-        } else {
-          // Merging of operands (associative merge) returned false.
-          // The user merge_operator does not know how to merge these operands.
-          // So we just stack them up until we find a Put/Delete or end of key.
-          break;
-        }
-      }
-      continue;
     }
   }
 
@@ -192,6 +174,23 @@ void MergeHelper::MergeUntil(Iterator* iter, SequenceNumber stop_before,
       RecordTick(stats, NUMBER_MERGE_FAILURES);
       // Do nothing if not success_. Leave keys() and operands() as they are.
     }
+  } else {
+    // We haven't seen the beginning of the key nor a Put/Delete.
+    // Attempt to use the user's associative merge function to
+    // merge the stacked merge operands into a single operand.
+
+    if (operands_.size() >= 2 &&
+        operands_.size() >= min_partial_merge_operands_ &&
+        user_merge_operator_->PartialMergeMulti(
+            ikey.user_key,
+            std::deque<Slice>(operands_.begin(), operands_.end()),
+            &merge_result, logger_)) {
+      // Merging of operands (associative merge) was successful.
+      // Replace operands with the merge result
+      operands_.clear();
+      operands_.push_front(std::move(merge_result));
+      keys_.erase(keys_.begin(), keys_.end() - 1);
+    }
   }
 }
 

db/merge_helper.h

@@ -22,12 +22,13 @@ class Statistics;
 class MergeHelper {
  public:
   MergeHelper(const Comparator* user_comparator,
-              const MergeOperator* user_merge_operator,
-              Logger* logger,
+              const MergeOperator* user_merge_operator, Logger* logger,
+              unsigned min_partial_merge_operands,
               bool assert_valid_internal_key)
       : user_comparator_(user_comparator),
         user_merge_operator_(user_merge_operator),
         logger_(logger),
+        min_partial_merge_operands_(min_partial_merge_operands),
         assert_valid_internal_key_(assert_valid_internal_key),
         keys_(),
         operands_(),
@@ -88,6 +89,7 @@ class MergeHelper {
   const Comparator* user_comparator_;
   const MergeOperator* user_merge_operator_;
   Logger* logger_;
+  unsigned min_partial_merge_operands_;
   bool assert_valid_internal_key_; // enforce no internal key corruption?
 
   // the scratch area that holds the result of MergeUntil

db/merge_operator.cc

@@ -11,6 +11,28 @@
 
 namespace rocksdb {
 
+// The default implementation of PartialMergeMulti, which invokes
+// PartialMerge multiple times internally and merges two operands at
+// a time.
+bool MergeOperator::PartialMergeMulti(const Slice& key,
+                                      const std::deque<Slice>& operand_list,
+                                      std::string* new_value,
+                                      Logger* logger) const {
+  // Simply loop through the operands
+  std::string temp_value;
+  Slice temp_slice;
+  for (const auto& operand : operand_list) {
+    if (!PartialMerge(key, temp_slice, operand, &temp_value, logger)) {
+      return false;
+    }
+    swap(temp_value, *new_value);
+    temp_slice = Slice(*new_value);
+  }
+
+  // The result will be in *new_value. All merges succeeded.
+  return true;
+}
+
 // Given a "real" merge from the library, call the user's
 // associative merge function one-by-one on each of the operands.
 // NOTE: It is assumed that the client's merge-operator will handle any errors.
@@ -46,7 +68,6 @@ bool AssociativeMergeOperator::PartialMerge(
     const Slice& right_operand,
     std::string* new_value,
     Logger* logger) const {
-
   return Merge(key, &left_operand, right_operand, new_value, logger);
 }