You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tsfile.apache.org by co...@apache.org on 2026/05/31 03:27:09 UTC
(tsfile) 02/05: fix MSVC build + ASan failures
This is an automated email from the ASF dual-hosted git repository.
ColinLeeo pushed a commit to branch colin_tsfile_dataframe_init
in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit 65b1b532a9935959a96e05d80f0c2217d8b53578
Author: ColinLee <sh...@163.com>
AuthorDate: Sun May 31 09:20:35 2026 +0800
fix MSVC build + ASan failures
Three independent regressions the batch-opt snapshot introduced that
only surface on the MSVC toolchain:
1. reader/result_set.h missed #include <cctype>. v141 doesn't pull the
one-arg std::tolower(int) overload in transitively, so the
CaseInsensitiveHash / CaseInsensitiveEqual call sites fail to find
it and fall through to the locale-aware two-arg template
(C2672/C2780). Add the include.
2. common/tablet.h had no copy/move semantics, so MSVC Debug
(which doesn't always perform NRVO) was shallow-copying a Tablet
on `return tablet;` in test helpers like gen_tablet. The source
destructor then freed value_matrix_[c].string_col while the
returned copy still pointed at it -- heap-use-after-free in
find_all_device_boundaries on every Debug+ASan table-mode test.
Delete the copy ctor/assign and add a pointer-stealing move
ctor/assign so return-by-value is safe.
3. encoding/int{32,64}_rle_decoder.h dropped the RLE-run decode
branch in the snapshot, leaving only bit-packing. The encoder
still emits varint headers with the low bit indicating bit-packing
vs run, so dictionary tests (which encode a single-entry vocabulary
as an RLE run) read garbage codes and index past entry_index_ ->
heap-buffer-overflow in std::basic_string ctor on Release+ASan.
Restore develop's full decoder (RLE-run + varint header + out-param
read_int API) and update the three callers (DictionaryDecoder,
Int32SprintzDecoder, Int64SprintzDecoder) to the out-param API.
493 C++ tests still pass on macOS.
---
cpp/src/common/tablet.h | 55 ++++++++++++++++++
cpp/src/encoding/dictionary_decoder.h | 3 +-
cpp/src/encoding/int32_rle_decoder.h | 83 ++++++++++++++++++++------
cpp/src/encoding/int32_sprintz_decoder.h | 3 +-
cpp/src/encoding/int64_rle_decoder.h | 99 +++++++++++++++++++++++++-------
cpp/src/encoding/int64_sprintz_decoder.h | 3 +-
cpp/src/reader/result_set.h | 1 +
7 files changed, 205 insertions(+), 42 deletions(-)
diff --git a/cpp/src/common/tablet.h b/cpp/src/common/tablet.h
index 418f8ca73..ebbef9477 100644
--- a/cpp/src/common/tablet.h
+++ b/cpp/src/common/tablet.h
@@ -227,6 +227,61 @@ class Tablet {
~Tablet() { destroy(); }
+ // Tablet owns several heap buffers (timestamps_, value_matrix_ with its
+ // StringColumn::buffer/offsets, bitmaps_) that ~Tablet frees. The default
+ // copy ctor / copy-assign shallow-copies the raw pointers, so any copy
+ // path (e.g. `return tablet;` without NRVO under MSVC Debug) leaves the
+ // source Tablet's destructor freeing buffers the copy still points at,
+ // triggering heap-use-after-free in code like
+ // Tablet::find_all_device_boundaries. Make Tablet move-only with a
+ // pointer-stealing move ctor / move-assign so return-by-value is safe.
+ Tablet(const Tablet&) = delete;
+ Tablet& operator=(const Tablet&) = delete;
+
+ Tablet(Tablet&& other) noexcept
+ : err_code_(other.err_code_),
+ max_row_num_(other.max_row_num_),
+ cur_row_size_(other.cur_row_size_),
+ insert_target_name_(std::move(other.insert_target_name_)),
+ schema_vec_(std::move(other.schema_vec_)),
+ schema_map_(std::move(other.schema_map_)),
+ timestamps_(other.timestamps_),
+ value_matrix_(other.value_matrix_),
+ bitmaps_(other.bitmaps_),
+ column_categories_(std::move(other.column_categories_)),
+ id_column_indexes_(std::move(other.id_column_indexes_)),
+ single_device_(other.single_device_) {
+ other.timestamps_ = nullptr;
+ other.value_matrix_ = nullptr;
+ other.bitmaps_ = nullptr;
+ other.cur_row_size_ = 0;
+ // Leaving other.schema_vec_ moved-from is fine; destroy() only
+ // touches the heap buffers above, which we've now nulled out.
+ }
+
+ Tablet& operator=(Tablet&& other) noexcept {
+ if (this != &other) {
+ destroy();
+ err_code_ = other.err_code_;
+ max_row_num_ = other.max_row_num_;
+ cur_row_size_ = other.cur_row_size_;
+ insert_target_name_ = std::move(other.insert_target_name_);
+ schema_vec_ = std::move(other.schema_vec_);
+ schema_map_ = std::move(other.schema_map_);
+ timestamps_ = other.timestamps_;
+ value_matrix_ = other.value_matrix_;
+ bitmaps_ = other.bitmaps_;
+ column_categories_ = std::move(other.column_categories_);
+ id_column_indexes_ = std::move(other.id_column_indexes_);
+ single_device_ = other.single_device_;
+ other.timestamps_ = nullptr;
+ other.value_matrix_ = nullptr;
+ other.bitmaps_ = nullptr;
+ other.cur_row_size_ = 0;
+ }
+ return *this;
+ }
+
const std::string& get_table_name() const { return insert_target_name_; }
void set_table_name(const std::string& table_name) {
insert_target_name_ = table_name;
diff --git a/cpp/src/encoding/dictionary_decoder.h b/cpp/src/encoding/dictionary_decoder.h
index 5f64b5873..2962c66ba 100644
--- a/cpp/src/encoding/dictionary_decoder.h
+++ b/cpp/src/encoding/dictionary_decoder.h
@@ -73,7 +73,8 @@ class DictionaryDecoder : public Decoder {
if (entry_index_.empty()) {
init_map(buffer);
}
- int code = value_decoder_.read_int(buffer);
+ int32_t code = 0;
+ value_decoder_.read_int(code, buffer);
return entry_index_[code];
}
diff --git a/cpp/src/encoding/int32_rle_decoder.h b/cpp/src/encoding/int32_rle_decoder.h
index aee9048a1..ef7b6f095 100644
--- a/cpp/src/encoding/int32_rle_decoder.h
+++ b/cpp/src/encoding/int32_rle_decoder.h
@@ -37,6 +37,8 @@ class Int32RleDecoder : public Decoder {
int bitpacking_num_;
bool is_length_and_bitwidth_readed_;
int current_count_;
+ bool is_rle_run_;
+ int32_t rle_value_;
common::ByteStream byte_cache_{common::MOD_DECODER_OBJ};
int32_t* current_buffer_;
Int32Packer* packer_;
@@ -49,6 +51,8 @@ class Int32RleDecoder : public Decoder {
bitpacking_num_(0),
is_length_and_bitwidth_readed_(false),
current_count_(0),
+ is_rle_run_(false),
+ rle_value_(0),
byte_cache_(1024, common::MOD_DECODER_OBJ),
current_buffer_(nullptr),
packer_(nullptr),
@@ -60,13 +64,14 @@ class Int32RleDecoder : public Decoder {
}
int read_boolean(bool& ret_value, common::ByteStream& in) override {
int32_t bool_value;
- read_int32(bool_value, in);
- ret_value = bool_value == 0 ? false : true;
- return common::E_OK;
+ int ret = read_int32(bool_value, in);
+ if (ret == common::E_OK) {
+ ret_value = bool_value != 0;
+ }
+ return ret;
}
int read_int32(int32_t& ret_value, common::ByteStream& in) override {
- ret_value = static_cast<int32_t>(read_int(in));
- return common::E_OK;
+ return read_int(ret_value, in);
}
int read_int64(int64_t& ret_value, common::ByteStream& in) override {
return common::E_TYPE_NOT_MATCH;
@@ -89,6 +94,8 @@ class Int32RleDecoder : public Decoder {
bit_width_ = 0;
bitpacking_num_ = 0;
current_count_ = 0;
+ is_rle_run_ = false;
+ rle_value_ = 0;
}
bool has_next(common::ByteStream& buffer) {
@@ -103,30 +110,69 @@ class Int32RleDecoder : public Decoder {
return current_count_ > 0 || byte_cache_.remaining_size() > 0;
}
- int32_t read_int(common::ByteStream& buffer) {
+ int read_int(int32_t& result, common::ByteStream& buffer) {
+ int ret = common::E_OK;
if (!is_length_and_bitwidth_readed_) {
// start to reader a new rle+bit-packing pattern
- read_length_and_bitwidth(buffer);
+ if (RET_FAIL(read_length_and_bitwidth(buffer))) {
+ return ret;
+ }
}
if (current_count_ == 0) {
- uint8_t header;
- int ret = common::E_OK;
- if (RET_FAIL(
- common::SerializationUtil::read_ui8(header, byte_cache_))) {
+ // The header is encoded as an unsigned varint where:
+ // low bit = 0 => RLE run: header_value >> 1 is the run
+ // count low bit = 1 => bit-packing: header_value >> 1 is the
+ // group count
+ uint32_t header_value = 0;
+ if (RET_FAIL(common::SerializationUtil::read_var_uint(
+ header_value, byte_cache_))) {
return ret;
}
- call_read_bit_packing_buffer(header);
+ if (header_value & 1) {
+ if (RET_FAIL(call_read_bit_packing_buffer(header_value))) {
+ return ret;
+ }
+ } else {
+ if (RET_FAIL(call_read_rle_run(header_value))) {
+ return ret;
+ }
+ }
}
--current_count_;
- int32_t result = current_buffer_[bitpacking_num_ - current_count_ - 1];
+ result = is_rle_run_
+ ? rle_value_
+ : current_buffer_[bitpacking_num_ - current_count_ - 1];
if (!has_next_package()) {
is_length_and_bitwidth_readed_ = false;
}
- return result;
+ return ret;
}
- int call_read_bit_packing_buffer(uint8_t header) {
- int bit_packed_group_count = (int)(header >> 1);
+ int call_read_rle_run(uint32_t header_value) {
+ int ret = common::E_OK;
+ int run_length = (int)(header_value >> 1);
+ if (run_length <= 0) {
+ return common::E_DECODE_ERR;
+ }
+ int byte_width = (bit_width_ + 7) / 8;
+ // Read the repeated value (stored as byte_width bytes, little-endian)
+ int32_t value = 0;
+ for (int i = 0; i < byte_width; i++) {
+ uint8_t b;
+ if (RET_FAIL(common::SerializationUtil::read_ui8(b, byte_cache_))) {
+ return ret;
+ }
+ value |= ((int32_t)b) << (i * 8);
+ }
+ rle_value_ = value;
+ is_rle_run_ = true;
+ current_count_ = run_length;
+ bitpacking_num_ = run_length;
+ return ret;
+ }
+
+ int call_read_bit_packing_buffer(uint32_t header_value) {
+ int bit_packed_group_count = (int)(header_value >> 1);
// in last bit-packing group, there may be some useless value,
// lastBitPackedNum indicates how many values is useful
uint8_t last_bit_packed_num;
@@ -139,6 +185,7 @@ class Int32RleDecoder : public Decoder {
current_count_ =
(bit_packed_group_count - 1) * 8 + last_bit_packed_num;
bitpacking_num_ = current_count_;
+ is_rle_run_ = false;
} else {
return common::E_DECODE_ERR;
}
@@ -236,8 +283,10 @@ class Int32RleDecoder : public Decoder {
bitpacking_num_ = 0;
is_length_and_bitwidth_readed_ = false;
current_count_ = 0;
+ is_rle_run_ = false;
+ rle_value_ = 0;
if (current_buffer_) {
- delete[] current_buffer_;
+ common::mem_free(current_buffer_);
current_buffer_ = nullptr;
}
if (packer_) {
diff --git a/cpp/src/encoding/int32_sprintz_decoder.h b/cpp/src/encoding/int32_sprintz_decoder.h
index 3d15597ee..500a3238b 100644
--- a/cpp/src/encoding/int32_sprintz_decoder.h
+++ b/cpp/src/encoding/int32_sprintz_decoder.h
@@ -125,7 +125,8 @@ class Int32SprintzDecoder : public SprintzDecoder {
decode_size_ = bit_width_ & ~(1 << 7);
Int32RleDecoder decoder;
for (int i = 0; i < decode_size_; ++i) {
- current_buffer_[i] = decoder.read_int(input);
+ int ret = decoder.read_int(current_buffer_[i], input);
+ if (ret != common::E_OK) return ret;
}
} else {
decode_size_ = block_size_ + 1;
diff --git a/cpp/src/encoding/int64_rle_decoder.h b/cpp/src/encoding/int64_rle_decoder.h
index 8010fe0f7..df8e17838 100644
--- a/cpp/src/encoding/int64_rle_decoder.h
+++ b/cpp/src/encoding/int64_rle_decoder.h
@@ -37,6 +37,8 @@ class Int64RleDecoder : public Decoder {
int bitpacking_num_;
bool is_length_and_bitwidth_readed_;
int current_count_;
+ bool is_rle_run_;
+ int64_t rle_value_;
common::ByteStream byte_cache_{common::MOD_DECODER_OBJ};
int64_t* current_buffer_;
Int64Packer* packer_;
@@ -49,6 +51,8 @@ class Int64RleDecoder : public Decoder {
bitpacking_num_(0),
is_length_and_bitwidth_readed_(false),
current_count_(0),
+ is_rle_run_(false),
+ rle_value_(0),
byte_cache_(1024, common::MOD_DECODER_OBJ),
current_buffer_(nullptr),
packer_(nullptr),
@@ -65,8 +69,7 @@ class Int64RleDecoder : public Decoder {
return common::E_TYPE_NOT_MATCH;
}
int read_int64(int64_t& ret_value, common::ByteStream& in) override {
- ret_value = read_int(in);
- return common::E_OK;
+ return read_int(ret_value, in);
}
int read_float(float& ret_value, common::ByteStream& in) override {
return common::E_TYPE_NOT_MATCH;
@@ -86,6 +89,8 @@ class Int64RleDecoder : public Decoder {
bit_width_ = 0;
bitpacking_num_ = 0;
current_count_ = 0;
+ is_rle_run_ = false;
+ rle_value_ = 0;
}
bool has_next(common::ByteStream& buffer) {
@@ -100,30 +105,69 @@ class Int64RleDecoder : public Decoder {
return current_count_ > 0 || byte_cache_.remaining_size() > 0;
}
- int64_t read_int(common::ByteStream& buffer) {
+ int read_int(int64_t& result, common::ByteStream& buffer) {
+ int ret = common::E_OK;
if (!is_length_and_bitwidth_readed_) {
// start to reader a new rle+bit-packing pattern
- read_length_and_bitwidth(buffer);
+ if (RET_FAIL(read_length_and_bitwidth(buffer))) {
+ return ret;
+ }
}
if (current_count_ == 0) {
- uint8_t header;
- int ret = common::E_OK;
- if (RET_FAIL(
- common::SerializationUtil::read_ui8(header, byte_cache_))) {
+ // The header is encoded as an unsigned varint where:
+ // low bit = 0 => RLE run: header_value >> 1 is the run
+ // count low bit = 1 => bit-packing: header_value >> 1 is the
+ // group count
+ uint32_t header_value = 0;
+ if (RET_FAIL(common::SerializationUtil::read_var_uint(
+ header_value, byte_cache_))) {
return ret;
}
- call_read_bit_packing_buffer(header);
+ if (header_value & 1) {
+ if (RET_FAIL(call_read_bit_packing_buffer(header_value))) {
+ return ret;
+ }
+ } else {
+ if (RET_FAIL(call_read_rle_run(header_value))) {
+ return ret;
+ }
+ }
}
--current_count_;
- int64_t result = current_buffer_[bitpacking_num_ - current_count_ - 1];
+ result = is_rle_run_
+ ? rle_value_
+ : current_buffer_[bitpacking_num_ - current_count_ - 1];
if (!has_next_package()) {
is_length_and_bitwidth_readed_ = false;
}
- return result;
+ return ret;
}
- int call_read_bit_packing_buffer(uint8_t header) {
- int bit_packed_group_count = (int)(header >> 1);
+ int call_read_rle_run(uint32_t header_value) {
+ int ret = common::E_OK;
+ int run_length = (int)(header_value >> 1);
+ if (run_length <= 0) {
+ return common::E_DECODE_ERR;
+ }
+ int byte_width = (bit_width_ + 7) / 8;
+ // Read the repeated value (stored as byte_width bytes, little-endian)
+ int64_t value = 0;
+ for (int i = 0; i < byte_width; i++) {
+ uint8_t b;
+ if (RET_FAIL(common::SerializationUtil::read_ui8(b, byte_cache_))) {
+ return ret;
+ }
+ value |= ((int64_t)b) << (i * 8);
+ }
+ rle_value_ = value;
+ is_rle_run_ = true;
+ current_count_ = run_length;
+ bitpacking_num_ = run_length;
+ return ret;
+ }
+
+ int call_read_bit_packing_buffer(uint32_t header_value) {
+ int bit_packed_group_count = (int)(header_value >> 1);
// in last bit-packing group, there may be some useless value,
// lastBitPackedNum indicates how many values is useful
uint8_t last_bit_packed_num;
@@ -136,25 +180,27 @@ class Int64RleDecoder : public Decoder {
current_count_ =
(bit_packed_group_count - 1) * 8 + last_bit_packed_num;
bitpacking_num_ = current_count_;
+ is_rle_run_ = false;
} else {
- printf(
- "tsfile-encoding IntRleDecoder: bit_packed_group_count %d, "
- "smaller "
- "than 1",
- bit_packed_group_count);
+ return common::E_DECODE_ERR;
}
- read_bit_packing_buffer(bit_packed_group_count, last_bit_packed_num);
+ ret = read_bit_packing_buffer(bit_packed_group_count,
+ last_bit_packed_num);
return ret;
}
- void read_bit_packing_buffer(int bit_packed_group_count,
- int last_bit_packed_num) {
+ int read_bit_packing_buffer(int bit_packed_group_count,
+ int last_bit_packed_num) {
+ int ret = common::E_OK;
if (current_buffer_ != nullptr) {
common::mem_free(current_buffer_);
}
current_buffer_ = static_cast<int64_t*>(
common::mem_alloc(sizeof(int64_t) * bit_packed_group_count * 8,
common::MOD_DECODER_OBJ));
+ if (IS_NULL(current_buffer_)) {
+ return common::E_OOM;
+ }
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();
@@ -162,13 +208,17 @@ class Int64RleDecoder : public Decoder {
std::vector<unsigned char> bytes(bytes_to_read);
for (int i = 0; i < bytes_to_read; i++) {
- common::SerializationUtil::read_ui8(bytes[i], byte_cache_);
+ if (RET_FAIL(common::SerializationUtil::read_ui8(bytes[i],
+ byte_cache_))) {
+ return ret;
+ }
}
// save all int values in currentBuffer
packer_->unpack_all_values(
bytes.data(), bytes_to_read,
current_buffer_); // decode from bytes, save in currentBuffer
+ return ret;
}
int read_length_and_bitwidth(common::ByteStream& buffer) {
@@ -177,6 +227,9 @@ class Int64RleDecoder : public Decoder {
common::SerializationUtil::read_var_uint(length_, buffer))) {
return common::E_PARTIAL_READ;
} else {
+ if (tmp_buf_) {
+ common::mem_free(tmp_buf_);
+ }
tmp_buf_ =
(uint8_t*)common::mem_alloc(length_, common::MOD_DECODER_OBJ);
if (tmp_buf_ == nullptr) {
@@ -222,6 +275,8 @@ class Int64RleDecoder : public Decoder {
bitpacking_num_ = 0;
is_length_and_bitwidth_readed_ = false;
current_count_ = 0;
+ is_rle_run_ = false;
+ rle_value_ = 0;
if (current_buffer_) {
common::mem_free(current_buffer_);
current_buffer_ = nullptr;
diff --git a/cpp/src/encoding/int64_sprintz_decoder.h b/cpp/src/encoding/int64_sprintz_decoder.h
index a7e3fdd27..21de3f3f7 100644
--- a/cpp/src/encoding/int64_sprintz_decoder.h
+++ b/cpp/src/encoding/int64_sprintz_decoder.h
@@ -124,7 +124,8 @@ class Int64SprintzDecoder : public SprintzDecoder {
decode_size_ = bit_width_ & ~(1 << 7);
Int64RleDecoder decoder;
for (int i = 0; i < decode_size_; ++i) {
- current_buffer_[i] = decoder.read_int(input);
+ int ret = decoder.read_int(current_buffer_[i], input);
+ if (ret != common::E_OK) return ret;
}
} else {
decode_size_ = block_size_ + 1;
diff --git a/cpp/src/reader/result_set.h b/cpp/src/reader/result_set.h
index c230e0a84..016087ef0 100644
--- a/cpp/src/reader/result_set.h
+++ b/cpp/src/reader/result_set.h
@@ -21,6 +21,7 @@
#define READER_QUERY_DATA_SET_H
#include <algorithm>
+#include <cctype>
#include <string>
#include <unordered_map>