Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 22 additions & 10 deletions cpp/src/common/allocator/byte_stream.h
Original file line number Diff line number Diff line change
Expand Up @@ -1241,25 +1241,37 @@ class SerializationUtil {
str = nullptr;
int ret = common::E_OK;
int32_t len = 0;
int32_t read_len = 0;
if (RET_FAIL(read_var_int(len, in))) {
return ret;
} else {
if (len == storage::NO_STR_TO_READ) {
return ret;
} else if (len < 0) {
return E_OUT_OF_RANGE;
} else if (len == 0) {
str = new std::string();
} else if (static_cast<uint64_t>(len) > in.remaining_size()) {
// Reject a corrupt length before resize() zero-fills it: a
// bad 4-byte field could otherwise request a 2 GiB string
// that is thrown away as soon as the read fails.
return E_BUF_NOT_ENOUGH;
} else {
char* tmp_buf =
static_cast<char*>(mem_alloc(len, in.get_mid()));
if (tmp_buf == nullptr) return E_OOM;
if (RET_FAIL(in.read_buf(tmp_buf, len, read_len))) {
mem_free(tmp_buf);
// Read straight into the string's own buffer instead of a
// temporary mem_alloc'd buffer that is copied and freed right
// away. One allocation per segment instead of three.
std::string* tmp_str = new std::string();
tmp_str->resize(static_cast<size_t>(len));
uint32_t read_len = 0;
if (RET_FAIL(
in.read_buf(reinterpret_cast<uint8_t*>(&(*tmp_str)[0]),
static_cast<uint32_t>(len), read_len))) {
delete tmp_str;
return ret;
} else if (len != read_len) {
mem_free(tmp_buf);
} else if (static_cast<uint32_t>(len) != read_len) {
delete tmp_str;
ret = E_BUF_NOT_ENOUGH;
} else {
str = new std::string(tmp_buf, len);
mem_free(tmp_buf);
str = tmp_str;
}
}
}
Expand Down
4 changes: 3 additions & 1 deletion cpp/src/common/tsfile_common.h
Original file line number Diff line number Diff line change
Expand Up @@ -877,7 +877,9 @@ struct DeviceMetaIndexEntry : IMetaIndexEntry {
int deserialize_from(common::ByteStream& in,
common::PageArena* pa) override {
int ret = common::E_OK;
device_id_ = std::make_shared<StringArrayDeviceID>("init");
// Default-construct (no "init" segment) so deserialize() does not
// allocate and then immediately free a placeholder string.
device_id_ = std::make_shared<StringArrayDeviceID>();
if (RET_FAIL(device_id_->deserialize(in))) {
} else if (RET_FAIL(common::SerializationUtil::read_i64(offset_, in))) {
}
Expand Down
105 changes: 105 additions & 0 deletions cpp/test/common/device_id_test.cc
Original file line number Diff line number Diff line change
Expand Up @@ -119,4 +119,109 @@ TEST(DeviceIdTest, SplitTableNameIsIdempotent) {
}
}
}

TEST(DeviceIdTest, SerializeDeserializeRoundTrip) {
std::vector<std::string*> segments;
segments.push_back(new std::string("root.table"));
// Longer than any small-string-optimisation buffer, so the segment owns a
// heap buffer inside the std::string.
segments.push_back(new std::string(64, 'x'));
segments.push_back(new std::string(""));
// The constructor deep-copies every segment, so the caller keeps
// ownership of (and must free) the pointers it passed in.
StringArrayDeviceID device(segments);
for (std::string* segment : segments) {
delete segment;
}

common::ByteStream out(1024, common::MOD_TSFILE_READER);
ASSERT_EQ(device.serialize(out), common::E_OK);
char* buffer = common::get_bytes_from_bytestream(out);
ASSERT_NE(buffer, nullptr);

common::ByteStream in;
in.wrap_from(buffer, static_cast<int32_t>(out.total_size()));
StringArrayDeviceID decoded;
ASSERT_EQ(decoded.deserialize(in), common::E_OK);
EXPECT_EQ(decoded.segment_num(), device.segment_num());
EXPECT_EQ(decoded.get_device_name(), device.get_device_name());
ASSERT_TRUE(decoded == device);

common::mem_free(buffer);
}

TEST(DeviceIdTest, SerializeDeserializeNullSegment) {
std::vector<std::string*> segments;
segments.push_back(new std::string("root.table"));
segments.push_back(nullptr);
StringArrayDeviceID device(segments);
for (std::string* segment : segments) {
delete segment;
}

common::ByteStream out(1024, common::MOD_TSFILE_READER);
ASSERT_EQ(device.serialize(out), common::E_OK);
char* buffer = common::get_bytes_from_bytestream(out);
ASSERT_NE(buffer, nullptr);

common::ByteStream in;
in.wrap_from(buffer, static_cast<int32_t>(out.total_size()));
StringArrayDeviceID decoded;
ASSERT_EQ(decoded.deserialize(in), common::E_OK);
ASSERT_EQ(decoded.segment_num(), 2);
EXPECT_EQ(decoded.get_segments()[1], nullptr);
ASSERT_TRUE(decoded == device);

common::mem_free(buffer);
}

TEST(DeviceIdTest, ReadVarCharPtrHandlesNullEmptyAndInvalidLength) {
common::ByteStream out(1024, common::MOD_TSFILE_READER);
ASSERT_EQ(common::SerializationUtil::write_var_char_ptr(nullptr, out),
common::E_OK);
std::string empty;
ASSERT_EQ(common::SerializationUtil::write_var_char_ptr(&empty, out),
common::E_OK);
// A negative length other than the null sentinel is corrupt input.
ASSERT_EQ(common::SerializationUtil::write_var_int(-2, out), common::E_OK);

char* buffer = common::get_bytes_from_bytestream(out);
ASSERT_NE(buffer, nullptr);
common::ByteStream in;
in.wrap_from(buffer, static_cast<int32_t>(out.total_size()));

std::string* decoded = reinterpret_cast<std::string*>(0x1);
ASSERT_EQ(common::SerializationUtil::read_var_char_ptr(decoded, in),
common::E_OK);
EXPECT_EQ(decoded, nullptr);

ASSERT_EQ(common::SerializationUtil::read_var_char_ptr(decoded, in),
common::E_OK);
ASSERT_NE(decoded, nullptr);
EXPECT_TRUE(decoded->empty());
delete decoded;

EXPECT_EQ(common::SerializationUtil::read_var_char_ptr(decoded, in),
common::E_OUT_OF_RANGE);

common::mem_free(buffer);

// A length that exceeds the remaining bytes is rejected before the
// string allocates (and zero-fills) that many bytes.
common::ByteStream truncated(1024, common::MOD_TSFILE_READER);
ASSERT_EQ(common::SerializationUtil::write_var_int(1 << 20, truncated),
common::E_OK);
char* truncated_buffer = common::get_bytes_from_bytestream(truncated);
ASSERT_NE(truncated_buffer, nullptr);
common::ByteStream truncated_in;
truncated_in.wrap_from(truncated_buffer,
static_cast<int32_t>(truncated.total_size()));
decoded = reinterpret_cast<std::string*>(0x1);
EXPECT_EQ(
common::SerializationUtil::read_var_char_ptr(decoded, truncated_in),
common::E_BUF_NOT_ENOUGH);
EXPECT_EQ(decoded, nullptr);
common::mem_free(truncated_buffer);
}

} // namespace storage
19 changes: 19 additions & 0 deletions cpp/test/reader/tsfile_reader_test.cc
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
#include <gtest/gtest.h>
#include <sys/stat.h>

#include <atomic>
#include <cmath>
#include <cstring>
#include <fstream>
Expand All @@ -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"
Expand Down Expand Up @@ -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;
}

Expand Down
Loading