Publish/support format_version=7, related enhancements (#13713)

Summary:
* Make new format_version=7 a supported setting.
* Fix a bug in compressed_secondary_cache.cc that is newly exercised by custom compression types and showing up in crash test with tiered secondary cache
* Small change to handling of disabled compression in fv=7: use empty compression manager compatibility name.
* Get rid of GetDefaultBuiltinCompressionManager() in public API because it could cause unexpected+unsafe schema change on a user's CompressionManager if built upon the default built-in manager and we add a new built-in schema. Now must be referenced by explicit compression schema version in the public API. (That notion was already exposed in compressed secondary cache API, for better or worse.)
* Improve some error messages for compression misconfiguration
* Improve testing with ObjectLibrary and CompressionManagers
* Improve testing of compression_name table property in BlockBasedTableTest.BlockBasedTableProperties2
* Improve some comments

Pull Request resolved: https://github.com/facebook/rocksdb/pull/13713

Test Plan: existing and updated tests. Notably, the crash test has already been running with (unpublished) format_version=7

Reviewed By: mszeszko-meta, hx235

Differential Revision: D77035482

Pulled By: pdillinger

fbshipit-source-id: 95278de8734a79706a22361bff2184b1edb230ca

cache/compressed_secondary_cache.cc

@@ -80,15 +80,16 @@ std::unique_ptr<SecondaryCacheResultHandle> CompressedSecondaryCache::Lookup(
     ptr = reinterpret_cast<CacheAllocationPtr*>(handle_value);
     handle_value_charge = cache_->GetCharge(lru_handle);
     data_ptr = ptr->get();
-    data_ptr = GetVarint32Ptr(data_ptr, data_ptr + 1,
-                              static_cast<uint32_t*>(&type_32));
+    const char* limit = ptr->get() + handle_value_charge;
+    data_ptr =
+        GetVarint32Ptr(data_ptr, limit, static_cast<uint32_t*>(&type_32));
     type = static_cast<CompressionType>(type_32);
-    data_ptr = GetVarint32Ptr(data_ptr, data_ptr + 1,
-                              static_cast<uint32_t*>(&source_32));
+    data_ptr =
+        GetVarint32Ptr(data_ptr, limit, static_cast<uint32_t*>(&source_32));
     source = static_cast<CacheTier>(source_32);
     uint64_t data_size = 0;
-    data_ptr = GetVarint64Ptr(data_ptr, ptr->get() + handle_value_charge,
-                              static_cast<uint64_t*>(&data_size));
+    data_ptr =
+        GetVarint64Ptr(data_ptr, limit, static_cast<uint64_t*>(&data_size));
     assert(handle_value_charge > data_size);
     handle_value_charge = data_size;
   }

db/column_family.cc

@@ -110,32 +110,48 @@ void GetInternalTblPropCollFactory(
   }
 }
 
-bool CompressionSupportedWithManager(CompressionType type,
-                                     UnownedPtr<CompressionManager> mgr) {
-  return mgr ? mgr->SupportsCompressionType(type)
-             : CompressionTypeSupported(type);
+Status CheckCompressionSupportedWithManager(
+    CompressionType type, UnownedPtr<CompressionManager> mgr) {
+  if (mgr) {
+    if (!mgr->SupportsCompressionType(type)) {
+      return Status::NotSupported("Compression type " +
+                                  CompressionTypeToString(type) +
+                                  " is not recognized/supported by this "
+                                  "version of CompressionManager " +
+                                  mgr->GetId());
+    }
+  } else {
+    if (!CompressionTypeSupported(type)) {
+      if (type <= kLastBuiltinCompression) {
+        return Status::InvalidArgument("Compression type " +
+                                       CompressionTypeToString(type) +
+                                       " is not linked with the binary.");
+      } else {
+        return Status::NotSupported(
+            "Compression type " + CompressionTypeToString(type) +
+            " is not recognized/supported by built-in CompressionManager.");
+      }
+    }
+  }
+  return Status::OK();
 }
 
 Status CheckCompressionSupported(const ColumnFamilyOptions& cf_options) {
   if (!cf_options.compression_per_level.empty()) {
     for (size_t level = 0; level < cf_options.compression_per_level.size();
          ++level) {
-      if (!CompressionSupportedWithManager(
-              cf_options.compression_per_level[level],
-              cf_options.compression_manager.get())) {
-        return Status::InvalidArgument(
-            "Compression type " +
-            CompressionTypeToString(cf_options.compression_per_level[level]) +
-            " is not linked with the binary.");
+      Status s = CheckCompressionSupportedWithManager(
+          cf_options.compression_per_level[level],
+          cf_options.compression_manager.get());
+      if (!s.ok()) {
+        return s;
       }
     }
   } else {
-    if (!CompressionSupportedWithManager(
-            cf_options.compression, cf_options.compression_manager.get())) {
-      return Status::InvalidArgument(
-          "Compression type " +
-          CompressionTypeToString(cf_options.compression) +
-          " is not linked with the binary.");
+    Status s = CheckCompressionSupportedWithManager(
+        cf_options.compression, cf_options.compression_manager.get());
+    if (!s.ok()) {
+      return s;
     }
   }
   if (cf_options.compression_opts.zstd_max_train_bytes > 0) {

db/db_block_cache_test.cc

@@ -833,7 +833,6 @@ TEST_F(DBBlockCacheTest, CacheCompressionDict) {
       GetSupportedDictCompressions();
   Random rnd(301);
   // Format version before and after compression handling changes
-  TEST_AllowUnsupportedFormatVersion() = true;
   for (int format_version : {6, 7}) {
     // Test all supported compression types because (at least historically)
     // dictionary compression could be enabled and a dictionary block saved

db/db_test.cc

@@ -6209,6 +6209,7 @@ TEST_F(DBTest, L0L1L2AndUpHitCounter) {
 }
 
 TEST_F(DBTest, EncodeDecompressedBlockSizeTest) {
+  // Allow testing format_version=1
   bool& allow_unsupported_fv = TEST_AllowUnsupportedFormatVersion();
   SaveAndRestore guard(&allow_unsupported_fv);
   ASSERT_FALSE(allow_unsupported_fv);

db/db_test2.cc

@@ -24,6 +24,7 @@
 #include "rocksdb/persistent_cache.h"
 #include "rocksdb/trace_record.h"
 #include "rocksdb/trace_record_result.h"
+#include "rocksdb/utilities/object_registry.h"
 #include "rocksdb/utilities/replayer.h"
 #include "rocksdb/wal_filter.h"
 #include "test_util/testutil.h"
@@ -1885,8 +1886,8 @@ TEST_F(DBTest2, CompressionOptions) {
 
 TEST_F(DBTest2, RoundRobinManager) {
   if (ZSTD_Supported()) {
-    auto mgr = std::make_shared<RoundRobinManager>(
-        GetDefaultBuiltinCompressionManager());
+    auto mgr =
+        std::make_shared<RoundRobinManager>(GetBuiltinV2CompressionManager());
 
     std::vector<std::string> values;
     for (bool use_wrapper : {true}) {
@@ -1934,7 +1935,7 @@ TEST_F(DBTest2, RoundRobinManager) {
 TEST_F(DBTest2, RandomMixedCompressionManager) {
   if (ZSTD_Supported()) {
     auto mgr = std::make_shared<RandomMixedCompressionManager>(
-        GetDefaultBuiltinCompressionManager());
+        GetBuiltinV2CompressionManager());
     std::vector<std::string> values;
     for (bool use_wrapper : {true}) {
       SCOPED_TRACE((use_wrapper ? "With " : "No ") + std::string("wrapper"));
@@ -2026,7 +2027,7 @@ TEST_F(DBTest2, CompressionManagerWrapper) {
           wrapped_->GetCompressorForSST(context, opts, preferred));
     }
   };
-  auto mgr = std::make_shared<MyManager>(GetDefaultBuiltinCompressionManager());
+  auto mgr = std::make_shared<MyManager>(GetBuiltinV2CompressionManager());
 
   for (CompressionType type : GetSupportedCompressions()) {
     for (bool use_wrapper : {false, true}) {
@@ -2117,8 +2118,7 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     bool SupportsCompressionType(CompressionType type) const override {
       return type == kCustomCompression8A || type == kCustomCompression8B ||
              type == kCustomCompression8C ||
-             GetDefaultBuiltinCompressionManager()->SupportsCompressionType(
-                 type);
+             GetBuiltinV2CompressionManager()->SupportsCompressionType(type);
     }
 
     int used_compressor8A_count_ = 0;
@@ -2139,8 +2139,7 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
           return std::make_unique<Compressor8C>();
         // Also support built-in compression algorithms
         default:
-          return GetDefaultBuiltinCompressionManager()->GetCompressor(opts,
-                                                                      type);
+          return GetBuiltinV2CompressionManager()->GetCompressor(opts, type);
       }
     }
 
@@ -2148,13 +2147,14 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
       return std::make_shared<test::DecompressorCustomAlg>();
     }
 
-    CompressionType last_specific_decompressor_type_ = kNoCompression;
+    RelaxedAtomic<CompressionType> last_specific_decompressor_type_{
+        kNoCompression};
 
     std::shared_ptr<Decompressor> GetDecompressorForTypes(
         const CompressionType* types_begin,
         const CompressionType* types_end) override {
       assert(types_end > types_begin);
-      last_specific_decompressor_type_ = *types_begin;
+      last_specific_decompressor_type_.StoreRelaxed(*types_begin);
       auto decomp = std::make_shared<test::DecompressorCustomAlg>();
       decomp->SetAllowedTypes(types_begin, types_end);
       return decomp;
@@ -2191,9 +2191,11 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     // respect their distinct compatibility names and treat them as incompatible
     // (or else risk processing data incorrectly)
     // NOTE: these are not registered in ObjectRegistry to test what happens
-    // when the original CompressionManager might not be available.
+    // when the original CompressionManager might not be available, but
+    // mgr_bar will be registered during the test, with different names to
+    // prevent interference between iterations.
     auto mgr_foo = std::make_shared<MyManager>("Foo");
-    auto mgr_bar = std::make_shared<MyManager>("Bar");
+    auto mgr_bar = std::make_shared<MyManager>(use_dict ? "Bar1" : "Bar2");
 
     // And this one claims to be fully compatible with the built-in compression
     // manager when it's not fully compatible (for custom CompressionTypes)
@@ -2256,10 +2258,7 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     options.compression = kLZ4Compression;
     ASSERT_EQ(TryReopen(options).code(), Status::Code::kInvalidArgument);
 
-    // TODO: eliminate this hack when format_version=7 is published
-    SaveAndRestore guard(&TEST_AllowUnsupportedFormatVersion(), true);
-
-    // Set new format version
+    // Set format version supporting custom compression
     bbto.format_version = 7;
     options.table_factory.reset(NewBlockBasedTableFactory(bbto));
 
@@ -2269,8 +2268,13 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     options.compression = kCustomCompression8B;
     ASSERT_EQ(TryReopen(options).code(), Status::Code::kInvalidArgument);
 
-    // Using a built-in compression type with fv=7 but named custom schema
+    // Custom compression schema, but specifying a custom compression type it
+    // doesn't support.
     options.compression_manager = mgr_foo;
+    options.compression = kCustomCompressionF0;
+    ASSERT_EQ(TryReopen(options).code(), Status::Code::kNotSupported);
+
+    // Using a built-in compression type with fv=7 but named custom schema
     options.compression = kLZ4Compression;
     Reopen(options);
     ASSERT_OK(
@@ -2279,7 +2283,7 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     ASSERT_EQ(NumTableFilesAtLevel(0), 2);
     ASSERT_EQ(Get("b"), value);
 
-    // Verify it was compressed with snappy
+    // Verify it was compressed with LZ4
     r = {"b", "b0"};
     tables_properties.clear();
     ASSERT_OK(db_->GetPropertiesOfTablesInRange(db_->DefaultColumnFamily(), &r,
@@ -2288,7 +2292,8 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     EXPECT_LT(tables_properties.begin()->second->data_size, kValueSize / 2);
     // Uses new format for "compression_name" property
     EXPECT_EQ(tables_properties.begin()->second->compression_name, "Foo;04;");
-    EXPECT_EQ(mgr_foo->last_specific_decompressor_type_, kLZ4Compression);
+    EXPECT_EQ(mgr_foo->last_specific_decompressor_type_.LoadRelaxed(),
+              kLZ4Compression);
 
     // Custom compression type
     options.compression = kCustomCompression8A;
@@ -2309,7 +2314,8 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     ASSERT_EQ(tables_properties.size(), 1U);
     EXPECT_LT(tables_properties.begin()->second->data_size, kValueSize / 2);
     EXPECT_EQ(tables_properties.begin()->second->compression_name, "Foo;8A;");
-    EXPECT_EQ(mgr_foo->last_specific_decompressor_type_, kCustomCompression8A);
+    EXPECT_EQ(mgr_foo->last_specific_decompressor_type_.LoadRelaxed(),
+              kCustomCompression8A);
 
     // Also dynamically changeable, because the compression manager will respect
     // the current setting as reported under the legacy logic
@@ -2320,7 +2326,7 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     ASSERT_EQ(NumTableFilesAtLevel(0), 4);
     ASSERT_EQ(Get("d"), value);
 
-    // Verify it was compressed with snappy
+    // Verify it was compressed with LZ4
     r = {"d", "d0"};
     tables_properties.clear();
     ASSERT_OK(db_->GetPropertiesOfTablesInRange(db_->DefaultColumnFamily(), &r,
@@ -2328,7 +2334,8 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     ASSERT_EQ(tables_properties.size(), 1U);
     EXPECT_LT(tables_properties.begin()->second->data_size, kValueSize / 2);
     EXPECT_EQ(tables_properties.begin()->second->compression_name, "Foo;04;");
-    EXPECT_EQ(mgr_foo->last_specific_decompressor_type_, kLZ4Compression);
+    EXPECT_EQ(mgr_foo->last_specific_decompressor_type_.LoadRelaxed(),
+              kLZ4Compression);
 
     // Dynamically changeable to custom compressions also
     ASSERT_OK(dbfull()->SetOptions({{"compression", "kCustomCompression8B"}}));
@@ -2346,7 +2353,8 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     ASSERT_EQ(tables_properties.size(), 1U);
     EXPECT_LT(tables_properties.begin()->second->data_size, kValueSize / 2);
     EXPECT_EQ(tables_properties.begin()->second->compression_name, "Foo;8B;");
-    EXPECT_EQ(mgr_foo->last_specific_decompressor_type_, kCustomCompression8B);
+    EXPECT_EQ(mgr_foo->last_specific_decompressor_type_.LoadRelaxed(),
+              kCustomCompression8B);
 
     // Fails to re-open with incompatible compression manager (can't find
     // compression manager Foo because it's not registered nor known by Bar)
@@ -2365,10 +2373,50 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     ASSERT_EQ(Get("d").size(), kValueSize);
     ASSERT_EQ(Get("e").size(), kValueSize);
 
-    // TODO: mix of compatibility names in same DB
+    // Add a file using mgr_bar
+    ASSERT_OK(
+        Put("f", test::CompressibleString(&rnd, 0.1, kValueSize, &value)));
+    ASSERT_OK(Flush());
+    ASSERT_EQ(NumTableFilesAtLevel(0), 6);
+    ASSERT_EQ(Get("f"), value);
+
+    // Verify it was compressed appropriately
+    r = {"f", "f0"};
+    tables_properties.clear();
+    ASSERT_OK(db_->GetPropertiesOfTablesInRange(db_->DefaultColumnFamily(), &r,
+                                                1, &tables_properties));
+    ASSERT_EQ(tables_properties.size(), 1U);
+    EXPECT_LT(tables_properties.begin()->second->data_size, kValueSize / 2);
+    EXPECT_EQ(mgr_bar->last_specific_decompressor_type_.LoadRelaxed(),
+              kLZ4Compression);
+
+    // Fails to re-open with incompatible compression manager (can't find
+    // compression manager Bar because it's not registered nor known by Foo)
+    options.compression_manager = mgr_foo;
+    ASSERT_EQ(TryReopen(options).code(), Status::Code::kNotSupported);
+
+    // Register and re-open
+    auto& library = *ObjectLibrary::Default();
+    library.AddFactory<CompressionManager>(
+        mgr_bar->CompatibilityName(),
+        [mgr_bar](const std::string& /*uri*/,
+                  std::unique_ptr<CompressionManager>* guard,
+                  std::string* /*errmsg*/) {
+          *guard = std::make_unique<MyManager>(mgr_bar->CompatibilityName());
+          return guard->get();
+        });
+    Reopen(options);
+
+    // Can still read everything
+    ASSERT_EQ(Get("a").size(), kValueSize);
+    ASSERT_EQ(Get("b").size(), kValueSize);
+    ASSERT_EQ(Get("c").size(), kValueSize);
+    ASSERT_EQ(Get("d").size(), kValueSize);
+    ASSERT_EQ(Get("e").size(), kValueSize);
+    ASSERT_EQ(Get("f").size(), kValueSize);
+
     // TODO: test old version of a compression manager unable to read a
     // compression type
-    // TODO: test getting compression manager from object registry
   }
 }
 

db_stress_tool/db_stress_compression_manager.h

@@ -41,7 +41,7 @@ class DbStressCustomCompressionManager : public CompressionManager {
             test::CompressorCustomAlg<kCustomCompressionAC>>();
       // Also support built-in compression algorithms
       default:
-        return GetDefaultBuiltinCompressionManager()->GetCompressor(opts, type);
+        return GetBuiltinV2CompressionManager()->GetCompressor(opts, type);
     }
   }
 
@@ -59,7 +59,7 @@ class DbStressCustomCompressionManager : public CompressionManager {
 
  protected:
   std::shared_ptr<CompressionManager> default_ =
-      GetDefaultBuiltinCompressionManager();
+      GetBuiltinV2CompressionManager();
 };
 
 }  // namespace ROCKSDB_NAMESPACE

db_stress_tool/db_stress_test_base.cc

@@ -3435,15 +3435,15 @@ void StressTest::Open(SharedState* shared, bool reopen) {
     options_.compression_manager =
         std::make_shared<DbStressCustomCompressionManager>();
   } else if (!strcasecmp(FLAGS_compression_manager.c_str(), "mixed")) {
-    options_.compression_manager = std::make_shared<RoundRobinManager>(
-        GetDefaultBuiltinCompressionManager());
+    options_.compression_manager =
+        std::make_shared<RoundRobinManager>(GetBuiltinV2CompressionManager());
   } else if (!strcasecmp(FLAGS_compression_manager.c_str(), "randommixed")) {
     options_.compression_manager =
         std::make_shared<RandomMixedCompressionManager>(
-            GetDefaultBuiltinCompressionManager());
+            GetBuiltinV2CompressionManager());
   } else if (!strcasecmp(FLAGS_compression_manager.c_str(), "autoskip")) {
     options_.compression_manager =
-        CreateAutoSkipCompressionManager(GetDefaultBuiltinCompressionManager());
+        CreateAutoSkipCompressionManager(GetBuiltinV2CompressionManager());
   } else if (!strcasecmp(FLAGS_compression_manager.c_str(), "none")) {
     // Nothing to do using default compression manager
   } else {

include/rocksdb/advanced_compression.h

@@ -371,8 +371,9 @@ class CompressionManager
                                  const std::string& id,
                                  std::shared_ptr<CompressionManager>* result);
 
-  // Will this compression type be used if requested in calling
-  // GetCompressor/GetCompressorForSST?
+  // Returns false iff a configuration that would pass the given compression
+  // type to GetCompressor/GetCompressorForSST should be rejected (not
+  // supported)
   virtual bool SupportsCompressionType(CompressionType type) const = 0;
 
   // TODO: function to check compatibility with or sanitize CompressionOptions
@@ -577,11 +578,17 @@ class CompressionManagerWrapper : public CompressionManager {
   std::shared_ptr<CompressionManager> wrapped_;
 };
 
-// Compression manager that implements built-in compression strategy. The
-// behavior of compression_manager=nullptr is essentially equivalent to
-// using this compression manager.
-const std::shared_ptr<CompressionManager>&
-GetDefaultBuiltinCompressionManager();
+// Compression manager that implements the second schema for RocksDB built-in
+// compression support. (The first schema is intentionally not provided here.)
+// *** CURRENT STATE ***
+// This is currently the latest schema for built-in compression, and the
+// compression manager used when compression_manager=nullptr.
+const std::shared_ptr<CompressionManager>& GetBuiltinV2CompressionManager();
+
+// NOTE: No GetLatestBuiltinCompressionManager() is provided because that could
+// lead to unexpected schema changes for user CompressionManagers building on
+// the built-in schema, in the unlikely/rare case of a new built-in schema.
+
 // Gets CompressionManager designed for the automated compression strategy.
 // This may include deciding to compress or not.
 // In future should be able to select compression algorithm based on the CPU

include/rocksdb/compression_type.h

@@ -32,6 +32,7 @@ enum CompressionType : unsigned char {
 
   // For use by user custom CompressionManagers
   kCustomCompression80 = 0x80,
+  kFirstCustomCompression = kCustomCompression80,
   kCustomCompression81 = 0x81,
   kCustomCompression82 = 0x82,
   kCustomCompression83 = 0x83,
@@ -158,6 +159,7 @@ enum CompressionType : unsigned char {
   kCustomCompressionFC = 0xFC,
   kCustomCompressionFD = 0xFD,
   kCustomCompressionFE = 0xFE,
+  kLastCustomCompression = kCustomCompressionFE,
 
   // kDisableCompressionOption is used to disable some compression options.
   kDisableCompressionOption = 0xff,

include/rocksdb/table.h

@@ -561,6 +561,10 @@ struct BlockBasedTableOptions {
   // misplaced within or between files is as likely to fail checksum
   // verification as random corruption. Also checksum-protects SST footer.
   // Can be read by RocksDB versions >= 8.6.0.
+  // 7 -- Support for custom compression algorithms with a CompressionManager
+  // using a non-built-in CompatibilityName(). See `compression_manager` in
+  // ColumnFamilyOptions. Also changes the format of TableProperties field
+  // `compression_name`. Can be read by RocksDB versions >= 10.4.0.
   //
   // Using the default setting of format_version is strongly recommended, so
   // that available enhancements are adopted eventually and automatically. The

include/rocksdb/table_properties.h

@@ -355,7 +355,13 @@ struct TableProperties {
   // behave, this must be set to "ZSTD" if any blocks are compressed
   // with zstd and must NOT be set to "NoCompression" if any blocks are
   // compressed.
-  // * For format_version >= 7, it is ...
+  // * For format_version >= 7, the format is
+  //   <compatibility_name>;<hex-coded compression types>;<future use>
+  // where <compatibility_name> is the CompatibilityName() of the
+  // CompressionManager used for the file, or empty if compression was
+  // disabled; <hex-coded compression types> represents a sorted set of
+  // CompressionType values used in the file other than kNoCompression, each
+  // as 2-digit hex, e.g. 04 for LZ$, 07 for ZSTD, etc.
   std::string compression_name;
 
   // Compression options used to compress the SST files.

options/options_settable_test.cc

@@ -705,10 +705,9 @@ TEST_F(OptionsSettableTest, ColumnFamilyOptionsAllFieldsSettable) {
       12345);
   // TODO: try to enhance ObjectLibrary to support singletons
   // ASSERT_EQ(new_options->compression_manager,
-  //           GetBuiltinCompressionManager(/*compression_format_version*/ 2));
-  ASSERT_STREQ(
-      new_options->compression_manager->Name(),
-      GetBuiltinCompressionManager(/*compression_format_version*/ 2)->Name());
+  //           GetBuiltinV2CompressionManager());
+  ASSERT_STREQ(new_options->compression_manager->Name(),
+               GetBuiltinV2CompressionManager()->Name());
 
   ColumnFamilyOptions rnd_filled_options = *new_options;
 

table/block_based/block_based_table_builder.cc

@@ -977,7 +977,10 @@ struct BlockBasedTableBuilder::Rep {
       assert(mgr);
       // Use newer compression_name property
       props.compression_name.reserve(32);
-      props.compression_name.append(mgr->CompatibilityName());
+      // If compression is disabled, use empty manager name
+      if (basic_compressor) {
+        props.compression_name.append(mgr->CompatibilityName());
+      }
       props.compression_name.push_back(';');
       // Rest of property to be filled out at the end of building the file
     } else {

table/block_based/block_based_table_reader.cc

@@ -606,7 +606,8 @@ Status GetDecompressor(const std::string& compression_name,
       if (!s.ok()) {
         return s;
       }
-      assert(mgr_to_use);
+      assert(mgr_to_use || compatibility_name == kNullptrString ||
+             compatibility_name.empty());
     }
 
     // Second field is set of compression types actually used in the file
@@ -632,9 +633,15 @@ Status GetDecompressor(const std::string& compression_name,
       }
       ctypes[i] = static_cast<CompressionType>(val);
     }
-    *out_decompressor =
-        mgr_to_use->GetDecompressorForTypes(ctypes.get(), ctypes.get() + count);
-    assert(*out_decompressor || count == 0);
+    if (mgr_to_use) {
+      *out_decompressor = mgr_to_use->GetDecompressorForTypes(
+          ctypes.get(), ctypes.get() + count);
+      assert(*out_decompressor || count == 0);
+    } else {
+      // Compression/decompression disabled
+      *out_decompressor = nullptr;
+      assert(count == 0);
+    }
     // Can ignore possible additional future fields
   } else {
     // No explicit CompressionManager, e.g. legacy file support where

table/format.h

@@ -157,10 +157,15 @@ inline uint32_t GetCompressFormatForVersion(uint32_t format_version) {
   // As of format_version 2, we encode compressed block with
   // compress_format_version == 2. Before that, the version is 1.
   // DO NOT CHANGE THIS FUNCTION, it affects disk format
+  // As of format_version 7 and opening up to custom compression, the
+  // compression format version is essentially independent of the block-based
+  // table format version, and encoded in the compression_name table property.
+  // Thus, this function can go away once we remove support for reading
+  // format_version=1.
   return format_version >= 2 ? 2 : 1;
 }
 
-constexpr uint32_t kLatestFormatVersion = 6;
+constexpr uint32_t kLatestFormatVersion = 7;
 
 inline bool IsSupportedFormatVersion(uint32_t version) {
   return version <= kLatestFormatVersion;

table/table_test.cc

@@ -1795,18 +1795,23 @@ TEST_P(BlockBasedTableTest, IndexUncompressed) {
 #endif  // SNAPPY
 
 TEST_P(BlockBasedTableTest, BlockBasedTableProperties2) {
-  TableConstructor c(&reverse_key_comparator);
+  TableConstructor c(&reverse_key_comparator,
+                     true /* convert_to_internal_key_ */);
   std::vector<std::string> keys;
   stl_wrappers::KVMap kvmap;
 
-  {
+  for (CompressionType ct : {kNoCompression, kSnappyCompression}) {
+    if (!Snappy_Supported() && ct == kSnappyCompression) {
+      continue;
+    }
     Options options;
-    options.compression = CompressionType::kNoCompression;
+    options.compression = ct;
     BlockBasedTableOptions table_options = GetBlockBasedTableOptions();
     options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 
     const ImmutableOptions ioptions(options);
     const MutableCFOptions moptions(options);
+    c.Add("blah", std::string(200, 'x'));  // something to compress
     c.Finish(options, ioptions, moptions, table_options,
              GetPlainInternalComparator(options.comparator), &keys, &kvmap);
 
@@ -1823,7 +1828,13 @@ TEST_P(BlockBasedTableTest, BlockBasedTableProperties2) {
     // No filter policy is used
     ASSERT_EQ("", props.filter_policy_name);
     // Compression type == that set:
-    ASSERT_EQ("NoCompression", props.compression_name);
+    if (FormatVersionUsesCompressionManagerName(table_options.format_version)) {
+      ASSERT_EQ(ct == kNoCompression ? ";;" : "BuiltinV2;01;",
+                props.compression_name);
+    } else {
+      ASSERT_EQ(ct == kNoCompression ? "NoCompression" : "Snappy",
+                props.compression_name);
+    }
     c.ResetTableReader();
   }
 

test_util/testutil.h

@@ -740,9 +740,9 @@ template <CompressionType kCompression>
 struct CompressorCustomAlg : public CompressorWrapper {
   static bool Supported() { return LZ4_Supported(); }
 
-  explicit CompressorCustomAlg(std::unique_ptr<Compressor> wrapped =
-                                   GetDefaultBuiltinCompressionManager()
-                                       ->GetCompressor({}, kLZ4Compression))
+  explicit CompressorCustomAlg(
+      std::unique_ptr<Compressor> wrapped =
+          GetBuiltinV2CompressionManager()->GetCompressor({}, kLZ4Compression))
       : CompressorWrapper(std::move(wrapped)),
         dictionary_hash_(GetSliceHash(wrapped_->GetSerializedDict())) {
     static_assert(kCompression > kLastBuiltinCompression);
@@ -750,12 +750,16 @@ struct CompressorCustomAlg : public CompressorWrapper {
 
   const char* Name() const override { return "CompressorCustomAlg"; }
 
+  CompressionType GetPreferredCompressionType() const override {
+    return kCompression;
+  }
+
   Status CompressBlock(Slice uncompressed_data, std::string* compressed_output,
                        CompressionType* out_compression_type,
                        ManagedWorkingArea* working_area) override {
     Status s = wrapped_->CompressBlock(uncompressed_data, compressed_output,
                                        out_compression_type, working_area);
-    if (*out_compression_type != kNoCompression) {
+    if (s.ok() && *out_compression_type != kNoCompression) {
       assert(*out_compression_type == kLZ4Compression);
       std::string header(/*size=*/5, 0);
       header[0] = lossless_cast<char>(kCompression);
@@ -783,9 +787,8 @@ struct CompressorCustomAlg : public CompressorWrapper {
 struct DecompressorCustomAlg : public DecompressorWrapper {
   using TypeSet = SmallEnumSet<CompressionType, kDisableCompressionOption>;
 
-  DecompressorCustomAlg(
-      std::shared_ptr<Decompressor> wrapped =
-          GetDefaultBuiltinCompressionManager()->GetDecompressor())
+  DecompressorCustomAlg(std::shared_ptr<Decompressor> wrapped =
+                            GetBuiltinV2CompressionManager()->GetDecompressor())
       : DecompressorWrapper(std::move(wrapped)),
         dictionary_hash_(GetSliceHash(wrapped_->GetSerializedDict())),
         allowed_types_(TypeSet::All()) {}
@@ -808,7 +811,8 @@ struct DecompressorCustomAlg : public DecompressorWrapper {
   }
 
   Status ExtractUncompressedSize(Args& args) override {
-    if (args.compression_type > kLastBuiltinCompression) {
+    if (args.compression_type >= kFirstCustomCompression &&
+        args.compression_type <= kLastCustomCompression) {
       assert(args.compressed_data.size() > 0);
       assert(args.compressed_data[0] ==
              lossless_cast<char>(args.compression_type));
@@ -828,7 +832,8 @@ struct DecompressorCustomAlg : public DecompressorWrapper {
   }
 
   Status DecompressBlock(const Args& args, char* uncompressed_output) override {
-    if (args.compression_type > kLastBuiltinCompression) {
+    if (args.compression_type >= kFirstCustomCompression &&
+        args.compression_type <= kLastCustomCompression) {
       // Also allowed to copy args and modify
       Args modified_args = args;
       modified_args.compression_type = kLZ4Compression;

tools/db_bench_tool.cc

@@ -4653,7 +4653,7 @@ class Benchmark {
         options.bottommost_compression = kZSTD;
 
         mgr = std::make_shared<RoundRobinManager>(
-            GetDefaultBuiltinCompressionManager());
+            GetBuiltinV2CompressionManager());
       } else if (!strcasecmp(FLAGS_compression_manager.c_str(), "autoskip")) {
         options.compression = FLAGS_compression_type_e;
         if (FLAGS_compression_type_e == kNoCompression) {
@@ -4662,8 +4662,8 @@ class Benchmark {
                   "autoskip");
           ErrorExit();
         }
-        mgr = CreateAutoSkipCompressionManager(
-            GetDefaultBuiltinCompressionManager());
+        mgr =
+            CreateAutoSkipCompressionManager(GetBuiltinV2CompressionManager());
       } else {
         // not defined -> exit with error
         fprintf(stderr, "Requested compression manager not supported");

tools/ldb_cmd.cc

@@ -870,8 +870,8 @@ bool LDBCommand::ParseCompressionTypeOption(
       }
       options_.compression = kZSTD;
       options_.bottommost_compression = kZSTD;
-      auto mgr = std::make_shared<RoundRobinManager>(
-          GetDefaultBuiltinCompressionManager());
+      auto mgr =
+          std::make_shared<RoundRobinManager>(GetBuiltinV2CompressionManager());
       options_.compression_manager = mgr;
 
       // Need to list zstd in the compression_name table property if it's

unreleased_history/new_features/format_version_7.md

@@ -0,0 +1 @@
+* Add new `format_version=7` to aid experimental support of custom compression algorithms with CompressionManager and block-based table. This format version includes changing the format of `TableProperties::compression_name`.

util/auto_skip_compressor.cc

@@ -126,6 +126,6 @@ std::unique_ptr<Compressor> AutoSkipCompressorManager::GetCompressorForSST(
 std::shared_ptr<CompressionManagerWrapper> CreateAutoSkipCompressionManager(
     std::shared_ptr<CompressionManager> wrapped) {
   return std::make_shared<AutoSkipCompressorManager>(
-      wrapped == nullptr ? GetDefaultBuiltinCompressionManager() : wrapped);
+      wrapped == nullptr ? GetBuiltinV2CompressionManager() : wrapped);
 }
 }  // namespace ROCKSDB_NAMESPACE

util/compression.cc

@@ -1043,8 +1043,7 @@ const std::shared_ptr<CompressionManager>& GetBuiltinCompressionManager(
   }
 }
 
-const std::shared_ptr<CompressionManager>&
-GetDefaultBuiltinCompressionManager() {
+const std::shared_ptr<CompressionManager>& GetBuiltinV2CompressionManager() {
   return GetBuiltinCompressionManager(2);
 }
 

util/compression.h

@@ -754,8 +754,11 @@ inline std::string CompressionTypeToString(CompressionType compression_type) {
     case kDisableCompressionOption:
       return "DisableOption";
     default: {
-      char c = lossless_cast<char>(compression_type);
-      return "Custom" + Slice(&c, 1).ToString(/*hex=*/true);
+      bool is_custom = compression_type >= kFirstCustomCompression &&
+                       compression_type <= kLastCustomCompression;
+      unsigned char c = lossless_cast<unsigned char>(compression_type);
+      return (is_custom ? "Custom" : "Reserved") +
+             ToBaseCharsString<16>(2, c, /*uppercase=*/true);
     }
   }
 }

util/compression_test.cc

@@ -158,7 +158,7 @@ class DBAutoSkip : public DBTestBase {
         rnd_(231),
         key_index_(0) {
     options.compression_manager =
-        CreateAutoSkipCompressionManager(GetDefaultBuiltinCompressionManager());
+        CreateAutoSkipCompressionManager(GetBuiltinV2CompressionManager());
     auto statistics = ROCKSDB_NAMESPACE::CreateDBStatistics();
     options.statistics = statistics;
     options.statistics->set_stats_level(StatsLevel::kExceptTimeForMutex);

util/simple_mixed_compressor.cc

@@ -17,7 +17,8 @@ namespace ROCKSDB_NAMESPACE {
 // MultiCompressorWrapper implementation
 MultiCompressorWrapper::MultiCompressorWrapper(const CompressionOptions& opts,
                                                CompressionDict&& dict) {
-  auto builtInManager = GetDefaultBuiltinCompressionManager();
+  // TODO: make the compression manager a field
+  auto builtInManager = GetBuiltinV2CompressionManager();
   const auto& compressions = GetSupportedCompressions();
   for (auto type : compressions) {
     if (type == kNoCompression) {

util/slice_test.cc

@@ -14,6 +14,7 @@
 #include "test_util/testharness.h"
 #include "test_util/testutil.h"
 #include "util/cast_util.h"
+#include "util/string_util.h"
 
 namespace ROCKSDB_NAMESPACE {
 
@@ -410,6 +411,17 @@ TEST(UnownedPtrTest, Tests) {
   }
 }
 
+TEST(ToBaseCharsStringTest, Tests) {
+  using ROCKSDB_NAMESPACE::ToBaseCharsString;
+  // Base 16
+  ASSERT_EQ(ToBaseCharsString<16>(5, 0, true), "00000");
+  ASSERT_EQ(ToBaseCharsString<16>(5, 42, true), "0002A");
+  ASSERT_EQ(ToBaseCharsString<16>(5, 42, false), "0002a");
+  ASSERT_EQ(ToBaseCharsString<16>(2, 255, false), "ff");
+  // Base 32
+  ASSERT_EQ(ToBaseCharsString<32>(2, 255, false), "7v");
+}
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

util/string_util.h

@@ -40,6 +40,16 @@ inline void PutBaseChars(char** buf, size_t n, uint64_t v, bool uppercase) {
   *buf += n;
 }
 
+// Construct a string of n digits from v in base kBase
+template <size_t kBase>
+inline std::string ToBaseCharsString(size_t n, uint64_t v, bool uppercase) {
+  std::string result;
+  result.resize(n);
+  char* buf = &result[0];
+  PutBaseChars<kBase>(&buf, n, v, uppercase);
+  return result;
+}
+
 // Parse n digits from *buf in base kBase to *v and advance *buf to the
 // position after what was read. On success, true is returned. On failure,
 // false is returned, *buf is placed at the first bad character, and *v