Adding new table properties

Summary:
This diff introduce new table properties that will be written for block based tables
These properties are
  - comparator name
  - merge operator name
  - property collectors names

Test Plan:
  - Added a new unit test to verify that these tests are written/read correctly
  - Running all other tests right now (wont land until all tests finish)

Reviewers: rven, kradhakrishnan, igor, sdong, anthony, yhchiang

Reviewed By: yhchiang

Subscribers: dhruba

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

include/rocksdb/table_properties.h

@@ -53,6 +53,16 @@ struct TableProperties {
   // The name of the filter policy used in this table.
   // If no filter policy is used, `filter_policy_name` will be an empty string.
   std::string filter_policy_name;
+  // The name of the comparator used in this table.
+  std::string comparator_name;
+  // The name of the merge operator used in this table.
+  // If no merge operator is used, `merge_operator_name` will be
+  // an empty string.
+  std::string merge_operator_name;
+  // The names of the property collectors factories used in this table
+  // separated by commas
+  // {collector_name[1]},{collector_name[2]},{collector_name[3]} ..
+  std::string property_collectors_names;
 
   // user collected properties
   UserCollectedProperties user_collected_properties;
@@ -80,6 +90,9 @@ struct TablePropertiesNames {
   static const std::string kFormatVersion;
   static const std::string kFixedKeyLen;
   static const std::string kFilterPolicy;
+  static const std::string kComparator;
+  static const std::string kMergeOperator;
+  static const std::string kPropertyCollectors;
 };
 
 extern const std::string kPropertiesBlock;

table/block_based_table_builder.cc

@@ -26,6 +26,7 @@
 #include "rocksdb/env.h"
 #include "rocksdb/filter_policy.h"
 #include "rocksdb/flush_block_policy.h"
+#include "rocksdb/merge_operator.h"
 #include "rocksdb/table.h"
 
 #include "table/block.h"
@@ -788,6 +789,23 @@ Status BlockBasedTableBuilder::Finish() {
           r->table_options.filter_policy->Name() : "";
       r->props.index_size =
           r->index_builder->EstimatedSize() + kBlockTrailerSize;
+      r->props.comparator_name = r->ioptions.comparator != nullptr
+                                     ? r->ioptions.comparator->Name()
+                                     : "N/A";
+      r->props.merge_operator_name = r->ioptions.merge_operator != nullptr
+                                         ? r->ioptions.merge_operator->Name()
+                                         : "N/A";
+
+      std::string property_collectors_names = "";
+      for (uint32_t i = 0;
+           i < r->ioptions.table_properties_collector_factories.size(); ++i) {
+        if (i != 0) {
+          property_collectors_names += ",";
+        }
+        property_collectors_names +=
+            r->ioptions.table_properties_collector_factories[i]->Name();
+      }
+      r->props.property_collectors_names = property_collectors_names;
 
       // Add basic properties
       property_block_builder.AddTableProperty(r->props);

table/meta_blocks.cc

@@ -71,8 +71,20 @@ void PropertyBlockBuilder::AddTableProperty(const TableProperties& props) {
   Add(TablePropertiesNames::kFixedKeyLen, props.fixed_key_len);
 
   if (!props.filter_policy_name.empty()) {
-    Add(TablePropertiesNames::kFilterPolicy,
-        props.filter_policy_name);
+    Add(TablePropertiesNames::kFilterPolicy, props.filter_policy_name);
+  }
+
+  if (!props.comparator_name.empty()) {
+    Add(TablePropertiesNames::kComparator, props.comparator_name);
+  }
+
+  if (!props.merge_operator_name.empty()) {
+    Add(TablePropertiesNames::kMergeOperator, props.merge_operator_name);
+  }
+
+  if (!props.property_collectors_names.empty()) {
+    Add(TablePropertiesNames::kPropertyCollectors,
+        props.property_collectors_names);
   }
 }
 
@@ -203,6 +215,12 @@ Status ReadProperties(const Slice& handle_value, RandomAccessFileReader* file,
       *(pos->second) = val;
     } else if (key == TablePropertiesNames::kFilterPolicy) {
       new_table_properties->filter_policy_name = raw_val.ToString();
+    } else if (key == TablePropertiesNames::kComparator) {
+      new_table_properties->comparator_name = raw_val.ToString();
+    } else if (key == TablePropertiesNames::kMergeOperator) {
+      new_table_properties->merge_operator_name = raw_val.ToString();
+    } else if (key == TablePropertiesNames::kPropertyCollectors) {
+      new_table_properties->property_collectors_names = raw_val.ToString();
     } else {
       // handle user-collected properties
       new_table_properties->user_collected_properties.insert(

table/table_properties.cc

@@ -75,6 +75,20 @@ std::string TableProperties::ToString(
       filter_policy_name.empty() ? std::string("N/A") : filter_policy_name,
       prop_delim, kv_delim);
 
+  AppendProperty(result, "comparator name",
+                 comparator_name.empty() ? std::string("N/A") : comparator_name,
+                 prop_delim, kv_delim);
+
+  AppendProperty(
+      result, "merge operator name",
+      merge_operator_name.empty() ? std::string("N/A") : merge_operator_name,
+      prop_delim, kv_delim);
+
+  AppendProperty(result, "property collectors names",
+                 property_collectors_names.empty() ? std::string("N/A")
+                                                   : property_collectors_names,
+                 prop_delim, kv_delim);
+
   return result;
 }
 
@@ -108,6 +122,11 @@ const std::string TablePropertiesNames::kFormatVersion =
     "rocksdb.format.version";
 const std::string TablePropertiesNames::kFixedKeyLen =
     "rocksdb.fixed.key.length";
+const std::string TablePropertiesNames::kComparator = "rocksdb.comparator";
+const std::string TablePropertiesNames::kMergeOperator =
+    "rocksdb.merge.operator";
+const std::string TablePropertiesNames::kPropertyCollectors =
+    "rocksdb.property.collectors";
 
 extern const std::string kPropertiesBlock = "rocksdb.properties";
 // Old property block name for backward compatibility

table/table_test.cc

@@ -48,6 +48,8 @@
 #include "util/testharness.h"
 #include "util/testutil.h"
 
+#include "utilities/merge_operators.h"
+
 namespace rocksdb {
 
 extern const uint64_t kLegacyBlockBasedTableMagicNumber;
@@ -57,6 +59,40 @@ extern const uint64_t kPlainTableMagicNumber;
 
 namespace {
 
+// DummyPropertiesCollector used to test BlockBasedTableProperties
+class DummyPropertiesCollector : public TablePropertiesCollector {
+ public:
+  const char* Name() const { return ""; }
+
+  Status Finish(UserCollectedProperties* properties) { return Status::OK(); }
+
+  Status Add(const Slice& user_key, const Slice& value) { return Status::OK(); }
+
+  virtual UserCollectedProperties GetReadableProperties() const {
+    return UserCollectedProperties{};
+  }
+};
+
+class DummyPropertiesCollectorFactory1
+    : public TablePropertiesCollectorFactory {
+ public:
+  virtual TablePropertiesCollector* CreateTablePropertiesCollector(
+      TablePropertiesCollectorFactory::Context context) {
+    return new DummyPropertiesCollector();
+  }
+  const char* Name() const { return "DummyPropertiesCollector1"; }
+};
+
+class DummyPropertiesCollectorFactory2
+    : public TablePropertiesCollectorFactory {
+ public:
+  virtual TablePropertiesCollector* CreateTablePropertiesCollector(
+      TablePropertiesCollectorFactory::Context context) {
+    return new DummyPropertiesCollector();
+  }
+  const char* Name() const { return "DummyPropertiesCollector2"; }
+};
+
 // Return reverse of "key".
 // Used to test non-lexicographic comparators.
 std::string Reverse(const Slice& key) {
@@ -1018,6 +1054,57 @@ TEST_F(BlockBasedTableTest, BasicBlockBasedTableProperties) {
   ASSERT_EQ(content.size() + kBlockTrailerSize, props.data_size);
 }
 
+TEST_F(BlockBasedTableTest, BlockBasedTableProperties2) {
+  TableConstructor c(&reverse_key_comparator);
+  std::vector<std::string> keys;
+  stl_wrappers::KVMap kvmap;
+
+  {
+    Options options;
+    BlockBasedTableOptions table_options;
+    options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+
+    const ImmutableCFOptions ioptions(options);
+    c.Finish(options, ioptions, table_options,
+             GetPlainInternalComparator(options.comparator), &keys, &kvmap);
+
+    auto& props = *c.GetTableReader()->GetTableProperties();
+
+    // Default comparator
+    ASSERT_EQ("leveldb.BytewiseComparator", props.comparator_name);
+    // No merge operator
+    ASSERT_EQ("N/A", props.merge_operator_name);
+    // No property collectors
+    ASSERT_EQ("", props.property_collectors_names);
+    // No filter policy is used
+    ASSERT_EQ("", props.filter_policy_name);
+  }
+
+  {
+    Options options;
+    BlockBasedTableOptions table_options;
+    options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+    options.comparator = &reverse_key_comparator;
+    options.merge_operator = MergeOperators::CreateUInt64AddOperator();
+    options.table_properties_collector_factories.emplace_back(
+        new DummyPropertiesCollectorFactory1());
+    options.table_properties_collector_factories.emplace_back(
+        new DummyPropertiesCollectorFactory2());
+
+    const ImmutableCFOptions ioptions(options);
+    c.Finish(options, ioptions, table_options,
+             GetPlainInternalComparator(options.comparator), &keys, &kvmap);
+
+    auto& props = *c.GetTableReader()->GetTableProperties();
+
+    ASSERT_EQ("rocksdb.ReverseBytewiseComparator", props.comparator_name);
+    ASSERT_EQ("UInt64AddOperator", props.merge_operator_name);
+    ASSERT_EQ("DummyPropertiesCollector1,DummyPropertiesCollector2",
+              props.property_collectors_names);
+    ASSERT_EQ("", props.filter_policy_name);  // no filter policy is used
+  }
+}
+
 TEST_F(BlockBasedTableTest, FilterPolicyNameProperties) {
   TableConstructor c(BytewiseComparator(), true);
   c.Add("a1", "val1");