Store SST file compression algorithm as a TableProperty

Summary: Store SST file compression algorithm as a TableProperty.

Test Plan: Modified and ran the table_test UT that checks for TableProperties

Reviewers: IslamAbdelRahman

Reviewed By: IslamAbdelRahman

Subscribers: lgalanis, andrewkr, dhruba, IslamAbdelRahman

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

include/rocksdb/table_properties.h

@@ -44,6 +44,7 @@ struct TablePropertiesNames {
   static const std::string kComparator;
   static const std::string kMergeOperator;
   static const std::string kPropertyCollectors;
+  static const std::string kCompression;
 };
 
 extern const std::string kPropertiesBlock;
@@ -170,6 +171,9 @@ struct TableProperties {
   // {collector_name[1]},{collector_name[2]},{collector_name[3]} ..
   std::string property_collectors_names;
 
+  // The compression algo used to compress the SST files.
+  std::string compression_name;
+
   // user collected properties
   UserCollectedProperties user_collected_properties;
   UserCollectedProperties readable_properties;

table/block_based_table_builder.cc

@@ -830,6 +830,7 @@ Status BlockBasedTableBuilder::Finish() {
       r->props.merge_operator_name = r->ioptions.merge_operator != nullptr
                                          ? r->ioptions.merge_operator->Name()
                                          : "nullptr";
+      r->props.compression_name = CompressionTypeToString(r->compression_type);
 
       std::string property_collectors_names = "[";
       property_collectors_names = "[";

table/meta_blocks.cc

@@ -88,6 +88,10 @@ void PropertyBlockBuilder::AddTableProperty(const TableProperties& props) {
   if (!props.column_family_name.empty()) {
     Add(TablePropertiesNames::kColumnFamilyName, props.column_family_name);
   }
+
+  if (!props.compression_name.empty()) {
+    Add(TablePropertiesNames::kCompression, props.compression_name);
+  }
 }
 
 Slice PropertyBlockBuilder::Finish() {
@@ -228,6 +232,8 @@ Status ReadProperties(const Slice& handle_value, RandomAccessFileReader* file,
       new_table_properties->merge_operator_name = raw_val.ToString();
     } else if (key == TablePropertiesNames::kPropertyCollectors) {
       new_table_properties->property_collectors_names = raw_val.ToString();
+    } else if (key == TablePropertiesNames::kCompression) {
+      new_table_properties->compression_name = raw_val.ToString();
     } else {
       // handle user-collected properties
       new_table_properties->user_collected_properties.insert(

table/table_properties.cc

@@ -113,6 +113,11 @@ std::string TableProperties::ToString(
                                                    : property_collectors_names,
                  prop_delim, kv_delim);
 
+  AppendProperty(
+      result, "SST file compression algo",
+      compression_name.empty() ? std::string("N/A") : compression_name,
+      prop_delim, kv_delim);
+
   return result;
 }
 
@@ -155,6 +160,7 @@ const std::string TablePropertiesNames::kMergeOperator =
     "rocksdb.merge.operator";
 const std::string TablePropertiesNames::kPropertyCollectors =
     "rocksdb.property.collectors";
+const std::string TablePropertiesNames::kCompression = "rocksdb.compression";
 
 extern const std::string kPropertiesBlock = "rocksdb.properties";
 // Old property block name for backward compatibility

table/table_test.cc

@@ -1071,6 +1071,7 @@ TEST_F(BlockBasedTableTest, BlockBasedTableProperties2) {
 
   {
     Options options;
+    options.compression = CompressionType::kNoCompression;
     BlockBasedTableOptions table_options;
     options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 
@@ -1088,6 +1089,8 @@ TEST_F(BlockBasedTableTest, BlockBasedTableProperties2) {
     ASSERT_EQ("[]", props.property_collectors_names);
     // No filter policy is used
     ASSERT_EQ("", props.filter_policy_name);
+    // Compression type == that set:
+    ASSERT_EQ("NoCompression", props.compression_name);
     c.ResetTableReader();
   }