Migrate blob handling to new compression APIs (#14234)

Summary:
as part of the effort to get rid of OLD_CompressData and OLD_UncompressData and the old implementations in compression.h.

It's unfortunate the the existing blob file schema doesn't allow storing blobs uncompressed when the compressed version is larger, so we have to work around that.

Note that use of GrowableBuffer in place of std::string is intended to avoid the potential performance overhead of zeroing out memory before overwriting it.

Also includes some cleanup of includes

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

Test Plan:
some unit test updates as needed. Crash test covers integrated blob support.

I'm not too concerned about performance, as until a future schema change, this code is committing the grave performance error of storing compressed data larger than uncompressed.

Reviewed By: mszeszko-meta, hx235

Differential Revision: D90544049

Pulled By: pdillinger

fbshipit-source-id: 2f2ed16de63990b797cc06c8dad36b5869dac302

db/blob/blob_file_builder.cc

@@ -67,6 +67,16 @@ BlobFileBuilder::BlobFileBuilder(
       min_blob_size_(mutable_cf_options->min_blob_size),
       blob_file_size_(mutable_cf_options->blob_file_size),
       blob_compression_type_(mutable_cf_options->blob_compression_type),
+      // TODO: support most CompressionOptions with a new CF option
+      // blob_compression_opts
+      // TODO with schema change: support custom compression manager and options
+      // such as max_compressed_bytes_per_kb
+      // NOTE: returns nullptr for kNoCompression
+      blob_compressor_(GetBuiltinV2CompressionManager()->GetCompressor(
+          CompressionOptions{}, blob_compression_type_)),
+      blob_compressor_wa_(blob_compressor_
+                              ? blob_compressor_->ObtainWorkingArea()
+                              : Compressor::ManagedWorkingArea{}),
       prepopulate_blob_cache_(mutable_cf_options->prepopulate_blob_cache),
       file_options_(file_options),
       write_options_(write_options),
@@ -113,7 +123,7 @@ Status BlobFileBuilder::Add(const Slice& key, const Slice& value,
   }
 
   Slice blob = value;
-  std::string compressed_blob;
+  GrowableBuffer compressed_blob;
 
   {
     const Status s = CompressBlobIfNeeded(&blob, &compressed_blob);
@@ -254,36 +264,27 @@ Status BlobFileBuilder::OpenBlobFileIfNeeded() {
 }
 
 Status BlobFileBuilder::CompressBlobIfNeeded(
-    Slice* blob, std::string* compressed_blob) const {
+    Slice* blob, GrowableBuffer* compressed_blob) const {
   assert(blob);
   assert(compressed_blob);
   assert(compressed_blob->empty());
   assert(immutable_options_);
 
-  if (blob_compression_type_ == kNoCompression) {
+  if (!blob_compressor_) {
+    assert(blob_compression_type_ == kNoCompression);
     return Status::OK();
   }
+  assert(blob_compression_type_ != kNoCompression);
 
-  // TODO: allow user CompressionOptions, including max_compressed_bytes_per_kb
-  CompressionOptions opts;
-  CompressionContext context(blob_compression_type_, opts);
+  // WART: always stored as compressed even when that increases the size.
 
-  CompressionInfo info(opts, context, CompressionDict::GetEmptyDict(),
-                       blob_compression_type_);
-
-  constexpr uint32_t compression_format_version = 2;
-
-  bool success = false;
-
-  {
-    StopWatch stop_watch(immutable_options_->clock, immutable_options_->stats,
-                         BLOB_DB_COMPRESSION_MICROS);
-    success = OLD_CompressData(*blob, info, compression_format_version,
-                               compressed_blob);
-  }
-
-  if (!success) {
-    return Status::Corruption("Error compressing blob");
+  Status s;
+  StopWatch stop_watch(immutable_options_->clock, immutable_options_->stats,
+                       BLOB_DB_COMPRESSION_MICROS);
+  s = LegacyForceBuiltinCompression(*blob_compressor_, &blob_compressor_wa_,
+                                    *blob, compressed_blob);
+  if (!s.ok()) {
+    return s;
   }
 
   *blob = Slice(*compressed_blob);

db/blob/blob_file_builder.h

@@ -10,12 +10,14 @@
 #include <string>
 #include <vector>
 
+#include "rocksdb/advanced_compression.h"
 #include "rocksdb/advanced_options.h"
 #include "rocksdb/compression_type.h"
 #include "rocksdb/env.h"
 #include "rocksdb/options.h"
 #include "rocksdb/rocksdb_namespace.h"
 #include "rocksdb/types.h"
+#include "util/aligned_buffer.h"
 
 namespace ROCKSDB_NAMESPACE {
 
@@ -76,7 +78,8 @@ class BlobFileBuilder {
  private:
   bool IsBlobFileOpen() const;
   Status OpenBlobFileIfNeeded();
-  Status CompressBlobIfNeeded(Slice* blob, std::string* compressed_blob) const;
+  Status CompressBlobIfNeeded(Slice* blob,
+                              GrowableBuffer* compressed_blob) const;
   Status WriteBlobToFile(const Slice& key, const Slice& blob,
                          uint64_t* blob_file_number, uint64_t* blob_offset);
   Status CloseBlobFile();
@@ -91,6 +94,8 @@ class BlobFileBuilder {
   uint64_t min_blob_size_;
   uint64_t blob_file_size_;
   CompressionType blob_compression_type_;
+  std::unique_ptr<Compressor> blob_compressor_;
+  mutable Compressor::ManagedWorkingArea blob_compressor_wa_;
   PrepopulateBlobCache prepopulate_blob_cache_;
   const FileOptions* file_options_;
   const WriteOptions* write_options_;

db/blob/blob_file_builder_test.cc

@@ -457,11 +457,12 @@ TEST_F(BlobFileBuilderTest, CompressionError) {
       nullptr /*IOTracer*/, nullptr /*BlobFileCompletionCallback*/,
       BlobFileCreationReason::kFlush, &blob_file_paths, &blob_file_additions);
 
-  SyncPoint::GetInstance()->SetCallBack("CompressData:TamperWithReturnValue",
-                                        [](void* arg) {
-                                          bool* ret = static_cast<bool*>(arg);
-                                          *ret = false;
-                                        });
+  SyncPoint::GetInstance()->SetCallBack(
+      "LegacyForceBuiltinCompression:TamperWithStatus", [](void* arg) {
+        Status* ret = static_cast<Status*>(arg);
+        ASSERT_OK(*ret);
+        *ret = Status::Corruption("Tampered result");
+      });
   SyncPoint::GetInstance()->EnableProcessing();
 
   constexpr char key[] = "1";
@@ -469,7 +470,7 @@ TEST_F(BlobFileBuilderTest, CompressionError) {
 
   std::string blob_index;
 
-  ASSERT_TRUE(builder.Add(key, value, &blob_index).IsCorruption());
+  ASSERT_EQ(builder.Add(key, value, &blob_index).code(), Status::kCorruption);
 
   SyncPoint::GetInstance()->DisableProcessing();
   SyncPoint::GetInstance()->ClearAllCallBacks();

db/blob/blob_file_reader.cc

@@ -17,10 +17,10 @@
 #include "rocksdb/file_system.h"
 #include "rocksdb/slice.h"
 #include "rocksdb/status.h"
+#include "table/format.h"
 #include "table/multiget_context.h"
 #include "test_util/sync_point.h"
 #include "util/compression.h"
-#include "util/crc32c.h"
 #include "util/stop_watch.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -69,9 +69,16 @@ Status BlobFileReader::Create(
     }
   }
 
-  blob_file_reader->reset(
-      new BlobFileReader(std::move(file_reader), file_size, compression_type,
-                         immutable_options.clock, statistics));
+  std::shared_ptr<Decompressor> decompressor;
+  if (compression_type != kNoCompression) {
+    // The blob format has always used compression format 2
+    decompressor = GetBuiltinV2CompressionManager()->GetDecompressorOptimizeFor(
+        compression_type);
+  }
+
+  blob_file_reader->reset(new BlobFileReader(
+      std::move(file_reader), file_size, compression_type,
+      std::move(decompressor), immutable_options.clock, statistics));
 
   return Status::OK();
 }
@@ -282,11 +289,13 @@ Status BlobFileReader::ReadFromFile(const RandomAccessFileReader* file_reader,
 
 BlobFileReader::BlobFileReader(
     std::unique_ptr<RandomAccessFileReader>&& file_reader, uint64_t file_size,
-    CompressionType compression_type, SystemClock* clock,
+    CompressionType compression_type,
+    std::shared_ptr<Decompressor> decompressor, SystemClock* clock,
     Statistics* statistics)
     : file_reader_(std::move(file_reader)),
       file_size_(file_size),
       compression_type_(compression_type),
+      decompressor_(std::move(decompressor)),
       clock_(clock),
       statistics_(statistics) {
   assert(file_reader_);
@@ -375,8 +384,9 @@ Status BlobFileReader::GetBlob(
   const Slice value_slice(record_slice.data() + adjustment, value_size);
 
   {
-    const Status s = UncompressBlobIfNeeded(
-        value_slice, compression_type, allocator, clock_, statistics_, result);
+    const Status s = UncompressBlobIfNeeded(value_slice, compression_type,
+                                            decompressor_.get(), allocator,
+                                            clock_, statistics_, result);
     if (!s.ok()) {
       return s;
     }
@@ -524,9 +534,9 @@ void BlobFileReader::MultiGetBlob(
 
     // Uncompress blob if needed
     Slice value_slice(record_slice.data() + adjustments[i], req->len);
-    *req->status =
-        UncompressBlobIfNeeded(value_slice, compression_type_, allocator,
-                               clock_, statistics_, &blob_reqs[i].second);
+    *req->status = UncompressBlobIfNeeded(
+        value_slice, compression_type_, decompressor_.get(), allocator, clock_,
+        statistics_, &blob_reqs[i].second);
     if (req->status->ok()) {
       total_bytes += record_slice.size();
     }
@@ -583,8 +593,8 @@ Status BlobFileReader::VerifyBlob(const Slice& record_slice,
 
 Status BlobFileReader::UncompressBlobIfNeeded(
     const Slice& value_slice, CompressionType compression_type,
-    MemoryAllocator* allocator, SystemClock* clock, Statistics* statistics,
-    std::unique_ptr<BlobContents>* result) {
+    Decompressor* decompressor, MemoryAllocator* allocator, SystemClock* clock,
+    Statistics* statistics, std::unique_ptr<BlobContents>* result) {
   assert(result);
 
   if (compression_type == kNoCompression) {
@@ -593,31 +603,33 @@ Status BlobFileReader::UncompressBlobIfNeeded(
     return Status::OK();
   }
 
-  UncompressionContext context(compression_type);
-  UncompressionInfo info(context, UncompressionDict::GetEmptyDict(),
-                         compression_type);
+  assert(decompressor);
 
-  size_t uncompressed_size = 0;
-  constexpr uint32_t compression_format_version = 2;
+  Decompressor::Args args;
+  args.compression_type = compression_type;
+  args.compressed_data = value_slice;
 
-  CacheAllocationPtr output;
+  Status s = decompressor->ExtractUncompressedSize(args);
+  if (!s.ok()) {
+    return Status::Corruption(s.ToString());
+  }
+
+  CacheAllocationPtr output = AllocateBlock(args.uncompressed_size, allocator);
 
   {
     PERF_TIMER_GUARD(blob_decompress_time);
     StopWatch stop_watch(clock, statistics, BLOB_DB_DECOMPRESSION_MICROS);
-    output = OLD_UncompressData(info, value_slice.data(), value_slice.size(),
-                                &uncompressed_size, compression_format_version,
-                                allocator);
+    s = decompressor->DecompressBlock(args, output.get());
   }
 
   TEST_SYNC_POINT_CALLBACK(
-      "BlobFileReader::UncompressBlobIfNeeded:TamperWithResult", &output);
+      "BlobFileReader::UncompressBlobIfNeeded:TamperWithResult", &s);
 
-  if (!output) {
-    return Status::Corruption("Unable to uncompress blob");
+  if (!s.ok()) {
+    return Status::Corruption(s.ToString());
   }
 
-  result->reset(new BlobContents(std::move(output), uncompressed_size));
+  result->reset(new BlobContents(std::move(output), args.uncompressed_size));
 
   return Status::OK();
 }

db/blob/blob_file_reader.h

@@ -10,6 +10,7 @@
 
 #include "db/blob/blob_read_request.h"
 #include "file/random_access_file_reader.h"
+#include "rocksdb/advanced_compression.h"
 #include "rocksdb/compression_type.h"
 #include "rocksdb/rocksdb_namespace.h"
 #include "util/autovector.h"
@@ -64,7 +65,8 @@ class BlobFileReader {
  private:
   BlobFileReader(std::unique_ptr<RandomAccessFileReader>&& file_reader,
                  uint64_t file_size, CompressionType compression_type,
-                 SystemClock* clock, Statistics* statistics);
+                 std::shared_ptr<Decompressor> decompressor, SystemClock* clock,
+                 Statistics* statistics);
 
   static Status OpenFile(const ImmutableOptions& immutable_options,
                          const FileOptions& file_opts,
@@ -96,6 +98,7 @@ class BlobFileReader {
 
   static Status UncompressBlobIfNeeded(const Slice& value_slice,
                                        CompressionType compression_type,
+                                       Decompressor* decompressor,
                                        MemoryAllocator* allocator,
                                        SystemClock* clock,
                                        Statistics* statistics,
@@ -104,6 +107,7 @@ class BlobFileReader {
   std::unique_ptr<RandomAccessFileReader> file_reader_;
   uint64_t file_size_;
   CompressionType compression_type_;
+  std::shared_ptr<Decompressor> decompressor_;
   SystemClock* clock_;
   Statistics* statistics_;
 };

db/blob/blob_file_reader_test.cc

@@ -65,7 +65,7 @@ void WriteBlobFile(const ImmutableOptions& immutable_options,
 
   ASSERT_OK(blob_log_writer.WriteHeader(WriteOptions(), header));
 
-  std::vector<std::string> compressed_blobs(num);
+  std::vector<GrowableBuffer> compressed_blobs(num);
   std::vector<Slice> blobs_to_write(num);
   if (kNoCompression == compression) {
     for (size_t i = 0; i < num; ++i) {
@@ -73,16 +73,13 @@ void WriteBlobFile(const ImmutableOptions& immutable_options,
       blob_sizes[i] = blobs[i].size();
     }
   } else {
-    CompressionOptions opts;
-    CompressionContext context(compression, opts);
-    CompressionInfo info(opts, context, CompressionDict::GetEmptyDict(),
-                         compression);
-
-    constexpr uint32_t compression_format_version = 2;
+    auto compressor =
+        GetBuiltinV2CompressionManager()->GetCompressor({}, compression);
 
     for (size_t i = 0; i < num; ++i) {
-      ASSERT_TRUE(OLD_CompressData(blobs[i], info, compression_format_version,
-                                   &compressed_blobs[i]));
+      ASSERT_OK(LegacyForceBuiltinCompression(*compressor,
+                                              /*working_area=*/nullptr,
+                                              blobs[i], &compressed_blobs[i]));
       blobs_to_write[i] = compressed_blobs[i];
       blob_sizes[i] = compressed_blobs[i].size();
     }
@@ -809,11 +806,10 @@ TEST_F(BlobFileReaderTest, UncompressionError) {
 
   SyncPoint::GetInstance()->SetCallBack(
       "BlobFileReader::UncompressBlobIfNeeded:TamperWithResult", [](void* arg) {
-        CacheAllocationPtr* const output =
-            static_cast<CacheAllocationPtr*>(arg);
-        assert(output);
+        auto* result = static_cast<Status*>(arg);
+        assert(result);
 
-        output->reset();
+        *result = Status::Corruption("Injected result");
       });
 
   SyncPoint::GetInstance()->EnableProcessing();
@@ -824,11 +820,12 @@ TEST_F(BlobFileReaderTest, UncompressionError) {
   std::unique_ptr<BlobContents> value;
   uint64_t bytes_read = 0;
 
-  ASSERT_TRUE(reader
-                  ->GetBlob(ReadOptions(), key, blob_offset, blob_size,
-                            kSnappyCompression, prefetch_buffer, allocator,
-                            &value, &bytes_read)
-                  .IsCorruption());
+  ASSERT_EQ(reader
+                ->GetBlob(ReadOptions(), key, blob_offset, blob_size,
+                          kSnappyCompression, prefetch_buffer, allocator,
+                          &value, &bytes_read)
+                .code(),
+            Status::Code::kCorruption);
   ASSERT_EQ(value, nullptr);
   ASSERT_EQ(bytes_read, 0);
 

db/blob/blob_source_test.cc

@@ -67,7 +67,7 @@ void WriteBlobFile(const ImmutableOptions& immutable_options,
 
   ASSERT_OK(blob_log_writer.WriteHeader(WriteOptions(), header));
 
-  std::vector<std::string> compressed_blobs(num);
+  std::vector<GrowableBuffer> compressed_blobs(num);
   std::vector<Slice> blobs_to_write(num);
   if (kNoCompression == compression) {
     for (size_t i = 0; i < num; ++i) {
@@ -75,16 +75,13 @@ void WriteBlobFile(const ImmutableOptions& immutable_options,
       blob_sizes[i] = blobs[i].size();
     }
   } else {
-    CompressionOptions opts;
-    CompressionContext context(compression, opts);
-    CompressionInfo info(opts, context, CompressionDict::GetEmptyDict(),
-                         compression);
-
-    constexpr uint32_t compression_format_version = 2;
+    auto compressor =
+        GetBuiltinV2CompressionManager()->GetCompressor({}, compression);
 
     for (size_t i = 0; i < num; ++i) {
-      ASSERT_TRUE(OLD_CompressData(blobs[i], info, compression_format_version,
-                                   &compressed_blobs[i]));
+      ASSERT_OK(LegacyForceBuiltinCompression(*compressor,
+                                              /*working_area=*/nullptr,
+                                              blobs[i], &compressed_blobs[i]));
       blobs_to_write[i] = compressed_blobs[i];
       blob_sizes[i] = compressed_blobs[i].size();
     }

table/block_based/block_based_table_builder.cc

@@ -1911,6 +1911,10 @@ Status BlockBasedTableBuilder::CompressAndVerifyBlock(
       assert(type == kNoCompression ||
              r->table_options.verify_compression == (verify_decomp != nullptr));
 
+      TEST_SYNC_POINT_CALLBACK(
+          "BlockBasedTableBuilder::CompressAndVerifyBlock:TamperWithResultType",
+          &type);
+
       // Some of the compression algorithms are known to be unreliable. If
       // the verify_compression flag is set then try to de-compress the
       // compressed data and compare to the input.

table/block_based/index_builder.h

@@ -20,6 +20,7 @@
 #include "table/block_based/block_builder.h"
 #include "table/block_based/flush_block_policy_impl.h"
 #include "table/format.h"
+#include "util/atomic.h"
 
 namespace ROCKSDB_NAMESPACE {
 // The interface for building index.

util/compression.cc

@@ -1776,4 +1776,40 @@ const std::shared_ptr<CompressionManager>& GetBuiltinV2CompressionManager() {
 // END built-in implementation of customization interface
 // ***********************************************************************
 
+Status LegacyForceBuiltinCompression(
+    Compressor& builtin_compressor,
+    Compressor::ManagedWorkingArea* working_area, Slice from,
+    GrowableBuffer* to) {
+  // For legacy cases that store compressed data even when it's larger than the
+  // uncompressed data (!!!), we need a reliable upper bound on the compressed
+  // size. This is based on consulting various algorithms documentation etc.
+  // and adding ~4 bytes for encoded uncompressed size. (Snappy is the worst
+  // case for multiplicative overhead at n + n/6, bounded by 19*n/16 to avoid
+  // costly division. Bzip2 is the worst case for additive overhead at 600
+  // bytes.)
+  size_t n = from.size();
+  size_t upper_bound = ((19 * n) >> 4) + 604;
+  // The upper bound has only been established considering built-in compression
+  // types through kZSTD. (Might need updating if this fails.)
+  assert(builtin_compressor.GetPreferredCompressionType() <= kZSTD);
+
+  to->ResetForSize(upper_bound);
+  CompressionType actual_type = kNoCompression;
+  Status s = builtin_compressor.CompressBlock(
+      from, to->data(), &to->MutableSize(), &actual_type, working_area);
+  TEST_SYNC_POINT_CALLBACK("LegacyForceBuiltinCompression:TamperWithStatus",
+                           &s);
+
+  if (!s.ok()) {
+    return s;
+  }
+  if (actual_type == kNoCompression) {
+    // abort in debug builds
+    assert(actual_type != kNoCompression);
+    return Status::Corruption("Compression unexpectedly declined or aborted");
+  }
+  assert(actual_type == builtin_compressor.GetPreferredCompressionType());
+  return Status::OK();
+}
+
 }  // namespace ROCKSDB_NAMESPACE

util/compression.h

@@ -27,7 +27,7 @@
 #include "rocksdb/options.h"
 #include "table/block_based/block_type.h"
 #include "test_util/sync_point.h"
-#include "util/atomic.h"
+#include "util/aligned_buffer.h"
 #include "util/cast_util.h"
 #include "util/coding.h"
 #include "util/compression_context_cache.h"
@@ -1831,6 +1831,17 @@ const std::shared_ptr<CompressionManager>& GetBuiltinCompressionManager(
 // END built-in implementation of customization interface
 // ***********************************************************************
 
+// The new compression APIs intentionally make it difficult to generate
+// compressed data larger than the original. (It is better to store the
+// uncompressed version in that case.) For legacy cases that must store
+// compressed data even when larger than the uncompressed, this is a convenient
+// wrapper to support that, with a compressor from BuiltinCompressionManager and
+// a GrowableBuffer.
+Status LegacyForceBuiltinCompression(
+    Compressor& builtin_compressor,
+    Compressor::ManagedWorkingArea* working_area, Slice from,
+    GrowableBuffer* to);
+
 // Records the compression type for subsequent WAL records.
 class CompressionTypeRecord {
  public:

util/compression_test.cc

@@ -829,9 +829,10 @@ TEST_P(CompressionFailuresTest, CompressionFailures) {
 
   if (compression_failure_type_ == kTestCompressionFail) {
     ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
-        "CompressData:TamperWithReturnValue", [](void* arg) {
-          bool* ret = static_cast<bool*>(arg);
-          *ret = false;
+        "BlockBasedTableBuilder::CompressAndVerifyBlock:TamperWithResultType",
+        [](void* arg) {
+          CompressionType* ret = static_cast<CompressionType*>(arg);
+          *ret = kNoCompression;
         });
   } else if (compression_failure_type_ == kTestDecompressionFail) {
     ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(

util/simple_mixed_compressor.cc

@@ -15,8 +15,7 @@
 namespace ROCKSDB_NAMESPACE {
 
 // MultiCompressorWrapper implementation
-MultiCompressorWrapper::MultiCompressorWrapper(const CompressionOptions& opts,
-                                               CompressionDict&& dict)
+MultiCompressorWrapper::MultiCompressorWrapper(const CompressionOptions& opts)
     : opts_(opts) {
   // TODO: make the compression manager a field
   auto builtInManager = GetBuiltinV2CompressionManager();
@@ -27,7 +26,6 @@ MultiCompressorWrapper::MultiCompressorWrapper(const CompressionOptions& opts,
     }
     compressors_.push_back(builtInManager->GetCompressor(opts, type));
   }
-  (void)dict;
 }
 
 size_t MultiCompressorWrapper::GetMaxSampleSizeIfWantDict(

util/simple_mixed_compressor.h

@@ -10,15 +10,14 @@
 #include <memory>
 #include <vector>
 
-#include "compression.h"
 #include "rocksdb/advanced_compression.h"
+#include "util/atomic.h"
 
 namespace ROCKSDB_NAMESPACE {
 
 class MultiCompressorWrapper : public Compressor {
  public:
-  explicit MultiCompressorWrapper(const CompressionOptions& opts,
-                                  CompressionDict&& dict = {});
+  explicit MultiCompressorWrapper(const CompressionOptions& opts);
 
   size_t GetMaxSampleSizeIfWantDict(CacheEntryRole block_type) const override;
   Slice GetSerializedDict() const override;

utilities/blob_db/blob_compaction_filter.cc

@@ -123,10 +123,14 @@ CompactionFilter::Decision BlobIndexCompactionFilterBase::HandleValueChange(
     return Decision::kIOError;
   }
   Slice new_blob_value(*new_value);
-  std::string compression_output;
+  GrowableBuffer compressed_output;
   if (blob_db_impl->bdb_options_.compression != kNoCompression) {
-    new_blob_value =
-        blob_db_impl->GetCompressedSlice(new_blob_value, &compression_output);
+    Status s = blob_db_impl->CompressBlob(new_blob_value, &compressed_output);
+    if (!s.ok()) {
+      // Best approximation
+      return Decision::kIOError;
+    }
+    new_blob_value = compressed_output.AsSlice();
   }
   uint64_t new_blob_file_number = 0;
   uint64_t new_blob_offset = 0;
@@ -336,7 +340,7 @@ CompactionFilter::BlobDecision BlobIndexCompactionFilterGC::PrepareBlobOutput(
   assert(blob_db_impl->bdb_options_.enable_garbage_collection);
 
   BlobIndex blob_index;
-  const Status s = blob_index.DecodeFrom(existing_value);
+  Status s = blob_index.DecodeFrom(existing_value);
   if (!s.ok()) {
     gc_stats_.SetError();
     return BlobDecision::kCorruption;
@@ -369,7 +373,7 @@ CompactionFilter::BlobDecision BlobIndexCompactionFilterGC::PrepareBlobOutput(
 
   PinnableSlice blob;
   CompressionType compression_type = kNoCompression;
-  std::string compression_output;
+  GrowableBuffer compressed_output;
   if (!ReadBlobFromOldFile(key, blob_index, &blob, false, &compression_type)) {
     gc_stats_.SetError();
     return BlobDecision::kIOError;
@@ -387,9 +391,11 @@ CompactionFilter::BlobDecision BlobIndexCompactionFilterGC::PrepareBlobOutput(
       }
     }
     if (blob_db_impl->bdb_options_.compression != kNoCompression) {
-      blob_db_impl->GetCompressedSlice(blob, &compression_output);
-      blob = PinnableSlice(&compression_output);
-      blob.PinSelf();
+      s = blob_db_impl->CompressBlob(blob, &compressed_output);
+      if (!s.ok()) {
+        return BlobDecision::kCorruption;
+      }
+      blob.PinSelf(compressed_output.AsSlice());
     }
   }
 

utilities/blob_db/blob_db_impl.cc

@@ -41,10 +41,6 @@
 #include "utilities/blob_db/blob_db_iterator.h"
 #include "utilities/blob_db/blob_db_listener.h"
 
-namespace {
-int kBlockBasedTableVersionFormat = 2;
-}  // end namespace
-
 namespace ROCKSDB_NAMESPACE::blob_db {
 
 bool BlobFileComparator::operator()(
@@ -87,7 +83,10 @@ BlobDBImpl::BlobDBImpl(const std::string& dbname,
       live_sst_size_(0),
       fifo_eviction_seq_(0),
       evict_expiration_up_to_(0),
-      debug_level_(0) {
+      debug_level_(0),
+      // NOTE: returns nullptr for kNoCompression
+      blob_compressor_(GetBuiltinV2CompressionManager()->GetCompressor(
+          CompressionOptions{}, bdb_options_.compression)) {
   clock_ = env_->GetSystemClock().get();
   blob_dir_ = (bdb_options_.path_relative)
                   ? dbname + "/" + bdb_options_.blob_dir
@@ -1087,18 +1086,32 @@ Status BlobDBImpl::PutBlobValue(const WriteOptions& write_options,
       RecordTick(statistics_, BLOB_DB_WRITE_INLINED_TTL);
     }
   } else {
-    std::string compression_output;
-    Slice value_compressed = GetCompressedSlice(value, &compression_output);
+    GrowableBuffer compression_output;
+    Slice value_maybe_compressed;
+    if (blob_compressor_) {
+      assert(bdb_options_.compression != kNoCompression);
+      assert(bdb_options_.compression ==
+             blob_compressor_->GetPreferredCompressionType());
+      s = CompressBlob(value, &compression_output);
+      if (!s.ok()) {
+        return s;
+      }
+      value_maybe_compressed = compression_output.AsSlice();
+    } else {
+      assert(bdb_options_.compression == kNoCompression);
+      value_maybe_compressed = value;
+    }
 
     std::string headerbuf;
-    BlobLogWriter::ConstructBlobHeader(&headerbuf, key, value_compressed,
+    BlobLogWriter::ConstructBlobHeader(&headerbuf, key, value_maybe_compressed,
                                        expiration);
 
     // Check DB size limit before selecting blob file to
     // Since CheckSizeAndEvictBlobFiles() can close blob files, it needs to be
     // done before calling SelectBlobFile().
     s = CheckSizeAndEvictBlobFiles(
-        write_options, headerbuf.size() + key.size() + value_compressed.size());
+        write_options,
+        headerbuf.size() + key.size() + value_maybe_compressed.size());
     if (!s.ok()) {
       return s;
     }
@@ -1112,8 +1125,8 @@ Status BlobDBImpl::PutBlobValue(const WriteOptions& write_options,
     if (s.ok()) {
       assert(blob_file != nullptr);
       assert(blob_file->GetCompressionType() == bdb_options_.compression);
-      s = AppendBlob(write_options, blob_file, headerbuf, key, value_compressed,
-                     expiration, &index_entry);
+      s = AppendBlob(write_options, blob_file, headerbuf, key,
+                     value_maybe_compressed, expiration, &index_entry);
     }
     if (s.ok()) {
       if (expiration != kNoExpiration) {
@@ -1150,26 +1163,16 @@ Status BlobDBImpl::PutBlobValue(const WriteOptions& write_options,
   return s;
 }
 
-Slice BlobDBImpl::GetCompressedSlice(const Slice& raw,
-                                     std::string* compression_output) const {
-  if (bdb_options_.compression == kNoCompression) {
-    return raw;
-  }
+Status BlobDBImpl::CompressBlob(const Slice& raw,
+                                GrowableBuffer* compression_output) const {
   StopWatch compression_sw(clock_, statistics_, BLOB_DB_COMPRESSION_MICROS);
-  CompressionType type = bdb_options_.compression;
-  CompressionOptions opts;
-  CompressionContext context(type, opts);
-  CompressionInfo info(opts, context, CompressionDict::GetEmptyDict(), type);
-  OLD_CompressData(raw, info,
-                   GetCompressFormatForVersion(kBlockBasedTableVersionFormat),
-                   compression_output);
-  return *compression_output;
+  return LegacyForceBuiltinCompression(
+      *blob_compressor_, /*working_area=*/nullptr, raw, compression_output);
 }
 
 Decompressor& BlobDecompressor() {
-  static auto mgr = GetBuiltinCompressionManager(
-      GetCompressFormatForVersion(kBlockBasedTableVersionFormat));
-  static auto decompressor = mgr->GetDecompressor();
+  static auto decompressor =
+      GetBuiltinV2CompressionManager()->GetDecompressor();
 
   return *decompressor;
 }

utilities/blob_db/blob_db_impl.h

@@ -234,8 +234,8 @@ class BlobDBImpl : public BlobDB {
                             PinnableSlice* value,
                             CompressionType* compression_type);
 
-  Slice GetCompressedSlice(const Slice& raw,
-                           std::string* compression_output) const;
+  Status CompressBlob(const Slice& raw,
+                      GrowableBuffer* compression_output) const;
 
   Status DecompressSlice(const Slice& compressed_value,
                          CompressionType compression_type,
@@ -507,6 +507,8 @@ class BlobDBImpl : public BlobDB {
   int disable_file_deletions_ = 0;
 
   uint32_t debug_level_;
+
+  std::unique_ptr<Compressor> blob_compressor_;
 };
 
 Decompressor& BlobDecompressor();