This is an automated email from the ASF dual-hosted git repository. colinlee pushed a commit to branch support_memory_detect in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit b5d74783415a6a45a37c3a4ed98b50a311a6774c Author: ColinLee <[email protected]> AuthorDate: Thu Mar 19 09:32:54 2026 +0800 enable memory detect. --- cpp/CMakeLists.txt | 5 ++ cpp/src/common/allocator/alloc_base.h | 86 ++++++++++------------ cpp/src/common/allocator/byte_stream.h | 31 +++++--- cpp/src/common/allocator/mem_alloc.cc | 65 +++++++++------- cpp/src/common/tablet.cc | 84 ++++++++++++--------- cpp/src/compress/gzip_compressor.h | 4 +- cpp/src/encoding/int32_rle_encoder.h | 2 +- cpp/src/encoding/int64_rle_decoder.h | 12 +-- cpp/src/encoding/int64_rle_encoder.h | 2 +- cpp/src/reader/tsfile_series_scan_iterator.cc | 4 +- .../writer/table_view/tsfile_writer_table_test.cc | 54 +++++++++++++- 11 files changed, 216 insertions(+), 133 deletions(-) diff --git a/cpp/CMakeLists.txt b/cpp/CMakeLists.txt index c85150d8f..f89675a7d 100755 --- a/cpp/CMakeLists.txt +++ b/cpp/CMakeLists.txt @@ -58,6 +58,11 @@ if (${COV_ENABLED}) message("add_definitions -DCOV_ENABLED=1") endif () +if (ENABLE_MEM_STAT) + add_definitions(-DENABLE_MEM_STAT) + message("add_definitions -DENABLE_MEM_STAT") +endif () + if (NOT CMAKE_BUILD_TYPE) set(CMAKE_BUILD_TYPE "Release" CACHE STRING "Choose the type of build." FORCE) diff --git a/cpp/src/common/allocator/alloc_base.h b/cpp/src/common/allocator/alloc_base.h index effe267d0..d34c9cff7 100644 --- a/cpp/src/common/allocator/alloc_base.h +++ b/cpp/src/common/allocator/alloc_base.h @@ -30,46 +30,33 @@ namespace common { enum AllocModID { __FIRST_MOD_ID = 0, - // if you are sure you will not consume too much memory, you can use - // MOD_DEFAULT. MOD_DEFAULT = 0, - MOD_MEMTABLE = 1, - MOD_SCHEMA = 2, - MOD_SQL = 3, - MOD_NET = 4, - MOD_NET_DATA = 5, - MOD_TVLIST_DATA = 6, - MOD_TVLIST_OBJ = 7, - MOD_TSBLOCK = 8, - MOD_CONTAINER = 9, - MOD_TSSTORE_OBJ = 10, - MOD_FLUSH_TASK_OBJ = 11, - MOD_PAGE_WRITER_OUTPUT_STREAM = 12, - MOD_CW_PAGES_DATA = 13, - MOD_CHUNK_WRITER_OBJ = 14, - MOD_STATISTIC_OBJ = 15, - MOD_ENCODER_OBJ = 16, - MOD_DECODER_OBJ = 17, - MOD_TSFILE_WRITER_META = 18, - MOD_TSFILE_WRITE_STREAM = 19, - MOD_TIMESERIES_INDEX_OBJ = 20, - MOD_BLOOM_FILTER = 21, - MOD_OPEN_FILE_OBJ = 22, - MOD_TSFILE_READER = 23, - MOD_CHUNK_READER = 24, - MOD_COMPRESSOR_OBJ = 25, - MOD_ARRAY = 26, - MOD_HASH_TABLE = 27, - MOD_WRITER_INDEX_NODE = 28, - MOD_TS2DIFF_OBJ = 29, - MOD_BITENCODE_OBJ = 30, - MOD_DICENCODE_OBJ = 31, - MOD_ZIGZAG_OBJ = 32, - MOD_DEVICE_META_ITER = 33, - MOD_DEVICE_TASK_ITER = 34, - MOD_DEVICE_ORDER_TSBLOCK_READER = 35, - __LAST_MOD_ID = 36, // prev + 1, - __MAX_MOD_ID = 127, // leave 1 bit to detect header size + MOD_TVLIST_DATA = 1, + MOD_TSBLOCK = 2, + MOD_PAGE_WRITER_OUTPUT_STREAM = 3, + MOD_CW_PAGES_DATA = 4, + MOD_STATISTIC_OBJ = 5, + MOD_ENCODER_OBJ = 6, + MOD_DECODER_OBJ = 7, + MOD_TSFILE_WRITER_META = 8, + MOD_TSFILE_WRITE_STREAM = 9, + MOD_TIMESERIES_INDEX_OBJ = 10, + MOD_BLOOM_FILTER = 11, + MOD_TSFILE_READER = 12, + MOD_CHUNK_READER = 13, + MOD_COMPRESSOR_OBJ = 14, + MOD_ARRAY = 15, + MOD_HASH_TABLE = 16, + MOD_WRITER_INDEX_NODE = 17, + MOD_TS2DIFF_OBJ = 18, + MOD_BITENCODE_OBJ = 19, + MOD_DICENCODE_OBJ = 20, + MOD_ZIGZAG_OBJ = 21, + MOD_DEVICE_META_ITER = 22, + MOD_DEVICE_TASK_ITER = 23, + MOD_TABLET = 24, + __LAST_MOD_ID = 25, + __MAX_MOD_ID = 127, }; extern const char* g_mod_names[__LAST_MOD_ID]; @@ -84,22 +71,27 @@ class ModStat { ModStat() : stat_arr_(NULL) {} static ModStat& get_instance() { - /* - * This is the singleton of Mod Memory Statistic. - * gms is short for Global Mod Statistic - */ static ModStat gms; +#ifdef ENABLE_MEM_STAT + if (UNLIKELY(gms.stat_arr_ == NULL)) { + gms.init(); + } +#endif return gms; } void init(); void destroy(); INLINE void update_alloc(AllocModID mid, int32_t size) { - // ASSERT(mid < __LAST_MOD_ID); - // ATOMIC_FAA(get_item(mid), size); +#ifdef ENABLE_MEM_STAT + ASSERT(mid < __LAST_MOD_ID); + ATOMIC_FAA(get_item(mid), size); +#endif } void update_free(AllocModID mid, uint32_t size) { - // ASSERT(mid < __LAST_MOD_ID); - // ATOMIC_FAA(get_item(mid), 0 - size); +#ifdef ENABLE_MEM_STAT + ASSERT(mid < __LAST_MOD_ID); + ATOMIC_FAA(get_item(mid), 0 - size); +#endif } void print_stat(); diff --git a/cpp/src/common/allocator/byte_stream.h b/cpp/src/common/allocator/byte_stream.h index 4e1029ea4..b9c8ca1a0 100644 --- a/cpp/src/common/allocator/byte_stream.h +++ b/cpp/src/common/allocator/byte_stream.h @@ -896,10 +896,11 @@ class SerializationUtil { FORCE_INLINE static int chunk_read_all_data(ByteStream& in, ByteStream& out, size_t chunk_size = 4096) { - char* buffer = new char[chunk_size]; + char* buffer = static_cast<char*>( + mem_alloc(chunk_size, MOD_DEFAULT)); + if (buffer == nullptr) return E_OOM; int ret = common::E_OK; while (in.remaining_size() > 0) { - // Adjust read size based on remaining input size uint32_t bytes_to_read = static_cast<uint32_t>( std::min(chunk_size, static_cast<size_t>(in.remaining_size()))); @@ -913,7 +914,7 @@ class SerializationUtil { break; } } - delete[] buffer; + mem_free(buffer); return ret; } @@ -1172,16 +1173,18 @@ class SerializationUtil { str = nullptr; return ret; } else { - char* tmp_buf = static_cast<char*>(malloc(len)); + char* tmp_buf = static_cast<char*>( + mem_alloc(len, MOD_DEFAULT)); + if (tmp_buf == nullptr) return E_OOM; if (RET_FAIL(in.read_buf(tmp_buf, len, read_len))) { - free(tmp_buf); + mem_free(tmp_buf); return ret; } else if (len != read_len) { - free(tmp_buf); + mem_free(tmp_buf); ret = E_BUF_NOT_ENOUGH; } else { str = new std::string(tmp_buf, len); - free(tmp_buf); + mem_free(tmp_buf); } } } @@ -1194,7 +1197,9 @@ class SerializationUtil { int32_t read_len = 0; if (RET_FAIL(read_var_int(len, in))) { } else { - char* tmp_buf = (char*)malloc(len + 1); + char* tmp_buf = static_cast<char*>( + mem_alloc(len + 1, MOD_DEFAULT)); + if (tmp_buf == nullptr) return E_OOM; tmp_buf[len] = '\0'; if (RET_FAIL(in.read_buf(tmp_buf, len, read_len))) { } else if (len != read_len) { @@ -1202,7 +1207,7 @@ class SerializationUtil { } else { str = std::string(tmp_buf); } - free(tmp_buf); + mem_free(tmp_buf); } return ret; } @@ -1220,7 +1225,9 @@ class SerializationUtil { if (RET_FAIL(read_i32(len, in))) { } else { int32_t read_len = 0; - char* tmp_buf = static_cast<char*>(malloc(len + 1)); + char* tmp_buf = static_cast<char*>( + mem_alloc(len + 1, MOD_DEFAULT)); + if (tmp_buf == nullptr) return E_OOM; tmp_buf[len] = '\0'; if (RET_FAIL(in.read_buf(tmp_buf, len, read_len))) { } else if (len != read_len) { @@ -1228,7 +1235,7 @@ class SerializationUtil { } else { str = std::string(tmp_buf); } - free(tmp_buf); + mem_free(tmp_buf); } return ret; } @@ -1308,7 +1315,7 @@ FORCE_INLINE char* get_bytes_from_bytestream(ByteStream& bs) { return nullptr; } uint32_t size = bs.total_size(); - char* ret_buf = (char*)malloc(size); + char* ret_buf = static_cast<char*>(mem_alloc(size, MOD_DEFAULT)); if (ret_buf == nullptr) { return nullptr; } diff --git a/cpp/src/common/allocator/mem_alloc.cc b/cpp/src/common/allocator/mem_alloc.cc index c79e78858..d077483b9 100644 --- a/cpp/src/common/allocator/mem_alloc.cc +++ b/cpp/src/common/allocator/mem_alloc.cc @@ -34,33 +34,30 @@ namespace common { const char* g_mod_names[__LAST_MOD_ID] = { /* 0 */ "DEFAULT", - /* 1 */ "MEMTABLE", - /* 2 */ "SCHEMA", - /* 3 */ "SQL", - /* 4 */ "NET", - /* 5 */ "NET_DATA", - /* 6 */ "TVLIST_DATA", - /* 7 */ "TVLIST_OBJ", - /* 8 */ "TSBLOCK", - /* 9 */ "CONTAINER", - /* 10 */ "TSSTORE_OBJ", - /* 11 */ "FLUSH_TASK_OBJ", - /* 12 */ "PAGE_WRITER_OUTPUT_STREAM", - /* 13 */ "CW_PAGES_DATA", - /* 14 */ "CHUNK_WRITER_OBJ", - /* 15 */ "STATISTIC_OBJ", - /* 16 */ "ENCODER_OBJ", - /* 17 */ "DECODER_OBJ", - /* 18 */ "TSFILE_WRITER_META", - /* 19 */ "TSFILE_WRITE_STREAM", - /* 20 */ "TIMESERIES_INDEX_OBJ", - /* 21 */ "BLOOM_FILTER", - /* 22 */ "OPEN_FILE_OBJ", - /* 23 */ "TSFILE_READER", - /* 24 */ "CHUNK_READER", - /* 25 */ "COMPRESSOR_OBJ", - /* 26 */ "ARRAY", - /* 27 */ "HASH_TABLE", + /* 1 */ "TVLIST_DATA", + /* 2 */ "TSBLOCK", + /* 3 */ "PAGE_WRITER_OUTPUT_STREAM", + /* 4 */ "CW_PAGES_DATA", + /* 5 */ "STATISTIC_OBJ", + /* 6 */ "ENCODER_OBJ", + /* 7 */ "DECODER_OBJ", + /* 8 */ "TSFILE_WRITER_META", + /* 9 */ "TSFILE_WRITE_STREAM", + /* 10 */ "TIMESERIES_INDEX_OBJ", + /* 11 */ "BLOOM_FILTER", + /* 12 */ "TSFILE_READER", + /* 13 */ "CHUNK_READER", + /* 14 */ "COMPRESSOR_OBJ", + /* 15 */ "ARRAY", + /* 16 */ "HASH_TABLE", + /* 17 */ "WRITER_INDEX_NODE", + /* 18 */ "TS2DIFF_OBJ", + /* 19 */ "BITENCODE_OBJ", + /* 20 */ "DICENCODE_OBJ", + /* 21 */ "ZIGZAG_OBJ", + /* 22 */ "DEVICE_META_ITER", + /* 23 */ "DEVICE_TASK_ITER", + /* 24 */ "TABLET", }; // Most modern CPUs (e.g., x86_64, Arm) support at least 8-byte alignment, @@ -97,6 +94,7 @@ void* mem_alloc(uint32_t size, AllocModID mid) { auto high4b = static_cast<uint32_t>(header >> 32); *reinterpret_cast<uint32_t*>(raw) = high4b; *reinterpret_cast<uint32_t*>(raw + 4) = low4b; + ModStat::get_instance().update_alloc(mid, static_cast<int32_t>(size)); return raw + header_size; } @@ -173,6 +171,19 @@ void ModStat::init() { void ModStat::destroy() { ::free(stat_arr_); } +void ModStat::print_stat() { + if (stat_arr_ == NULL) return; + std::cout << "=== Memory Statistics ===" << std::endl; + for (int i = 0; i < __LAST_MOD_ID; i++) { + int32_t val = ATOMIC_FAA(get_item(i), 0); + if (val != 0) { + const char* name = (i < __LAST_MOD_ID) ? g_mod_names[i] : "UNKNOWN"; + std::cout << " " << name << ": " << val << " bytes" << std::endl; + } + } + std::cout << "=========================" << std::endl; +} + BaseAllocator g_base_allocator; } // end namespace common \ No newline at end of file diff --git a/cpp/src/common/tablet.cc b/cpp/src/common/tablet.cc index 10489f67d..8348e3cfe 100644 --- a/cpp/src/common/tablet.cc +++ b/cpp/src/common/tablet.cc @@ -21,6 +21,7 @@ #include <cstdlib> +#include "allocator/alloc_base.h" #include "datatype/date_converter.h" #include "utils/errno_define.h" @@ -30,7 +31,9 @@ namespace storage { int Tablet::init() { ASSERT(timestamps_ == nullptr); - timestamps_ = (int64_t*)malloc(sizeof(int64_t) * max_row_num_); + timestamps_ = static_cast<int64_t*>( + common::mem_alloc(sizeof(int64_t) * max_row_num_, common::MOD_TABLET)); + if (timestamps_ == nullptr) return E_OOM; cur_row_size_ = 0; size_t schema_count = schema_vec_->size(); @@ -39,54 +42,58 @@ int Tablet::init() { ins_res = schema_map_.insert( std::make_pair(to_lower(schema_vec_->at(c).measurement_name_), c)); if (!ins_res.second) { - // maybe dup measurement_name return E_INVALID_ARG; } } ASSERT(schema_map_.size() == schema_count); - value_matrix_ = - (ValueMatrixEntry*)malloc(sizeof(ValueMatrixEntry) * schema_count); + value_matrix_ = static_cast<ValueMatrixEntry*>(common::mem_alloc( + sizeof(ValueMatrixEntry) * schema_count, common::MOD_TABLET)); + if (value_matrix_ == nullptr) return E_OOM; for (size_t c = 0; c < schema_count; ++c) { const MeasurementSchema& schema = schema_vec_->at(c); + uint32_t elem_size = get_data_type_size(schema.data_type_); + uint32_t alloc_size = elem_size * max_row_num_; switch (schema.data_type_) { case BOOLEAN: - value_matrix_[c].bool_data = (bool*)malloc( - get_data_type_size(schema.data_type_) * max_row_num_); - memset(value_matrix_[c].bool_data, 0, - get_data_type_size(schema.data_type_) * max_row_num_); + value_matrix_[c].bool_data = static_cast<bool*>( + common::mem_alloc(alloc_size, common::MOD_TABLET)); + if (value_matrix_[c].bool_data == nullptr) return E_OOM; + memset(value_matrix_[c].bool_data, 0, alloc_size); break; case DATE: case INT32: - value_matrix_[c].int32_data = (int32_t*)malloc( - get_data_type_size(schema.data_type_) * max_row_num_); - memset(value_matrix_[c].int32_data, 0, - get_data_type_size(schema.data_type_) * max_row_num_); + value_matrix_[c].int32_data = static_cast<int32_t*>( + common::mem_alloc(alloc_size, common::MOD_TABLET)); + if (value_matrix_[c].int32_data == nullptr) return E_OOM; + memset(value_matrix_[c].int32_data, 0, alloc_size); break; case TIMESTAMP: case INT64: - value_matrix_[c].int64_data = (int64_t*)malloc( - get_data_type_size(schema.data_type_) * max_row_num_); - memset(value_matrix_[c].int64_data, 0, - get_data_type_size(schema.data_type_) * max_row_num_); + value_matrix_[c].int64_data = static_cast<int64_t*>( + common::mem_alloc(alloc_size, common::MOD_TABLET)); + if (value_matrix_[c].int64_data == nullptr) return E_OOM; + memset(value_matrix_[c].int64_data, 0, alloc_size); break; case FLOAT: - value_matrix_[c].float_data = (float*)malloc( - get_data_type_size(schema.data_type_) * max_row_num_); - memset(value_matrix_[c].float_data, 0, - get_data_type_size(schema.data_type_) * max_row_num_); + value_matrix_[c].float_data = static_cast<float*>( + common::mem_alloc(alloc_size, common::MOD_TABLET)); + if (value_matrix_[c].float_data == nullptr) return E_OOM; + memset(value_matrix_[c].float_data, 0, alloc_size); break; case DOUBLE: - value_matrix_[c].double_data = (double*)malloc( - get_data_type_size(schema.data_type_) * max_row_num_); - memset(value_matrix_[c].double_data, 0, - get_data_type_size(schema.data_type_) * max_row_num_); + value_matrix_[c].double_data = static_cast<double*>( + common::mem_alloc(alloc_size, common::MOD_TABLET)); + if (value_matrix_[c].double_data == nullptr) return E_OOM; + memset(value_matrix_[c].double_data, 0, alloc_size); break; case BLOB: case TEXT: case STRING: { value_matrix_[c].string_data = - (common::String*)malloc(sizeof(String) * max_row_num_); + static_cast<common::String*>(common::mem_alloc( + sizeof(String) * max_row_num_, common::MOD_TABLET)); + if (value_matrix_[c].string_data == nullptr) return E_OOM; break; } default: @@ -95,8 +102,11 @@ int Tablet::init() { } } - bitmaps_ = new BitMap[schema_count]; + bitmaps_ = static_cast<BitMap*>(common::mem_alloc( + sizeof(BitMap) * schema_count, common::MOD_TABLET)); + if (bitmaps_ == nullptr) return E_OOM; for (size_t c = 0; c < schema_count; c++) { + new (&bitmaps_[c]) BitMap(); bitmaps_[c].init(max_row_num_, false); } return E_OK; @@ -104,7 +114,7 @@ int Tablet::init() { void Tablet::destroy() { if (timestamps_ != nullptr) { - free(timestamps_); + common::mem_free(timestamps_); timestamps_ = nullptr; } @@ -114,36 +124,40 @@ void Tablet::destroy() { switch (schema.data_type_) { case DATE: case INT32: - free(value_matrix_[c].int32_data); + common::mem_free(value_matrix_[c].int32_data); break; case TIMESTAMP: case INT64: - free(value_matrix_[c].int64_data); + common::mem_free(value_matrix_[c].int64_data); break; case FLOAT: - free(value_matrix_[c].float_data); + common::mem_free(value_matrix_[c].float_data); break; case DOUBLE: - free(value_matrix_[c].double_data); + common::mem_free(value_matrix_[c].double_data); break; case BOOLEAN: - free(value_matrix_[c].bool_data); + common::mem_free(value_matrix_[c].bool_data); break; case BLOB: case TEXT: case STRING: - free(value_matrix_[c].string_data); + common::mem_free(value_matrix_[c].string_data); break; default: break; } } - free(value_matrix_); + common::mem_free(value_matrix_); value_matrix_ = nullptr; } if (bitmaps_ != nullptr) { - delete[] bitmaps_; + size_t schema_count = schema_vec_->size(); + for (size_t c = 0; c < schema_count; c++) { + bitmaps_[c].~BitMap(); + } + common::mem_free(bitmaps_); bitmaps_ = nullptr; } } diff --git a/cpp/src/compress/gzip_compressor.h b/cpp/src/compress/gzip_compressor.h index 677b72663..a803107ac 100644 --- a/cpp/src/compress/gzip_compressor.h +++ b/cpp/src/compress/gzip_compressor.h @@ -46,7 +46,7 @@ class GzipCompressor { void destroy() { end_zstream(); } int compress(char* uncompressed_buf, uint32_t uncompressed_buf_len, char*& compressed_buf, uint32_t& compressed_buf_len); - void after_compress(char* compressed_buf) { ::free(compressed_buf); } + void after_compress(char* compressed_buf) { common::mem_free(compressed_buf); } int compress_into_bytestream(char* uncompressed_buf, uint32_t uncompressed_buf_len, common::ByteStream& out); @@ -69,7 +69,7 @@ class GzipDeCompressor { void destroy() { end_zstream(); } int uncompress(char* compressed_buf, uint32_t compressed_buf_len, char*& uncompressed_buf, uint32_t& uncompressed_buf_len); - void after_uncompress(char* uncompressed_buf) { ::free(uncompressed_buf); } + void after_uncompress(char* uncompressed_buf) { common::mem_free(uncompressed_buf); } int decompress_into_bytestream(char* compressed_buf, uint32_t compressed_buf_len, common::ByteStream& out); diff --git a/cpp/src/encoding/int32_rle_encoder.h b/cpp/src/encoding/int32_rle_encoder.h index d9eecf5dd..6b9ae1634 100644 --- a/cpp/src/encoding/int32_rle_encoder.h +++ b/cpp/src/encoding/int32_rle_encoder.h @@ -146,7 +146,7 @@ class Int32RleEncoder : public Encoder { void convert_buffer() { // TODO: put the bytes on the stack instead on the heap unsigned char* bytes = (unsigned char*)common::mem_alloc( - bit_width_, common::MOD_BITENCODE_OBJ); + bit_width_, common::MOD_ENCODER_OBJ); int32_t tmp_buffer[8]; for (int i = 0; i < 8; i++) { tmp_buffer[i] = (int64_t)buffered_values_[i]; diff --git a/cpp/src/encoding/int64_rle_decoder.h b/cpp/src/encoding/int64_rle_decoder.h index b2f85ed1e..25d64b084 100644 --- a/cpp/src/encoding/int64_rle_decoder.h +++ b/cpp/src/encoding/int64_rle_decoder.h @@ -150,9 +150,11 @@ class Int64RleDecoder : public Decoder { void read_bit_packing_buffer(int bit_packed_group_count, int last_bit_packed_num) { if (current_buffer_ != nullptr) { - delete[] current_buffer_; + common::mem_free(current_buffer_); } - current_buffer_ = new int64_t[bit_packed_group_count * 8]; + current_buffer_ = static_cast<int64_t*>(common::mem_alloc( + sizeof(int64_t) * bit_packed_group_count * 8, + common::MOD_DECODER_OBJ)); int bytes_to_read = bit_packed_group_count * bit_width_; if (bytes_to_read > (int)byte_cache_.remaining_size()) { bytes_to_read = byte_cache_.remaining_size(); @@ -199,13 +201,13 @@ class Int64RleDecoder : public Decoder { void init_packer() { packer_ = new Int64Packer(bit_width_); } - void destroy() { /* do nothing for BitpackEncoder */ + void destroy() { if (packer_) { delete (packer_); packer_ = nullptr; } if (current_buffer_) { - delete[] current_buffer_; + common::mem_free(current_buffer_); current_buffer_ = nullptr; } if (tmp_buf_) { @@ -221,7 +223,7 @@ class Int64RleDecoder : public Decoder { is_length_and_bitwidth_readed_ = false; current_count_ = 0; if (current_buffer_) { - delete[] current_buffer_; + common::mem_free(current_buffer_); current_buffer_ = nullptr; } if (packer_) { diff --git a/cpp/src/encoding/int64_rle_encoder.h b/cpp/src/encoding/int64_rle_encoder.h index 82fd40f38..edd28180f 100644 --- a/cpp/src/encoding/int64_rle_encoder.h +++ b/cpp/src/encoding/int64_rle_encoder.h @@ -146,7 +146,7 @@ class Int64RleEncoder : public Encoder { void convert_buffer() { // TODO: put the bytes on the stack instead on the heap unsigned char* bytes = (unsigned char*)common::mem_alloc( - bit_width_, common::MOD_BITENCODE_OBJ); + bit_width_, common::MOD_ENCODER_OBJ); int64_t tmp_buffer[8]; for (int i = 0; i < 8; i++) { tmp_buffer[i] = (int64_t)buffered_values_[i]; diff --git a/cpp/src/reader/tsfile_series_scan_iterator.cc b/cpp/src/reader/tsfile_series_scan_iterator.cc index 5e78574e7..75655fb31 100644 --- a/cpp/src/reader/tsfile_series_scan_iterator.cc +++ b/cpp/src/reader/tsfile_series_scan_iterator.cc @@ -96,7 +96,7 @@ int TsFileSeriesScanIterator::init_chunk_reader() { int ret = E_OK; is_aligned_ = itimeseries_index_->get_data_type() == common::VECTOR; if (!is_aligned_) { - void* buf = common::mem_alloc(sizeof(ChunkReader), common::MOD_DEFAULT); + void* buf = common::mem_alloc(sizeof(ChunkReader), common::MOD_CHUNK_READER); chunk_reader_ = new (buf) ChunkReader; chunk_meta_cursor_ = itimeseries_index_->get_chunk_meta_list()->begin(); ChunkMeta* cm = chunk_meta_cursor_.get(); @@ -110,7 +110,7 @@ int TsFileSeriesScanIterator::init_chunk_reader() { } } else { void* buf = - common::mem_alloc(sizeof(AlignedChunkReader), common::MOD_DEFAULT); + common::mem_alloc(sizeof(AlignedChunkReader), common::MOD_CHUNK_READER); chunk_reader_ = new (buf) AlignedChunkReader; time_chunk_meta_cursor_ = itimeseries_index_->get_time_chunk_meta_list()->begin(); diff --git a/cpp/test/writer/table_view/tsfile_writer_table_test.cc b/cpp/test/writer/table_view/tsfile_writer_table_test.cc index 1f8c80ff6..8f1072eee 100644 --- a/cpp/test/writer/table_view/tsfile_writer_table_test.cc +++ b/cpp/test/writer/table_view/tsfile_writer_table_test.cc @@ -1095,4 +1095,56 @@ TEST_F(TsFileWriterTableTest, EncodingConfigIntegration) { delete[] literal; delete[] text_literal; delete table_schema; -} \ No newline at end of file +} + +#ifdef ENABLE_MEM_STAT +TEST_F(TsFileWriterTableTest, MemStatWriteAndVerify) { + TableSchema* table_schema = gen_table_schema(0, 2, 3); + auto tsfile_table_writer = + std::make_shared<TsFileTableWriter>(&write_file_, table_schema); + + const int num_devices = 10; + const int num_timestamps = 100; + + for (int flush = 0; flush < 3; flush++) { + Tablet tablet = gen_tablet(table_schema, + flush * num_devices * num_timestamps, + num_devices, num_timestamps); + ASSERT_EQ(tsfile_table_writer->write_table(tablet), E_OK); + ASSERT_EQ(tsfile_table_writer->flush(), E_OK); + + std::cout << "--- After flush " << flush << " ---" << std::endl; + ModStat::get_instance().print_stat(); + } + + ASSERT_EQ(tsfile_table_writer->close(), E_OK); + + std::cout << "--- After writer close ---" << std::endl; + ModStat::get_instance().print_stat(); + + TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + ResultSet* result_set = nullptr; + ASSERT_EQ(reader.query(table_schema->get_table_name(), + table_schema->get_measurement_names(), 0, + INT64_MAX, result_set), E_OK); + int row_count = 0; + bool has_next = false; + auto* table_result_set = static_cast<TableResultSet*>(result_set); + while (IS_SUCC(table_result_set->next(has_next)) && has_next) { + row_count++; + } + EXPECT_EQ(row_count, num_devices * num_timestamps * 3); + + std::cout << "--- After read ---" << std::endl; + ModStat::get_instance().print_stat(); + + reader.destroy_query_data_set(table_result_set); + ASSERT_EQ(reader.close(), E_OK); + + std::cout << "--- After cleanup ---" << std::endl; + ModStat::get_instance().print_stat(); + + delete table_schema; +} +#endif \ No newline at end of file
