This is an automated email from the ASF dual-hosted git repository. colinlee pushed a commit to branch colin_support_read_tree in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit 26ea7fb6f8fa355a62c6183d4f0fefb495da1c10 Author: ColinLee <[email protected]> AuthorDate: Thu Nov 20 10:34:17 2025 +0800 Support query on tree and modify clang-format. --- cpp/.clang-format | 3 +- cpp/src/common/tsblock/tsblock.h | 74 +++--- cpp/src/common/tsfile_common.h | 295 +++++++++++---------- cpp/src/file/tsfile_io_reader.cc | 154 +++++------ .../reader/block/single_device_tsblock_reader.cc | 10 +- cpp/src/reader/device_meta_iterator.cc | 7 +- cpp/src/reader/device_meta_iterator.h | 10 + cpp/src/reader/imeta_data_querier.h | 5 +- cpp/src/reader/meta_data_querier.cc | 6 + cpp/src/reader/meta_data_querier.h | 3 + cpp/src/reader/table_query_executor.cc | 125 ++++++++- cpp/src/reader/table_query_executor.h | 22 +- cpp/src/reader/task/device_task_iterator.h | 15 ++ cpp/src/reader/tsfile_reader.cc | 38 ++- cpp/src/reader/tsfile_reader.h | 45 ++-- .../reader/tree_view/tsfile_reader_tree_test.cc | 88 +++++- 16 files changed, 594 insertions(+), 306 deletions(-) diff --git a/cpp/.clang-format b/cpp/.clang-format index 26651a2c..da6eee61 100644 --- a/cpp/.clang-format +++ b/cpp/.clang-format @@ -65,7 +65,6 @@ ConstructorInitializerIndentWidth: 4 ContinuationIndentWidth: 4 Cpp11BracedListStyle: true DeriveLineEnding: true -DerivePointerAlignment: true DisableFormat: false EmptyLineAfterAccessModifier: Never EmptyLineBeforeAccessModifier: LogicalBlock @@ -207,7 +206,7 @@ SpacesInParentheses: false SpacesInSquareBrackets: false SpaceBeforeSquareBrackets: false BitFieldColonSpacing: Both -Standard: Auto +Standard: Cpp11 StatementAttributeLikeMacros: - Q_EMIT StatementMacros: diff --git a/cpp/src/common/tsblock/tsblock.h b/cpp/src/common/tsblock/tsblock.h index a0e94391..dce94f8a 100644 --- a/cpp/src/common/tsblock/tsblock.h +++ b/cpp/src/common/tsblock/tsblock.h @@ -44,7 +44,7 @@ class TsBlock { * information, such as insert scenarios, etc. Then we will use the given * number of rows */ - explicit TsBlock(TupleDesc *tupledesc, uint32_t max_row_count = 0) + explicit TsBlock(TupleDesc* tupledesc, uint32_t max_row_count = 0) : capacity_(g_config_value_.tsblock_max_memory_), row_count_(0), max_row_count_(max_row_count), @@ -60,9 +60,9 @@ class TsBlock { FORCE_INLINE uint32_t get_row_count() const { return row_count_; } - FORCE_INLINE TupleDesc *get_tuple_desc() const { return tuple_desc_; } + FORCE_INLINE TupleDesc* get_tuple_desc() const { return tuple_desc_; } - FORCE_INLINE Vector *get_vector(uint32_t index) { return vectors_[index]; } + FORCE_INLINE Vector* get_vector(uint32_t index) { return vectors_[index]; } FORCE_INLINE uint32_t get_column_count() const { return tuple_desc_->get_column_count(); @@ -104,8 +104,8 @@ class TsBlock { row_count_ = 0; } - FORCE_INLINE static int create_tsblock(TupleDesc *tupledesc, - TsBlock *&ret_tsblock, + FORCE_INLINE static int create_tsblock(TupleDesc* tupledesc, + TsBlock*& ret_tsblock, uint32_t max_row_count = 0) { int ret = common::E_OK; if (ret_tsblock == nullptr) { @@ -119,13 +119,13 @@ class TsBlock { } int init(); - void tsblock_to_json(ByteStream *byte_stream); + void tsblock_to_json(ByteStream* byte_stream); std::string debug_string(); private: int build_vector(common::TSDataType type, uint32_t row_count); - void write_data(ByteStream *__restrict byte_stream, char *__restrict val, + void write_data(ByteStream* __restrict byte_stream, char* __restrict val, uint32_t len, bool has_null, TSDataType type); private: @@ -134,13 +134,13 @@ class TsBlock { uint32_t max_row_count_; common::BitMap select_list_; - TupleDesc *tuple_desc_; - std::vector<Vector *> vectors_; + TupleDesc* tuple_desc_; + std::vector<Vector*> vectors_; }; class RowAppender { public: - explicit RowAppender(TsBlock *tsblock) : tsblock_(tsblock) {} + explicit RowAppender(TsBlock* tsblock) : tsblock_(tsblock) {} ~RowAppender() {} // todo:(yanghao) maybe need to consider select-list @@ -157,25 +157,37 @@ class RowAppender { tsblock_->row_count_--; } - FORCE_INLINE void append(uint32_t slot_index, const char *value, + FORCE_INLINE void append(uint32_t slot_index, const char* value, uint32_t len) { ASSERT(slot_index < tsblock_->tuple_desc_->get_column_count()); - Vector *vec = tsblock_->vectors_[slot_index]; - vec->append(value, len); + Vector* vec = tsblock_->vectors_[slot_index]; + // TODO(Colin): Refine this. + TSDataType datatype = vec->get_vector_type(); + if (len == 4 && datatype == INT64) { + int32_t int32_val = *reinterpret_cast<const int32_t*>(value); + int64_t int64_val = static_cast<int64_t>(int32_val); + vec->append(reinterpret_cast<const char*>(&int64_val), 8); + } else if (len == 4 && datatype == DOUBLE) { + float float_val = *reinterpret_cast<const float*>(value); + double double_val = static_cast<double>(float_val); + vec->append(reinterpret_cast<const char*>(&double_val), 8); + } else { + vec->append(value, len); + } } FORCE_INLINE void append_null(uint32_t slot_index) { - Vector *vec = tsblock_->vectors_[slot_index]; + Vector* vec = tsblock_->vectors_[slot_index]; vec->set_null(tsblock_->row_count_ - 1); } private: - TsBlock *tsblock_; + TsBlock* tsblock_; }; class ColAppender { public: - ColAppender(uint32_t column_index, TsBlock *tsblock) + ColAppender(uint32_t column_index, TsBlock* tsblock) : column_index_(column_index), column_row_count_(0), tsblock_(tsblock) { ASSERT(column_index < tsblock_->tuple_desc_->get_column_count()); vec_ = tsblock_->vectors_[column_index]; @@ -194,7 +206,7 @@ class ColAppender { } } - FORCE_INLINE void append(const char *value, uint32_t len) { + FORCE_INLINE void append(const char* value, uint32_t len) { vec_->append(value, len); } @@ -211,7 +223,7 @@ class ColAppender { } return E_OK; } - FORCE_INLINE int fill(const char *value, uint32_t len, uint32_t end_index) { + FORCE_INLINE int fill(const char* value, uint32_t len, uint32_t end_index) { while (column_row_count_ < end_index) { if (!add_row()) { return E_INVALID_ARG; @@ -225,14 +237,14 @@ class ColAppender { private: uint32_t column_index_; uint32_t column_row_count_; - TsBlock *tsblock_; - Vector *vec_; + TsBlock* tsblock_; + Vector* vec_; }; // todo:(yanghao) need to deal with select-list class RowIterator { public: - explicit RowIterator(TsBlock *tsblock) : tsblock_(tsblock), row_id_(0) { + explicit RowIterator(TsBlock* tsblock) : tsblock_(tsblock), row_id_(0) { column_count_ = tsblock_->tuple_desc_->get_column_count(); } @@ -264,17 +276,17 @@ class RowIterator { FORCE_INLINE void update_row_id() { row_id_++; } - FORCE_INLINE char *read(uint32_t column_index, uint32_t *__restrict len, - bool *__restrict null) { + FORCE_INLINE char* read(uint32_t column_index, uint32_t* __restrict len, + bool* __restrict null) { ASSERT(column_index < column_count_); - Vector *vec = tsblock_->vectors_[column_index]; + Vector* vec = tsblock_->vectors_[column_index]; return vec->read(len, null, row_id_); } std::string debug_string(); // for debug private: - TsBlock *tsblock_; + TsBlock* tsblock_; uint32_t row_id_; // The line number currently being reader uint32_t column_count_; }; @@ -282,7 +294,7 @@ class RowIterator { // todo:(yanghao) need to deal with select-list class ColIterator { public: - ColIterator(uint32_t column_index, const TsBlock *tsblock) + ColIterator(uint32_t column_index, const TsBlock* tsblock) : column_index_(column_index), row_id_(0), tsblock_(tsblock) { ASSERT(column_index < tsblock_->tuple_desc_->get_column_count()); vec_ = tsblock_->vectors_[column_index]; @@ -303,22 +315,22 @@ class ColIterator { FORCE_INLINE TSDataType get_data_type() { return vec_->get_vector_type(); } - FORCE_INLINE char *read(uint32_t *__restrict len, bool *__restrict null) { + FORCE_INLINE char* read(uint32_t* __restrict len, bool* __restrict null) { return vec_->read(len, null, row_id_); } - FORCE_INLINE char *read(uint32_t *len) { return vec_->read(len); } + FORCE_INLINE char* read(uint32_t* len) { return vec_->read(len); } FORCE_INLINE uint32_t get_column_index() { return column_index_; } private: uint32_t column_index_; uint32_t row_id_; - const TsBlock *tsblock_; - Vector *vec_; + const TsBlock* tsblock_; + Vector* vec_; }; -int merge_tsblock_by_row(TsBlock *sea, TsBlock *river); +int merge_tsblock_by_row(TsBlock* sea, TsBlock* river); } // end namespace common #endif // COMMON_TSBLOCK_TSBLOCK_H diff --git a/cpp/src/common/tsfile_common.h b/cpp/src/common/tsfile_common.h index dd22ca40..39cd027e 100644 --- a/cpp/src/common/tsfile_common.h +++ b/cpp/src/common/tsfile_common.h @@ -39,7 +39,7 @@ namespace storage { -extern const char *MAGIC_STRING_TSFILE; +extern const char* MAGIC_STRING_TSFILE; constexpr int MAGIC_STRING_TSFILE_LEN = 6; extern const char VERSION_NUM_BYTE; extern const char CHUNK_GROUP_HEADER_MARKER; @@ -60,7 +60,7 @@ typedef int64_t TsFileID; struct PageHeader { uint32_t uncompressed_size_; uint32_t compressed_size_; - Statistic *statistic_; + Statistic* statistic_; PageHeader() : uncompressed_size_(0), compressed_size_(0), statistic_(nullptr) {} @@ -73,7 +73,7 @@ struct PageHeader { uncompressed_size_ = 0; compressed_size_ = 0; } - int deserialize_from(common::ByteStream &in, bool deserialize_stat, + 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( @@ -99,7 +99,7 @@ struct PageHeader { } #ifndef NDEBUG - friend std::ostream &operator<<(std::ostream &os, const PageHeader &h) { + 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) { @@ -132,7 +132,7 @@ struct ChunkHeader { ~ChunkHeader() = default; - int serialize_to(common::ByteStream &out) { + 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_var_str( @@ -148,7 +148,7 @@ struct ChunkHeader { } return ret; } - int deserialize_from(common::ByteStream &in) { + 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))) { @@ -157,18 +157,18 @@ struct ChunkHeader { } else if (RET_FAIL(common::SerializationUtil::read_var_uint(data_size_, in))) { } else if (RET_FAIL(common::SerializationUtil::read_char( - (char &)data_type_, in))) { + (char&)data_type_, in))) { } else if (RET_FAIL(common::SerializationUtil::read_char( - (char &)compression_type_, in))) { + (char&)compression_type_, in))) { } else if (RET_FAIL(common::SerializationUtil::read_char( - (char &)encoding_type_, in))) { + (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) { + 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_ @@ -197,7 +197,7 @@ struct ChunkMeta { common::String measurement_name_; common::TSDataType data_type_; int64_t offset_of_chunk_header_; - Statistic *statistic_; + Statistic* statistic_; char mask_; common::TSEncoding encoding_; common::CompressionType compression_type_; @@ -209,10 +209,10 @@ struct ChunkMeta { statistic_(nullptr), mask_(0) {} - int init(const common::String &measurement_name, + int init(const common::String& measurement_name, common::TSDataType data_type, int64_t offset_of_chunk_header, - Statistic *stat, char mask, common::TSEncoding encoding, - common::CompressionType compression_type, common::PageArena &pa) { + Statistic* stat, 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; @@ -223,10 +223,10 @@ struct ChunkMeta { compression_type_ = compression_type; return common::E_OK; } - FORCE_INLINE void clone_statistic_from(Statistic *stat) { + FORCE_INLINE void clone_statistic_from(Statistic* stat) { clone_statistic(stat, statistic_, data_type_); } - FORCE_INLINE int clone_from(ChunkMeta &that, common::PageArena *pa) { + 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; @@ -244,7 +244,7 @@ struct ChunkMeta { mask_ = that.mask_; return ret; } - int serialize_to(common::ByteStream &out, bool serialize_statistic) { + 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))) { @@ -253,8 +253,8 @@ struct ChunkMeta { } return ret; } - int deserialize_from(common::ByteStream &in, bool deserialize_stat, - common::PageArena *pa) { + 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))) { @@ -270,7 +270,7 @@ struct ChunkMeta { return ret; } #ifndef NDEBUG - friend std::ostream &operator<<(std::ostream &os, const ChunkMeta &cm) { + 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_ @@ -287,16 +287,16 @@ struct ChunkMeta { struct ChunkGroupMeta { std::shared_ptr<IDeviceID> device_id_; - common::SimpleList<ChunkMeta *> chunk_meta_list_; + common::SimpleList<ChunkMeta*> chunk_meta_list_; - explicit ChunkGroupMeta(common::PageArena *pa_ptr) + 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) { + FORCE_INLINE int push(ChunkMeta* cm) { return chunk_meta_list_.push_back(cm); } }; @@ -305,13 +305,13 @@ class ITimeseriesIndex { public: ITimeseriesIndex() {} ~ITimeseriesIndex() {} - virtual common::SimpleList<ChunkMeta *> *get_chunk_meta_list() const { + virtual common::SimpleList<ChunkMeta*>* get_chunk_meta_list() const { return nullptr; } - virtual common::SimpleList<ChunkMeta *> *get_time_chunk_meta_list() const { + virtual common::SimpleList<ChunkMeta*>* get_time_chunk_meta_list() const { return nullptr; } - virtual common::SimpleList<ChunkMeta *> *get_value_chunk_meta_list() const { + virtual common::SimpleList<ChunkMeta*>* get_value_chunk_meta_list() const { return nullptr; } @@ -321,7 +321,7 @@ class ITimeseriesIndex { virtual common::TSDataType get_data_type() const { return common::INVALID_DATATYPE; } - virtual Statistic *get_statistic() const { return nullptr; } + virtual Statistic* get_statistic() const { return nullptr; } }; /* @@ -368,19 +368,18 @@ class TimeseriesIndex : public ITimeseriesIndex { } } - int add_chunk_meta(ChunkMeta *chunk_meta, bool serialize_statistic); - FORCE_INLINE int set_measurement_name(common::String &measurement_name, - common::PageArena &pa) { + 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) { + 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() const { return measurement_name_; } - virtual inline common::SimpleList<ChunkMeta *> *get_chunk_meta_list() - const { + 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) { @@ -405,7 +404,7 @@ class TimeseriesIndex : public ITimeseriesIndex { statistic_->reset(); return common::E_OK; } - virtual Statistic *get_statistic() const { return statistic_; } + virtual Statistic* get_statistic() const { return statistic_; } common::TsID get_ts_id() const { return ts_id_; } FORCE_INLINE void finish() { @@ -413,7 +412,7 @@ class TimeseriesIndex : public ITimeseriesIndex { chunk_meta_list_serialized_buf_.total_size(); } - int serialize_to(common::ByteStream &out) { + int serialize_to(common::ByteStream& out) { int ret = common::E_OK; if (RET_FAIL(common::SerializationUtil::write_char( timeseries_meta_type_, out))) { @@ -430,14 +429,14 @@ class TimeseriesIndex : public ITimeseriesIndex { return ret; } - int deserialize_from(common::ByteStream &in, common::PageArena *pa) { + 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))) { + (char&)data_type_, in))) { } else if (RET_FAIL(common::SerializationUtil::read_var_uint( chunk_meta_list_data_size_, in))) { } else if (nullptr == @@ -447,22 +446,22 @@ class TimeseriesIndex : public ITimeseriesIndex { } else if (RET_FAIL(statistic_->deserialize_from(in))) { } else { statistic_from_pa_ = true; - void *chunk_meta_list_buf = pa->alloc(sizeof(*chunk_meta_list_)); + 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); + 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)); + void* cm_buf = pa->alloc(sizeof(ChunkMeta)); if (IS_NULL(cm_buf)) { ret = common::E_OOM; } else { - ChunkMeta *cm = new (cm_buf) ChunkMeta; + ChunkMeta* cm = new (cm_buf) ChunkMeta; cm->measurement_name_.shallow_copy_from( this->measurement_name_); cm->data_type_ = this->data_type_; @@ -477,7 +476,7 @@ class TimeseriesIndex : public ITimeseriesIndex { return ret; } - int clone_from(const TimeseriesIndex &that, common::PageArena *pa) { + 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_; @@ -496,20 +495,20 @@ class TimeseriesIndex : public ITimeseriesIndex { } if (that.chunk_meta_list_ != nullptr) { - void *buf = pa->alloc(sizeof(*chunk_meta_list_)); + 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; + 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)); + 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; + 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))) { } @@ -519,8 +518,8 @@ class TimeseriesIndex : public ITimeseriesIndex { return ret; } #ifndef NDEBUG - friend std::ostream &operator<<(std::ostream &os, - const TimeseriesIndex &tsi) { + 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_ @@ -531,7 +530,7 @@ class TimeseriesIndex : public ITimeseriesIndex { if (tsi.chunk_meta_list_) { os << ", chunk_meta_list={"; int count = 0; - common::SimpleList<ChunkMeta *>::Iterator it = + common::SimpleList<ChunkMeta*>::Iterator it = tsi.chunk_meta_list_->begin(); for (; it != tsi.chunk_meta_list_->end(); it++, count++) { if (count != 0) { @@ -565,24 +564,24 @@ class TimeseriesIndex : public ITimeseriesIndex { * TimeseriesIndex.statistic_ is duplicated with ChunkMeta.statistic_. In * this case, we do not serialize ChunkMeta.statistic_. */ - Statistic *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 + common::SimpleList<ChunkMeta*>* chunk_meta_list_; // for deserialize_from }; class AlignedTimeseriesIndex : public ITimeseriesIndex { public: - TimeseriesIndex *time_ts_idx_; - TimeseriesIndex *value_ts_idx_; + TimeseriesIndex* time_ts_idx_; + TimeseriesIndex* value_ts_idx_; AlignedTimeseriesIndex() {} ~AlignedTimeseriesIndex() {} - virtual common::SimpleList<ChunkMeta *> *get_time_chunk_meta_list() const { + 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 { + virtual common::SimpleList<ChunkMeta*>* get_value_chunk_meta_list() const { return value_ts_idx_->get_chunk_meta_list(); } @@ -592,13 +591,13 @@ class AlignedTimeseriesIndex : public ITimeseriesIndex { virtual common::TSDataType get_data_type() const { return time_ts_idx_->get_data_type(); } - virtual Statistic *get_statistic() const { + virtual Statistic* get_statistic() const { return value_ts_idx_->get_statistic(); } #ifndef NDEBUG - friend std::ostream &operator<<(std::ostream &os, - const AlignedTimeseriesIndex &tsi) { + 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; @@ -609,7 +608,7 @@ class AlignedTimeseriesIndex : public ITimeseriesIndex { class TSMIterator { public: explicit TSMIterator( - common::SimpleList<ChunkGroupMeta *> &chunk_group_meta_list) + common::SimpleList<ChunkGroupMeta*>& chunk_group_meta_list) : chunk_group_meta_list_(chunk_group_meta_list), chunk_group_meta_iter_(), chunk_meta_iter_() {} @@ -617,38 +616,38 @@ class TSMIterator { // 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); + int get_next(std::shared_ptr<IDeviceID>& ret_device_id, + common::String& ret_measurement_name, + TimeseriesIndex& ret_ts_index); private: - common::SimpleList<ChunkGroupMeta *> &chunk_group_meta_list_; - common::SimpleList<ChunkGroupMeta *>::Iterator chunk_group_meta_iter_; - common::SimpleList<ChunkMeta *>::Iterator chunk_meta_iter_; + 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<std::shared_ptr<IDeviceID>, - std::map<common::String, std::vector<ChunkMeta *>>> + std::map<common::String, std::vector<ChunkMeta*>>> tsm_chunk_meta_info_; // device iterator std::map<std::shared_ptr<IDeviceID>, - std::map<common::String, std::vector<ChunkMeta *>>>::iterator + std::map<common::String, std::vector<ChunkMeta*>>>::iterator tsm_device_iter_; // measurement iterator - std::map<common::String, std::vector<ChunkMeta *>>::iterator + std::map<common::String, std::vector<ChunkMeta*>>::iterator tsm_measurement_iter_; }; /* =============== TsFile Index ================ */ struct IComparable { virtual ~IComparable() = default; - virtual bool operator<(const IComparable &other) const = 0; - virtual bool operator>(const IComparable &other) const = 0; - virtual bool operator==(const IComparable &other) const = 0; - virtual int compare(const IComparable &other) { + virtual bool operator<(const IComparable& other) const = 0; + virtual bool operator>(const IComparable& other) const = 0; + virtual bool operator==(const IComparable& other) const = 0; + virtual int compare(const IComparable& other) { if (this->operator<(other)) { return -1; } else if (this->operator==(other)) { @@ -663,27 +662,27 @@ struct IComparable { struct DeviceIDComparable : IComparable { std::shared_ptr<IDeviceID> device_id_; - explicit DeviceIDComparable(const std::shared_ptr<IDeviceID> &device_id) + explicit DeviceIDComparable(const std::shared_ptr<IDeviceID>& device_id) : device_id_(device_id) {} - bool operator<(const IComparable &other) const override { - const auto *other_device = - dynamic_cast<const DeviceIDComparable *>(&other); + bool operator<(const IComparable& other) const override { + const auto* other_device = + dynamic_cast<const DeviceIDComparable*>(&other); if (!other_device) throw std::runtime_error("Incompatible comparison"); return *device_id_ < *other_device->device_id_; } - bool operator>(const IComparable &other) const override { - const auto *other_device = - dynamic_cast<const DeviceIDComparable *>(&other); + bool operator>(const IComparable& other) const override { + const auto* other_device = + dynamic_cast<const DeviceIDComparable*>(&other); if (!other_device) throw std::runtime_error("Incompatible comparison"); return *device_id_ != *other_device->device_id_ && !(*device_id_ < *other_device->device_id_); } - bool operator==(const IComparable &other) const override { - const auto *other_device = - dynamic_cast<const DeviceIDComparable *>(&other); + bool operator==(const IComparable& other) const override { + const auto* other_device = + dynamic_cast<const DeviceIDComparable*>(&other); if (!other_device) throw std::runtime_error("Incompatible comparison"); return *device_id_ == *other_device->device_id_; } @@ -696,25 +695,25 @@ struct DeviceIDComparable : IComparable { struct StringComparable : IComparable { std::string value_; - explicit StringComparable(const std::string &value) : value_(value) {} + explicit StringComparable(const std::string& value) : value_(value) {} - bool operator<(const IComparable &other) const override { - const auto *other_string = - dynamic_cast<const StringComparable *>(&other); + bool operator<(const IComparable& other) const override { + const auto* other_string = + dynamic_cast<const StringComparable*>(&other); if (!other_string) throw std::runtime_error("Incompatible comparison"); return value_ < other_string->value_; } - bool operator>(const IComparable &other) const override { - const auto *other_string = - dynamic_cast<const StringComparable *>(&other); + bool operator>(const IComparable& other) const override { + const auto* other_string = + dynamic_cast<const StringComparable*>(&other); if (!other_string) throw std::runtime_error("Incompatible comparison"); return value_ > other_string->value_; } - bool operator==(const IComparable &other) const override { - const auto *other_string = - dynamic_cast<const StringComparable *>(&other); + bool operator==(const IComparable& other) const override { + const auto* other_string = + dynamic_cast<const StringComparable*>(&other); if (!other_string) throw std::runtime_error("Incompatible comparison"); return value_ == other_string->value_; } @@ -723,7 +722,7 @@ struct StringComparable : IComparable { }; struct IMetaIndexEntry { - static void self_destructor(IMetaIndexEntry *ptr) { + static void self_destructor(IMetaIndexEntry* ptr) { if (ptr) { ptr->~IMetaIndexEntry(); } @@ -731,9 +730,9 @@ struct IMetaIndexEntry { IMetaIndexEntry() = default; virtual ~IMetaIndexEntry() = default; - virtual int serialize_to(common::ByteStream &out) { return common::E_OK; } - virtual int deserialize_from(common::ByteStream &out, - common::PageArena *pa) { + virtual int serialize_to(common::ByteStream& out) { return common::E_OK; } + virtual int deserialize_from(common::ByteStream& out, + common::PageArena* pa) { return common::E_NOT_SUPPORT; } virtual int64_t get_offset() const = 0; @@ -743,11 +742,11 @@ struct IMetaIndexEntry { } virtual common::String get_name() const { return {}; } virtual std::shared_ptr<IDeviceID> get_device_id() const { return nullptr; } - virtual std::shared_ptr<IMetaIndexEntry> clone(common::PageArena *pa) = 0; + virtual std::shared_ptr<IMetaIndexEntry> clone(common::PageArena* pa) = 0; #ifndef NDEBUG - virtual void print(std::ostream &os) const {} - friend std::ostream &operator<<(std::ostream &os, - const IMetaIndexEntry &entry) { + virtual void print(std::ostream& os) const {} + friend std::ostream& operator<<(std::ostream& os, + const IMetaIndexEntry& entry) { entry.print(os); return os; } @@ -760,19 +759,19 @@ struct DeviceMetaIndexEntry : IMetaIndexEntry { DeviceMetaIndexEntry() = default; - DeviceMetaIndexEntry(const std::shared_ptr<IDeviceID> &device_id, + DeviceMetaIndexEntry(const std::shared_ptr<IDeviceID>& device_id, const int64_t offset) : device_id_(device_id), offset_(offset) {} ~DeviceMetaIndexEntry() override = default; - static void self_deleter(DeviceMetaIndexEntry *ptr) { + static void self_deleter(DeviceMetaIndexEntry* ptr) { if (ptr) { ptr->~DeviceMetaIndexEntry(); } } - int serialize_to(common::ByteStream &out) override { + int serialize_to(common::ByteStream& out) override { int ret = common::E_OK; if (RET_FAIL(device_id_->serialize(out))) { } else if (RET_FAIL( @@ -781,10 +780,10 @@ struct DeviceMetaIndexEntry : IMetaIndexEntry { return ret; } - std::shared_ptr<IDeviceID> &get_device_id() { return device_id_; } + std::shared_ptr<IDeviceID>& get_device_id() { return device_id_; } - int deserialize_from(common::ByteStream &in, - common::PageArena *pa) override { + int deserialize_from(common::ByteStream& in, + common::PageArena* pa) override { int ret = common::E_OK; device_id_ = std::make_shared<StringArrayDeviceID>("init"); if (RET_FAIL(device_id_->deserialize(in))) { @@ -804,11 +803,11 @@ struct DeviceMetaIndexEntry : IMetaIndexEntry { std::shared_ptr<IDeviceID> get_device_id() const override { return device_id_; } - std::shared_ptr<IMetaIndexEntry> clone(common::PageArena *pa) override { + std::shared_ptr<IMetaIndexEntry> clone(common::PageArena* pa) override { return std::make_shared<DeviceMetaIndexEntry>(device_id_, offset_); } #ifndef NDEBUG - void print(std::ostream &os) const override { + void print(std::ostream& os) const override { os << "name=" << device_id_ << ", offset=" << offset_; } #endif @@ -821,19 +820,19 @@ struct MeasurementMetaIndexEntry : IMetaIndexEntry { ~MeasurementMetaIndexEntry() override = default; MeasurementMetaIndexEntry() = default; - MeasurementMetaIndexEntry(const common::String &name, const int64_t offset, - common::PageArena &pa) { + MeasurementMetaIndexEntry(const common::String& name, const int64_t offset, + common::PageArena& pa) { offset_ = offset; name_.dup_from(name, pa); } - FORCE_INLINE int init(const std::string &str, const int64_t offset, - common::PageArena &pa) { + FORCE_INLINE int init(const std::string& str, const int64_t offset, + common::PageArena& pa) { offset_ = offset; return name_.dup_from(str, pa); } - int serialize_to(common::ByteStream &out) override { + int serialize_to(common::ByteStream& out) override { int ret = common::E_OK; if (RET_FAIL(common::SerializationUtil::write_mystring(name_, out))) { } else if (RET_FAIL( @@ -842,8 +841,8 @@ struct MeasurementMetaIndexEntry : IMetaIndexEntry { return ret; } - int deserialize_from(common::ByteStream &in, - common::PageArena *pa) override { + int deserialize_from(common::ByteStream& in, + common::PageArena* pa) override { int ret = common::E_OK; if (RET_FAIL(common::SerializationUtil::read_mystring(name_, pa, in))) { } else if (RET_FAIL(common::SerializationUtil::read_i64(offset_, in))) { @@ -863,11 +862,11 @@ struct MeasurementMetaIndexEntry : IMetaIndexEntry { std::shared_ptr<IDeviceID> get_device_id() const override { return nullptr; } - std::shared_ptr<IMetaIndexEntry> clone(common::PageArena *pa) override { + std::shared_ptr<IMetaIndexEntry> clone(common::PageArena* pa) override { return std::make_shared<MeasurementMetaIndexEntry>(name_, offset_, *pa); } #ifndef NDEBUG - void print(std::ostream &os) const override { + void print(std::ostream& os) const override { os << "name=" << name_ << ", offset=" << offset_; } #endif @@ -881,7 +880,7 @@ enum MetaIndexNodeType { INVALID_META_NODE_TYPE = 4, }; #ifndef NDEBUG -static const char *meta_index_node_type_names[5] = { +static const char* meta_index_node_type_names[5] = { "INTERNAL_DEVICE", "LEAF_DEVICE", "INTERNAL_MEASUREMENT", "LEAF_MEASUREMENT", "INVALID_META_NODE_TYPE"}; #endif @@ -892,9 +891,9 @@ struct MetaIndexNode { std::vector<std::shared_ptr<IMetaIndexEntry>> children_; int64_t end_offset_; MetaIndexNodeType node_type_; - common::PageArena *pa_; + common::PageArena* pa_; - explicit MetaIndexNode(common::PageArena *pa) + explicit MetaIndexNode(common::PageArena* pa) : children_(), end_offset_(0), node_type_(), pa_(pa) {} std::shared_ptr<IMetaIndexEntry> peek() { @@ -906,7 +905,7 @@ struct MetaIndexNode { ~MetaIndexNode() {} - static void self_deleter(MetaIndexNode *ptr) { + static void self_deleter(MetaIndexNode* ptr) { if (ptr) { ptr->~MetaIndexNode(); } @@ -914,10 +913,10 @@ struct MetaIndexNode { int binary_search_children( std::shared_ptr<IComparable> key, bool exact_search, - std::shared_ptr<IMetaIndexEntry> &ret_index_entry, - int64_t &ret_end_offset); + std::shared_ptr<IMetaIndexEntry>& ret_index_entry, + int64_t& ret_end_offset); - int serialize_to(common::ByteStream &out) { + int serialize_to(common::ByteStream& out) { int ret = common::E_OK; #if DEBUG_SE int64_t start_pos = out.total_size(); @@ -946,12 +945,12 @@ struct MetaIndexNode { return ret; } - int deserialize_from(const char *buf, int len) { + int deserialize_from(const char* buf, int len) { common::ByteStream bs; bs.wrap_from(buf, len); return deserialize_from(bs); } - int deserialize_from(common::ByteStream &in) { + int deserialize_from(common::ByteStream& in) { int ret = common::E_OK; uint32_t children_size = 0; if (RET_FAIL( @@ -959,7 +958,7 @@ struct MetaIndexNode { return ret; } for (uint32_t i = 0; i < children_size && IS_SUCC(ret); i++) { - void *entry_buf = pa_->alloc(sizeof(MeasurementMetaIndexEntry)); + void* entry_buf = pa_->alloc(sizeof(MeasurementMetaIndexEntry)); if (IS_NULL(entry_buf)) { return common::E_OOM; } @@ -987,12 +986,12 @@ struct MetaIndexNode { #endif return ret; } - int device_deserialize_from(const char *buf, int len) { + int device_deserialize_from(const char* buf, int len) { common::ByteStream bs; bs.wrap_from(buf, len); return device_deserialize_from(bs); } - int device_deserialize_from(common::ByteStream &in) { + int device_deserialize_from(common::ByteStream& in) { int ret = common::E_OK; uint32_t children_size = 0; if (RET_FAIL( @@ -1000,11 +999,11 @@ struct MetaIndexNode { return ret; } for (uint32_t i = 0; i < children_size && IS_SUCC(ret); i++) { - void *entry_buf = pa_->alloc(sizeof(DeviceMetaIndexEntry)); + void* entry_buf = pa_->alloc(sizeof(DeviceMetaIndexEntry)); if (IS_NULL(entry_buf)) { return common::E_OOM; } - auto *entry_ptr = new (entry_buf) DeviceMetaIndexEntry(); + auto* entry_ptr = new (entry_buf) DeviceMetaIndexEntry(); auto entry = std::shared_ptr<DeviceMetaIndexEntry>( entry_ptr, DeviceMetaIndexEntry::self_deleter); if (RET_FAIL(entry->deserialize_from(in, pa_))) { @@ -1030,8 +1029,8 @@ struct MetaIndexNode { } #ifndef NDEBUG - friend std::ostream &operator<<(std::ostream &os, - const MetaIndexNode &node) { + friend std::ostream& operator<<(std::ostream& os, + const MetaIndexNode& node) { os << "end_offset=" << node.end_offset_ << ", node_type=" << meta_index_node_type_names[node.node_type_]; @@ -1073,16 +1072,16 @@ struct TsFileMeta { DeviceNodeMap; std::map<std::string, std::shared_ptr<MetaIndexNode>> table_metadata_index_node_map_; - std::unordered_map<std::string, std::string *> tsfile_properties_; + std::unordered_map<std::string, std::string*> tsfile_properties_; typedef std::unordered_map<std::string, std::shared_ptr<TableSchema>> TableSchemasMap; TableSchemasMap table_schemas_; int64_t meta_offset_; - BloomFilter *bloom_filter_; - common::PageArena *page_arena_; + BloomFilter* bloom_filter_; + common::PageArena* page_arena_; - int get_table_metaindex_node(const std::string &table_name, - MetaIndexNode *&ret_node) { + int get_table_metaindex_node(const std::string& table_name, + MetaIndexNode*& ret_node) { std::map<std::string, std::shared_ptr<MetaIndexNode>>::iterator it = table_metadata_index_node_map_.find(table_name); if (it == table_metadata_index_node_map_.end()) { @@ -1092,8 +1091,8 @@ struct TsFileMeta { return common::E_OK; } - int get_table_schema(const std::string &table_name, - std::shared_ptr<TableSchema> &ret_schema) { + int get_table_schema(const std::string& table_name, + std::shared_ptr<TableSchema>& ret_schema) { TableSchemasMap::iterator it = table_schemas_.find(table_name); if (it == table_schemas_.end()) { return common::E_TABLE_NOT_EXIST; @@ -1105,7 +1104,7 @@ struct TsFileMeta { TsFileMeta() : meta_offset_(0), bloom_filter_(nullptr), page_arena_(nullptr) {} - explicit TsFileMeta(common::PageArena *pa) + explicit TsFileMeta(common::PageArena* pa) : meta_offset_(0), bloom_filter_(nullptr), page_arena_(pa) {} ~TsFileMeta() { if (bloom_filter_ != nullptr) { @@ -1114,19 +1113,21 @@ struct TsFileMeta { for (auto properties : tsfile_properties_) { if (properties.second != nullptr) { delete properties.second; + properties.second = nullptr; } } + tsfile_properties_.clear(); table_metadata_index_node_map_.clear(); table_schemas_.clear(); } - int serialize_to(common::ByteStream &out); + int serialize_to(common::ByteStream& out); - int deserialize_from(common::ByteStream &in); + int deserialize_from(common::ByteStream& in); #ifndef NDEBUG - friend std::ostream &operator<<(std::ostream &os, - const TsFileMeta &tsfile_meta) { + friend std::ostream& operator<<(std::ostream& os, + const TsFileMeta& tsfile_meta) { os << "meta_offset=" << tsfile_meta.meta_offset_; return os; } diff --git a/cpp/src/file/tsfile_io_reader.cc b/cpp/src/file/tsfile_io_reader.cc index 273f09a4..92c05c92 100644 --- a/cpp/src/file/tsfile_io_reader.cc +++ b/cpp/src/file/tsfile_io_reader.cc @@ -24,7 +24,7 @@ using namespace common; namespace storage { -int TsFileIOReader::init(const std::string &file_path) { +int TsFileIOReader::init(const std::string& file_path) { int ret = E_OK; read_file_ = new ReadFile; read_file_created_ = true; @@ -33,7 +33,7 @@ int TsFileIOReader::init(const std::string &file_path) { return ret; } -int TsFileIOReader::init(ReadFile *read_file) { +int TsFileIOReader::init(ReadFile* read_file) { if (IS_NULL(read_file)) { ASSERT(false); return E_INVALID_ARG; @@ -56,9 +56,9 @@ void TsFileIOReader::reset() { } int TsFileIOReader::alloc_ssi(std::shared_ptr<IDeviceID> device_id, - const std::string &measurement_name, - TsFileSeriesScanIterator *&ssi, - common::PageArena &pa, Filter *time_filter) { + const std::string& measurement_name, + TsFileSeriesScanIterator*& ssi, + common::PageArena& pa, Filter* time_filter) { int ret = E_OK; if (RET_FAIL(load_tsfile_meta_if_necessary())) { } else { @@ -80,7 +80,7 @@ int TsFileIOReader::alloc_ssi(std::shared_ptr<IDeviceID> device_id, return ret; } -void TsFileIOReader::revert_ssi(TsFileSeriesScanIterator *ssi) { +void TsFileIOReader::revert_ssi(TsFileSeriesScanIterator* ssi) { if (ssi != nullptr) { ssi->destroy(); delete ssi; @@ -89,12 +89,12 @@ void TsFileIOReader::revert_ssi(TsFileSeriesScanIterator *ssi) { int TsFileIOReader::get_device_timeseries_meta_without_chunk_meta( std::shared_ptr<IDeviceID> device_id, - std::vector<ITimeseriesIndex *> ×eries_indexs, PageArena &pa) { + std::vector<ITimeseriesIndex*>& timeseries_indexs, PageArena& pa) { int ret = E_OK; load_tsfile_meta_if_necessary(); std::shared_ptr<IMetaIndexEntry> meta_index_entry; int64_t end_offset; - std::vector<std::pair<std::shared_ptr<IMetaIndexEntry>, int64_t> > + std::vector<std::pair<std::shared_ptr<IMetaIndexEntry>, int64_t>> meta_index_entry_list; if (RET_FAIL(load_device_index_entry( std::make_shared<DeviceIDComparable>(device_id), meta_index_entry, @@ -108,8 +108,8 @@ int TsFileIOReader::get_device_timeseries_meta_without_chunk_meta( return ret; } -bool TsFileIOReader::filter_stasify(ITimeseriesIndex *ts_index, - Filter *time_filter) { +bool TsFileIOReader::filter_stasify(ITimeseriesIndex* ts_index, + Filter* time_filter) { ASSERT(ts_index->get_statistic() != nullptr); return time_filter->satisfy(ts_index->get_statistic()); } @@ -141,7 +141,7 @@ int TsFileIOReader::load_tsfile_meta() { // Step 1: reader the tsfile_meta_size // 1.1 prepare reader buffer int32_t alloc_size = UTIL_MIN(TSFILE_READ_IO_SIZE, file_size()); - char *read_buf = (char *)mem_alloc(alloc_size, MOD_TSFILE_READER); + char* read_buf = (char*)mem_alloc(alloc_size, MOD_TSFILE_READER); if (IS_NULL(read_buf)) { return E_OOM; } @@ -159,7 +159,7 @@ int TsFileIOReader::load_tsfile_meta() { // 1.3 deserialize tsfile_meta_size if (IS_SUCC(ret)) { // deserialize tsfile_meta_size - char *size_buf = read_buf + alloc_size - TAIL_MAGIC_AND_META_SIZE_SIZE; + char* size_buf = read_buf + alloc_size - TAIL_MAGIC_AND_META_SIZE_SIZE; tsfile_meta_size = SerializationUtil::read_ui32(size_buf); ASSERT(tsfile_meta_size > 0 && tsfile_meta_size <= (1ll << 20)); } @@ -167,12 +167,12 @@ int TsFileIOReader::load_tsfile_meta() { // Step 2: reader TsFileMeta if (IS_SUCC(ret)) { // 2.1 prepare enough buffer (use the previous buffer if can). - char *tsfile_meta_buf = nullptr; + char* tsfile_meta_buf = nullptr; if (tsfile_meta_size + TAIL_MAGIC_AND_META_SIZE_SIZE > (uint32_t)alloc_size) { // prepare buffer to re-reader from start of tsfile_meta - char *old_read_buf = read_buf; - read_buf = (char *)mem_realloc(read_buf, tsfile_meta_size); + char* old_read_buf = read_buf; + read_buf = (char*)mem_realloc(read_buf, tsfile_meta_size); if (IS_NULL(read_buf)) { read_buf = old_read_buf; ret = E_OOM; @@ -211,8 +211,8 @@ int TsFileIOReader::load_tsfile_meta() { } int TsFileIOReader::load_timeseries_index_for_ssi( - std::shared_ptr<IDeviceID> device_id, const std::string &measurement_name, - TsFileSeriesScanIterator *&ssi) { + std::shared_ptr<IDeviceID> device_id, const std::string& measurement_name, + TsFileSeriesScanIterator*& ssi) { int ret = E_OK; std::shared_ptr<IMetaIndexEntry> device_index_entry; int64_t device_ie_end_offset = 0; @@ -224,19 +224,19 @@ int TsFileIOReader::load_timeseries_index_for_ssi( device_ie_end_offset))) { return ret; } - auto &pa = ssi->timeseries_index_pa_; + auto& pa = ssi->timeseries_index_pa_; int start_offset = device_index_entry->get_offset(), end_offset = device_ie_end_offset; ASSERT(start_offset < end_offset); const int32_t read_size = end_offset - start_offset; int32_t ret_read_len = 0; - char *data_buf = (char *)pa.alloc(read_size); - void *m_idx_node_buf = pa.alloc(sizeof(MetaIndexNode)); + char* data_buf = (char*)pa.alloc(read_size); + void* m_idx_node_buf = pa.alloc(sizeof(MetaIndexNode)); if (IS_NULL(data_buf) || IS_NULL(m_idx_node_buf)) { return E_OOM; } - auto *top_node_ptr = new (m_idx_node_buf) MetaIndexNode(&pa); + auto* top_node_ptr = new (m_idx_node_buf) MetaIndexNode(&pa); auto top_node = std::shared_ptr<MetaIndexNode>(top_node_ptr, MetaIndexNode::self_deleter); @@ -248,7 +248,7 @@ int TsFileIOReader::load_timeseries_index_for_ssi( } bool is_aligned = is_aligned_device(top_node); - TimeseriesIndex *timeseries_index = nullptr; + TimeseriesIndex* timeseries_index = nullptr; if (is_aligned) { if (RET_FAIL( get_time_column_metadata(top_node, timeseries_index, pa))) { @@ -267,8 +267,8 @@ int TsFileIOReader::load_timeseries_index_for_ssi( return ret; } if (is_aligned) { - auto *aligned_timeseries_index = - dynamic_cast<AlignedTimeseriesIndex *>(ssi->itimeseries_index_); + auto* aligned_timeseries_index = + dynamic_cast<AlignedTimeseriesIndex*>(ssi->itimeseries_index_); if (aligned_timeseries_index) { aligned_timeseries_index->time_ts_idx_ = timeseries_index; } @@ -277,10 +277,10 @@ int TsFileIOReader::load_timeseries_index_for_ssi( #if DEBUG_SE if (measurement_index_entry.name_.len_) { std::cout << "load timeseries index: " - << *((TimeseriesIndex *)ssi->itimeseries_index_) << std::endl; + << *((TimeseriesIndex*)ssi->itimeseries_index_) << std::endl; } else { std::cout << "load aligned timeseries index: " - << *((AlignedTimeseriesIndex *)ssi->itimeseries_index_) + << *((AlignedTimeseriesIndex*)ssi->itimeseries_index_) << std::endl; } #endif @@ -289,7 +289,7 @@ int TsFileIOReader::load_timeseries_index_for_ssi( int TsFileIOReader::load_device_index_entry( std::shared_ptr<IComparable> device_name, - std::shared_ptr<IMetaIndexEntry> &device_index_entry, int64_t &end_offset) { + std::shared_ptr<IMetaIndexEntry>& device_index_entry, int64_t& end_offset) { int ret = E_OK; std::shared_ptr<DeviceIDComparable> device_id_comparable = std::dynamic_pointer_cast<DeviceIDComparable>(device_name); @@ -322,10 +322,10 @@ int TsFileIOReader::load_device_index_entry( } int TsFileIOReader::load_measurement_index_entry( - const std::string &measurement_name_str, + const std::string& measurement_name_str, std::shared_ptr<MetaIndexNode> top_node, - std::shared_ptr<IMetaIndexEntry> &ret_measurement_index_entry, - int64_t &ret_end_offset) { + std::shared_ptr<IMetaIndexEntry>& ret_measurement_index_entry, + int64_t& ret_end_offset) { int ret = E_OK; // search from top_node in top-down way auto measurement_name = @@ -346,9 +346,9 @@ int TsFileIOReader::load_measurement_index_entry( } int TsFileIOReader::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) { + 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) { #if DEBUG_SE std::cout << "load_measurement_index_entry: measurement_name_str= " << ", start_offset=" << start_offset @@ -359,12 +359,12 @@ int TsFileIOReader::load_all_measurement_index_entry( // 1. load top measuremnt_index_node const int32_t read_size = (int32_t)(end_offset - start_offset); int32_t ret_read_len = 0; - char *data_buf = (char *)pa.alloc(read_size); - void *m_idx_node_buf = pa.alloc(sizeof(MetaIndexNode)); + char* data_buf = (char*)pa.alloc(read_size); + void* m_idx_node_buf = pa.alloc(sizeof(MetaIndexNode)); if (IS_NULL(data_buf) || IS_NULL(m_idx_node_buf)) { return E_OOM; } - auto *top_node_ptr = new (m_idx_node_buf) MetaIndexNode(&pa); + auto* top_node_ptr = new (m_idx_node_buf) MetaIndexNode(&pa); auto top_node = std::shared_ptr<MetaIndexNode>(top_node_ptr, MetaIndexNode::self_deleter); if (RET_FAIL(read_file_->read(start_offset, data_buf, read_size, @@ -389,15 +389,15 @@ int TsFileIOReader::load_all_measurement_index_entry( int TsFileIOReader::read_device_meta_index(int32_t start_offset, int32_t end_offset, - common::PageArena &pa, - MetaIndexNode *&device_meta_index, + common::PageArena& pa, + MetaIndexNode*& device_meta_index, bool leaf) { int ret = E_OK; ASSERT(start_offset < end_offset); const int32_t read_size = (int32_t)(end_offset - start_offset); int32_t ret_read_len = 0; - char *data_buf = (char *)pa.alloc(read_size); - void *m_idx_node_buf = pa.alloc(sizeof(MetaIndexNode)); + char* data_buf = (char*)pa.alloc(read_size); + void* m_idx_node_buf = pa.alloc(sizeof(MetaIndexNode)); if (IS_NULL(data_buf) || IS_NULL(m_idx_node_buf)) { return E_OOM; } @@ -415,8 +415,8 @@ int TsFileIOReader::read_device_meta_index(int32_t start_offset, int TsFileIOReader::get_timeseries_indexes( std::shared_ptr<IDeviceID> device_id, - const std::unordered_set<std::string> &measurement_names, - std::vector<ITimeseriesIndex *> ×eries_indexs, common::PageArena &pa) { + const std::unordered_set<std::string>& measurement_names, + std::vector<ITimeseriesIndex*>& timeseries_indexs, common::PageArena& pa) { int ret = E_OK; std::shared_ptr<IMetaIndexEntry> device_index_entry; int64_t device_ie_end_offset = 0; @@ -433,12 +433,12 @@ int TsFileIOReader::get_timeseries_indexes( ASSERT(start_offset < end_offset); const int32_t read_size = end_offset - start_offset; int32_t ret_read_len = 0; - char *data_buf = (char *)pa.alloc(read_size); - void *m_idx_node_buf = pa.alloc(sizeof(MetaIndexNode)); + char* data_buf = (char*)pa.alloc(read_size); + void* m_idx_node_buf = pa.alloc(sizeof(MetaIndexNode)); if (IS_NULL(data_buf) || IS_NULL(m_idx_node_buf)) { return E_OOM; } - auto *top_node_ptr = new (m_idx_node_buf) MetaIndexNode(&pa); + auto* top_node_ptr = new (m_idx_node_buf) MetaIndexNode(&pa); auto top_node = std::shared_ptr<MetaIndexNode>(top_node_ptr, MetaIndexNode::self_deleter); @@ -450,13 +450,13 @@ int TsFileIOReader::get_timeseries_indexes( } bool is_aligned = is_aligned_device(top_node); - TimeseriesIndex *timeseries_index = nullptr; + TimeseriesIndex* timeseries_index = nullptr; if (is_aligned) { get_time_column_metadata(top_node, timeseries_index, pa); } int64_t idx = 0; - for (const auto &measurement_name : measurement_names) { + for (const auto& measurement_name : measurement_names) { if (RET_FAIL(load_measurement_index_entry(measurement_name, top_node, measurement_index_entry, measurement_ie_end_offset))) { @@ -466,8 +466,8 @@ int TsFileIOReader::get_timeseries_indexes( is_aligned))) { } if (is_aligned) { - AlignedTimeseriesIndex *aligned_timeseries_index = - dynamic_cast<AlignedTimeseriesIndex *>(timeseries_indexs[idx]); + AlignedTimeseriesIndex* aligned_timeseries_index = + dynamic_cast<AlignedTimeseriesIndex*>(timeseries_indexs[idx]); if (aligned_timeseries_index) { aligned_timeseries_index->time_ts_idx_ = timeseries_index; } @@ -485,8 +485,8 @@ int TsFileIOReader::get_timeseries_indexes( int TsFileIOReader::search_from_leaf_node( std::shared_ptr<IComparable> target_name, std::shared_ptr<MetaIndexNode> index_node, - std::shared_ptr<IMetaIndexEntry> &ret_index_entry, - int64_t &ret_end_offset) { + std::shared_ptr<IMetaIndexEntry>& ret_index_entry, + int64_t& ret_end_offset) { int ret = E_OK; ret = index_node->binary_search_children(target_name, true, ret_index_entry, ret_end_offset); @@ -496,8 +496,8 @@ int TsFileIOReader::search_from_leaf_node( int TsFileIOReader::search_from_internal_node( std::shared_ptr<IComparable> target_name, bool is_device, std::shared_ptr<MetaIndexNode> index_node, - std::shared_ptr<IMetaIndexEntry> &ret_index_entry, - int64_t &ret_end_offset) { + std::shared_ptr<IMetaIndexEntry>& ret_index_entry, + int64_t& ret_end_offset) { int ret = E_OK; std::shared_ptr<IMetaIndexEntry> index_entry; int64_t end_offset = 0; @@ -519,12 +519,12 @@ int TsFileIOReader::search_from_internal_node( #endif ASSERT(read_size > 0 && read_size < (1 << 30)); PageArena cur_level_index_node_pa; - void *buf = cur_level_index_node_pa.alloc(sizeof(MetaIndexNode)); - char *data_buf = (char *)cur_level_index_node_pa.alloc(read_size); + void* buf = cur_level_index_node_pa.alloc(sizeof(MetaIndexNode)); + char* data_buf = (char*)cur_level_index_node_pa.alloc(read_size); if (IS_NULL(buf) || IS_NULL(data_buf)) { return E_OOM; } - MetaIndexNode *cur_level_index_node = + MetaIndexNode* cur_level_index_node = new (buf) MetaIndexNode(&cur_level_index_node_pa); int32_t ret_read_len = 0; if (RET_FAIL(read_file_->read(index_entry->get_offset(), data_buf, @@ -569,12 +569,12 @@ bool TsFileIOReader::is_aligned_device( int TsFileIOReader::get_time_column_metadata( std::shared_ptr<MetaIndexNode> measurement_node, - TimeseriesIndex *&ret_timeseries_index, PageArena &pa) { + TimeseriesIndex*& ret_timeseries_index, PageArena& pa) { int ret = E_OK; if (!is_aligned_device(measurement_node)) { return ret; } - char *ti_buf = nullptr; + char* ti_buf = nullptr; int start_idx = 0, end_idx = 0; int ret_read_len = 0; if (measurement_node->node_type_ == LEAF_MEASUREMENT) { @@ -597,7 +597,7 @@ int TsFileIOReader::get_time_column_metadata( } } buffer.wrap_from(ti_buf, end_idx - start_idx); - void *buf = pa.alloc(sizeof(TimeseriesIndex)); + void* buf = pa.alloc(sizeof(TimeseriesIndex)); if (IS_NULL(buf)) { return E_OOM; } @@ -621,14 +621,14 @@ int TsFileIOReader::get_time_column_metadata( } int TsFileIOReader::do_load_timeseries_index( - const std::string &measurement_name_str, int64_t start_offset, - int64_t end_offset, PageArena &in_timeseries_index_pa, - ITimeseriesIndex *&ret_timeseries_index, bool is_aligned) { + const std::string& measurement_name_str, int64_t start_offset, + int64_t end_offset, PageArena& in_timeseries_index_pa, + ITimeseriesIndex*& ret_timeseries_index, bool is_aligned) { ASSERT(end_offset > start_offset); int ret = E_OK; int32_t read_size = (int32_t)(end_offset - start_offset); int32_t ret_read_len = 0; - char *ti_buf = (char *)mem_alloc(read_size, MOD_TSFILE_READER); + char* ti_buf = (char*)mem_alloc(read_size, MOD_TSFILE_READER); if (IS_NULL(ti_buf)) { return E_OOM; } @@ -638,7 +638,7 @@ int TsFileIOReader::do_load_timeseries_index( ByteStream bs; bs.wrap_from(ti_buf, read_size); const String target_measurement_name( - (char *)measurement_name_str.c_str(), + (char*)measurement_name_str.c_str(), strlen(measurement_name_str.c_str())); bool found = false; #if DEBUG_SE @@ -654,12 +654,12 @@ int TsFileIOReader::do_load_timeseries_index( } else if (is_aligned && cur_timeseries_index.get_measurement_name().equal_to( target_measurement_name)) { - void *buf = in_timeseries_index_pa.alloc( + void* buf = in_timeseries_index_pa.alloc( sizeof(AlignedTimeseriesIndex)); if (IS_NULL(buf)) { return E_OOM; } - AlignedTimeseriesIndex *aligned_ts_idx = + AlignedTimeseriesIndex* aligned_ts_idx = new (buf) AlignedTimeseriesIndex; buf = in_timeseries_index_pa.alloc(sizeof(TimeseriesIndex)); if (IS_NULL(buf)) { @@ -674,7 +674,7 @@ int TsFileIOReader::do_load_timeseries_index( } else if (!is_aligned && cur_timeseries_index.get_measurement_name().equal_to( target_measurement_name)) { - void *buf = + void* buf = in_timeseries_index_pa.alloc(sizeof(TimeseriesIndex)); auto ts_idx = new (buf) TimeseriesIndex; ts_idx->clone_from(cur_timeseries_index, @@ -693,12 +693,12 @@ int TsFileIOReader::do_load_timeseries_index( } int TsFileIOReader::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) { + 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 ret = E_OK; - for (const auto &index_node_entry : index_node_entry_list) { + for (const auto& index_node_entry : index_node_entry_list) { int64_t start_offset = index_node_entry.first->get_offset(), end_offset = index_node_entry.second; int32_t read_size = (int32_t)(end_offset - start_offset); @@ -729,8 +729,8 @@ int TsFileIOReader::do_load_all_timeseries_index( int TsFileIOReader::get_all_leaf( std::shared_ptr<MetaIndexNode> index_node, - std::vector<std::pair<std::shared_ptr<IMetaIndexEntry>, int64_t> > - &index_node_entry_list) { + std::vector<std::pair<std::shared_ptr<IMetaIndexEntry>, int64_t>>& + index_node_entry_list) { int ret = E_OK; if (index_node->node_type_ == LEAF_MEASUREMENT || index_node->node_type_ == LEAF_DEVICE) { @@ -760,12 +760,12 @@ int TsFileIOReader::get_all_leaf( #endif ASSERT(read_size > 0 && read_size < (1 << 30)); PageArena cur_level_index_node_pa; - void *buf = cur_level_index_node_pa.alloc(sizeof(MetaIndexNode)); - char *data_buf = (char *)cur_level_index_node_pa.alloc(read_size); + void* buf = cur_level_index_node_pa.alloc(sizeof(MetaIndexNode)); + char* data_buf = (char*)cur_level_index_node_pa.alloc(read_size); if (IS_NULL(buf) || IS_NULL(data_buf)) { return E_OOM; } - auto *cur_level_index_node_ptr = + auto* cur_level_index_node_ptr = new (buf) MetaIndexNode(&cur_level_index_node_pa); auto cur_level_index_node = std::shared_ptr<MetaIndexNode>( cur_level_index_node_ptr, MetaIndexNode::self_deleter); diff --git a/cpp/src/reader/block/single_device_tsblock_reader.cc b/cpp/src/reader/block/single_device_tsblock_reader.cc index 1df563cd..4137006e 100644 --- a/cpp/src/reader/block/single_device_tsblock_reader.cc +++ b/cpp/src/reader/block/single_device_tsblock_reader.cc @@ -214,8 +214,14 @@ int SingleDeviceTsBlockReader::fill_ids() { const auto& id_column_context = entry.second; for (int32_t pos : id_column_context.pos_in_result_) { std::string* device_tag = nullptr; - device_tag = device_query_task_->get_device_id()->get_segments().at( - id_column_context.pos_in_device_id_); + const auto& segments = + device_query_task_->get_device_id()->get_segments(); + int32_t pos_in_device_id = id_column_context.pos_in_device_id_; + if (pos_in_device_id >= 0 && + static_cast<size_t>(pos_in_device_id - 1) < segments.size()) { + device_tag = segments[pos_in_device_id - 1]; + } + if (device_tag == nullptr) { ret = col_appenders_[pos + 1]->fill_null( current_block_->get_row_count()); diff --git a/cpp/src/reader/device_meta_iterator.cc b/cpp/src/reader/device_meta_iterator.cc index a59f511d..53eaaeb2 100644 --- a/cpp/src/reader/device_meta_iterator.cc +++ b/cpp/src/reader/device_meta_iterator.cc @@ -45,7 +45,7 @@ int DeviceMetaIterator::next( } int DeviceMetaIterator::load_results() { - bool is_root_idx_node = true; + int root_num = meta_index_nodes_.size(); while (!meta_index_nodes_.empty()) { // To avoid ASan overflow. // using `const auto&` creates a reference @@ -60,12 +60,9 @@ int DeviceMetaIterator::load_results() { } else { return common::E_INVALID_NODE_TYPE; } - // The first MetaIndexNode is the root and is not loaded here, so no - // need to destruct it here. - if (!is_root_idx_node) { + if (root_num-- <= 0) { meta_data_index_node->~MetaIndexNode(); } - is_root_idx_node = false; } return common::E_OK; diff --git a/cpp/src/reader/device_meta_iterator.h b/cpp/src/reader/device_meta_iterator.h index 55f20913..8b39006b 100644 --- a/cpp/src/reader/device_meta_iterator.h +++ b/cpp/src/reader/device_meta_iterator.h @@ -37,6 +37,16 @@ class DeviceMetaIterator { pa_.init(512, common::MOD_DEVICE_META_ITER); } + DeviceMetaIterator(TsFileIOReader *io_reader, + const std::vector<MetaIndexNode *> &meta_index_node_list, + const Filter *id_filter) + : io_reader_(io_reader), id_filter_(id_filter) { + for (auto meta_index_node : meta_index_node_list) { + meta_index_nodes_.push(meta_index_node); + } + pa_.init(512, common::MOD_DEVICE_META_ITER); + } + ~DeviceMetaIterator() { pa_.destroy(); } bool has_next(); diff --git a/cpp/src/reader/imeta_data_querier.h b/cpp/src/reader/imeta_data_querier.h index 73a005e8..c034f915 100644 --- a/cpp/src/reader/imeta_data_querier.h +++ b/cpp/src/reader/imeta_data_querier.h @@ -57,7 +57,10 @@ class IMetadataQuerier { virtual std::unique_ptr<DeviceMetaIterator> device_iterator( MetaIndexNode* root, const Filter* id_filter) = 0; -}; + // FIXME(Colin): refine this. + virtual std::unique_ptr<DeviceMetaIterator> device_iterator( + std::vector<MetaIndexNode*> root, const Filter* id_filter) = 0; +}; } // end namespace storage #endif // READER_IMETA_DATA_QUERIER_H diff --git a/cpp/src/reader/meta_data_querier.cc b/cpp/src/reader/meta_data_querier.cc index 5a32b922..0accbdde 100644 --- a/cpp/src/reader/meta_data_querier.cc +++ b/cpp/src/reader/meta_data_querier.cc @@ -98,6 +98,12 @@ std::unique_ptr<DeviceMetaIterator> MetadataQuerier::device_iterator( new DeviceMetaIterator(io_reader_, root, id_filter)); } +std::unique_ptr<DeviceMetaIterator> MetadataQuerier::device_iterator( + std::vector<MetaIndexNode*> root, const Filter* id_filter) { + return std::unique_ptr<DeviceMetaIterator>( + new DeviceMetaIterator(io_reader_, root, id_filter)); +} + int MetadataQuerier::load_chunk_meta( const std::pair<IDeviceID, std::string>& key, std::vector<ChunkMeta*>& chunk_meta_list) { diff --git a/cpp/src/reader/meta_data_querier.h b/cpp/src/reader/meta_data_querier.h index b4eed350..525ecf86 100644 --- a/cpp/src/reader/meta_data_querier.h +++ b/cpp/src/reader/meta_data_querier.h @@ -61,6 +61,9 @@ class MetadataQuerier : public IMetadataQuerier { std::unique_ptr<DeviceMetaIterator> device_iterator( MetaIndexNode* root, const Filter* id_filter) override; + std::unique_ptr<DeviceMetaIterator> device_iterator( + std::vector<MetaIndexNode*> root, const Filter* id_filter) override; + void clear() override; private: diff --git a/cpp/src/reader/table_query_executor.cc b/cpp/src/reader/table_query_executor.cc index d09a5c90..6fbd4f0f 100644 --- a/cpp/src/reader/table_query_executor.cc +++ b/cpp/src/reader/table_query_executor.cc @@ -19,17 +19,19 @@ #include "reader/table_query_executor.h" +#include "utils/db_utils.h" + namespace storage { -int TableQueryExecutor::query(const std::string &table_name, - const std::vector<std::string> &columns, - Filter *time_filter, Filter *id_filter, - Filter *field_filter, ResultSet *&ret_qds) { +int TableQueryExecutor::query(const std::string& table_name, + const std::vector<std::string>& columns, + Filter* time_filter, Filter* id_filter, + Filter* field_filter, ResultSet*& ret_qds) { int ret = common::E_OK; - TsFileMeta *file_metadata = nullptr; + TsFileMeta* file_metadata = nullptr; file_metadata = tsfile_io_reader_->get_tsfile_meta(); common::PageArena pa; pa.init(512, common::MOD_TSFILE_READER); - MetaIndexNode *table_root = nullptr; + MetaIndexNode* table_root = nullptr; std::shared_ptr<TableSchema> table_schema; if (RET_FAIL( file_metadata->get_table_metaindex_node(table_name, table_root))) { @@ -42,7 +44,7 @@ int TableQueryExecutor::query(const std::string &table_name, return ret; } std::vector<std::string> lower_case_column_names(columns); - for (auto &column : lower_case_column_names) { + for (auto& column : lower_case_column_names) { to_lowercase_inplace(column); } std::shared_ptr<ColumnMapping> column_mapping = @@ -85,6 +87,113 @@ int TableQueryExecutor::query(const std::string &table_name, return ret; } -void TableQueryExecutor::destroy_query_data_set(ResultSet *qds) { delete qds; } +int TableQueryExecutor::query_on_tree( + const std::vector<std::shared_ptr<IDeviceID>>& devices, + const std::vector<std::string>& tag_columns, + const std::vector<std::string>& field_columns, Filter* time_filter, + ResultSet*& ret_qds) { + common::PageArena pa; + pa.init(512, common::MOD_TSFILE_READER); + int ret = common::E_OK; + TsFileMeta* file_meta = tsfile_io_reader_->get_tsfile_meta(); + std::vector<MetaIndexNode*> table_inodes; + for (auto const& device : devices) { + MetaIndexNode* table_inode; + if (RET_FAIL(file_meta->get_table_metaindex_node( + device->get_table_name(), table_inode))) { + }; + table_inodes.push_back(table_inode); + } + + std::vector<common::ColumnSchema> col_schema; + for (auto const& tag : tag_columns) { + col_schema.emplace_back(tag, common::TSDataType::STRING, + common::ColumnCategory::TAG); + } + + std::unordered_map<std::string, common::TSDataType> column_types_map; + + for (auto const& device : devices) { + bool all_collected = true; + for (const auto& field_col : field_columns) { + if (column_types_map.find(field_col) == column_types_map.end()) { + all_collected = false; + break; + } + } + if (all_collected) { + break; + } + + std::unordered_set<std::string> measurements(field_columns.begin(), + field_columns.end()); + std::vector<ITimeseriesIndex*> index(measurements.size()); + if (RET_FAIL(tsfile_io_reader_->get_timeseries_indexes( + device, measurements, index, pa))) { + assert(0); + } + + for (auto* ts_index : index) { + if (ts_index != nullptr) { + std::string measurement_name = + ts_index->get_measurement_name().to_std_string(); + if (column_types_map.find(measurement_name) == + column_types_map.end()) { + common::TSDataType type = ts_index->get_data_type(); + if (type == common::TSDataType::INT32 || + type == common::TSDataType::INT64 || + type == common::TSDataType::TIMESTAMP || + type == common::TSDataType::DATE) { + type = common::TSDataType::INT64; + } else if (type == common::TSDataType::FLOAT) { + type = common::TSDataType::DOUBLE; + } + column_types_map[measurement_name] = type; + } + } + } + } + + for (const auto& field_col : field_columns) { + if (column_types_map.find(field_col) != column_types_map.end()) { + col_schema.emplace_back(field_col, column_types_map[field_col], + common::ColumnCategory::FIELD); + } else { + col_schema.emplace_back(field_col, + common::TSDataType::INVALID_DATATYPE, + common::ColumnCategory::FIELD); + } + } + + auto schema = std::make_shared<TableSchema>("default", col_schema); + std::shared_ptr<ColumnMapping> column_mapping = + std::make_shared<ColumnMapping>(); + for (size_t i = 0; i < col_schema.size(); ++i) { + column_mapping->add(col_schema[i].column_name_, i, *schema); + } + std::vector<common::TSDataType> datatypes = schema->get_data_types(); + auto device_task_iterator = + std::unique_ptr<DeviceTaskIterator>(new DeviceTaskIterator( + schema->get_measurement_names(), table_inodes, column_mapping, + meta_data_querier_, nullptr, schema)); + std::unique_ptr<TsBlockReader> tsblock_reader; + switch (table_query_ordering_) { + case TableQueryOrdering::DEVICE: + tsblock_reader = std::unique_ptr<DeviceOrderedTsBlockReader>( + new DeviceOrderedTsBlockReader( + std::move(device_task_iterator), meta_data_querier_, + block_size_, tsfile_io_reader_, time_filter, nullptr)); + break; + case TableQueryOrdering::TIME: + default: + ret = common::E_UNSUPPORTED_ORDER; + } + assert(tsblock_reader != nullptr); + ret_qds = new TableResultSet(std::move(tsblock_reader), + schema->get_measurement_names(), + schema->get_data_types()); + return ret; +} +void TableQueryExecutor::destroy_query_data_set(ResultSet* qds) { delete qds; } } // end namespace storage diff --git a/cpp/src/reader/table_query_executor.h b/cpp/src/reader/table_query_executor.h index 83a82fe5..974e6b45 100644 --- a/cpp/src/reader/table_query_executor.h +++ b/cpp/src/reader/table_query_executor.h @@ -37,15 +37,15 @@ class TableQueryExecutor { public: enum class TableQueryOrdering { TIME, DEVICE }; - TableQueryExecutor(IMetadataQuerier *meta_data_querier, - TsFileIOReader *tsfile_io_reader, + TableQueryExecutor(IMetadataQuerier* meta_data_querier, + TsFileIOReader* tsfile_io_reader, TableQueryOrdering table_query_ordering, int block_size = 1024) : meta_data_querier_(meta_data_querier), tsfile_io_reader_(tsfile_io_reader), table_query_ordering_(table_query_ordering), block_size_(block_size) {} - TableQueryExecutor(ReadFile *read_file) { + TableQueryExecutor(ReadFile* read_file) { tsfile_io_reader_ = new TsFileIOReader(); tsfile_io_reader_->init(read_file); meta_data_querier_ = new MetadataQuerier(tsfile_io_reader_); @@ -62,14 +62,18 @@ class TableQueryExecutor { tsfile_io_reader_ = nullptr; } } - int query(const std::string &table_name, - const std::vector<std::string> &columns, Filter *time_filter, - Filter *id_filter, Filter *field_filter, ResultSet *&ret_qds); - void destroy_query_data_set(ResultSet *qds); + int query(const std::string& table_name, + const std::vector<std::string>& columns, Filter* time_filter, + Filter* id_filter, Filter* field_filter, ResultSet*& ret_qds); + int query_on_tree(const std::vector<std::shared_ptr<IDeviceID>>& devices, + const std::vector<std::string>& tag_columns, + const std::vector<std::string>& field_columns, + Filter* time_filter, ResultSet*& ret_qds); + void destroy_query_data_set(ResultSet* qds); private: - IMetadataQuerier *meta_data_querier_; - TsFileIOReader *tsfile_io_reader_; + IMetadataQuerier* meta_data_querier_; + TsFileIOReader* tsfile_io_reader_; TableQueryOrdering table_query_ordering_; int32_t block_size_; }; diff --git a/cpp/src/reader/task/device_task_iterator.h b/cpp/src/reader/task/device_task_iterator.h index a5079877..ec30a472 100644 --- a/cpp/src/reader/task/device_task_iterator.h +++ b/cpp/src/reader/task/device_task_iterator.h @@ -43,6 +43,21 @@ class DeviceTaskIterator { table_schema_(table_schema) { pa_.init(512, common::MOD_DEVICE_TASK_ITER); } + + DeviceTaskIterator(std::vector<std::string> column_names, + std::vector<MetaIndexNode *> index_roots, + std::shared_ptr<ColumnMapping> column_mapping, + IMetadataQuerier *metadata_querier, + const Filter *id_filter, + std::shared_ptr<TableSchema> table_schema) + : column_names_(column_names), + column_mapping_(column_mapping), + device_meta_iterator_( + metadata_querier->device_iterator(index_roots, id_filter)), + table_schema_(table_schema) { + pa_.init(512, common::MOD_DEVICE_TASK_ITER); + } + ~DeviceTaskIterator() { pa_.destroy(); } bool has_next() const; diff --git a/cpp/src/reader/tsfile_reader.cc b/cpp/src/reader/tsfile_reader.cc index 2ac45d84..8199ad28 100644 --- a/cpp/src/reader/tsfile_reader.cc +++ b/cpp/src/reader/tsfile_reader.cc @@ -107,8 +107,6 @@ int TsFileReader::query(const std::string& table_name, return E_TABLE_NOT_EXIST; } - std::vector<TSDataType> data_types = table_schema->get_data_types(); - Filter* time_filter = new TimeBetween(start_time, end_time, false); ret = table_query_executor_->query(to_lower(table_name), columns_names, time_filter, tag_filter, nullptr, @@ -116,6 +114,42 @@ int TsFileReader::query(const std::string& table_name, return ret; } +int TsFileReader::query_table_on_tree( + const std::vector<std::string>& measurement_names, int64_t star_time, + int64_t end_time, ResultSet*& result_set) { + int ret = E_OK; + TsFileMeta* tsfile_meta = tsfile_executor_->get_tsfile_meta(); + if (tsfile_meta == nullptr) { + return E_TSFILE_WRITER_META_ERR; + } + auto device_names = this->get_all_device_ids(); + std::vector<std::shared_ptr<IDeviceID>> device_ids; + size_t max_len = 0; + for (auto& device_name : device_names) { + std::vector<MeasurementSchema> schemas; + this->get_timeseries_schema(device_name, schemas); + for (auto schema : schemas) { + if (std::find(measurement_names.begin(), measurement_names.end(), + schema.measurement_name_) != + measurement_names.end()) { + device_ids.push_back(device_name); + if (device_name->get_segments().size() > max_len) { + max_len = device_name->get_segments().size(); + } + break; + } + } + } + std::vector<std::string> columns_names(max_len); + for (int i = 0; i < max_len; i++) { + columns_names[i] = "l_" + std::to_string(i); + } + Filter* time_filter = new TimeBetween(star_time, end_time, false); + ret = table_query_executor_->query_on_tree( + device_ids, columns_names, measurement_names, time_filter, result_set); + return ret; +} + void TsFileReader::destroy_query_data_set(storage::ResultSet* qds) { tsfile_executor_->destroy_query_data_set(qds); } diff --git a/cpp/src/reader/tsfile_reader.h b/cpp/src/reader/tsfile_reader.h index 55e5f2c4..8a6ba226 100644 --- a/cpp/src/reader/tsfile_reader.h +++ b/cpp/src/reader/tsfile_reader.h @@ -54,7 +54,7 @@ class TsFileReader { * @param file_path the path of the tsfile which will be opened * @return Returns 0 on success, or a non-zero error code on failure. */ - int open(const std::string &file_path); + int open(const std::string& file_path); /** * @brief close the tsfile, this method should be called after the * query is finished @@ -70,7 +70,7 @@ class TsFileReader { * @param [out] ret_qds the result set * @return Returns 0 on success, or a non-zero error code on failure. */ - int query(storage::QueryExpression *qe, ResultSet *&ret_qds); + int query(storage::QueryExpression* qe, ResultSet*& ret_qds); /** * @brief query the tsfile by the path list, start time and end time * this method is used to query the tsfile by the tree model. @@ -80,8 +80,8 @@ class TsFileReader { * @param [in] end_time the end time * @param [out] result_set the result set */ - int query(std::vector<std::string> &path_list, int64_t start_time, - int64_t end_time, ResultSet *&result_set); + int query(std::vector<std::string>& path_list, int64_t start_time, + int64_t end_time, ResultSet*& result_set); /** * @brief query the tsfile by the table name, columns names, start time * and end time. this method is used to query the tsfile by the table @@ -93,9 +93,9 @@ class TsFileReader { * @param [in] end_time the end time * @param [out] result_set the result set */ - int query(const std::string &table_name, - const std::vector<std::string> &columns_names, int64_t start_time, - int64_t end_time, ResultSet *&result_set); + int query(const std::string& table_name, + const std::vector<std::string>& columns_names, int64_t start_time, + int64_t end_time, ResultSet*& result_set); /** * @brief query the tsfile by the table name, columns names, start time @@ -109,20 +109,23 @@ class TsFileReader { * @param [in] tag_filter the tag filter * @param [out] result_set the result set */ - int query(const std::string &table_name, - const std::vector<std::string> &columns_names, int64_t start_time, - int64_t end_time, ResultSet *&result_set, Filter *tag_filter); + int query(const std::string& table_name, + const std::vector<std::string>& columns_names, int64_t start_time, + int64_t end_time, ResultSet*& result_set, Filter* tag_filter); + int query_table_on_tree(const std::vector<std::string>& measurement_names, + int64_t star_time, int64_t end_time, + ResultSet*& result_set); /** * @brief destroy the result set, this method should be called after the * query is finished and result_set * * @param qds the result set */ - void destroy_query_data_set(ResultSet *qds); - ResultSet *read_timeseries( - const std::shared_ptr<IDeviceID> &device_id, - const std::vector<std::string> &measurement_name); + void destroy_query_data_set(ResultSet* qds); + ResultSet* read_timeseries( + const std::shared_ptr<IDeviceID>& device_id, + const std::vector<std::string>& measurement_name); /** * @brief get all devices in the tsfile * @@ -148,7 +151,7 @@ class TsFileReader { * @return Returns 0 on success, or a non-zero error code on failure. */ int get_timeseries_schema(std::shared_ptr<IDeviceID> device_id, - std::vector<MeasurementSchema> &result); + std::vector<MeasurementSchema>& result); /** * @brief get the table schema by the table name * @@ -156,7 +159,7 @@ class TsFileReader { * @return std::shared_ptr<TableSchema> the table schema */ std::shared_ptr<TableSchema> get_table_schema( - const std::string &table_name); + const std::string& table_name); /** * @brief get all table schemas in the tsfile * @@ -165,12 +168,12 @@ class TsFileReader { std::vector<std::shared_ptr<TableSchema>> get_all_table_schemas(); private: - int get_all_devices(std::vector<std::shared_ptr<IDeviceID>> &device_ids, + int get_all_devices(std::vector<std::shared_ptr<IDeviceID>>& device_ids, std::shared_ptr<MetaIndexNode> index_node, - common::PageArena &pa); - storage::ReadFile *read_file_; - storage::TsFileExecutor *tsfile_executor_; - storage::TableQueryExecutor *table_query_executor_; + common::PageArena& pa); + storage::ReadFile* read_file_; + storage::TsFileExecutor* tsfile_executor_; + storage::TableQueryExecutor* table_query_executor_; }; } // namespace storage diff --git a/cpp/test/reader/tree_view/tsfile_reader_tree_test.cc b/cpp/test/reader/tree_view/tsfile_reader_tree_test.cc index 477ab24b..e534a131 100644 --- a/cpp/test/reader/tree_view/tsfile_reader_tree_test.cc +++ b/cpp/test/reader/tree_view/tsfile_reader_tree_test.cc @@ -32,6 +32,7 @@ namespace storage { class QDSWithoutTimeGenerator; } + using namespace storage; using namespace common; @@ -49,7 +50,8 @@ class TsFileTreeReaderTest : public ::testing::Test { mode_t mode = 0666; write_file_.create(file_name_, flags, mode); } - void TearDown() override { remove(file_name_.c_str()); } + + void TearDown() override {} std::string file_name_; WriteFile write_file_; @@ -108,6 +110,90 @@ TEST_F(TsFileTreeReaderTest, BasicTest) { reader.close(); } +TEST_F(TsFileTreeReaderTest, ReadTreeByTable) { + TsFileTreeWriter writer(&write_file_); + std::vector<std::string> device_ids = {"root.db1.t1", "root.db2.t1", + "root.db3.t2.t3", "root.db3.t3", + "device"}; + std::vector<std::string> measurement_ids = {"temperature", "hudi", "level"}; + for (auto& device_id : device_ids) { + TsRecord record(device_id, 0); + TsRecord record1(device_id, 1); + for (auto const& measurement : measurement_ids) { + auto schema = + new storage::MeasurementSchema(measurement, TSDataType::INT32); + ASSERT_EQ(E_OK, writer.register_timeseries(device_id, schema)); + delete schema; + record.add_point(measurement, static_cast<int64_t>(1)); + record1.add_point(measurement, static_cast<int64_t>(2)); + } + ASSERT_EQ(E_OK, writer.write(record)); + ASSERT_EQ(E_OK, writer.write(record1)); + } + writer.flush(); + writer.close(); + + TsFileReader reader; + reader.open(file_name_); + ResultSet* result; + int ret = reader.query_table_on_tree({"temperature", "hudi"}, INT64_MIN, + INT64_MAX, result); + ASSERT_EQ(ret, E_OK); + + auto* table_result_set = (storage::TableResultSet*)result; + bool has_next = false; + int num = table_result_set->get_metadata()->get_column_count(); + std::unordered_map<std::string, std::string> res; + res["root.db1"] = "t1"; + res["root.db2"] = "t1"; + res["root.db3.t2"] = "t3"; + res["root.db3"] = "t3"; + res["device"] = "null"; + int cnt = 0; + while (IS_SUCC(table_result_set->next(has_next)) && has_next) { + auto t = table_result_set->get_value<int64_t>(1); + ASSERT_TRUE(t == 0 || t == 1); + std::string key = ""; + std::string value = ""; + for (int i = 1; i < num + 1; ++i) { + switch (table_result_set->get_metadata()->get_column_type(i)) { + case INT64: + ASSERT_TRUE(table_result_set->get_value<int64_t>(i) == 1 || + table_result_set->get_value<int64_t>(i) == 0); + break; + case INT32: + ASSERT_TRUE(table_result_set->get_value<int32_t>(i) == 1 || + table_result_set->get_value<int32_t>(i) == 2); + break; + case STRING: { + common::String* str = + table_result_set->get_value<common::String*>(i); + if (i == 2) { + key = std::string(str->buf_, str->len_); + ASSERT_TRUE(res.find(key) != res.end()); + } + if (i == 3) { + if (str == nullptr) { + value = "null"; + } else { + value = std::string(str->buf_, str->len_); + } + ASSERT_TRUE(res.find(key) != res.end()); + ASSERT_TRUE(res[key] == value); + } + } break; + default: + break; + } + } + std::cout << std::endl; + cnt++; + } + ASSERT_EQ(cnt, 10); + reader.destroy_query_data_set(result); + reader.close(); +} + TEST_F(TsFileTreeReaderTest, ExtendedRowsAndColumnsTest) { TsFileTreeWriter writer(&write_file_); std::vector<std::string> device_ids = {"device_1", "device_2", "device_3"};
