Compression API clarifcations/minor fixes (facebook#13775)

Summary:
* A number of comments clarifying contracts, etc.
* Make ReleaseWorkingArea public instead of protected because there are some limited cases where a wrapper implementation might want to call it directly
* Check non-empty dictionary precondition on MaybeCloneForDict
* Expand testing of wrapped WorkingAreas
* Random documentation improvement in block_builder.cc

Pull Request resolved: facebook#13775

Test Plan: existing and expanded tests and assertions

Reviewed By: hx235

Differential Revision: D78304550

Pulled By: pdillinger

fbshipit-source-id: e5f064e8405a5a49be123ee13145cb3626bbbfbf

Author: pdillinger

include/rocksdb/advanced_compression.h

@@ -50,6 +50,10 @@ struct FilterBuildingContext;
 // a number of built-in CompressionTypes that ignore any dictionary block in
 // the file; therefore they cannot accommodate dictionary compression in the
 // future without a schema change / extension.)
+//
+// Exceptions MUST NOT propagate out of overridden functions into RocksDB,
+// because RocksDB is not exception-safe. This could cause undefined behavior
+// including data loss, unreported corruption, deadlocks, and more.
 class Compressor {
  public:
   Compressor() = default;
@@ -134,15 +138,15 @@ class Compressor {
   // EXTENSIBLE or reinterpret_cast-able by custom Compressor implementations
   struct WorkingArea {};
 
- protected:
   // To allow for flexible re-use / reclaimation, we have explicit Get and
   // Release functions, and usually wrap in a special RAII smart pointer.
   // For example, a WorkingArea could be saved/recycled in thread-local or
   // core-local storage, or heap managed, etc., though an explicit WorkingArea
   // is only advised for repeated compression (by a single thread).
+  // ReleaseWorkingArea() in not intended to be called directly, but used by
+  // ManagedWorkingArea.
   virtual void ReleaseWorkingArea(WorkingArea*) {}
 
- public:
   using ManagedWorkingArea =
       ManagedPtr<WorkingArea, Compressor, &Compressor::ReleaseWorkingArea>;
 
@@ -221,6 +225,10 @@ class Compressor {
 // decompressed into part of a single buffer allocated to hold a block's
 // uncompressed contents along with an in-memory object representation of the
 // block (to reduce fragmentation and other overheads of separate objects).
+//
+// Exceptions MUST NOT propagate out of overridden functions into RocksDB,
+// because RocksDB is not exception-safe. This could cause undefined behavior
+// including data loss, unreported corruption, deadlocks, and more.
 class Decompressor {
  public:
   Decompressor() = default;
@@ -278,6 +286,9 @@ class Decompressor {
   // supported for this kind of Decompressor. Corruption - dictionary is
   // malformed (though many implementations will accept any data as a
   // dictionary)
+  //
+  // RocksDB promises not to call this function with an empty dictionary slice
+  // (equivalent to no dictionary).
   virtual Status MaybeCloneForDict(const Slice& /*serialized_dict*/,
                                    std::unique_ptr<Decompressor>* /*out*/) {
     return Status::NotSupported(
@@ -339,6 +350,10 @@ class Decompressor {
 //   (because that would break backward compatibility, potential quiet
 //   corruption)
 // TODO: consider adding optional streaming compression support (low priority)
+//
+// Exceptions MUST NOT propagate out of overridden functions into RocksDB,
+// because RocksDB is not exception-safe. This could cause undefined behavior
+// including data loss, unreported corruption, deadlocks, and more.
 class CompressionManager
     : public std::enable_shared_from_this<CompressionManager>,
       public Customizable {
@@ -466,6 +481,10 @@ class CompressorWrapper : public Compressor {
     return wrapped_->ObtainWorkingArea();
   }
 
+  // NOTE: Don't need to override ReleaseWorkingArea() here because
+  // ManagedWorkingArea takes care of calling it on the Compressor that created
+  // the WorkingArea.
+
   Status CompressBlock(Slice uncompressed_data, std::string* compressed_output,
                        CompressionType* out_compression_type,
                        ManagedWorkingArea* working_area) override {
@@ -491,6 +510,10 @@ class DecompressorWrapper : public Decompressor {
     wrapped_->ReleaseWorkingArea(wa);
   }
 
+  // NOTE: Don't need to override ReleaseWorkingArea() here because
+  // ManagedWorkingArea takes care of calling it on the Decompressor that
+  // created the WorkingArea.
+
   ManagedWorkingArea ObtainWorkingArea(CompressionType preferred) override {
     return wrapped_->ObtainWorkingArea(preferred);
   }

table/block_based/block_builder.cc

@@ -21,15 +21,19 @@
 // An entry for a particular key-value pair has the form:
 //     shared_bytes: varint32
 //     unshared_bytes: varint32
-//     value_length: varint32
+//     value_length: varint32 (NOTE1)
 //     key_delta: char[unshared_bytes]
 //     value: char[value_length]
-// shared_bytes == 0 for restart points.
+// shared_bytes == 0 (explicitly stored) for restart points.
 //
 // The trailer of the block has the form:
 //     restarts: uint32[num_restarts]
 //     num_restarts: uint32
 // restarts[i] contains the offset within the block of the ith restart point.
+//
+// NOTE1: omitted for format_version >= 4 index blocks, because the value is
+// composed of one (shared_bytes > 0) or two (shared_bytes == 0) varints, whose
+// length is self-describing.
 
 #include "table/block_based/block_builder.h"
 

util/compression.cc

@@ -746,6 +746,8 @@ class BuiltinDecompressorV2WithDict : public BuiltinDecompressorV2 {
 
 Status BuiltinDecompressorV2::MaybeCloneForDict(
     const Slice& dict, std::unique_ptr<Decompressor>* out) {
+  // Check RocksDB-promised precondition
+  assert(dict.size() > 0);
   // Because of unfortunate decisions in handling built-in compression types,
   // all the compression types before ZSTD that do not actually support
   // dictionary compression pretend to support it. Specifically, we have to be

util/compression.h

@@ -240,15 +240,25 @@ struct DecompressorDict {
 
  private:
   void Populate(Decompressor& from_decompressor, Slice dict) {
-    Status s = from_decompressor.MaybeCloneForDict(dict, &decompressor_);
-    if (decompressor_ == nullptr) {
+    if (UNLIKELY(dict.empty())) {
       dict_str_ = {};
       dict_allocation_ = {};
-      assert(!s.ok());
-      decompressor_ = std::make_unique<FailureDecompressor>(std::move(s));
+      // Appropriately reject bad files with empty dictionary block.
+      // It is longstanding not to write an empty dictionary block:
+      // https://github.com/facebook/rocksdb/blame/10.2.fb/table/block_based/block_based_table_builder.cc#L1841
+      decompressor_ = std::make_unique<FailureDecompressor>(
+          Status::Corruption("Decompression dictionary is empty"));
     } else {
-      assert(s.ok());
-      assert(decompressor_->GetSerializedDict() == dict);
+      Status s = from_decompressor.MaybeCloneForDict(dict, &decompressor_);
+      if (decompressor_ == nullptr) {
+        dict_str_ = {};
+        dict_allocation_ = {};
+        assert(!s.ok());
+        decompressor_ = std::make_unique<FailureDecompressor>(std::move(s));
+      } else {
+        assert(s.ok());
+        assert(decompressor_->GetSerializedDict() == dict);
+      }
     }
 
     memory_usage_ = sizeof(struct DecompressorDict);

util/compression_test.cc

@@ -1124,6 +1124,27 @@ TEST_F(DBCompressionTest, CompressionManagerWrapper) {
                                        out_compression_type, working_area);
       }
     }
+
+    // Also check WorkingArea handling
+    struct MyWorkingArea : public WorkingArea {
+      explicit MyWorkingArea(ManagedWorkingArea&& wrapped)
+          : wrapped_(std::move(wrapped)) {}
+      ManagedWorkingArea wrapped_;
+    };
+    ManagedWorkingArea ObtainWorkingArea() override {
+      ManagedWorkingArea rv{
+          new MyWorkingArea{CompressorWrapper::ObtainWorkingArea()}, this};
+      if (GetPreferredCompressionType() == kZSTD) {
+        // ZSTD should always use WorkingArea, so this is our chance to ensure
+        // CompressorWrapper::ObtainWorkingArea() is properly connected
+        assert(rv.get() != nullptr);
+      }
+      return rv;
+    }
+
+    void ReleaseWorkingArea(WorkingArea* wa) override {
+      delete static_cast<MyWorkingArea*>(wa);
+    }
   };
   struct MyManager : public CompressionManagerWrapper {
     using CompressionManagerWrapper::CompressionManagerWrapper;