perf(cpp): reuse the LZ4 decompression buffer across pages

cpp/src/compress/compressor_factory.h

@@ -114,7 +114,17 @@ class CompressorFactory {
         }
     }
 
-    static void free(Compressor* c) { common::mem_free(c); }
+    // Compressors are placement-new'd into mem_alloc() storage, so freeing
+    // that storage alone skips ~Compressor().  Any resource a compressor
+    // retains outside its own object (e.g. LZ4's reused decompression
+    // buffer) would leak; run the (virtual) destructor first.
+    static void free(Compressor* c) {
+        if (c == nullptr) {
+            return;
+        }
+        c->~Compressor();
+        common::mem_free(c);
+    }
 };
 
 }  // end namespace storage

cpp/src/compress/lz4_compressor.cc

@@ -18,6 +18,8 @@
  */
 #include "lz4_compressor.h"
 
+#include <limits>
+
 #include "common/allocator/alloc_base.h"
 
 using namespace common;
@@ -34,6 +36,7 @@ void LZ4Compressor::destroy() {
         mem_free(uncompressed_buf_);
         uncompressed_buf_ = nullptr;
     }
+    uncompressed_buf_capacity_ = 0;
 }
 
 int LZ4Compressor::reset(bool for_compress) {
@@ -109,42 +112,51 @@ int LZ4Compressor::uncompress(char* compressed_buf, uint32_t compressed_buf_len,
                               char*& uncompressed_buf,
                               uint32_t& uncompressed_buf_len, float ratio) {
     int ret = E_OK;
-    char* regen_buffer = nullptr;
-    int src_size = 0;
-    if (ratio > 0) {
-        src_size = (int)(compressed_buf_len * ratio);
-        regen_buffer = (char*)mem_alloc(src_size, MOD_COMPRESSOR_OBJ);
-    } else {
-        ret = E_COMPRESS_ERR;
+    if (ratio <= 0) {
+        return E_COMPRESS_ERR;
     }
-
-    if (regen_buffer != nullptr) {
-        int decompressed_size = LZ4_decompress_safe(
-            compressed_buf, regen_buffer, compressed_buf_len, src_size);
-
-        if (decompressed_size < 0) {
-            // Release regen_buffer since decompress failed
-            mem_free(regen_buffer);
-            regen_buffer = nullptr;
-            ret = E_COMPRESS_ERR;
-        } else {
-            uncompressed_buf = regen_buffer;
-            uncompressed_buf_ = regen_buffer;
-            uncompressed_buf_len = decompressed_size;
+    // Compute the requested capacity in double and check it before converting:
+    // compressed_buf_len * ratio can exceed INT_MAX (the last retry ratio is
+    // 255x), and an out-of-range double-to-int conversion is undefined.
+    const double requested_size =
+        static_cast<double>(compressed_buf_len) * ratio;
+    if (requested_size > static_cast<double>(std::numeric_limits<int>::max())) {
+        return E_COMPRESS_ERR;
+    }
+    int src_size = static_cast<int>(requested_size);
+    // Reuse the persistent buffer across pages; (re)allocate only when the
+    // requested capacity grows. This avoids one malloc + free per page, which
+    // dominates the read path's allocation churn.
+    if (uncompressed_buf_ == nullptr ||
+        src_size > static_cast<int>(uncompressed_buf_capacity_)) {
+        if (uncompressed_buf_ != nullptr) {
+            mem_free(uncompressed_buf_);
+            uncompressed_buf_ = nullptr;
         }
+        uncompressed_buf_ =
+            (char*)mem_alloc((size_t)src_size, MOD_COMPRESSOR_OBJ);
+        if (uncompressed_buf_ == nullptr) {
+            uncompressed_buf_capacity_ = 0;
+            return E_OOM;
+        }
+        uncompressed_buf_capacity_ = static_cast<uint32_t>(src_size);
+    }
+
+    int decompressed_size = LZ4_decompress_safe(
+        compressed_buf, uncompressed_buf_, compressed_buf_len, src_size);
+    if (decompressed_size < 0) {
+        ret = E_COMPRESS_ERR;
     } else {
-        ret = E_OOM;
+        uncompressed_buf = uncompressed_buf_;
+        uncompressed_buf_len = static_cast<uint32_t>(decompressed_size);
     }
     return ret;
 }
 
 void LZ4Compressor::after_uncompress(char* uncompressed_buf) {
-    if (uncompressed_buf != nullptr) {
-        mem_free(uncompressed_buf);
-        if (uncompressed_buf_ == uncompressed_buf) {
-            uncompressed_buf_ = nullptr;
-        }
-    }
+    // The decompression buffer is owned by this compressor and reused across
+    // pages; it is freed in destroy(). Do not free it per page.
+    (void)uncompressed_buf;
 }
 
 }  // end namespace storage

cpp/src/compress/lz4_compressor.h

@@ -38,10 +38,16 @@ namespace storage {
 
 class LZ4Compressor : public Compressor {
    public:
-    LZ4Compressor() : compressed_buf_(nullptr), uncompressed_buf_(nullptr){};
-    ~LZ4Compressor(){};
+    LZ4Compressor()
+        : compressed_buf_(nullptr),
+          uncompressed_buf_(nullptr),
+          uncompressed_buf_capacity_(0){};
+    ~LZ4Compressor() { destroy(); }
+    // Owns the reused decompression buffer; a copy would double-free it.
+    LZ4Compressor(const LZ4Compressor&) = delete;
+    LZ4Compressor& operator=(const LZ4Compressor&) = delete;
     // @for_compress
-    //  true  - for compressiom
+    //  true  - for compression
     //  false - for uncompression
     int reset(bool for_compress) OVERRIDE;
     void destroy() OVERRIDE;
@@ -55,7 +61,10 @@ class LZ4Compressor : public Compressor {
 
    private:
     char* compressed_buf_;
+    // Reused across uncompress() calls; owned by this compressor and freed in
+    // destroy(). uncompressed_buf_capacity_ tracks its allocated size.
     char* uncompressed_buf_;
+    uint32_t uncompressed_buf_capacity_;
 
     int uncompress(char* compressed_buf, uint32_t compressed_buf_len,
                    char*& uncompressed_buf, uint32_t& uncompressed_buf_len,

cpp/src/reader/aligned_chunk_reader.cc

@@ -161,12 +161,10 @@ void AlignedChunkReader::destroy() {
         value_decoder_ = nullptr;
     }
     if (time_compressor_ != nullptr) {
-        time_compressor_->~Compressor();
         CompressorFactory::free(time_compressor_);
         time_compressor_ = nullptr;
     }
     if (value_compressor_ != nullptr) {
-        value_compressor_->~Compressor();
         CompressorFactory::free(value_compressor_);
         value_compressor_ = nullptr;
     }
@@ -196,7 +194,6 @@ void AlignedChunkReader::destroy() {
             col->decoder = nullptr;
         }
         if (col->compressor != nullptr) {
-            col->compressor->~Compressor();
             CompressorFactory::free(col->compressor);
             col->compressor = nullptr;
         }
@@ -236,7 +233,6 @@ void AlignedChunkReader::destroy() {
     std::vector<Decoder*>().swap(time_decoder_pool_);
     for (auto* c : time_compressor_pool_) {
         if (c != nullptr) {
-            c->~Compressor();
             CompressorFactory::free(c);
         }
     }

cpp/src/reader/chunk_reader.cc

@@ -81,7 +81,6 @@ void ChunkReader::destroy() {
         value_decoder_ = nullptr;
     }
     if (compressor_ != nullptr) {
-        compressor_->~Compressor();
         CompressorFactory::free(compressor_);
         compressor_ = nullptr;
     }

cpp/test/compress/lz4_compressor_test.cc

@@ -26,6 +26,8 @@
 #include <string>
 #include <vector>
 
+#include "compress/compressor_factory.h"
+
 namespace {
 
 class LZ4Test : public ::testing::Test {
@@ -162,4 +164,112 @@ TEST_F(LZ4Test, AfterUncompressFreesParamNotMember) {
     compressor.after_compress(compressed_a);
     compressor.after_compress(compressed_b);
 }
+
+TEST_F(LZ4Test, UncompressReusesBufferAcrossDifferentPageSizes) {
+    storage::LZ4Compressor compressor;
+    // Small page, then a much larger one, then a small one again: the reused
+    // buffer must grow for the large page (new allocation) and must be handed
+    // out again, unchanged, for the following smaller page.
+    const std::vector<std::string> payloads = {std::string(1024, 'a'),
+                                               std::string(200000, 'b'),
+                                               std::string(4096, 'c')};
+
+    char* out_ptr[3] = {nullptr, nullptr, nullptr};
+    for (size_t i = 0; i < payloads.size(); ++i) {
+        std::string input = payloads[i];
+        char* compressed = nullptr;
+        uint32_t compressed_len = 0;
+        ASSERT_EQ(compressor.reset(true), common::E_OK);
+        ASSERT_EQ(compressor.compress(&input[0], input.size(), compressed,
+                                      compressed_len),
+                  common::E_OK);
+
+        char* uncompressed = nullptr;
+        uint32_t uncompressed_len = 0;
+        ASSERT_EQ(compressor.reset(false), common::E_OK);
+        ASSERT_EQ(compressor.uncompress(compressed, compressed_len,
+                                        uncompressed, uncompressed_len),
+                  common::E_OK);
+        ASSERT_NE(uncompressed, nullptr);
+        EXPECT_EQ(uncompressed_len, input.size());
+        EXPECT_EQ(std::string(uncompressed, uncompressed_len), input);
+        out_ptr[i] = uncompressed;
+
+        compressor.after_uncompress(uncompressed);
+        compressor.after_compress(compressed);
+    }
+
+    // Growth frees the old buffer and allocates a new one; the allocator may
+    // reuse the same address, so only the guaranteed reuse for the smaller
+    // page is asserted.
+    EXPECT_EQ(out_ptr[1], out_ptr[2]);
+}
+
+// CompressorFactory::free() must run ~Compressor(): compressors are
+// placement-new'd into mem_alloc() storage, and LZ4 retains its decompression
+// buffer there.  Before the fix the buffer leaked on every recovered chunk
+// (LeakSanitizer caught it in the file-recovery tests).
+TEST_F(LZ4Test, FactoryFreeReleasesRetainedBuffer) {
+    storage::Compressor* compressor =
+        storage::CompressorFactory::alloc_compressor(common::LZ4);
+    ASSERT_NE(nullptr, compressor);
+
+    std::string input(100000, 'z');
+    char* compressed = nullptr;
+    uint32_t compressed_len = 0;
+    ASSERT_EQ(compressor->reset(true), common::E_OK);
+    ASSERT_EQ(compressor->compress(&input[0], input.size(), compressed,
+                                   compressed_len),
+              common::E_OK);
+    char* uncompressed = nullptr;
+    uint32_t uncompressed_len = 0;
+    ASSERT_EQ(compressor->reset(false), common::E_OK);
+    ASSERT_EQ(compressor->uncompress(compressed, compressed_len, uncompressed,
+                                     uncompressed_len),
+              common::E_OK);
+    EXPECT_EQ(std::string(uncompressed, uncompressed_len), input);
+    compressor->after_uncompress(uncompressed);
+    compressor->after_compress(compressed);
+    storage::CompressorFactory::free(compressor);
+}
+
+TEST_F(LZ4Test, UncompressAfterPreviousPageNoLongerInvalidated) {
+    storage::LZ4Compressor compressor;
+    std::string input_a(2048, 'x');
+    std::string input_b(8192, 'y');
+    char* compressed_a = nullptr;
+    char* compressed_b = nullptr;
+    uint32_t compressed_a_len = 0;
+    uint32_t compressed_b_len = 0;
+    ASSERT_EQ(compressor.reset(true), common::E_OK);
+    ASSERT_EQ(compressor.compress(&input_a[0], input_a.size(), compressed_a,
+                                  compressed_a_len),
+              common::E_OK);
+    ASSERT_EQ(compressor.compress(&input_b[0], input_b.size(), compressed_b,
+                                  compressed_b_len),
+              common::E_OK);
+
+    char* out_a = nullptr;
+    uint32_t out_a_len = 0;
+    ASSERT_EQ(compressor.reset(false), common::E_OK);
+    ASSERT_EQ(
+        compressor.uncompress(compressed_a, compressed_a_len, out_a, out_a_len),
+        common::E_OK);
+    // after_uncompress() releases ownership back to the compressor but must
+    // leave the compressor usable for the next page.
+    compressor.after_uncompress(out_a);
+
+    char* out_b = nullptr;
+    uint32_t out_b_len = 0;
+    ASSERT_EQ(
+        compressor.uncompress(compressed_b, compressed_b_len, out_b, out_b_len),
+        common::E_OK);
+    ASSERT_EQ(out_b_len, input_b.size());
+    EXPECT_EQ(std::string(out_b, out_b_len), input_b);
+
+    compressor.after_uncompress(out_b);
+    compressor.after_compress(compressed_a);
+    compressor.after_compress(compressed_b);
+}
+
 }  // namespace

cpp/test/reader/tsfile_reader_test.cc

@@ -21,6 +21,7 @@
 #include <gtest/gtest.h>
 #include <sys/stat.h>
 
+#include <atomic>
 #include <cmath>
 #include <cstring>
 #include <fstream>
@@ -43,6 +44,12 @@
 #include "reader/block/single_device_tsblock_reader.h"
 #include "reader/filter/time_operator.h"
 #include "reader/qds_without_timegenerator.h"
+
+#ifdef _WIN32
+#include <process.h>
+#else
+#include <unistd.h>
+#endif
 #include "reader/tsfile_series_scan_iterator.h"
 #include "writer/chunk_writer.h"
 #include "writer/tsfile_writer.h"
@@ -118,6 +125,18 @@ class TsFileReaderTest : public ::testing::Test {
             random_string += chars[dis(gen)];
         }
 
+        // CTest runs test cases in parallel processes, and two of them can
+        // seed the generator in the same clock tick.  Without a per-process
+        // unique suffix one process can remove the file another is using,
+        // which surfaced as E_ALREADY_EXIST from TsFileWriter::open.
+#ifdef _WIN32
+        const auto process_id = static_cast<uint64_t>(_getpid());
+#else
+        const auto process_id = static_cast<uint64_t>(getpid());
+#endif
+        static std::atomic<uint64_t> counter{0};
+        random_string += "_" + std::to_string(process_id) + "_" +
+                         std::to_string(counter.fetch_add(1));
         return random_string;
     }