jt2594838 commented on code in PR #345:
URL: https://github.com/apache/tsfile/pull/345#discussion_r1922997832
##########
cpp/src/common/tsblock/tuple_desc.cc:
##########
@@ -44,6 +44,10 @@ uint32_t TupleDesc::get_single_row_len(int *erro_code) {
totol_len += sizeof(double);
break;
}
+ case common::STRING: {
+ totol_len += DEFAULT_RESERVED_SIZE_OF_STRING + STRING_LEN;
Review Comment:
totol_len -> total_len
##########
cpp/src/common/row_record.h:
##########
@@ -108,6 +116,15 @@ struct Field {
return -1; // when data type is unknown
}
+ FORCE_INLINE common::String *get_string_value() {
+ if (type_ == common::STRING) {
+ return value_.strval_;
+ } else {
+ std::cout << "not String type" << std::endl;
Review Comment:
Remove
##########
cpp/src/common/tablet.cc:
##########
@@ -44,11 +44,41 @@ int Tablet::init() {
}
ASSERT(schema_map_.size() == schema_count);
- value_matrix_ = (void **)malloc(sizeof(void *) * schema_count);
- for (size_t c = 0; c < schema_count; c++) {
+ // value_matrix_ = (void **)malloc(sizeof(void *) * schema_count);
+ // for (size_t c = 0; c < schema_count; c++) {
+ // const MeasurementSchema &schema = schema_vec_->at(c);
+ // value_matrix_[c] =
+ // malloc(get_data_type_size(schema.data_type_) * max_row_num_);
+ // }
Review Comment:
remove
##########
cpp/src/file/tsfile_io_writer.cc:
##########
@@ -274,7 +283,7 @@ int TsFileIOWriter::write_log_index_range() {
#if DEBUG_SE
void debug_print_chunk_group_meta(ChunkGroupMeta *cgm) {
std::cout << "ChunkGroupMeta = {insert_target_name_="
- << cgm->insert_target_name_ << ", chunk_meta_list_={";
+ << cgm->device_id_ << ", chunk_meta_list_={";
Review Comment:
{insert_target_name_= -> {device_id_=
##########
cpp/src/common/tablet.cc:
##########
@@ -82,84 +137,118 @@ int Tablet::add_timestamp(uint32_t row_index, int64_t
timestamp) {
return E_OUT_OF_RANGE;
}
timestamps_[row_index] = timestamp;
- cur_row_size_++;
+ cur_row_size_ = std::max(row_index, cur_row_size_);
return E_OK;
}
-template <typename T>
-void Tablet::process_val(uint32_t row_index, uint32_t schema_index, T val) {
- T *column_values = (T *)value_matrix_[schema_index];
- column_values[row_index] = val;
- bitmaps_[schema_index].set(row_index); /* mark as non-null*/
-}
-
-template <typename T>
-int Tablet::add_value(uint32_t row_index, uint32_t schema_index, T val) {
- int ret = common::E_OK;
- if (LIKELY(schema_index >= schema_vec_->size())) {
- ASSERT(false);
- ret = common::E_OUT_OF_RANGE;
- } else {
- const MeasurementSchema &schema = schema_vec_->at(schema_index);
- if (UNLIKELY(GetDataTypeFromTemplateType<T>() != schema.data_type_)) {
- if (GetDataTypeFromTemplateType<T>() == common::INT32 &&
- schema.data_type_ == common::INT64) {
- process_val(row_index, schema_index,
static_cast<int64_t>(val));
- } else if (GetDataTypeFromTemplateType<T>() == common::FLOAT &&
- schema.data_type_ == common::DOUBLE) {
- process_val(row_index, schema_index, static_cast<double>(val));
- } else {
- ASSERT(false);
- return E_TYPE_NOT_MATCH;
- }
- } else {
- process_val(row_index, schema_index, val);
- }
- }
- return ret;
-}
-
void* Tablet::get_value(int row_index, uint32_t schema_index,
common::TSDataType& data_type) const {
- if (LIKELY(schema_index >= schema_vec_->size())) {
+ if (UNLIKELY(schema_index >= schema_vec_->size())) {
return nullptr;
}
const MeasurementSchema& schema = schema_vec_->at(schema_index);
- void* column_values = value_matrix_[schema_index];
+ ValueMatrixEntry column_values = value_matrix_[schema_index];
data_type = schema.data_type_;
if (!bitmaps_[schema_index].test(row_index)) {
return nullptr;
}
switch (schema.data_type_) {
case BOOLEAN: {
- bool* bool_values = static_cast<bool*>(column_values);
+ bool* bool_values = column_values.bool_data;
return &bool_values[row_index];
}
case INT32: {
- int32_t* int32_values = static_cast<int32_t*>(column_values);
+ int32_t* int32_values = column_values.int32_data;
return &int32_values[row_index];
}
case INT64: {
- int64_t* int64_values = static_cast<int64_t*>(column_values);
+ int64_t* int64_values = column_values.int64_data;
return &int64_values[row_index];
}
case FLOAT: {
- float* float_values = static_cast<float*>(column_values);
+ float* float_values = column_values.float_data;
return &float_values[row_index];
}
case DOUBLE: {
- double* double_values = static_cast<double*>(column_values);
+ double* double_values = column_values.double_data;
return &double_values[row_index];
}
- case TEXT: {
- std::string* string_values =
static_cast<std::string*>(column_values);
+ case STRING: {
+ auto string_values = column_values.string_data;
return &string_values[row_index];
}
default:
return nullptr;
}
}
+template <>
+void Tablet::process_val(uint32_t row_index, uint32_t schema_index,
common::String val) {
+ value_matrix_[schema_index].string_data[row_index].dup_from(val,
page_arena_);
+ bitmaps_[schema_index].set(row_index); /* mark as non-null */
Review Comment:
The bitmaps of a Tablet in the Java edition are marked when the value is
null. I think we should make them consistent.
##########
cpp/src/file/tsfile_io_reader.h:
##########
@@ -30,162 +30,176 @@
#include "utils/storage_utils.h"
namespace storage {
+ class TsFileSeriesScanIterator;
-class TsFileSeriesScanIterator;
+ /*
+ * TODO:
+ * TsFileIOReader correspond to one tsfile.
+ * It may be shared by many query.
+ */
+ class TsFileIOReader {
+ public:
+ TsFileIOReader()
+ : read_file_(nullptr),
+ tsfile_meta_page_arena_(),
+ tsfile_meta_(&tsfile_meta_page_arena_),
+ tsfile_meta_ready_(false),
+ read_file_created_(false) {
+ tsfile_meta_page_arena_.init(512, common::MOD_DEFAULT);
+ }
-/*
- * TODO:
- * TsFileIOReader correspond to one tsfile.
- * It may be shared by many query.
- */
-class TsFileIOReader {
- public:
- TsFileIOReader()
- : read_file_(nullptr),
- tsfile_meta_page_arena_(),
- tsfile_meta_(&tsfile_meta_page_arena_),
- tsfile_meta_ready_(false),
- read_file_created_(false) {
- tsfile_meta_page_arena_.init(512, common::MOD_DEFAULT);
- }
-
- int init(const std::string &file_path);
- int init(ReadFile *read_file);
- void reset();
-
- int alloc_ssi(const std::string &device_path,
- const std::string &measurement_name,
- TsFileSeriesScanIterator *&ssi,
- Filter *time_filter = nullptr);
- void revert_ssi(TsFileSeriesScanIterator *ssi);
- std::string get_file_path() const { return read_file_->file_path(); }
- TsFileMeta *get_tsfile_meta() {
- load_tsfile_meta_if_necessary();
- return &tsfile_meta_;
- }
-
- int get_device_timeseries_meta_without_chunk_meta(
- std::string device_id,
- std::vector<ITimeseriesIndex *> ×eries_indexs,
- common::PageArena &pa);
-
- private:
- FORCE_INLINE int32_t file_size() const { return read_file_->file_size(); }
- int load_tsfile_meta();
- int load_tsfile_meta_if_necessary();
- int load_device_index_entry(const std::string &device_path,
- MetaIndexEntry &device_index_entry,
- int64_t &end_offset);
- int load_measurement_index_entry(
- const std::string &measurement_name, int64_t start_offset,
- int64_t end_offset, MetaIndexEntry &ret_measurement_index_entry,
- int64_t &ret_end_offset);
- int load_all_measurement_index_entry(
- int64_t start_offset, int64_t end_offset, common::PageArena &pa,
- std::vector<std::pair<MetaIndexEntry *, int64_t>>
+ int init(const std::string &file_path);
+
+ int init(ReadFile *read_file);
+
+ void reset();
+
+ int alloc_ssi(const std::string &device_path,
+ const std::string &measurement_name,
+ TsFileSeriesScanIterator *&ssi, common::PageArena &pa,
+ Filter *time_filter = nullptr);
+
+ void revert_ssi(TsFileSeriesScanIterator *ssi);
+
+ std::string get_file_path() const { return read_file_->file_path(); }
+
+ TsFileMeta *get_tsfile_meta() {
+ load_tsfile_meta_if_necessary();
+ return &tsfile_meta_;
+ }
+
+ int get_device_timeseries_meta_without_chunk_meta(
+ std::shared_ptr<IDeviceID> device_id,
+ std::vector<ITimeseriesIndex *> ×eries_indexs,
+ common::PageArena &pa);
+
+ private:
+ FORCE_INLINE int32_t file_size() const { return
read_file_->file_size(); }
+
+ int load_tsfile_meta();
+
+ int load_tsfile_meta_if_necessary();
+
+ int load_device_index_entry(std::shared_ptr<IComparable> target_name,
+ IMetaIndexEntry &device_index_entry,
+ int64_t &end_offset);
+
+ int load_measurement_index_entry(
+ const std::string &measurement_name, int64_t start_offset,
+ int64_t end_offset, IMetaIndexEntry &ret_measurement_index_entry,
+ int64_t &ret_end_offset);
+
+ int load_all_measurement_index_entry(
+ int64_t start_offset, int64_t end_offset, common::PageArena &pa,
+ std::vector<std::pair<std::shared_ptr<IMetaIndexEntry>, int64_t> >
&ret_measurement_index_entry);
- int do_load_timeseries_index(const std::string &measurement_name_str,
- int64_t start_offset, int64_t end_offset,
- common::PageArena &pa,
- ITimeseriesIndex *&ts_index);
- int do_load_all_timeseries_index(
- std::vector<std::pair<MetaIndexEntry *, int64_t>>
+
+ int do_load_timeseries_index(const std::string &measurement_name_str,
+ int64_t start_offset, int64_t end_offset,
+ common::PageArena &pa,
+ ITimeseriesIndex *&ts_index);
+
+ int do_load_all_timeseries_index(
+ std::vector<std::pair<std::shared_ptr<IMetaIndexEntry>, int64_t> >
&index_node_entry_list,
- common::PageArena &in_timeseries_index_pa,
- std::vector<ITimeseriesIndex *> &ts_indexs);
- int load_timeseries_index_for_ssi(const std::string &device_path,
- const std::string &measurement_name,
- TsFileSeriesScanIterator *&ssi);
- int search_from_leaf_node(const common::String &target_name,
- MetaIndexNode *index_node,
- MetaIndexEntry &ret_index_entry,
- int64_t &ret_end_offset);
- int search_from_internal_node(const common::String &target_name,
- MetaIndexNode *index_node,
- MetaIndexEntry &ret_index_entry,
+ common::PageArena &in_timeseries_index_pa,
+ std::vector<ITimeseriesIndex *> &ts_indexs);
+
+ int load_timeseries_index_for_ssi(const std::string &device_path,
+ const std::string &measurement_name,
+ TsFileSeriesScanIterator *&ssi);
+
+ int search_from_leaf_node(std::shared_ptr<IComparable> target_name,
+ std::shared_ptr<MetaIndexNode> index_node,
+ IMetaIndexEntry &ret_index_entry,
int64_t &ret_end_offset);
- bool filter_stasify(ITimeseriesIndex *ts_index, Filter *time_filter);
- int get_all_leaf(MetaIndexNode *index_node,
- std::vector<std::pair<MetaIndexEntry *, int64_t>>
+ int search_from_internal_node(std::shared_ptr<IComparable> target_name,
+ std::shared_ptr<MetaIndexNode>
index_node,
+ IMetaIndexEntry &ret_index_entry,
+ int64_t &ret_end_offset);
+
+ bool filter_stasify(ITimeseriesIndex *ts_index, Filter *time_filter);
+
+ int get_all_leaf(std::shared_ptr<MetaIndexNode> index_node,
+
std::vector<std::pair<std::shared_ptr<IMetaIndexEntry>, int64_t> >
&index_node_entry_list);
- private:
- ReadFile *read_file_;
- common::PageArena tsfile_meta_page_arena_;
- TsFileMeta tsfile_meta_;
- bool tsfile_meta_ready_;
- bool read_file_created_;
-};
-
-// class TsFileIOReaderSet
-// {
-// public:
-// TsFileIOReaderSet() : readers_() {}
-// ~TsFileIOReaderSet() {}
-//
-// static TsFileIOReaderSet& get_instance()
-// {
-// static TsFileIOReaderSet g_tfrs;
-// return g_tfrs;
-// }
-//
-// int get_or_create(const std::string &file_path,
-// ReadFile *read_file,
-// TsFileIOReader *&reader)
-// {
-// ASSERT(read_file != nullptr);
-// do {
-// {
-// MutexGuard g(&mutex_);
-// MapIterator find_iter = readers_.find(file_path);
-// if (find_iter != readers_.end()) {
-// reader = find_iter->second;
-// reader->ref();
-// ASSERT(reader->ref() >= 2);
-// return common::E_OK;
-// }
-// }
-//
-// reader = new TsFileIOReader;
-// reader->ref();
-// ASSERT(reader->get_ref() == 1);
-// reader->init(read_file);
-//
-// MutexGuard g(&mutex_);
-// std::pair<MapIterator, bool> ins_res =
-// readers_.insert(std::make_pair(file_path, reader)); if
(ins_res.second)
-// {
-// return common::E_OK;
-// } else {
-// delete reader;
-// continue;
-// }
-// } while (false);
-// }
-//
-// void revert(TsFileIOReader *reader)
-// {
-// const std::string &file_path = reader->get_file_path();
-// MutexGuard g(&mutex_);
-// MapIterator find_iter = readers_.find(file_path);
-// if (find_iter != readers_.end()) {
-// reader = find_iter->second;
-// if (0 == reader->unref()) {
-// readers_.erase(file_path);
-// delete reader;
-// }
-// } else {
-// ASSERT(false);
-// }
-// }
-//
-// private:
-// typedef std::map<std::string, TsFileIOReader*>::iterator MapIterator;
-// private:
-// std::map<std::string, TsFileIOReader*> readers_;
-// common::Mutex mutex_;
-// };
-
-} // end namespace storage
+ private:
+ ReadFile *read_file_;
+ common::PageArena tsfile_meta_page_arena_;
+ TsFileMeta tsfile_meta_;
+ bool tsfile_meta_ready_;
+ bool read_file_created_;
+ };
+
+ // class TsFileIOReaderSet
Review Comment:
Remove
##########
cpp/src/common/tsfile_common.h:
##########
@@ -58,655 +58,855 @@ typedef int64_t TsFileID;
// one page exists in the chunk but we know that fact after we writer
// the first page.
struct PageHeader {
- uint32_t uncompressed_size_;
- uint32_t compressed_size_;
- Statistic *statistic_;
-
- PageHeader()
- : uncompressed_size_(0), compressed_size_(0), statistic_(nullptr) {}
- ~PageHeader() { reset(); }
- void reset() {
- if (statistic_ != nullptr) {
- StatisticFactory::free(statistic_);
- statistic_ = nullptr;
- }
- uncompressed_size_ = 0;
- compressed_size_ = 0;
- }
- int deserialize_from(common::ByteStream &in, bool deserialize_stat,
- common::TSDataType data_type) {
- int ret = common::E_OK;
- if (RET_FAIL(common::SerializationUtil::read_var_uint(
- uncompressed_size_, in))) {
- } else if (RET_FAIL(common::SerializationUtil::read_var_uint(
- compressed_size_, in))) {
- } else if (deserialize_stat) {
- statistic_ = StatisticFactory::alloc_statistic(data_type);
- if (IS_NULL(statistic_)) {
- return common::E_OOM;
- } else if (RET_FAIL(statistic_->deserialize_from(in))) {
- }
- }
- return ret;
+ uint32_t uncompressed_size_;
+ uint32_t compressed_size_;
+ Statistic *statistic_;
+
+ PageHeader()
+ : uncompressed_size_(0), compressed_size_(0), statistic_(nullptr) {}
+ ~PageHeader() { reset(); }
+ void reset() {
+ if (statistic_ != nullptr) {
+ StatisticFactory::free(statistic_);
+ statistic_ = nullptr;
}
-
- /** max page header size without statistics. */
- static int estimat_max_page_header_size_without_statistics() {
- // uncompressedSize, compressedSize
- // because we use unsigned varInt to encode these two integer, each
- // unsigned varInt will cost at most 5 bytes
- return 2 * (4 + 1);
+ uncompressed_size_ = 0;
+ compressed_size_ = 0;
+ }
+ int deserialize_from(common::ByteStream &in, bool deserialize_stat,
+ common::TSDataType data_type) {
+ int ret = common::E_OK;
+ if (RET_FAIL(common::SerializationUtil::read_var_uint(
+ uncompressed_size_, in))) {
+ } else if (RET_FAIL(common::SerializationUtil::read_var_uint(
+ compressed_size_, in))) {
+ } else if (deserialize_stat) {
+ statistic_ = StatisticFactory::alloc_statistic(data_type);
+ if (IS_NULL(statistic_)) {
+ return common::E_OOM;
+ } else if (RET_FAIL(statistic_->deserialize_from(in))) {
+ }
}
+ return ret;
+ }
+
+ /** max page header size without statistics. */
+ static int estimat_max_page_header_size_without_statistics() {
+ // uncompressedSize, compressedSize
+ // because we use unsigned varInt to encode these two integer, each
+ // unsigned varInt will cost at most 5 bytes
+ return 2 * (4 + 1);
+ }
#ifndef NDEBUG
- friend std::ostream &operator<<(std::ostream &os, const PageHeader &h) {
- os << "{uncompressed_size_=" << h.uncompressed_size_
- << ", compressed_size_=" << h.uncompressed_size_;
- if (h.statistic_ == nullptr) {
- os << ", stat=nil}";
- } else {
- os << ", stat=" << h.statistic_->to_string() << "}";
- }
- return os;
+ friend std::ostream &operator<<(std::ostream &os, const PageHeader &h) {
+ os << "{uncompressed_size_=" << h.uncompressed_size_
+ << ", compressed_size_=" << h.uncompressed_size_;
+ if (h.statistic_ == nullptr) {
+ os << ", stat=nil}";
+ } else {
+ os << ", stat=" << h.statistic_->to_string() << "}";
}
+ return os;
+ }
#endif
};
struct ChunkHeader {
- ChunkHeader()
- : measurement_name_(""),
- data_size_(0),
- data_type_(common::INVALID_DATATYPE),
- compression_type_(common::INVALID_COMPRESSION),
- encoding_type_(common::INVALID_ENCODING),
- num_of_pages_(0),
- serialized_size_(0),
- chunk_type_(0) {}
-
- void reset() {
- measurement_name_.clear();
- data_size_ = 0;
- data_type_ = common::INVALID_DATATYPE;
- compression_type_ = common::INVALID_COMPRESSION;
- encoding_type_ = common::INVALID_ENCODING;
- num_of_pages_ = 0;
- serialized_size_ = 0;
- chunk_type_ = 0;
- }
-
- int serialize_to(common::ByteStream &out) {
- int ret = common::E_OK;
- if (RET_FAIL(common::SerializationUtil::write_char(chunk_type_, out)))
{
- } else if (RET_FAIL(common::SerializationUtil::write_str(
- measurement_name_, out))) {
- } else if (RET_FAIL(common::SerializationUtil::write_var_uint(
- data_size_, out))) {
- } else if (RET_FAIL(common::SerializationUtil::write_char(data_type_,
- out))) {
- } else if (RET_FAIL(common::SerializationUtil::write_char(
- compression_type_, out))) {
- } else if (RET_FAIL(common::SerializationUtil::write_char(
- encoding_type_, out))) {
- }
- return ret;
- }
- int deserialize_from(common::ByteStream &in) {
- int ret = common::E_OK;
- in.mark_read_pos();
- if (RET_FAIL(common::SerializationUtil::read_char(chunk_type_, in))) {
- } else if (RET_FAIL(common::SerializationUtil::read_str(
- measurement_name_, in))) {
- } else if
(RET_FAIL(common::SerializationUtil::read_var_uint(data_size_,
- in))) {
- } else if (RET_FAIL(common::SerializationUtil::read_char(
- (char &)data_type_, in))) {
- } else if (RET_FAIL(common::SerializationUtil::read_char(
- (char &)compression_type_, in))) {
- } else if (RET_FAIL(common::SerializationUtil::read_char(
- (char &)encoding_type_, in))) {
- } else {
- serialized_size_ = in.get_mark_len();
- }
- return ret;
+ ChunkHeader()
+ : measurement_name_(""),
+ data_size_(0),
+ data_type_(common::INVALID_DATATYPE),
+ compression_type_(common::INVALID_COMPRESSION),
+ encoding_type_(common::INVALID_ENCODING),
+ num_of_pages_(0),
+ serialized_size_(0),
+ chunk_type_(0) {}
+
+ void reset() {
+ measurement_name_.clear();
+ data_size_ = 0;
+ data_type_ = common::INVALID_DATATYPE;
+ compression_type_ = common::INVALID_COMPRESSION;
+ encoding_type_ = common::INVALID_ENCODING;
+ num_of_pages_ = 0;
+ serialized_size_ = 0;
+ chunk_type_ = 0;
+ }
+
+ ~ChunkHeader() = default;
+
+ int serialize_to(common::ByteStream &out) {
+ int ret = common::E_OK;
+ if (RET_FAIL(common::SerializationUtil::write_char(chunk_type_, out))) {
+ } else if (RET_FAIL(common::SerializationUtil::write_str(
+ measurement_name_, out))) {
+ } else if (RET_FAIL(common::SerializationUtil::write_var_uint(
+ data_size_, out))) {
+ } else if (RET_FAIL(common::SerializationUtil::write_char(data_type_,
+ out))) {
+ } else if (RET_FAIL(common::SerializationUtil::write_char(
+ compression_type_, out))) {
+ } else if (RET_FAIL(common::SerializationUtil::write_char(
+ encoding_type_, out))) {
}
-#ifndef NDEBUG
- friend std::ostream &operator<<(std::ostream &os, const ChunkHeader &h) {
- os << "{measurement_name=" << h.measurement_name_
- << ", data_size=" << h.data_size_ << ", data_type=" << h.data_type_
- << ", compression_type=" << h.compression_type_
- << ", encoding_type=" << h.encoding_type_
- << ", num_of_pages=" << h.num_of_pages_
- << ", serialized_size=" << h.serialized_size_
- << ", chunk_type=" << (int)h.chunk_type_ << "}";
- return os;
+ return ret;
+ }
+ int deserialize_from(common::ByteStream &in) {
+ int ret = common::E_OK;
+ in.mark_read_pos();
+ if (RET_FAIL(common::SerializationUtil::read_char(chunk_type_, in))) {
+ } else if (RET_FAIL(common::SerializationUtil::read_str(
+ measurement_name_, in))) {
+ } else if (RET_FAIL(common::SerializationUtil::read_var_uint(data_size_,
+ in))) {
+ } else if (RET_FAIL(common::SerializationUtil::read_char(
+ (char &) data_type_, in))) {
+ } else if (RET_FAIL(common::SerializationUtil::read_char(
+ (char &) compression_type_, in))) {
+ } else if (RET_FAIL(common::SerializationUtil::read_char(
+ (char &) encoding_type_, in))) {
+ } else {
+ serialized_size_ = in.get_mark_len();
}
+ return ret;
+ }
+#ifndef NDEBUG
+ friend std::ostream &operator<<(std::ostream &os, const ChunkHeader &h) {
+ os << "{measurement_name=" << h.measurement_name_
+ << ", data_size=" << h.data_size_ << ", data_type=" << h.data_type_
+ << ", compression_type=" << h.compression_type_
+ << ", encoding_type=" << h.encoding_type_
+ << ", num_of_pages=" << h.num_of_pages_
+ << ", serialized_size=" << h.serialized_size_
+ << ", chunk_type=" << (int) h.chunk_type_ << "}";
+ return os;
+ }
#endif
- std::string measurement_name_;
- uint32_t data_size_;
- common::TSDataType data_type_;
- common::CompressionType compression_type_;
- common::TSEncoding encoding_type_;
- int32_t num_of_pages_;
- int32_t serialized_size_; // TODO seems no usage
- char chunk_type_; // TODO give a description here
+ std::string measurement_name_;
+ uint32_t data_size_;
+ common::TSDataType data_type_;
+ common::CompressionType compression_type_;
+ common::TSEncoding encoding_type_;
+ int32_t num_of_pages_;
+ int32_t serialized_size_; // TODO seems no usage
+ char chunk_type_; // TODO give a description here
- static const int MIN_SERIALIZED_SIZE = 7;
+ static const int MIN_SERIALIZED_SIZE = 7;
};
struct ChunkMeta {
- // std::string measurement_name_;
- common::String measurement_name_;
- common::TSDataType data_type_;
- int64_t offset_of_chunk_header_;
- Statistic *statistic_;
- common::TsID ts_id_;
- char mask_;
- common::TSEncoding encoding_;
- common::CompressionType compression_type_;
-
- ChunkMeta()
- : measurement_name_(),
- data_type_(),
- offset_of_chunk_header_(0),
- statistic_(nullptr),
- ts_id_(),
- mask_(0) {}
-
- int init(const common::String &measurement_name,
- common::TSDataType data_type, int64_t offset_of_chunk_header,
- Statistic *stat, const common::TsID &ts_id, char mask,
- common::TSEncoding encoding,
- common::CompressionType compression_type, common::PageArena &pa) {
- // TODO check parameter valid
- measurement_name_.dup_from(measurement_name, pa);
- data_type_ = data_type;
- offset_of_chunk_header_ = offset_of_chunk_header;
- statistic_ = stat;
- ts_id_ = ts_id;
- mask_ = mask;
- encoding_ = encoding;
- compression_type_ = compression_type;
- return common::E_OK;
- }
- FORCE_INLINE void clone_statistic_from(Statistic *stat) {
- clone_statistic(stat, statistic_, data_type_);
- }
- FORCE_INLINE int clone_from(ChunkMeta &that, common::PageArena *pa) {
- int ret = common::E_OK;
- if (RET_FAIL(measurement_name_.dup_from(that.measurement_name_, *pa)))
{
- return ret;
- }
- data_type_ = that.data_type_;
- offset_of_chunk_header_ = that.offset_of_chunk_header_;
- if (that.statistic_ != nullptr) {
- statistic_ =
- StatisticFactory::alloc_statistic_with_pa(data_type_, pa);
- if (IS_NULL(statistic_)) {
- return common::E_OOM;
- }
- clone_statistic_from(that.statistic_);
- }
- ts_id_ = that.ts_id_;
- mask_ = that.mask_;
- return ret;
- }
- int serialize_to(common::ByteStream &out, bool serialize_statistic) {
- int ret = common::E_OK;
- if (RET_FAIL(common::SerializationUtil::write_i64(
- offset_of_chunk_header_, out))) {
- } else if (serialize_statistic) {
- ret = statistic_->serialize_to(out);
- }
- return ret;
- }
- int deserialize_from(common::ByteStream &in, bool deserialize_stat,
- common::PageArena *pa) {
- int ret = common::E_OK;
- if (RET_FAIL(common::SerializationUtil::read_i64(
- offset_of_chunk_header_, in))) {
- } else if (deserialize_stat) {
- statistic_ =
- StatisticFactory::alloc_statistic_with_pa(data_type_, pa);
- if (IS_NULL(statistic_)) {
- ret = common::E_OOM;
- } else {
- ret = statistic_->deserialize_from(in);
- }
- }
- return ret;
+ // std::string measurement_name_;
+ common::String measurement_name_;
+ common::TSDataType data_type_;
+ int64_t offset_of_chunk_header_;
+ Statistic *statistic_;
+ common::TsID ts_id_;
+ char mask_;
+ common::TSEncoding encoding_;
+ common::CompressionType compression_type_;
+
+ ChunkMeta()
+ : measurement_name_(),
+ data_type_(),
+ offset_of_chunk_header_(0),
+ statistic_(nullptr),
+ ts_id_(),
+ mask_(0) {}
+
+ int init(const common::String &measurement_name,
+ common::TSDataType data_type, int64_t offset_of_chunk_header,
+ Statistic *stat, const common::TsID &ts_id, char mask,
+ common::TSEncoding encoding,
+ common::CompressionType compression_type, common::PageArena &pa) {
+ // TODO check parameter valid
+ measurement_name_.dup_from(measurement_name, pa);
+ data_type_ = data_type;
+ offset_of_chunk_header_ = offset_of_chunk_header;
+ statistic_ = stat;
+ ts_id_ = ts_id;
+ mask_ = mask;
+ encoding_ = encoding;
+ compression_type_ = compression_type;
+ return common::E_OK;
+ }
+ FORCE_INLINE void clone_statistic_from(Statistic *stat) {
+ clone_statistic(stat, statistic_, data_type_);
+ }
+ FORCE_INLINE int clone_from(ChunkMeta &that, common::PageArena *pa) {
+ int ret = common::E_OK;
+ if (RET_FAIL(measurement_name_.dup_from(that.measurement_name_, *pa))) {
+ return ret;
+ }
+ data_type_ = that.data_type_;
+ offset_of_chunk_header_ = that.offset_of_chunk_header_;
+ if (that.statistic_ != nullptr) {
+ statistic_ =
+ StatisticFactory::alloc_statistic_with_pa(data_type_, pa);
+ if (IS_NULL(statistic_)) {
+ return common::E_OOM;
+ }
+ clone_statistic_from(that.statistic_);
}
+ ts_id_ = that.ts_id_;
+ mask_ = that.mask_;
+ return ret;
+ }
+ int serialize_to(common::ByteStream &out, bool serialize_statistic) {
+ int ret = common::E_OK;
+ if (RET_FAIL(common::SerializationUtil::write_i64(
+ offset_of_chunk_header_, out))) {
+ } else if (serialize_statistic) {
+ ret = statistic_->serialize_to(out);
+ }
+ return ret;
+ }
+ int deserialize_from(common::ByteStream &in, bool deserialize_stat,
+ common::PageArena *pa) {
+ int ret = common::E_OK;
+ if (RET_FAIL(common::SerializationUtil::read_i64(
+ offset_of_chunk_header_, in))) {
+ } else if (deserialize_stat) {
+ statistic_ =
+ StatisticFactory::alloc_statistic_with_pa(data_type_, pa);
+ if (IS_NULL(statistic_)) {
+ ret = common::E_OOM;
+ } else {
+ ret = statistic_->deserialize_from(in);
+ }
+ }
+ return ret;
+ }
#ifndef NDEBUG
- friend std::ostream &operator<<(std::ostream &os, const ChunkMeta &cm) {
- os << "{measurement_name=" << cm.measurement_name_
- << ", data_type=" << cm.data_type_
- << ", offset_of_chunk_header=" << cm.offset_of_chunk_header_
- << ", ts_id=" << cm.ts_id_.to_string()
- << ", mask=" << ((int)cm.mask_);
- if (cm.statistic_ == nullptr) {
- os << ", statistic=nil}";
- } else {
- os << ", statistic=" << cm.statistic_->to_string() << "}";
- }
- return os;
+ friend std::ostream &operator<<(std::ostream &os, const ChunkMeta &cm) {
+ os << "{measurement_name=" << cm.measurement_name_
+ << ", data_type=" << cm.data_type_
+ << ", offset_of_chunk_header=" << cm.offset_of_chunk_header_
+ << ", ts_id=" << cm.ts_id_.to_string()
+ << ", mask=" << ((int) cm.mask_);
+ if (cm.statistic_ == nullptr) {
+ os << ", statistic=nil}";
+ } else {
+ os << ", statistic=" << cm.statistic_->to_string() << "}";
}
+ return os;
+ }
#endif
};
struct ChunkGroupMeta {
- std::shared_ptr<IDeviceID> device_name_;
- common::String device_name_str_;
- common::SimpleList<ChunkMeta *> chunk_meta_list_;
-
- explicit ChunkGroupMeta(common::PageArena *pa_ptr)
- : chunk_meta_list_(pa_ptr) {}
-
- FORCE_INLINE int init(std::shared_ptr<IDeviceID> device_id,
- common::PageArena &pa) {
- device_name_ = device_id;
- return device_name_str_.dup_from(device_id->get_device_name(), pa);
- }
- FORCE_INLINE int push(ChunkMeta *cm) {
- return chunk_meta_list_.push_back(cm);
- }
+ std::shared_ptr<IDeviceID> device_id_;
+ common::SimpleList<ChunkMeta *> chunk_meta_list_;
+
+ explicit ChunkGroupMeta(common::PageArena *pa_ptr)
+ : chunk_meta_list_(pa_ptr) {}
+
+ FORCE_INLINE int init(std::shared_ptr<IDeviceID> device_id) {
+ device_id_ = device_id;
+ return 0;
+ }
+ FORCE_INLINE int push(ChunkMeta *cm) {
+ return chunk_meta_list_.push_back(cm);
+ }
};
class ITimeseriesIndex {
- public:
- ITimeseriesIndex() {}
- ~ITimeseriesIndex() {}
- virtual common::SimpleList<ChunkMeta *> *get_chunk_meta_list() const {
- return nullptr;
- }
- virtual common::SimpleList<ChunkMeta *> *get_time_chunk_meta_list() const {
- return nullptr;
- }
- virtual common::SimpleList<ChunkMeta *> *get_value_chunk_meta_list() const
{
- return nullptr;
- }
-
- virtual common::String get_measurement_name() { return common::String(); }
- virtual common::TSDataType get_data_type() const {
- return common::INVALID_DATATYPE;
- }
- virtual Statistic *get_statistic() const { return nullptr; }
+public:
+ ITimeseriesIndex() {}
+ ~ITimeseriesIndex() {}
+ virtual common::SimpleList<ChunkMeta *> *get_chunk_meta_list() const {
+ return nullptr;
+ }
+ virtual common::SimpleList<ChunkMeta *> *get_time_chunk_meta_list() const {
+ return nullptr;
+ }
+ virtual common::SimpleList<ChunkMeta *> *get_value_chunk_meta_list() const {
+ return nullptr;
+ }
+
+ virtual common::String get_measurement_name() { return common::String(); }
+ virtual common::TSDataType get_data_type() const {
+ return common::INVALID_DATATYPE;
+ }
+ virtual Statistic *get_statistic() const { return nullptr; }
};
/*
* A TimeseriesIndex may have one or more chunk metas,
* that means we have such a map: <Timeseries, List<ChunkMeta>>.
*/
class TimeseriesIndex : public ITimeseriesIndex {
- public:
- static const uint32_t CHUNK_META_LIST_SERIALIZED_BUF_PAGE_SIZE = 128;
- static const uint32_t PAGE_ARENA_PAGE_SIZE = 256;
- static const common::AllocModID PAGE_ARENA_MOD_ID =
- common::MOD_TIMESERIES_INDEX_OBJ;
-
- public:
- TimeseriesIndex()
- : timeseries_meta_type_((char)255),
- chunk_meta_list_data_size_(0),
- measurement_name_(),
- ts_id_(),
- data_type_(common::INVALID_DATATYPE),
- statistic_(nullptr),
- statistic_from_pa_(false),
- chunk_meta_list_serialized_buf_(
- CHUNK_META_LIST_SERIALIZED_BUF_PAGE_SIZE, PAGE_ARENA_MOD_ID),
- chunk_meta_list_(nullptr) {
- // page_arena_.init(PAGE_ARENA_PAGE_SIZE, PAGE_ARENA_MOD_ID);
- }
- ~TimeseriesIndex() { destroy(); }
- void destroy() {
- // page_arena_.destroy();
- reset();
- }
- void reset() // FIXME reuse
- {
- timeseries_meta_type_ = 0;
- chunk_meta_list_data_size_ = 0;
- measurement_name_.reset();
- ts_id_.reset();
- data_type_ = common::VECTOR;
- chunk_meta_list_serialized_buf_.reset();
- if (statistic_ != nullptr && !statistic_from_pa_) {
- StatisticFactory::free(statistic_);
- statistic_ = nullptr;
- }
- }
-
- int add_chunk_meta(ChunkMeta *chunk_meta, bool serialize_statistic);
- FORCE_INLINE int set_measurement_name(common::String &measurement_name,
- common::PageArena &pa) {
- return measurement_name_.dup_from(measurement_name, pa);
- }
- FORCE_INLINE void set_measurement_name(common::String &measurement_name) {
- measurement_name_.shallow_copy_from(measurement_name);
+public:
+ static const uint32_t CHUNK_META_LIST_SERIALIZED_BUF_PAGE_SIZE = 128;
+ static const uint32_t PAGE_ARENA_PAGE_SIZE = 256;
+ static const common::AllocModID PAGE_ARENA_MOD_ID =
+ common::MOD_TIMESERIES_INDEX_OBJ;
+
+public:
+ TimeseriesIndex()
+ : timeseries_meta_type_((char) 255),
+ chunk_meta_list_data_size_(0),
+ measurement_name_(),
+ ts_id_(),
+ data_type_(common::INVALID_DATATYPE),
+ statistic_(nullptr),
+ statistic_from_pa_(false),
+ chunk_meta_list_serialized_buf_(
+ CHUNK_META_LIST_SERIALIZED_BUF_PAGE_SIZE, PAGE_ARENA_MOD_ID),
+ chunk_meta_list_(nullptr) {
+ // page_arena_.init(PAGE_ARENA_PAGE_SIZE, PAGE_ARENA_MOD_ID);
+ }
+ ~TimeseriesIndex() { destroy(); }
+ void destroy() {
+ // page_arena_.destroy();
+ reset();
+ }
+ void reset() // FIXME reuse
+ {
+ timeseries_meta_type_ = 0;
+ chunk_meta_list_data_size_ = 0;
+ measurement_name_.reset();
+ ts_id_.reset();
+ data_type_ = common::VECTOR;
+ chunk_meta_list_serialized_buf_.reset();
+ if (statistic_ != nullptr && !statistic_from_pa_) {
+ StatisticFactory::free(statistic_);
+ statistic_ = nullptr;
}
- FORCE_INLINE virtual common::String get_measurement_name() {
- return measurement_name_;
+ }
+
+ int add_chunk_meta(ChunkMeta *chunk_meta, bool serialize_statistic);
+ FORCE_INLINE int set_measurement_name(common::String &measurement_name,
+ common::PageArena &pa) {
+ return measurement_name_.dup_from(measurement_name, pa);
+ }
+ FORCE_INLINE void set_measurement_name(common::String &measurement_name) {
+ measurement_name_.shallow_copy_from(measurement_name);
+ }
+ FORCE_INLINE virtual common::String get_measurement_name() {
+ return measurement_name_;
+ }
+ virtual inline common::SimpleList<ChunkMeta *> *get_chunk_meta_list()
+ const {
+ return chunk_meta_list_;
+ }
+ FORCE_INLINE void set_ts_meta_type(char ts_meta_type) {
+ timeseries_meta_type_ = ts_meta_type;
+ }
+ FORCE_INLINE void set_data_type(common::TSDataType data_type) {
+ data_type_ = data_type;
+ }
+ FORCE_INLINE virtual common::TSDataType get_data_type() const {
+ return data_type_;
+ }
+ int init_statistic(common::TSDataType data_type) {
+ if (statistic_ != nullptr &&
+ !statistic_from_pa_) { // clear old statistic
+ StatisticFactory::free(statistic_);
+ statistic_ = nullptr;
}
- virtual inline common::SimpleList<ChunkMeta *> *get_chunk_meta_list()
- const {
- return chunk_meta_list_;
+ statistic_ = StatisticFactory::alloc_statistic(data_type);
+ if (IS_NULL(statistic_)) {
+ return common::E_OOM;
}
- FORCE_INLINE void set_ts_meta_type(char ts_meta_type) {
- timeseries_meta_type_ = ts_meta_type;
+ statistic_->reset();
+ return common::E_OK;
+ }
+ virtual Statistic *get_statistic() const { return statistic_; }
+ common::TsID get_ts_id() const { return ts_id_; }
+ void set_ts_id(const common::TsID &ts_id) {
+ ts_id_ = ts_id;
+
+ // TODO for debug only
+ if (chunk_meta_list_ != nullptr) {
+ common::SimpleList<ChunkMeta *>::Iterator it =
+ chunk_meta_list_->begin();
+ for (; it != chunk_meta_list_->end(); it++) {
+ it.get()->ts_id_ = ts_id;
+ }
}
- FORCE_INLINE void set_data_type(common::TSDataType data_type) {
- data_type_ = data_type;
+ }
+
+ FORCE_INLINE void finish() {
+ chunk_meta_list_data_size_ =
+ chunk_meta_list_serialized_buf_.total_size();
+ }
+
+ int serialize_to(common::ByteStream &out) {
+ int ret = common::E_OK;
+ if (RET_FAIL(common::SerializationUtil::write_char(
+ timeseries_meta_type_, out))) {
+ } else if (RET_FAIL(common::SerializationUtil::write_mystring(
+ measurement_name_, out))) {
+ } else if (RET_FAIL(common::SerializationUtil::write_char(data_type_,
+ out))) {
+ } else if (RET_FAIL(common::SerializationUtil::write_var_uint(
+ chunk_meta_list_data_size_, out))) {
+ } else if (RET_FAIL(statistic_->serialize_to(out))) {
+ } else if (RET_FAIL(merge_byte_stream(
+ out, chunk_meta_list_serialized_buf_))) {
}
- FORCE_INLINE virtual common::TSDataType get_data_type() const {
- return data_type_;
- }
- int init_statistic(common::TSDataType data_type) {
- statistic_ = StatisticFactory::alloc_statistic(data_type);
- if (IS_NULL(statistic_)) {
- return common::E_OOM;
- }
- statistic_->reset();
- return common::E_OK;
- }
- virtual Statistic *get_statistic() const { return statistic_; }
- common::TsID get_ts_id() const { return ts_id_; }
- void set_ts_id(const common::TsID &ts_id) {
- ts_id_ = ts_id;
-
- // TODO for debug only
- if (chunk_meta_list_ != nullptr) {
- common::SimpleList<ChunkMeta *>::Iterator it =
- chunk_meta_list_->begin();
- for (; it != chunk_meta_list_->end(); it++) {
- it.get()->ts_id_ = ts_id;
- }
- }
- }
-
- FORCE_INLINE void finish() {
- chunk_meta_list_data_size_ =
- chunk_meta_list_serialized_buf_.total_size();
- }
-
- int serialize_to(common::ByteStream &out) {
- int ret = common::E_OK;
- if (RET_FAIL(common::SerializationUtil::write_char(
- timeseries_meta_type_, out))) {
- } else if (RET_FAIL(common::SerializationUtil::write_mystring(
- measurement_name_, out))) {
- } else if (RET_FAIL(common::SerializationUtil::write_char(data_type_,
- out))) {
- } else if (RET_FAIL(common::SerializationUtil::write_var_uint(
- chunk_meta_list_data_size_, out))) {
- } else if (RET_FAIL(statistic_->serialize_to(out))) {
- } else if (RET_FAIL(merge_byte_stream(
- out, chunk_meta_list_serialized_buf_))) {
- }
- return ret;
- }
-
- int deserialize_from(common::ByteStream &in, common::PageArena *pa) {
- int ret = common::E_OK;
- if
(RET_FAIL(common::SerializationUtil::read_char(timeseries_meta_type_,
- in))) {
- } else if (RET_FAIL(common::SerializationUtil::read_mystring(
- measurement_name_, pa, in))) {
- } else if (RET_FAIL(common::SerializationUtil::read_char(
- (char &)data_type_, in))) {
- } else if (RET_FAIL(common::SerializationUtil::read_var_uint(
- chunk_meta_list_data_size_, in))) {
- } else if (nullptr ==
- (statistic_ = StatisticFactory::alloc_statistic_with_pa(
- data_type_, pa))) {
- ret = common::E_OOM;
- } else if (RET_FAIL(statistic_->deserialize_from(in))) {
+ return ret;
+ }
+
+ int deserialize_from(common::ByteStream &in, common::PageArena *pa) {
+ int ret = common::E_OK;
+ if (RET_FAIL(common::SerializationUtil::read_char(timeseries_meta_type_,
+ in))) {
+ } else if (RET_FAIL(common::SerializationUtil::read_mystring(
+ measurement_name_, pa, in))) {
+ } else if (RET_FAIL(common::SerializationUtil::read_char(
+ (char &) data_type_, in))) {
+ } else if (RET_FAIL(common::SerializationUtil::read_var_uint(
+ chunk_meta_list_data_size_, in))) {
+ } else if (nullptr ==
+ (statistic_ = StatisticFactory::alloc_statistic_with_pa(
+ data_type_, pa))) {
+ ret = common::E_OOM;
+ } else if (RET_FAIL(statistic_->deserialize_from(in))) {
+ } else {
+ statistic_from_pa_ = true;
+ void *chunk_meta_list_buf = pa->alloc(sizeof(*chunk_meta_list_));
+ if (IS_NULL(chunk_meta_list_buf)) {
+ return common::E_OOM;
+ }
+ const bool deserialize_chunk_meta_statistic =
+ (timeseries_meta_type_ & 0x3F); // TODO
+ chunk_meta_list_ =
+ new(chunk_meta_list_buf) common::SimpleList<ChunkMeta *>(pa);
+ uint32_t start_pos = in.read_pos();
+ while (IS_SUCC(ret) &&
+ in.read_pos() < start_pos + chunk_meta_list_data_size_) {
+ void *cm_buf = pa->alloc(sizeof(ChunkMeta));
+ if (IS_NULL(cm_buf)) {
+ ret = common::E_OOM;
} else {
- statistic_from_pa_ = true;
- void *chunk_meta_list_buf = pa->alloc(sizeof(*chunk_meta_list_));
- if (IS_NULL(chunk_meta_list_buf)) {
- return common::E_OOM;
- }
- const bool deserialize_chunk_meta_statistic =
- (timeseries_meta_type_ & 0x3F); // TODO
- chunk_meta_list_ =
- new (chunk_meta_list_buf) common::SimpleList<ChunkMeta *>(pa);
- uint32_t start_pos = in.read_pos();
- while (IS_SUCC(ret) &&
- in.read_pos() < start_pos + chunk_meta_list_data_size_) {
- void *cm_buf = pa->alloc(sizeof(ChunkMeta));
- if (IS_NULL(cm_buf)) {
- ret = common::E_OOM;
- } else {
- ChunkMeta *cm = new (cm_buf) ChunkMeta;
- cm->measurement_name_.shallow_copy_from(
- this->measurement_name_);
- cm->data_type_ = this->data_type_;
- cm->mask_ = 0; // TODO
- if (RET_FAIL(cm->deserialize_from(
- in, deserialize_chunk_meta_statistic, pa))) {
- } else if (RET_FAIL(chunk_meta_list_->push_back(cm))) {
- }
- }
- }
+ ChunkMeta *cm = new(cm_buf) ChunkMeta;
+ cm->measurement_name_.shallow_copy_from(
+ this->measurement_name_);
+ cm->data_type_ = this->data_type_;
+ cm->mask_ = 0; // TODO
+ if (RET_FAIL(cm->deserialize_from(
+ in, deserialize_chunk_meta_statistic, pa))) {
+ } else if (RET_FAIL(chunk_meta_list_->push_back(cm))) {
+ }
}
- return ret;
+ }
}
+ return ret;
+ }
+
+ int clone_from(const TimeseriesIndex &that, common::PageArena *pa) {
+ int ret = common::E_OK;
+ timeseries_meta_type_ = that.timeseries_meta_type_;
+ chunk_meta_list_data_size_ = that.chunk_meta_list_data_size_;
+ ts_id_ = that.ts_id_;
+ data_type_ = that.data_type_;
+
+ statistic_ = StatisticFactory::alloc_statistic_with_pa(data_type_, pa);
+ if (IS_NULL(statistic_)) {
+ return common::E_OOM;
+ }
+ clone_statistic(that.statistic_, this->statistic_, data_type_);
+ statistic_from_pa_ = true;
- int clone_from(const TimeseriesIndex &that, common::PageArena *pa) {
- int ret = common::E_OK;
- timeseries_meta_type_ = that.timeseries_meta_type_;
- chunk_meta_list_data_size_ = that.chunk_meta_list_data_size_;
- ts_id_ = that.ts_id_;
- data_type_ = that.data_type_;
-
- statistic_ = StatisticFactory::alloc_statistic_with_pa(data_type_, pa);
- if (IS_NULL(statistic_)) {
- return common::E_OOM;
- }
- clone_statistic(that.statistic_, this->statistic_, data_type_);
- statistic_from_pa_ = true;
+ if (RET_FAIL(measurement_name_.dup_from(that.measurement_name_, *pa))) {
+ return ret;
+ }
- if (RET_FAIL(measurement_name_.dup_from(that.measurement_name_, *pa)))
{
- return ret;
+ if (that.chunk_meta_list_ != nullptr) {
+ void *buf = pa->alloc(sizeof(*chunk_meta_list_));
+ if (IS_NULL(buf)) {
+ return common::E_OOM;
+ }
+ chunk_meta_list_ = new(buf) common::SimpleList<ChunkMeta *>(pa);
+ common::SimpleList<ChunkMeta *>::Iterator it;
+ for (it = that.chunk_meta_list_->begin();
+ IS_SUCC(ret) && it != that.chunk_meta_list_->end(); it++) {
+ ChunkMeta *cm = it.get();
+ void *cm_buf = pa->alloc(sizeof(ChunkMeta));
+ if (IS_NULL(cm_buf)) {
+ return common::E_OOM;
+ } else {
+ ChunkMeta *my_cm = new(cm_buf) ChunkMeta;
+ if (RET_FAIL(my_cm->clone_from(*cm, pa))) {
+ } else if (RET_FAIL(chunk_meta_list_->push_back(my_cm))) {
+ }
}
-
- if (that.chunk_meta_list_ != nullptr) {
- void *buf = pa->alloc(sizeof(*chunk_meta_list_));
- if (IS_NULL(buf)) {
- return common::E_OOM;
- }
- chunk_meta_list_ = new (buf) common::SimpleList<ChunkMeta *>(pa);
- common::SimpleList<ChunkMeta *>::Iterator it;
- for (it = that.chunk_meta_list_->begin();
- IS_SUCC(ret) && it != that.chunk_meta_list_->end(); it++) {
- ChunkMeta *cm = it.get();
- void *cm_buf = pa->alloc(sizeof(ChunkMeta));
- if (IS_NULL(cm_buf)) {
- return common::E_OOM;
- } else {
- ChunkMeta *my_cm = new (cm_buf) ChunkMeta;
- if (RET_FAIL(my_cm->clone_from(*cm, pa))) {
- } else if (RET_FAIL(chunk_meta_list_->push_back(my_cm))) {
- }
- }
- }
- } // end (that.chunk_meta_list_ != nullptr)
- return ret;
- }
+ }
+ } // end (that.chunk_meta_list_ != nullptr)
+ return ret;
+ }
#ifndef NDEBUG
- friend std::ostream &operator<<(std::ostream &os,
- const TimeseriesIndex &tsi) {
- os << "{meta_type=" << (int)tsi.timeseries_meta_type_
- << ", chunk_meta_list_data_size=" << tsi.chunk_meta_list_data_size_
- << ", measurement_name=" << tsi.measurement_name_
- << ", ts_id=" << tsi.ts_id_.to_string()
- << ", data_type=" << common::get_data_type_name(tsi.data_type_)
- << ", statistic=" << tsi.statistic_->to_string();
-
- if (tsi.chunk_meta_list_) {
- os << ", chunk_meta_list={";
- int count = 0;
- common::SimpleList<ChunkMeta *>::Iterator it =
- tsi.chunk_meta_list_->begin();
- for (; it != tsi.chunk_meta_list_->end(); it++, count++) {
- if (count != 0) {
- os << ", ";
- }
- os << "[" << count << "]={" << *it.get() << "}";
- }
- os << "}";
+ friend std::ostream &operator<<(std::ostream &os,
+ const TimeseriesIndex &tsi) {
+ os << "{meta_type=" << (int) tsi.timeseries_meta_type_
+ << ", chunk_meta_list_data_size=" << tsi.chunk_meta_list_data_size_
+ << ", measurement_name=" << tsi.measurement_name_
+ << ", ts_id=" << tsi.ts_id_.to_string()
+ << ", data_type=" << common::get_data_type_name(tsi.data_type_)
+ << ", statistic=" << tsi.statistic_->to_string();
+
+ if (tsi.chunk_meta_list_) {
+ os << ", chunk_meta_list={";
+ int count = 0;
+ common::SimpleList<ChunkMeta *>::Iterator it =
+ tsi.chunk_meta_list_->begin();
+ for (; it != tsi.chunk_meta_list_->end(); it++, count++) {
+ if (count != 0) {
+ os << ", ";
}
- return os;
+ os << "[" << count << "]={" << *it.get() << "}";
+ }
+ os << "}";
}
+ return os;
+ }
#endif
- private:
- /*
- * If this timeseries has more than one chunk meta, timeseries_meta_type_
- * is 1. Otherwise timeseries_meta_type_ is 0. It also should OR with mask
- * of chunk meta.
- */
- char timeseries_meta_type_;
-
- // Sum of chunk meta serialized size in List<ChunkMeta> of this timeseries.
- uint32_t chunk_meta_list_data_size_;
-
- // std::string measurement_name_;
- common::String measurement_name_;
- common::TsID ts_id_;
- common::TSDataType data_type_;
-
- /*
- * If TimeseriesIndex has only one ChunkMeta, then
- * TimeseriesIndex.statistic_ is duplicated with ChunkMeta.statistic_. In
- * this case, we do not serialize ChunkMeta.statistic_.
- */
- Statistic *statistic_;
- bool statistic_from_pa_;
- common::ByteStream chunk_meta_list_serialized_buf_;
- // common::PageArena page_arena_;
- common::SimpleList<ChunkMeta *> *chunk_meta_list_; // for deserialize_from
+private:
+ /*
+ * If this timeseries has more than one chunk meta, timeseries_meta_type_
+ * is 1. Otherwise timeseries_meta_type_ is 0. It also should OR with mask
+ * of chunk meta.
+ */
+ char timeseries_meta_type_;
+
+ // Sum of chunk meta serialized size in List<ChunkMeta> of this timeseries.
+ uint32_t chunk_meta_list_data_size_;
+
+ // std::string measurement_name_;
+ common::String measurement_name_;
+ common::TsID ts_id_;
+ common::TSDataType data_type_;
+
+ /*
+ * If TimeseriesIndex has only one ChunkMeta, then
+ * TimeseriesIndex.statistic_ is duplicated with ChunkMeta.statistic_. In
+ * this case, we do not serialize ChunkMeta.statistic_.
+ */
+ Statistic *statistic_;
+ bool statistic_from_pa_;
+ common::ByteStream chunk_meta_list_serialized_buf_;
+ // common::PageArena page_arena_;
+ common::SimpleList<ChunkMeta *> *chunk_meta_list_; // for deserialize_from
};
class AlignedTimeseriesIndex : public ITimeseriesIndex {
- public:
- TimeseriesIndex *time_ts_idx_;
- TimeseriesIndex *value_ts_idx_;
-
- AlignedTimeseriesIndex() {}
- ~AlignedTimeseriesIndex() {}
- virtual common::SimpleList<ChunkMeta *> *get_time_chunk_meta_list() const {
- return time_ts_idx_->get_chunk_meta_list();
- }
- virtual common::SimpleList<ChunkMeta *> *get_value_chunk_meta_list() const
{
- return value_ts_idx_->get_chunk_meta_list();
- }
-
- virtual common::String get_measurement_name() {
- return value_ts_idx_->get_measurement_name();
- }
- virtual common::TSDataType get_data_type() const {
- return time_ts_idx_->get_data_type();
- }
- virtual Statistic *get_statistic() const {
- return value_ts_idx_->get_statistic();
- }
+public:
+ TimeseriesIndex *time_ts_idx_;
+ TimeseriesIndex *value_ts_idx_;
+
+ AlignedTimeseriesIndex() {}
+ ~AlignedTimeseriesIndex() {}
+ virtual common::SimpleList<ChunkMeta *> *get_time_chunk_meta_list() const {
+ return time_ts_idx_->get_chunk_meta_list();
+ }
+ virtual common::SimpleList<ChunkMeta *> *get_value_chunk_meta_list() const {
+ return value_ts_idx_->get_chunk_meta_list();
+ }
+
+ virtual common::String get_measurement_name() {
+ return value_ts_idx_->get_measurement_name();
+ }
+ virtual common::TSDataType get_data_type() const {
+ return time_ts_idx_->get_data_type();
+ }
+ virtual Statistic *get_statistic() const {
+ return value_ts_idx_->get_statistic();
+ }
#ifndef NDEBUG
- friend std::ostream &operator<<(std::ostream &os,
- const AlignedTimeseriesIndex &tsi) {
- os << "time_ts_idx=" << *tsi.time_ts_idx_;
- os << ", value_ts_idx=" << *tsi.value_ts_idx_;
- return os;
- }
+ friend std::ostream &operator<<(std::ostream &os,
+ const AlignedTimeseriesIndex &tsi) {
+ os << "time_ts_idx=" << *tsi.time_ts_idx_;
+ os << ", value_ts_idx=" << *tsi.value_ts_idx_;
+ return os;
+ }
#endif
};
class TSMIterator {
- public:
- explicit TSMIterator(
- common::SimpleList<ChunkGroupMeta *> &chunk_group_meta_list)
- : chunk_group_meta_list_(chunk_group_meta_list),
- chunk_group_meta_iter_(),
- chunk_meta_iter_() {}
-
- // sort => iterate
- int init();
- bool has_next() const;
- int get_next(common::String &ret_device_name,
- common::String &ret_measurement_name,
- TimeseriesIndex &ret_ts_index);
-
- private:
- /*
- FORCE_INLINE bool is_same_measurement_name(const common::String &s1, const
- common::String &s2)
- {
- return s1.equal_to(s2);
- }
- */
- private:
- common::SimpleList<ChunkGroupMeta *> &chunk_group_meta_list_;
- common::SimpleList<ChunkGroupMeta *>::Iterator chunk_group_meta_iter_;
- common::SimpleList<ChunkMeta *>::Iterator chunk_meta_iter_;
-
- // timeseries measurenemnt chunk meta info
- // map <device_name, <measurement_name, vector<chunk_meta>>>
- std::map<common::String, std::map<common::String, std::vector<ChunkMeta
*>>>
- tsm_chunk_meta_info_;
-
- // device iterator
- std::map<common::String,
- std::map<common::String, std::vector<ChunkMeta *>>>::iterator
- tsm_device_iter_;
-
- // measurement iterator
- std::map<common::String, std::vector<ChunkMeta *>>::iterator
- tsm_measurement_iter_;
+public:
+ explicit TSMIterator(
+ common::SimpleList<ChunkGroupMeta *> &chunk_group_meta_list)
+ : chunk_group_meta_list_(chunk_group_meta_list),
+ chunk_group_meta_iter_(),
+ chunk_meta_iter_() {}
+
+ // sort => iterate
+ int init();
+ bool has_next() const;
+ int get_next(std::shared_ptr<IDeviceID> &ret_device_id,
+ common::String &ret_measurement_name,
+ TimeseriesIndex &ret_ts_index);
+
+private:
+ /*
+ FORCE_INLINE bool is_same_measurement_name(const common::String &s1, const
+ common::String &s2)
+ {
+ return s1.equal_to(s2);
+ }
+ */
+private:
Review Comment:
remove
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
