Fix regression in LZ4 compression performance since 10.6 (facebook#14017)

pdillinger · pdillinger · commit c44a28384747 · 2025-10-02T09:16:59.000-07:00
Summary: In RocksDB 10.6 with facebook#13805, due to inaccurate testing of an async system, it went undetected at the time that LZ4 compression was using more CPU despite making a change to reuse stream objects which dramatically improved LZ4HC compression efficiency. This change switches to using a basic LZ4 compress API which appears to be faster than all of these: * Legacy behavior of creating LZ4_stream_t for each compression * 10.6-10.7 behavior of re-using streams between compressions for the same file (with stream-as-WorkingArea) * using LZ4's extState APIs without streams (with extState-as-WorkingArea) (data not shown in below results) Also in this PR: more improvements to sst_dump --recompress, which is arguably the best SST construction benchmark right now since db_bench seems to be so noisy due to backgroun flush+compaction, even with no compaction (FIFO). Streamlined some output and added a SST read time test, mostly for decompression performance. Pull Request resolved: facebook#14017 Test Plan: Performance test using sst_dump --recompress with newer sst_dump back-ported to 10.5: ``` ./sst_dump --command=recompress --compression_types=kLZ4Compression test5.sst --compression_level_from=-6 --compression_level_to=-1 ``` and with default compression level. 10.5: ``` Cx level: -6 Cx size: 61608137 Write usec: 880404 Cx level: -5 Cx size: 60793749 Write usec: 840903 Cx level: -4 Cx size: 58134030 Write usec: 836365 Cx level: -3 Cx size: 55193773 Write usec: 857113 Cx level: -2 Cx size: 54013891 Write usec: 855642 Cx level: -1 Cx size: 50400393 Write usec: 865194 Cx level: 32767 Cx size: 50400393 Write usec: 886310 ``` Before this change (showing the regression, more time, from 10.6: ``` Cx level: -6 Cx size: 61608137 Write usec: 933448 Cx level: -5 Cx size: 60793749 Write usec: 893826 Cx level: -4 Cx size: 58134030 Write usec: 891138 Cx level: -3 Cx size: 55193773 Write usec: 898461 Cx level: -2 Cx size: 54013891 Write usec: 897485 Cx level: -1 Cx size: 50400393 Write usec: 936970 Cx level: 32767 Cx size: 50400393 Write usec: 958764 ``` After this change (faster than both the above): ``` Cx level: -6 Cx size: 63641883 Write usec: 874190 Cx level: -5 Cx size: 58860032 Write usec: 834662 Cx level: -4 Cx size: 57150188 Write usec: 832707 Cx level: -3 Cx size: 58791894 Write usec: 850305 Cx level: -2 Cx size: 53145885 Write usec: 839574 Cx level: -1 Cx size: 49809139 Write usec: 845639 Cx level: 32767 Cx size: 49809139 Write usec: 875199 ``` Similar tests with dictionary compression show essentially no difference (need to use stream APIs and reuse doesn't seem to matter). LZ4HC also unaffected (still improved vs. 10.5) Reviewed By: hx235 Differential Revision: D83722880 Pulled By: pdillinger fbshipit-source-id: 30149dd187686d5dd98321e6aa7d74bd7653a905
diff --git a/table/sst_file_dumper.cc b/table/sst_file_dumper.cc
@@ -231,7 +231,9 @@ Status SstFileDumper::DumpTable(const std::string& out_filename) {
 }
 
 Status SstFileDumper::CalculateCompressedTableSize(
-    const TableBuilderOptions& tb_options, TableProperties* props) {
+    const TableBuilderOptions& tb_options, TableProperties* props,
+    std::chrono::microseconds* write_time,
+    std::chrono::microseconds* read_time) {
   std::unique_ptr<Env> env(NewMemEnv(options_.env));
   std::unique_ptr<WritableFileWriter> dest_writer;
   Status s =
@@ -240,6 +242,8 @@ Status SstFileDumper::CalculateCompressedTableSize(
   if (!s.ok()) {
     return s;
   }
+  std::chrono::steady_clock::time_point start =
+      std::chrono::steady_clock::now();
   std::unique_ptr<TableBuilder> table_builder{
       tb_options.moptions.table_factory->NewTableBuilder(tb_options,
                                                          dest_writer.get())};
@@ -253,17 +257,69 @@ Status SstFileDumper::CalculateCompressedTableSize(
   if (!s.ok()) {
     return s;
   }
+  iter.reset();
   s = table_builder->Finish();
+  *write_time = std::chrono::duration_cast<std::chrono::microseconds>(
+      std::chrono::steady_clock::now() - start);
   if (!s.ok()) {
     return s;
   }
+  s = dest_writer->Close({});
+  if (!s.ok()) {
+    return s;
+  }
+  dest_writer.reset();
   *props = table_builder->GetTableProperties();
+  start = std::chrono::steady_clock::now();
+  TableReaderOptions reader_options(ioptions_, moptions_.prefix_extractor,
+                                    moptions_.compression_manager.get(),
+                                    soptions_, internal_comparator_,
+                                    0 /* block_protection_bytes_per_key */);
+  std::unique_ptr<RandomAccessFileReader> file_reader;
+  s = RandomAccessFileReader::Create(env->GetFileSystem(), testFileName,
+                                     soptions_, &file_reader, /*dbg=*/nullptr);
+  if (!s.ok()) {
+    return s;
+  }
+  std::unique_ptr<TableReader> table_reader;
+  s = tb_options.moptions.table_factory->NewTableReader(
+      reader_options, std::move(file_reader), table_builder->FileSize(),
+      &table_reader);
+  if (!s.ok()) {
+    return s;
+  }
+  iter.reset(table_reader->NewIterator(
+      read_options_, moptions_.prefix_extractor.get(), /*arena=*/nullptr,
+      /*skip_filters=*/false, TableReaderCaller::kSSTDumpTool));
+  for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
+  }
+  s = iter->status();
+  if (!s.ok()) {
+    return s;
+  }
+  iter.reset();
+  table_reader.reset();
+  file_reader.reset();
+  *read_time = std::chrono::duration_cast<std::chrono::microseconds>(
+      std::chrono::steady_clock::now() - start);
   return env->DeleteFile(testFileName);
 }
 
 Status SstFileDumper::ShowAllCompressionSizes(
     const std::vector<CompressionType>& compression_types,
     int32_t compress_level_from, int32_t compress_level_to) {
+#ifndef NDEBUG
+  fprintf(stdout,
+          "WARNING: Assertions are enabled; benchmarks unnecessarily slow\n");
+#endif
+  BlockBasedTableOptions bbto;
+  if (options_.table_factory->IsInstanceOf(
+          TableFactory::kBlockBasedTableName())) {
+    bbto = *(static_cast_with_check<BlockBasedTableFactory>(
+                 options_.table_factory.get()))
+                ->GetOptions<BlockBasedTableOptions>();
+  }
+
   for (CompressionType ctype : compression_types) {
     std::string cname;
     if (!GetStringFromCompressionType(&cname, ctype).ok()) {
@@ -273,10 +329,12 @@ Status SstFileDumper::ShowAllCompressionSizes(
     if (options_.compression_manager
             ? options_.compression_manager->SupportsCompressionType(ctype)
             : CompressionTypeSupported(ctype)) {
-      fprintf(stdout, "Compression: %-24s\n", cname.c_str());
       CompressionOptions compress_opt = options_.compression_opts;
+      fprintf(stdout,
+              "Compression: %-24s Block Size: %" PRIu64 "  Threads: %u\n",
+              cname.c_str(), bbto.block_size, compress_opt.parallel_threads);
       for (int32_t j = compress_level_from; j <= compress_level_to; j++) {
-        fprintf(stdout, "Compression level: %d", j);
+        fprintf(stdout, "Cx level: %d", j);
         compress_opt.level = j;
         Status s = ShowCompressionSize(ctype, compress_opt);
         if (!s.ok()) {
@@ -320,27 +378,26 @@ Status SstFileDumper::ShowCompressionSize(
       TablePropertiesCollectorFactory::Context::kUnknownColumnFamily,
       column_family_name, unknown_level, kUnknownNewestKeyTime);
   TableProperties props;
-  std::chrono::steady_clock::time_point start =
-      std::chrono::steady_clock::now();
-  Status s = CalculateCompressedTableSize(tb_opts, &props);
+  std::chrono::microseconds write_time;
+  std::chrono::microseconds read_time;
+  Status s =
+      CalculateCompressedTableSize(tb_opts, &props, &write_time, &read_time);
   if (!s.ok()) {
     return s;
   }
 
   uint64_t num_data_blocks = props.num_data_blocks;
 
-  std::chrono::steady_clock::time_point end = std::chrono::steady_clock::now();
-  fprintf(stdout, " Comp size: %10" PRIu64, props.data_size);
-  fprintf(stdout, " Uncompressed: %10" PRIu64, props.uncompressed_data_size);
+  fprintf(stdout, " Cx size: %10" PRIu64, props.data_size);
+  fprintf(stdout, " Uncx size: %10" PRIu64, props.uncompressed_data_size);
   fprintf(stdout, " Ratio: %10s",
           std::to_string(static_cast<double>(props.uncompressed_data_size) /
                          static_cast<double>(props.data_size))
               .c_str());
-  fprintf(stdout, " Microsecs: %10s ",
-          std::to_string(
-              std::chrono::duration_cast<std::chrono::microseconds>(end - start)
-                  .count())
-              .c_str());
+  fprintf(stdout, " Write usec: %10s ",
+          std::to_string(write_time.count()).c_str());
+  fprintf(stdout, " Read usec: %10s ",
+          std::to_string(read_time.count()).c_str());
   const uint64_t compressed_blocks =
       opts.statistics->getAndResetTickerCount(NUMBER_BLOCK_COMPRESSED);
   const uint64_t not_compressed_blocks =
@@ -370,11 +427,11 @@ Status SstFileDumper::ShowCompressionSize(
                              : ((static_cast<double>(not_compressed_blocks) /
                                  static_cast<double>(num_data_blocks)) *
                                 100.0);
-  fprintf(stdout, " Comp count: %6" PRIu64 " (%5.1f%%)", compressed_blocks,
+  fprintf(stdout, " Cx count: %6" PRIu64 " (%5.1f%%)", compressed_blocks,
           compressed_pcnt);
-  fprintf(stdout, " Not compressed (ratio): %6" PRIu64 " (%5.1f%%)",
+  fprintf(stdout, " Not cx for ratio: %6" PRIu64 " (%5.1f%%)",
           ratio_not_compressed_blocks, ratio_not_compressed_pcnt);
-  fprintf(stdout, " Not compressed (abort): %6" PRIu64 " (%5.1f%%)\n",
+  fprintf(stdout, " Not cx otherwise: %6" PRIu64 " (%5.1f%%)\n",
           not_compressed_blocks, not_compressed_pcnt);
   return Status::OK();
 }
diff --git a/table/sst_file_dumper.h b/table/sst_file_dumper.h
@@ -59,7 +59,9 @@ class SstFileDumper {
                              FilePrefetchBuffer* prefetch_buffer);
 
   Status CalculateCompressedTableSize(const TableBuilderOptions& tb_options,
-                                      TableProperties* props);
+                                      TableProperties* props,
+                                      std::chrono::microseconds* write_time,
+                                      std::chrono::microseconds* read_time);
 
   Status SetTableOptionsByMagicNumber(uint64_t table_magic_number);
   Status SetOldTableOptions();
diff --git a/tools/sst_dump_tool.cc b/tools/sst_dump_tool.cc
@@ -181,7 +181,6 @@ int SSTDumpTool::Run(int argc, char const* const* argv, Options options) {
   bool list_meta_blocks = false;
   bool has_compression_level_from = false;
   bool has_compression_level_to = false;
-  bool has_specified_compression_types = false;
   std::string from_key;
   std::string to_key;
   std::string block_size_str;
@@ -258,7 +257,6 @@ int SSTDumpTool::Run(int argc, char const* const* argv, Options options) {
       std::string compression_types_csv = argv[i] + 20;
       std::istringstream iss(compression_types_csv);
       std::string compression_type;
-      has_specified_compression_types = true;
 
       while (std::getline(iss, compression_type, ',')) {
         auto iter =
@@ -392,12 +390,7 @@ int SSTDumpTool::Run(int argc, char const* const* argv, Options options) {
     }
   }
 
-  if (has_compression_level_from && has_compression_level_to) {
-    if (!has_specified_compression_types || compression_types.size() != 1) {
-      fprintf(stderr, "Specify one compression type.\n\n");
-      exit(1);
-    }
-  } else if (has_compression_level_from || has_compression_level_to) {
+  if (has_compression_level_from ^ has_compression_level_to) {
     fprintf(stderr,
             "Specify both --compression_level_from and "
             "--compression_level_to.\n\n");
@@ -536,14 +529,20 @@ int SSTDumpTool::Run(int argc, char const* const* argv, Options options) {
     }
 
     if (command == "recompress") {
-      fprintf(stdout, "Block Size: %zu  Threads: %u\n", block_size,
-              (unsigned)compression_parallel_threads);
-      // TODO: consider getting supported compressions from the compression
-      // manager
+      if (compression_types.empty()) {
+        if (options.compression_manager != nullptr) {
+          for (int c = 0; c < kDisableCompressionOption; ++c) {
+            if (options.compression_manager->SupportsCompressionType(
+                    static_cast<CompressionType>(c))) {
+              compression_types.emplace_back(static_cast<CompressionType>(c));
+            }
+          }
+        } else {
+          compression_types = GetSupportedCompressions();
+        }
+      }
       st = dumper.ShowAllCompressionSizes(
-          compression_types.empty() ? GetSupportedCompressions()
-                                    : compression_types,
-          compress_level_from, compress_level_to);
+          compression_types, compress_level_from, compress_level_to);
       if (!st.ok()) {
         fprintf(stderr, "Failed to recompress: %s\n", st.ToString().c_str());
         exit(1);
diff --git a/unreleased_history/performance_improvements/lz4.md b/unreleased_history/performance_improvements/lz4.md
@@ -0,0 +1 @@
+* Fixed a performance regression in LZ4 compression that started in version 10.6.0
diff --git a/util/compression.cc b/util/compression.cc
@@ -516,7 +516,7 @@ class BuiltinBZip2CompressorV2 : public CompressorWithSimpleDictBase {
   }
 };
 
-class BuiltinLZ4CompressorV2 : public CompressorWithSimpleDictBase {
+class BuiltinLZ4CompressorV2WithDict : public CompressorWithSimpleDictBase {
  public:
   using CompressorWithSimpleDictBase::CompressorWithSimpleDictBase;
 
@@ -527,8 +527,8 @@ class BuiltinLZ4CompressorV2 : public CompressorWithSimpleDictBase {
   }
 
   std::unique_ptr<Compressor> CloneForDict(std::string&& dict_data) override {
-    return std::make_unique<BuiltinLZ4CompressorV2>(opts_,
-                                                    std::move(dict_data));
+    return std::make_unique<BuiltinLZ4CompressorV2WithDict>(
+        opts_, std::move(dict_data));
   }
 
   ManagedWorkingArea ObtainWorkingArea() override {
@@ -611,6 +611,72 @@ class BuiltinLZ4CompressorV2 : public CompressorWithSimpleDictBase {
   }
 };
 
+class BuiltinLZ4CompressorV2NoDict : public BuiltinLZ4CompressorV2WithDict {
+ public:
+  BuiltinLZ4CompressorV2NoDict(const CompressionOptions& opts)
+      : BuiltinLZ4CompressorV2WithDict(opts, /*dict_data=*/{}) {}
+
+  ManagedWorkingArea ObtainWorkingArea() override {
+    // Using an LZ4_stream_t between compressions and resetting with
+    // LZ4_resetStream_fast is actually slower than using a fresh LZ4_stream_t
+    // each time, or not involving a stream at all. Similarly, using an extState
+    // does not seem to offer a performance boost, perhaps a small regression.
+    return {};
+  }
+
+  void ReleaseWorkingArea(WorkingArea* wa) override {
+    // Should not be called
+    (void)wa;
+    assert(wa == nullptr);
+  }
+
+  Status CompressBlock(Slice uncompressed_data, char* compressed_output,
+                       size_t* compressed_output_size,
+                       CompressionType* out_compression_type,
+                       ManagedWorkingArea* wa) override {
+#ifdef LZ4
+    (void)wa;
+    auto [alg_output, alg_max_output_size] = StartCompressBlockV2(
+        uncompressed_data, compressed_output, *compressed_output_size);
+    if (alg_max_output_size == 0) {
+      // Compression bypassed
+      *compressed_output_size = 0;
+      *out_compression_type = kNoCompression;
+      return Status::OK();
+    }
+    int acceleration;
+    if (opts_.level < 0) {
+      acceleration = -opts_.level;
+    } else {
+      acceleration = 1;
+    }
+    auto outlen =
+        LZ4_compress_fast(uncompressed_data.data(), alg_output,
+                          static_cast<int>(uncompressed_data.size()),
+                          static_cast<int>(alg_max_output_size), acceleration);
+    if (outlen > 0) {
+      // Compression kept/successful
+      size_t output_size = static_cast<size_t>(
+          outlen + /*header size*/ (alg_output - compressed_output));
+      assert(output_size <= *compressed_output_size);
+      *compressed_output_size = output_size;
+      *out_compression_type = kLZ4Compression;
+      return Status::OK();
+    }
+    // Compression rejected
+    *compressed_output_size = 1;
+#else
+    (void)uncompressed_data;
+    (void)compressed_output;
+    (void)wa;
+    // Compression bypassed (not supported)
+    *compressed_output_size = 0;
+#endif
+    *out_compression_type = kNoCompression;
+    return Status::OK();
+  }
+};
+
 class BuiltinLZ4HCCompressorV2 : public CompressorWithSimpleDictBase {
  public:
   using CompressorWithSimpleDictBase::CompressorWithSimpleDictBase;
@@ -1508,7 +1574,7 @@ class BuiltinCompressionManagerV2 : public CompressionManager {
       case kBZip2Compression:
         return std::make_unique<BuiltinBZip2CompressorV2>(opts);
       case kLZ4Compression:
-        return std::make_unique<BuiltinLZ4CompressorV2>(opts);
+        return std::make_unique<BuiltinLZ4CompressorV2NoDict>(opts);
       case kLZ4HCCompression:
         return std::make_unique<BuiltinLZ4HCCompressorV2>(opts);
       case kXpressCompression:

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	`+* Fixed a performance regression in LZ4 compression that started in version 10.6.0`