Added "number of merge operands" to statistics in ssts.

Summary:
A couple of notes from the diff:
  - The namespace block I added at the top of table_properties_collector.cc was in reaction to an issue i was having with PutVarint64 and reusing the "val" string.  I'm not sure this is the cleanest way of doing this, but abstracting this out at least results in the correct behavior.
  - I chose "rocksdb.merge.operands" as the property name.  I am open to suggestions for better names.
  - The change to sst_dump_tool.cc seems a bit inelegant to me.  Is there a better way to do the if-else block?

Test Plan:
I added a test case in table_properties_collector_test.cc.  It adds two merge operands and checks to make sure that both of them are reflected by GetMergeOperands.  It also checks to make sure the wasPropertyPresent bool is properly set in the method.

Running both of these tests should pass:
./table_properties_collector_test
./sst_dump_test

Reviewers: IslamAbdelRahman

Reviewed By: IslamAbdelRahman

Subscribers: andrewkr, dhruba

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

db/table_properties_collector.cc

@@ -23,6 +23,8 @@ Status InternalKeyPropertiesCollector::InternalAdd(const Slice& key,
   if (ikey.type == ValueType::kTypeDeletion ||
       ikey.type == ValueType::kTypeSingleDeletion) {
     ++deleted_keys_;
+  } else if (ikey.type == ValueType::kTypeMerge) {
+    ++merge_operands_;
   }
 
   return Status::OK();
@@ -33,19 +35,26 @@ Status InternalKeyPropertiesCollector::Finish(
   assert(properties);
   assert(properties->find(
         InternalKeyTablePropertiesNames::kDeletedKeys) == properties->end());
-  std::string val;
+  assert(properties->find(InternalKeyTablePropertiesNames::kMergeOperands) ==
+         properties->end());
 
-  PutVarint64(&val, deleted_keys_);
-  properties->insert({ InternalKeyTablePropertiesNames::kDeletedKeys, val });
+  std::string val_deleted_keys;
+  PutVarint64(&val_deleted_keys, deleted_keys_);
+  properties->insert(
+      {InternalKeyTablePropertiesNames::kDeletedKeys, val_deleted_keys});
+
+  std::string val_merge_operands;
+  PutVarint64(&val_merge_operands, merge_operands_);
+  properties->insert(
+      {InternalKeyTablePropertiesNames::kMergeOperands, val_merge_operands});
 
   return Status::OK();
 }
 
 UserCollectedProperties
 InternalKeyPropertiesCollector::GetReadableProperties() const {
-  return {
-    { "kDeletedKeys", ToString(deleted_keys_) }
-  };
+  return {{"kDeletedKeys", ToString(deleted_keys_)},
+          {"kMergeOperands", ToString(merge_operands_)}};
 }
 
 namespace {
@@ -65,6 +74,20 @@ EntryType GetEntryType(ValueType value_type) {
   }
 }
 
+uint64_t GetUint64Property(const UserCollectedProperties& props,
+                           const std::string property_name,
+                           bool* property_present) {
+  auto pos = props.find(property_name);
+  if (pos == props.end()) {
+    *property_present = false;
+    return 0;
+  }
+  Slice raw = pos->second;
+  uint64_t val = 0;
+  *property_present = true;
+  return GetVarint64(&raw, &val) ? val : 0;
+}
+
 }  // namespace
 
 Status UserKeyTablePropertiesCollector::InternalAdd(const Slice& key,
@@ -92,16 +115,20 @@ UserKeyTablePropertiesCollector::GetReadableProperties() const {
 
 const std::string InternalKeyTablePropertiesNames::kDeletedKeys
   = "rocksdb.deleted.keys";
+const std::string InternalKeyTablePropertiesNames::kMergeOperands =
+    "rocksdb.merge.operands";
 
 uint64_t GetDeletedKeys(
     const UserCollectedProperties& props) {
-  auto pos = props.find(InternalKeyTablePropertiesNames::kDeletedKeys);
-  if (pos == props.end()) {
-    return 0;
-  }
-  Slice raw = pos->second;
-  uint64_t val = 0;
-  return GetVarint64(&raw, &val) ? val : 0;
+  bool property_present_ignored;
+  return GetUint64Property(props, InternalKeyTablePropertiesNames::kDeletedKeys,
+                           &property_present_ignored);
+}
+
+uint64_t GetMergeOperands(const UserCollectedProperties& props,
+                          bool* property_present) {
+  return GetUint64Property(
+      props, InternalKeyTablePropertiesNames::kMergeOperands, property_present);
 }
 
 }  // namespace rocksdb

db/table_properties_collector.h

@@ -16,6 +16,7 @@ namespace rocksdb {
 
 struct InternalKeyTablePropertiesNames {
   static const std::string kDeletedKeys;
+  static const std::string kMergeOperands;
 };
 
 // Base class for internal table properties collector.
@@ -65,6 +66,7 @@ class InternalKeyPropertiesCollector : public IntTblPropCollector {
 
  private:
   uint64_t deleted_keys_ = 0;
+  uint64_t merge_operands_ = 0;
 };
 
 class InternalKeyPropertiesCollectorFactory

db/table_properties_collector_test.cc

@@ -368,6 +368,8 @@ void TestInternalKeyPropertiesCollector(
       InternalKey("Y       ", 5, ValueType::kTypeDeletion),
       InternalKey("Z       ", 6, ValueType::kTypeDeletion),
       InternalKey("a       ", 7, ValueType::kTypeSingleDeletion),
+      InternalKey("b       ", 8, ValueType::kTypeMerge),
+      InternalKey("c       ", 9, ValueType::kTypeMerge),
   };
 
   std::unique_ptr<TableBuilder> builder;
@@ -423,6 +425,11 @@ void TestInternalKeyPropertiesCollector(
     uint64_t deleted = GetDeletedKeys(user_collected);
     ASSERT_EQ(5u, deleted);  // deletes + single-deletes
 
+    bool property_present;
+    uint64_t merges = GetMergeOperands(user_collected, &property_present);
+    ASSERT_TRUE(property_present);
+    ASSERT_EQ(2u, merges);
+
     if (sanitized) {
       uint32_t starts_with_A = 0;
       ASSERT_NE(user_collected.find("Count"), user_collected.end());

include/rocksdb/table_properties.h

@@ -193,5 +193,7 @@ struct TableProperties {
 // itself. Especially some properties regarding to the internal keys (which
 // is unknown to `table`).
 extern uint64_t GetDeletedKeys(const UserCollectedProperties& props);
+extern uint64_t GetMergeOperands(const UserCollectedProperties& props,
+                                 bool* property_present);
 
 }  // namespace rocksdb

tools/sst_dump_tool.cc

@@ -557,6 +557,15 @@ int SSTDumpTool::Run(int argc, char** argv) {
         fprintf(stdout, "# deleted keys: %" PRIu64 "\n",
                 rocksdb::GetDeletedKeys(
                     table_properties->user_collected_properties));
+
+        bool property_present;
+        uint64_t merge_operands = rocksdb::GetMergeOperands(
+            table_properties->user_collected_properties, &property_present);
+        if (property_present) {
+          fprintf(stdout, "  # merge operands: %" PRIu64 "\n", merge_operands);
+        } else {
+          fprintf(stdout, "  # merge operands: UNKNOWN\n");
+        }
       }
     }
   }