Add support for Zstd-based compression in LevelDB.

This change implements support for Zstd-based compression in LevelDB. Building
up from the Snappy compression (which has been supported since inception), this
change adds Zstd as an alternate compression algorithm.

We are implementing this to provide alternative options for users who might
have different performance and efficiency requirements. For instance, the
Zstandard website (https://facebook.github.io/zstd/) claims that the Zstd
algorithm can achieve around 30% higher compression ratios than Snappy, with
relatively smaller (~10%) slowdowns in de/compression speeds.

Benchmarking results:

$ blaze-bin/third_party/leveldb/db_bench
LevelDB:    version 1.23
Date:       Thu Feb  2 18:50:06 2023
CPU:        56 * Intel(R) Xeon(R) CPU E5-2690 v4 @ 2.60GHz
CPUCache:   35840 KB
Keys:       16 bytes each
Values:     100 bytes each (50 bytes after compression)
Entries:    1000000
RawSize:    110.6 MB (estimated)
FileSize:   62.9 MB (estimated)
------------------------------------------------
fillseq      :       2.613 micros/op;   42.3 MB/s
fillsync     :    3924.432 micros/op;    0.0 MB/s (1000 ops)
fillrandom   :       3.609 micros/op;   30.7 MB/s
overwrite    :       4.508 micros/op;   24.5 MB/s
readrandom   :       6.136 micros/op; (864322 of 1000000 found)
readrandom   :       5.446 micros/op; (864083 of 1000000 found)
readseq      :       0.180 micros/op;  613.3 MB/s
readreverse  :       0.321 micros/op;  344.7 MB/s
compact      :  827043.000 micros/op;
readrandom   :       4.603 micros/op; (864105 of 1000000 found)
readseq      :       0.169 micros/op;  656.3 MB/s
readreverse  :       0.315 micros/op;  350.8 MB/s
fill100K     :     854.009 micros/op;  111.7 MB/s (1000 ops)
crc32c       :       1.227 micros/op; 3184.0 MB/s (4K per op)
snappycomp   :       3.610 micros/op; 1081.9 MB/s (output: 55.2%)
snappyuncomp :       0.691 micros/op; 5656.3 MB/s
zstdcomp     :      15.731 micros/op;  248.3 MB/s (output: 44.1%)
zstduncomp   :       4.218 micros/op;  926.2 MB/s
PiperOrigin-RevId: 509957778

3 files changed, 56 insertions, 8 deletions

diff --git a/table/format.cc b/table/format.cc
index e183977..7647372 100644
--- a/table/format.cc
+++ b/table/format.cc
@@ -5,6 +5,7 @@
 #include "table/format.h"
 
 #include "leveldb/env.h"
+#include "leveldb/options.h"
 #include "port/port.h"
 #include "table/block.h"
 #include "util/coding.h"
@@ -116,13 +117,31 @@ Status ReadBlock(RandomAccessFile* file, const ReadOptions& options,
       size_t ulength = 0;
       if (!port::Snappy_GetUncompressedLength(data, n, &ulength)) {
         delete[] buf;
-        return Status::Corruption("corrupted compressed block contents");
+        return Status::Corruption("corrupted snappy compressed block length");
       }
       char* ubuf = new char[ulength];
       if (!port::Snappy_Uncompress(data, n, ubuf)) {
         delete[] buf;
         delete[] ubuf;
-        return Status::Corruption("corrupted compressed block contents");
+        return Status::Corruption("corrupted snappy compressed block contents");
+      }
+      delete[] buf;
+      result->data = Slice(ubuf, ulength);
+      result->heap_allocated = true;
+      result->cachable = true;
+      break;
+    }
+    case kZstdCompression: {
+      size_t ulength = 0;
+      if (!port::Zstd_GetUncompressedLength(data, n, &ulength)) {
+        delete[] buf;
+        return Status::Corruption("corrupted zstd compressed block length");
+      }
+      char* ubuf = new char[ulength];
+      if (!port::Zstd_Uncompress(data, n, ubuf)) {
+        delete[] buf;
+        delete[] ubuf;
+        return Status::Corruption("corrupted zstd compressed block contents");
       }
       delete[] buf;
       result->data = Slice(ubuf, ulength);
diff --git a/table/table_builder.cc b/table/table_builder.cc
index 29a619d..ba3df9e 100644
--- a/table/table_builder.cc
+++ b/table/table_builder.cc
@@ -168,6 +168,20 @@ void TableBuilder::WriteBlock(BlockBuilder* block, BlockHandle* handle) {
       }
       break;
     }
+
+    case kZstdCompression: {
+      std::string* compressed = &r->compressed_output;
+      if (port::Zstd_Compress(raw.data(), raw.size(), compressed) &&
+          compressed->size() < raw.size() - (raw.size() / 8u)) {
+        block_contents = *compressed;
+      } else {
+        // Zstd not supported, or compressed less than 12.5%, so just
+        // store uncompressed form
+        block_contents = raw;
+        type = kNoCompression;
+      }
+      break;
+    }
   }
   WriteRawBlock(block_contents, type, handle);
   r->compressed_output.clear();
diff --git a/table/table_test.cc b/table/table_test.cc
index a405586..b3baf95 100644
--- a/table/table_test.cc
+++ b/table/table_test.cc
@@ -14,6 +14,7 @@
 #include "leveldb/db.h"
 #include "leveldb/env.h"
 #include "leveldb/iterator.h"
+#include "leveldb/options.h"
 #include "leveldb/table_builder.h"
 #include "table/block.h"
 #include "table/block_builder.h"
@@ -784,15 +785,29 @@ TEST(TableTest, ApproximateOffsetOfPlain) {
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("xyz"), 610000, 612000));
 }
 
-static bool SnappyCompressionSupported() {
+static bool CompressionSupported(CompressionType type) {
   std::string out;
   Slice in = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
-  return port::Snappy_Compress(in.data(), in.size(), &out);
+  if (type == kSnappyCompression) {
+    return port::Snappy_Compress(in.data(), in.size(), &out);
+  } else if (type == kZstdCompression) {
+    return port::Zstd_Compress(in.data(), in.size(), &out);
+  }
+  return false;
 }
 
-TEST(TableTest, ApproximateOffsetOfCompressed) {
-  if (!SnappyCompressionSupported())
-    GTEST_SKIP() << "skipping compression tests";
+class CompressionTableTest
+    : public ::testing::TestWithParam<std::tuple<CompressionType>> {};
+
+INSTANTIATE_TEST_SUITE_P(CompressionTests, CompressionTableTest,
+                         ::testing::Values(kSnappyCompression,
+                                           kZstdCompression));
+
+TEST_P(CompressionTableTest, ApproximateOffsetOfCompressed) {
+  CompressionType type = ::testing::get<0>(GetParam());
+  if (!CompressionSupported(type)) {
+    GTEST_SKIP() << "skipping compression test: " << type;
+  }
 
   Random rnd(301);
   TableConstructor c(BytewiseComparator());
@@ -805,7 +820,7 @@ TEST(TableTest, ApproximateOffsetOfCompressed) {
   KVMap kvmap;
   Options options;
   options.block_size = 1024;
-  options.compression = kSnappyCompression;
+  options.compression = type;
   c.Finish(options, &keys, &kvmap);
 
   // Expected upper and lower bounds of space used by compressible strings.