fix compressor mem problem

deps/oblib/src/lib/compress/ob_compressor.h

@@ -40,6 +40,7 @@ public:
                          int64_t &dst_data_size) = 0;
   virtual int get_max_overflow_size(const int64_t src_data_size,
                                     int64_t &max_overflow_size) const = 0;
+  virtual void reset_mem() {}
 
   virtual const char *get_compressor_name() const = 0;
   virtual ObCompressorType get_compressor_type() const = 0;

deps/oblib/src/lib/compress/zstd/ob_zstd_compressor.cpp

@@ -65,6 +65,11 @@ void ObZstdCtxAllocator::reuse()
   allocator_.reuse();
 }
 
+void ObZstdCtxAllocator::reset()
+{
+  allocator_.reset();
+}
+
 /**
  * ----------------------------ObZstdCompressor---------------------------
  */
@@ -147,6 +152,14 @@ int ObZstdCompressor::decompress(const char *src_buffer,
   return ret;
 }
 
+void ObZstdCompressor::reset_mem()
+{
+  ObZstdCtxAllocator *zstd_allocator = GET_TSI_MULT(ObZstdCtxAllocator, 1);
+  if (NULL != zstd_allocator) {
+    zstd_allocator->reset();
+  }
+}
+
 const char *ObZstdCompressor::get_compressor_name() const
 {
   return all_compressor_name[ObCompressorType::ZSTD_COMPRESSOR];

deps/oblib/src/lib/compress/zstd/ob_zstd_compressor.h

@@ -31,6 +31,7 @@ public:
   void *alloc(size_t size);
   void free(void *addr);
   void reuse();
+  void reset();
 private:
   ObArenaAllocator allocator_;
 };
@@ -54,6 +55,7 @@ public:
   ObCompressorType get_compressor_type() const;
   int get_max_overflow_size(const int64_t src_data_size,
                                    int64_t &max_overflow_size) const;
+  void reset_mem();
 
 };
 } // namespace zstd

deps/oblib/src/lib/compress/zstd_1_3_8/ob_zstd_compressor_1_3_8.cpp

@@ -66,6 +66,11 @@ void ObZstdCtxAllocator::reuse()
   allocator_.reuse();
 }
 
+void ObZstdCtxAllocator::reset()
+{
+  allocator_.reset();
+}
+
 /**
  * ----------------------------ObZstdCompressor---------------------------
  */
@@ -148,6 +153,14 @@ int ObZstdCompressor_1_3_8::decompress(const char *src_buffer,
   return ret;
 }
 
+void ObZstdCompressor_1_3_8::reset_mem()
+{
+  ObZstdCtxAllocator *zstd_allocator = GET_TSI_MULT(ObZstdCtxAllocator, 1);
+  if (NULL != zstd_allocator) {
+    zstd_allocator->reset();
+  }
+}
+
 const char *ObZstdCompressor_1_3_8::get_compressor_name() const
 {
   return all_compressor_name[ObCompressorType::ZSTD_1_3_8_COMPRESSOR];

deps/oblib/src/lib/compress/zstd_1_3_8/ob_zstd_compressor_1_3_8.h

@@ -31,6 +31,7 @@ public:
   void *alloc(size_t size);
   void free(void *addr);
   void reuse();
+  void reset();
 private:
   ObArenaAllocator allocator_;
 };
@@ -54,6 +55,7 @@ public:
   ObCompressorType get_compressor_type() const;
   int get_max_overflow_size(const int64_t src_data_size,
                             int64_t &max_overflow_size) const;
+  void reset_mem();
 
 };
 } // namespace zstd_1_3_8

src/share/stat/ob_opt_stat_sql_service.cpp

@@ -1486,6 +1486,9 @@ int ObOptStatSqlService::get_compressed_llc_bitmap(ObIAllocator &allocator,
       comp_buf = const_cast<char*>(bitmap_buf);
       comp_size = bitmap_size;
     }
+    if (compressor != nullptr) {
+      compressor->reset_mem();
+    }
   }
   return ret;
 }
@@ -1523,6 +1526,8 @@ int ObOptStatSqlService::get_decompressed_llc_bitmap(ObIAllocator &allocator,
     LOG_WARN("decompress bitmap buffer failed.",
                KP(comp_buf), K(comp_size), KP(bitmap_buf),
                K(max_bitmap_size), K(bitmap_size), K(ret));
+  } else {
+    compressor->reset_mem();
   }
   return ret;
 }

src/storage/blocksstable/ob_macro_block.cpp

@@ -374,13 +374,19 @@ ObMicroBlockCompressor::ObMicroBlockCompressor()
 
 ObMicroBlockCompressor::~ObMicroBlockCompressor()
 {
+  if (compressor_ != nullptr) {
+    compressor_->reset_mem();
+  }
 }
 
 void ObMicroBlockCompressor::reset()
 {
   is_none_ = false;
   micro_block_size_ = 0;
-  compressor_ = NULL;
+  if (compressor_ != nullptr) {
+    compressor_->reset_mem();
+    compressor_ = nullptr;
+  }
   comp_buf_.reuse();
   decomp_buf_.reuse();
 }

src/storage/blocksstable/ob_macro_block_reader.cpp

@@ -49,6 +49,9 @@ ObMacroBlockReader::~ObMacroBlockReader()
     encryption_->~ObMicroBlockEncryption();
     encryption_ = nullptr;
   }
+  if (nullptr != compressor_) {
+    compressor_->reset_mem();
+  }
 }