Add pluggable compression

Author: lucagiac81

db/blob/blob_file_builder.h

@@ -12,10 +12,10 @@
 
 #include "rocksdb/advanced_options.h"
 #include "rocksdb/compression_type.h"
+#include "rocksdb/compressor.h"
 #include "rocksdb/env.h"
 #include "rocksdb/rocksdb_namespace.h"
 #include "rocksdb/types.h"
-#include "util/compressor.h"
 
 namespace ROCKSDB_NAMESPACE {
 

db/blob/blob_file_reader.h

@@ -13,7 +13,6 @@
 #include "rocksdb/compression_type.h"
 #include "rocksdb/rocksdb_namespace.h"
 #include "util/autovector.h"
-#include "util/compressor.h"
 
 namespace ROCKSDB_NAMESPACE {
 

db/blob/blob_read_request.h

@@ -8,10 +8,10 @@
 #include <cinttypes>
 
 #include "rocksdb/compression_type.h"
+#include "rocksdb/compressor.h"
 #include "rocksdb/slice.h"
 #include "rocksdb/status.h"
 #include "util/autovector.h"
-#include "util/compressor.h"
 
 namespace ROCKSDB_NAMESPACE {
 

db/blob/blob_source.h

@@ -14,10 +14,10 @@
 #include "db/blob/blob_file_cache.h"
 #include "db/blob/blob_read_request.h"
 #include "rocksdb/cache.h"
+#include "rocksdb/compressor.h"
 #include "rocksdb/rocksdb_namespace.h"
 #include "table/block_based/cachable_entry.h"
 #include "util/autovector.h"
-#include "util/compressor.h"
 
 namespace ROCKSDB_NAMESPACE {
 

db/compact_files_test.cc

@@ -11,12 +11,12 @@
 
 #include "db/db_impl/db_impl.h"
 #include "port/port.h"
+#include "rocksdb/compressor.h"
 #include "rocksdb/db.h"
 #include "rocksdb/env.h"
 #include "test_util/sync_point.h"
 #include "test_util/testharness.h"
 #include "util/cast_util.h"
-#include "util/compressor.h"
 #include "util/string_util.h"
 
 namespace ROCKSDB_NAMESPACE {

db/db_basic_test.cc

@@ -12,6 +12,7 @@
 #include "db/db_test_util.h"
 #include "options/options_helper.h"
 #include "port/stack_trace.h"
+#include "rocksdb/compressor.h"
 #include "rocksdb/filter_policy.h"
 #include "rocksdb/flush_block_policy.h"
 #include "rocksdb/merge_operator.h"
@@ -21,7 +22,6 @@
 #include "table/block_based/block_based_table_reader.h"
 #include "table/block_based/block_builder.h"
 #include "test_util/sync_point.h"
-#include "util/compressor.h"
 #include "util/file_checksum_helper.h"
 #include "util/random.h"
 #include "utilities/counted_fs.h"

db/db_block_cache_test.cc

@@ -20,13 +20,13 @@
 #include "db/db_test_util.h"
 #include "env/unique_id_gen.h"
 #include "port/stack_trace.h"
+#include "rocksdb/compressor.h"
 #include "rocksdb/persistent_cache.h"
 #include "rocksdb/statistics.h"
 #include "rocksdb/table.h"
 #include "rocksdb/table_properties.h"
 #include "table/block_based/block_based_table_reader.h"
 #include "table/unique_id_impl.h"
-#include "util/compressor.h"
 #include "util/defer.h"
 #include "util/hash.h"
 #include "util/math.h"

db/db_options_test.cc

@@ -1107,6 +1107,8 @@ TEST_F(DBOptionsTest, ChangeCompression) {
       "LevelCompactionPicker::PickCompaction:Return", [&](void* arg) {
         Compaction* c = reinterpret_cast<Compaction*>(arg);
         compression_used = c->output_compressor()->GetCompressionType();
+        compression_opt_used =
+            *(c->output_compressor()->GetOptions<CompressionOptions>());
         compacted = true;
       });
   SyncPoint::GetInstance()->EnableProcessing();
@@ -1127,7 +1129,8 @@ TEST_F(DBOptionsTest, ChangeCompression) {
   compression_used = CompressionType::kLZ4Compression;
   compacted = false;
   ASSERT_OK(dbfull()->SetOptions(
-      {{"bottommost_compression", "kSnappyCompression"},
+      {{"bottommost_compressor", "nullptr"},
+       {"bottommost_compression", "kSnappyCompression"},
        {"bottommost_compression_opts", "0:6:0:0:4:true"}}));
   ASSERT_OK(Put("foo", "foofoofoo"));
   ASSERT_OK(Put("bar", "foofoofoo"));
@@ -1138,8 +1141,30 @@ TEST_F(DBOptionsTest, ChangeCompression) {
   ASSERT_OK(dbfull()->TEST_WaitForCompact());
   ASSERT_TRUE(compacted);
   ASSERT_EQ(CompressionType::kSnappyCompression, compression_used);
+  // Snappy compressor does not define level option. Default is returned.
+  ASSERT_EQ(32767, compression_opt_used.level);
   // Right now parallel_level is not yet allowed to be changed.
 
+  if (!Zlib_Supported()) {
+    return;
+  }
+  compression_used = CompressionType::kLZ4Compression;
+  compacted = false;
+  ASSERT_OK(dbfull()->SetOptions(
+      {{"bottommost_compressor", "nullptr"},
+       {"bottommost_compression", "kZlibCompression"},
+       {"bottommost_compression_opts", "0:6:0:0:4:true"}}));
+  ASSERT_OK(Put("foo", "foofoofoo"));
+  ASSERT_OK(Put("bar", "foofoofoo"));
+  ASSERT_OK(Flush());
+  ASSERT_OK(Put("foo", "foofoofoo"));
+  ASSERT_OK(Put("bar", "foofoofoo"));
+  ASSERT_OK(Flush());
+  ASSERT_OK(dbfull()->TEST_WaitForCompact());
+  ASSERT_TRUE(compacted);
+  ASSERT_EQ(CompressionType::kZlibCompression, compression_used);
+  ASSERT_EQ(6, compression_opt_used.level);
+
   SyncPoint::GetInstance()->DisableProcessing();
 }
 

db/db_test2.cc

@@ -18,6 +18,7 @@
 #include "options/options_helper.h"
 #include "port/port.h"
 #include "port/stack_trace.h"
+#include "rocksdb/compressor.h"
 #include "rocksdb/experimental.h"
 #include "rocksdb/iostats_context.h"
 #include "rocksdb/persistent_cache.h"
@@ -26,7 +27,6 @@
 #include "rocksdb/utilities/replayer.h"
 #include "rocksdb/wal_filter.h"
 #include "test_util/testutil.h"
-#include "util/compressor.h"
 #include "util/random.h"
 #include "utilities/fault_injection_env.h"
 

include/rocksdb/advanced_options.h

@@ -17,6 +17,7 @@
 
 namespace ROCKSDB_NAMESPACE {
 
+class Compressor;
 class Slice;
 class SliceTransform;
 class TablePropertiesCollectorFactory;
@@ -526,6 +527,19 @@ struct AdvancedColumnFamilyOptions {
   // Dynamically changeable through SetOptions() API
   std::vector<CompressionType> compression_per_level;
 
+  // Similar to compression_per_level, but the algorithms are encapsulated in
+  // Compressor objects. This adds the ability to select custom compressors,
+  // beyond the built-in ones provided through CompressionType.
+  //
+  // If compressor_per_level is specified (not empty), it overrides
+  // compression_per_level.
+  //
+  // If compressor_per_level is not specified (empty),
+  // compression_per_level is applied as described for that option.
+  //
+  // Default: empty
+  std::vector<std::shared_ptr<Compressor>> compressor_per_level;
+
   // Number of levels for this database
   int num_levels = 7;
 
@@ -1013,6 +1027,19 @@ struct AdvancedColumnFamilyOptions {
   // Dynamically changeable through the SetOptions() API
   CompressionType blob_compression_type = kNoCompression;
 
+  // Similar to blob_compression_type, but the algorithm is encapsulated in a
+  // Compressor class. This adds the ability to select plugin compressors,
+  // beyond the built-in ones provided through CompressionType.
+  //
+  // If blob_compressor is specified (not null), it overrides
+  // blob_compression_type.
+  //
+  // If blob_compressor is not specified (null), blob_compression_type is
+  // applied.
+  //
+  // Default: nullptr
+  std::shared_ptr<Compressor> blob_compressor = nullptr;
+
   // Enables garbage collection of blobs. Blob GC is performed as part of
   // compaction. Valid blobs residing in blob files older than a cutoff get
   // relocated to new files as they are encountered during compaction, which