diff --git a/cpp/src/common/record.h b/cpp/src/common/record.h index 370508a35..8c729f688 100644 --- a/cpp/src/common/record.h +++ b/cpp/src/common/record.h @@ -125,6 +125,11 @@ struct TsRecord { pa.init(512, common::MOD_TSFILE_READER); } + TsRecord(const std::string &device_name, const int64_t ×tamp) + : device_id_(device_name), timestamp_(timestamp) { + pa.init(512, common::MOD_TSFILE_READER); + } + TsRecord(int64_t timestamp, const std::string &device_name, int32_t point_count_in_row = 0) : timestamp_(timestamp), device_id_(device_name), points_() { diff --git a/cpp/src/reader/qds_with_timegenerator.h b/cpp/src/reader/qds_with_timegenerator.h index 648a1ce83..699168685 100644 --- a/cpp/src/reader/qds_with_timegenerator.h +++ b/cpp/src/reader/qds_with_timegenerator.h @@ -109,8 +109,7 @@ struct Node { class QDSWithTimeGenerator : public ResultSet { public: QDSWithTimeGenerator() - : row_record_(nullptr), - result_set_metadata_(nullptr), + : result_set_metadata_(nullptr), io_reader_(nullptr), qe_(nullptr), tree_(nullptr), @@ -129,7 +128,6 @@ class QDSWithTimeGenerator : public ResultSet { int construct_node_tree(Expression *expr, Node *&node); private: - RowRecord *row_record_; std::shared_ptr result_set_metadata_; TsFileIOReader *io_reader_; QueryExpression *qe_; diff --git a/cpp/src/reader/qds_without_timegenerator.h b/cpp/src/reader/qds_without_timegenerator.h index 5c461cf45..ae1922faf 100644 --- a/cpp/src/reader/qds_without_timegenerator.h +++ b/cpp/src/reader/qds_without_timegenerator.h @@ -32,8 +32,7 @@ namespace storage { class QDSWithoutTimeGenerator : public ResultSet { public: QDSWithoutTimeGenerator() - : row_record_(nullptr), - result_set_metadata_(nullptr), + : result_set_metadata_(nullptr), io_reader_(nullptr), qe_(nullptr), ssi_vec_(), @@ -54,7 +53,6 @@ class QDSWithoutTimeGenerator : public ResultSet { int get_next_tsblock(uint32_t index, bool alloc_mem); private: - RowRecord *row_record_; std::shared_ptr result_set_metadata_; TsFileIOReader *io_reader_; QueryExpression *qe_; diff --git a/cpp/src/reader/result_set.h b/cpp/src/reader/result_set.h index b0d45b019..da9891f2e 100644 --- a/cpp/src/reader/result_set.h +++ b/cpp/src/reader/result_set.h @@ -21,7 +21,6 @@ #define READER_QUERY_DATA_SET_H #include -#include #include #include @@ -84,6 +83,8 @@ class ResultSetMetadata { std::vector column_types_; }; +class ResultSetIterator; + /** * @brief ResultSet is the query result of the TsfileReader. It provides access * to the results. @@ -96,7 +97,7 @@ class ResultSetMetadata { * it should be QDSWithTimeGenerator. * @note If the query uses the table model, the cast should be TableResultSet */ -class ResultSet { +class ResultSet : std::enable_shared_from_this { public: ResultSet() {} virtual ~ResultSet() {} @@ -122,6 +123,11 @@ class ResultSet { */ virtual bool is_null(uint32_t column_index) = 0; + /** + * @brief Simple iterator for ResultSet with smart pointers + */ + virtual ResultSetIterator iterator(); + /** * @brief Get the value of the column by column name * @@ -196,6 +202,7 @@ class ResultSet { std::unordered_map index_lookup_; + RowRecord* row_record_ = nullptr; common::PageArena pa_; }; @@ -231,6 +238,65 @@ inline std::tm ResultSet::get_value(uint32_t column_index) { return row_record->get_field(column_index)->get_date_value(); } +/** + * @brief Simple iterator for ResultSet with smart pointers + */ +class ResultSetIterator { + public: + explicit ResultSetIterator(ResultSet* result_set) + : result_set_(result_set) {} + + /** + * @brief Check if there is a next row available + */ + bool hasNext() { + if (cached_record_ != nullptr) { + return true; + } + if (exhausted_) { + return false; + } + + bool has_next = false; + if (result_set_) { + int ret = result_set_->next(has_next); + ASSERT(ret == 0); + if (has_next) { + cached_record_ = result_set_->get_row_record(); + } else { + exhausted_ = true; + } + } + return has_next; + } + + /** + * @brief Get the next row record + */ + RowRecord* next() { + if (!hasNext()) { + return nullptr; + } + RowRecord* ret = cached_record_; + cached_record_ = nullptr; + return ret; + } + + /** + * @brief Get the underlying ResultSet for direct access + */ + ResultSet* getResultSet() const { return result_set_; } + + private: + ResultSet* result_set_ = nullptr; + RowRecord* cached_record_ = nullptr; + bool exhausted_ = false; +}; + +inline ResultSetIterator ResultSet::iterator() { + return ResultSetIterator(this); +} + } // namespace storage #endif // READER_QUERY_DATA_SET_H diff --git a/cpp/src/reader/table_result_set.h b/cpp/src/reader/table_result_set.h index 71e57c47d..4192f7c2f 100644 --- a/cpp/src/reader/table_result_set.h +++ b/cpp/src/reader/table_result_set.h @@ -47,7 +47,6 @@ class TableResultSet : public ResultSet { std::unique_ptr tsblock_reader_; common::RowIterator* row_iterator_ = nullptr; common::TsBlock* tsblock_ = nullptr; - RowRecord* row_record_ = nullptr; std::shared_ptr result_set_metadata_; std::vector> tsblock_readers_; std::vector column_names_; diff --git a/cpp/src/reader/tsfile_reader.cc b/cpp/src/reader/tsfile_reader.cc index 5dba982f6..6da094303 100644 --- a/cpp/src/reader/tsfile_reader.cc +++ b/cpp/src/reader/tsfile_reader.cc @@ -127,6 +127,20 @@ std::vector> TsFileReader::get_all_devices( return device_ids; } +std::vector> TsFileReader::get_all_device_ids() { + TsFileMeta* tsfile_meta = tsfile_executor_->get_tsfile_meta(); + std::vector> device_ids; + if (tsfile_meta != nullptr) { + PageArena pa; + pa.init(512, MOD_TSFILE_READER); + for (auto entry : tsfile_meta->table_metadata_index_node_map_) { + auto index_node = entry.second; + get_all_devices(device_ids, index_node, pa); + } + } + return device_ids; +} + int TsFileReader::get_all_devices( std::vector>& device_ids, std::shared_ptr index_node, PageArena& pa) { diff --git a/cpp/src/reader/tsfile_reader.h b/cpp/src/reader/tsfile_reader.h index dc5105c0b..eb6a7b70e 100644 --- a/cpp/src/reader/tsfile_reader.h +++ b/cpp/src/reader/tsfile_reader.h @@ -114,6 +114,14 @@ class TsFileReader { */ std::vector> get_all_devices( std::string table_name); + + /** + * @brief get all devices in the tsfile + * + * @return std::vector> the device id list + */ + std::vector> get_all_device_ids(); + /** * @brief get the timeseries schema by the device id and measurement name * diff --git a/cpp/src/reader/tsfile_tree_reader.cc b/cpp/src/reader/tsfile_tree_reader.cc new file mode 100644 index 000000000..c7c9004a3 --- /dev/null +++ b/cpp/src/reader/tsfile_tree_reader.cc @@ -0,0 +1,71 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * License); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "reader/tsfile_tree_reader.h" + +namespace storage { + +TsFileTreeReader::TsFileTreeReader() { + tsfile_reader_ = std::make_shared(); +} + +TsFileTreeReader::~TsFileTreeReader() = default; + +int TsFileTreeReader::open(const std::string &file_path) { + return tsfile_reader_->open(file_path); +} + +int TsFileTreeReader::close() { return tsfile_reader_->close(); } + +int TsFileTreeReader::query(const std::vector &device_ids, + const std::vector &measurement_names, + int64_t start_time, int64_t end_time, + ResultSet *&result_set) { + std::vector path_list; + for (auto &device_id : device_ids) { + for (auto &measurement : measurement_names) { + path_list.emplace_back(device_id + PATH_SEPARATOR_CHAR + + measurement); + } + } + return tsfile_reader_->query(path_list, start_time, end_time, result_set); +} + +void TsFileTreeReader::destroy_query_data_set(ResultSet *qds) { + tsfile_reader_->destroy_query_data_set(qds); +} + +std::vector TsFileTreeReader::get_device_schema( + const std::string &device_id) { + std::vector schemas; + tsfile_reader_->get_timeseries_schema( + std::make_shared(device_id), schemas); + return schemas; +} + +std::vector TsFileTreeReader::get_all_device_ids() { + std::vector ret_device_ids; + auto device_ids = tsfile_reader_->get_all_device_ids(); + for (auto device_id : device_ids) { + ret_device_ids.emplace_back(device_id->get_device_name()); + } + return ret_device_ids; +} + +} // namespace storage \ No newline at end of file diff --git a/cpp/src/reader/tsfile_tree_reader.h b/cpp/src/reader/tsfile_tree_reader.h new file mode 100644 index 000000000..1d43c3b01 --- /dev/null +++ b/cpp/src/reader/tsfile_tree_reader.h @@ -0,0 +1,107 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * License); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#ifndef READER_TSFILE_TREE_READER_H +#define READER_TSFILE_TREE_READER_H + +#include +#include +#include + +#include "tsfile_reader.h" + +namespace storage { + +class TsFileTreeReader { + public: + TsFileTreeReader(); + ~TsFileTreeReader(); + + /** + * @brief open the tsfile + * + * @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); + + /** + * @brief close the tsfile, this method should be called after the + * query is finished + * + * @return Returns 0 on success, or a non-zero error code on failure. + */ + int close(); + + /** + * @brief Query time series data for specified devices and measurements + * within a time range + * + * @param device_ids List of device identifiers to query + * @param measurement_names List of measurement names to query + * @param start_time Start timestamp of the query range (inclusive) + * @param end_time End timestamp of the query range (inclusive) + * @param[out] result_set Pointer to the ResultSet that will contain query + * results + * @return Returns 0 on success, or a non-zero error code on failure. + * The caller is responsible for destroying the result set using + * destroy_query_data_set() + */ + int query(const std::vector &device_ids, + const std::vector &measurement_names, + int64_t start_time, int64_t end_time, ResultSet *&result_set); + + /** + * @brief Destroy and deallocate the query result set + * + * @param result_set Pointer to the ResultSet to be destroyed + * @note This method should be called after the result set is no longer + * needed to prevent memory leaks + */ + void destroy_query_data_set(ResultSet *result_set); + + /** + * @brief Get the measurement schema for a specific device + * + * @param device_id The device identifier + * @return Pointer to the MeasurementSchema for the device, or nullptr if + * not found + * @note The caller should not delete the returned pointer as it's managed + * by the reader + */ + std::vector get_device_schema( + const std::string &device_id); + + /** + * @brief Get all device identifiers in the TsFile + * + * @return Vector containing all device identifiers found in the TsFile + * @note The returned vector will be empty if no devices are found or file + * is not opened + */ + std::vector get_all_device_ids(); + + private: + std::shared_ptr + tsfile_reader_; ///< Underlying TsFile reader implementation +}; + +} // namespace storage + +#endif // READER_TSFILE_TREE_READER_H \ No newline at end of file diff --git a/cpp/src/writer/tsfile_tree_writer.cc b/cpp/src/writer/tsfile_tree_writer.cc new file mode 100644 index 000000000..59c11914d --- /dev/null +++ b/cpp/src/writer/tsfile_tree_writer.cc @@ -0,0 +1,53 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * License); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "writer/tsfile_tree_writer.h" + +namespace storage { + +TsFileTreeWriter::TsFileTreeWriter(storage::WriteFile* writer_file, + uint64_t memory_threshold) { + tsfile_writer_ = std::make_shared(); + tsfile_writer_->init(writer_file); + common::g_config_value_.chunk_group_size_threshold_ = memory_threshold; +} + +int TsFileTreeWriter::register_timeseries(std::string& device_id, + MeasurementSchema* schema) { + return tsfile_writer_->register_timeseries(device_id, *schema); +} + +int TsFileTreeWriter::register_timeseries( + std::string& device_id, std::vector schemas) { + return tsfile_writer_->register_aligned_timeseries(device_id, schemas); +} + +int TsFileTreeWriter::write(const Tablet& tablet) { + return tsfile_writer_->write_tree(tablet); +} + +int TsFileTreeWriter::write(const TsRecord& record) { + return tsfile_writer_->write_tree(record); +} + +int TsFileTreeWriter::flush() { return tsfile_writer_->flush(); } + +int TsFileTreeWriter::close() { return tsfile_writer_->close(); } + +} // namespace storage \ No newline at end of file diff --git a/cpp/src/writer/tsfile_tree_writer.h b/cpp/src/writer/tsfile_tree_writer.h new file mode 100644 index 000000000..90ef0d76f --- /dev/null +++ b/cpp/src/writer/tsfile_tree_writer.h @@ -0,0 +1,134 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * License); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +#ifndef WRITER_TSFILE_TREE_WRITER_H +#define WRITER_TSFILE_TREE_WRITER_H + +#include +#include +#include + +#include "tsfile_writer.h" + +namespace storage { + +/** + * @brief Provides an interface for writing hierarchical (tree-structured) + * time-series data into a TsFile. + * + * The TsFileTreeWriter class is designed for writing time-series data organized + * in a device–measurement hierarchy into a TsFile. It allows registration of + * single or aligned time series and supports writing data in both batch + * (Tablet) and row (TsRecord) formats. + * + * This writer is particularly suitable when data is naturally organized by + * device paths, and measurements are associated with those devices. The class + * also allows controlling memory usage via a configurable threshold. + */ +class TsFileTreeWriter { + public: + /** + * Constructs a TsFileTreeWriter instance to write hierarchical time-series + * data into the specified file, with an optional memory threshold for + * buffered data. + * + * @param writer_file Target file where the TsFile data will be written. + * Must not be null. + * @param memory_threshold Optional parameter specifying the memory usage + * threshold for buffered data before automatic flush occurs. Default is + * 128MB. + */ + explicit TsFileTreeWriter(storage::WriteFile* writer_file, + uint64_t memory_threshold = 128 * 1024 * 1024); + + /** + * Registers a single (non-aligned) time series under the given device ID. + * + * @param device_id The ID or path of the device to which the time series + * belongs. + * @param schema The measurement schema defining the data type and encoding + * of the time series. Must not be null. + * @return Returns 0 on success, or a non-zero error code on failure. + */ + int register_timeseries(std::string& device_id, MeasurementSchema* schema); + + /** + * Registers multiple aligned time series under the same device ID. + * + * @param device_id The ID or path of the device to which the aligned time + * series belong. + * @param schemas A vector of measurement schema pointers representing the + * aligned measurements. Must not be empty or contain null pointers. + * @return Returns 0 on success, or a non-zero error code on failure. + */ + int register_timeseries(std::string& device_id, + std::vector schemas); + + /** + * Writes a batch of data points (tablet) for one or more time series into + * the TsFile. + * + * @param tablet The Tablet object containing multiple rows of data to be + * written. Must not be null. + * @return Returns 0 on success, or a non-zero error code on failure. + */ + int write(const Tablet& tablet); + + /** + * Writes a single record (row) of data for a specific device and timestamp + * into the TsFile. + * + * @param record The TsRecord object containing the device ID, timestamp, + * and measurement values. Must not be null. + * @return Returns 0 on success, or a non-zero error code on failure. + */ + int write(const TsRecord& record); + + /** + * @brief Flushes all buffered data to the underlying storage + * + * This method forces any buffered data to be written to the TsFile. It is + * useful for ensuring data persistence at specific points in the writing + * process, such as after writing a significant amount of data or before + * performing critical operations. + * + * @return Returns 0 on success, or a non-zero error code on failure. + * @note After a successful flush, buffered data is cleared and memory + * usage is reduced. The writer continues to accept new data after + * flushing. + */ + int flush(); + + /** + * Closes the writer and ensures all buffered data is flushed to disk. + * After this method is called, no further operations should be performed on + * this writer instance. + * + * @return Returns 0 on success, or a non-zero error code on failure. + */ + int close(); + + private: + // Underlying TsFile writer responsible for the actual file I/O and + // serialization logic. + std::shared_ptr tsfile_writer_; +}; + +} // namespace storage + +#endif // WRITER_TSFILE_TREE_WRITER_H \ No newline at end of file diff --git a/cpp/src/writer/tsfile_writer.cc b/cpp/src/writer/tsfile_writer.cc index 54c1be020..3ae5cf28a 100644 --- a/cpp/src/writer/tsfile_writer.cc +++ b/cpp/src/writer/tsfile_writer.cc @@ -749,6 +749,31 @@ int TsFileWriter::write_tablet(const Tablet &tablet) { return ret; } +int TsFileWriter::write_tree(const Tablet &tablet) { + auto device_id = + std::make_shared(tablet.insert_target_name_); + if (schemas_.find(device_id) != schemas_.end()) { + auto schema = schemas_[device_id]; + if (schema->is_aligned_) { + return this->write_tablet_aligned(tablet); + } + return this->write_tablet(tablet); + } + return E_NOT_EXIST; +} + +int TsFileWriter::write_tree(const TsRecord &record) { + auto device_id = std::make_shared(record.device_id_); + if (schemas_.find(device_id) != schemas_.end()) { + auto schema = schemas_[device_id]; + if (schema->is_aligned_) { + return this->write_record_aligned(record); + } + return this->write_record(record); + } + return E_NOT_EXIST; +} + int TsFileWriter::write_table(Tablet &tablet) { int ret = E_OK; if (io_writer_->get_schema()->table_schema_map_.find( diff --git a/cpp/src/writer/tsfile_writer.h b/cpp/src/writer/tsfile_writer.h index bad92381c..db16ae335 100644 --- a/cpp/src/writer/tsfile_writer.h +++ b/cpp/src/writer/tsfile_writer.h @@ -73,6 +73,8 @@ class TsFileWriter { int write_tablet(const Tablet &tablet); int write_record_aligned(const TsRecord &record); int write_tablet_aligned(const Tablet &tablet); + int write_tree(const Tablet &tablet); + int write_tree(const TsRecord &record); int write_table(Tablet &tablet); typedef std::map, MeasurementSchemaGroup *, diff --git a/cpp/test/reader/tree_view/tsfile_reader_tree_test.cc b/cpp/test/reader/tree_view/tsfile_reader_tree_test.cc new file mode 100644 index 000000000..e7466c7b9 --- /dev/null +++ b/cpp/test/reader/tree_view/tsfile_reader_tree_test.cc @@ -0,0 +1,226 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * License); you may not use this file except in compliance + * with the License. You may obtain a copy of the License a + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +#include + +#include + +#include "common/record.h" +#include "common/schema.h" +#include "common/tablet.h" +#include "file/write_file.h" +#include "reader/tsfile_reader.h" +#include "reader/tsfile_tree_reader.h" +#include "writer/tsfile_table_writer.h" +#include "writer/tsfile_tree_writer.h" + +namespace storage { +class QDSWithoutTimeGenerator; +} +using namespace storage; +using namespace common; + +class TsFileTreeReaderTest : public ::testing::Test { + protected: + void SetUp() override { + libtsfile_init(); + file_name_ = std::string("tsfile_writer_tree_test_") + + generate_random_string(10) + std::string(".tsfile"); + remove(file_name_.c_str()); + int flags = O_WRONLY | O_CREAT | O_TRUNC; +#ifdef _WIN32 + flags |= O_BINARY; +#endif + mode_t mode = 0666; + write_file_.create(file_name_, flags, mode); + } + void TearDown() override { remove(file_name_.c_str()); } + std::string file_name_; + WriteFile write_file_; + + public: + static std::string generate_random_string(int length) { + std::random_device rd; + std::mt19937 gen(rd()); + std::uniform_int_distribution<> dis(0, 61); + + const std::string chars = + "0123456789" + "abcdefghijklmnopqrstuvwxyz" + "ABCDEFGHIJKLMNOPQRSTUVWXYZ"; + + std::string random_string; + + for (int i = 0; i < length; ++i) { + random_string += chars[dis(gen)]; + } + + return random_string; + } +}; + +TEST_F(TsFileTreeReaderTest, BasicTest) { + TsFileTreeWriter writer(&write_file_); + std::string device_id = "test_device"; + std::string measurement_id = "test_measurement"; + auto* measurement = new MeasurementSchema(measurement_id, INT64); + writer.register_timeseries(device_id, measurement); + TsRecord record(device_id, 0); + record.add_point(measurement_id, static_cast(1)); + writer.write(record); + writer.flush(); + writer.close(); + + delete measurement; + TsFileTreeReader reader; + reader.open(file_name_); + auto device_ids = reader.get_all_device_ids(); + ASSERT_EQ(device_ids.size(), 1); + ASSERT_EQ(device_ids[0], device_id); + + std::vector measurement_ids{measurement_id}; + ResultSet* result; + int ret = + reader.query(device_ids, measurement_ids, INT64_MIN, INT64_MAX, result); + ASSERT_EQ(ret, E_OK); + auto iter = result->iterator(); + RowRecord* read_record; + while (iter.hasNext()) { + read_record = iter.next(); + EXPECT_EQ(read_record->get_field(1)->type_, INT64); + } + reader.destroy_query_data_set(result); + reader.close(); +} + +TEST_F(TsFileTreeReaderTest, ExtendedRowsAndColumnsTest) { + TsFileTreeWriter writer(&write_file_); + std::vector device_ids = {"device_1", "device_2", "device_3"}; + std::vector measurement_ids = {"temperature", "humidity", + "pressure", "voltage"}; + std::vector data_types = {INT64, DOUBLE, FLOAT, INT32}; + std::vector measurements; + for (size_t i = 0; i < measurement_ids.size(); ++i) { + auto* measurement = + new MeasurementSchema(measurement_ids[i], data_types[i]); + measurements.push_back(measurement); + } + for (auto& device_id : device_ids) { + for (auto measurement : measurements) { + writer.register_timeseries(device_id, measurement); + } + } + + const int NUM_ROWS = 100; + for (int row = 0; row < NUM_ROWS; ++row) { + for (const auto& device_id : device_ids) { + TsRecord record(device_id, row * 1000); + for (size_t i = 0; i < measurement_ids.size(); ++i) { + switch (data_types[i]) { + case INT64: + record.add_point(measurement_ids[i], + static_cast(row + i)); + break; + case DOUBLE: + record.add_point(measurement_ids[i], + static_cast(row * 1.5 + i)); + break; + case FLOAT: + record.add_point(measurement_ids[i], + static_cast(row * 0.8f + i)); + break; + case INT32: + record.add_point(measurement_ids[i], + static_cast(row * 2 + i)); + break; + default: + break; + } + } + writer.write(record); + } + } + + writer.flush(); + writer.close(); + + TsFileTreeReader reader; + reader.open(file_name_); + + auto read_device_ids = reader.get_all_device_ids(); + ASSERT_EQ(read_device_ids.size(), device_ids.size()); + for (size_t i = 0; i < device_ids.size(); ++i) { + EXPECT_EQ(read_device_ids[i], device_ids[i]); + } + + ResultSet* result; + int ret = + reader.query(device_ids, measurement_ids, 0, NUM_ROWS * 1000, result); + ASSERT_EQ(ret, E_OK); + + auto iter = result->iterator(); + int row_count = 0; + + while (iter.hasNext()) { + RowRecord* read_record = iter.next(); + row_count++; + EXPECT_EQ(read_record->get_fields()->size(), + device_ids.size() * measurement_ids.size() + 1); + + // device_id1 + for (size_t i = 0; i < measurement_ids.size(); ++i) { + Field* field = read_record->get_field(i + 1); + ASSERT_NE(field, nullptr); + EXPECT_EQ(field->type_, data_types[i]); + + int64_t timestamp = read_record->get_timestamp(); + int row_index = timestamp / 1000; + + switch (data_types[i]) { + case INT64: { + EXPECT_EQ(field->get_value(), + static_cast(row_index + i)); + break; + } + case DOUBLE: { + EXPECT_NEAR(field->get_value(), row_index * 1.5 + i, + 0.001); + break; + } + case FLOAT: { + EXPECT_NEAR(field->get_value(), row_index * 0.8f + i, + 0.001f); + break; + } + case INT32: { + EXPECT_EQ(field->get_value(), + static_cast(row_index * 2 + i)); + break; + } + default: + break; + } + } + } + EXPECT_EQ(row_count * device_ids.size(), NUM_ROWS * device_ids.size()); + reader.destroy_query_data_set(result); + reader.close(); + for (auto* measurement : measurements) { + delete measurement; + } +} diff --git a/java/examples/src/main/java/org/apache/tsfile/v4/TsFileTreeReaderExample.java b/java/examples/src/main/java/org/apache/tsfile/v4/TsFileTreeReaderExample.java index a317a22e4..ab59cf837 100644 --- a/java/examples/src/main/java/org/apache/tsfile/v4/TsFileTreeReaderExample.java +++ b/java/examples/src/main/java/org/apache/tsfile/v4/TsFileTreeReaderExample.java @@ -109,7 +109,7 @@ public static void main(String[] args) throws IOException { try (ResultSet resultSet = treeReader.query(deviceIds, measurements, startTime, endTime)) { System.out.println("=== Iterator-based query results ==="); - Iterator recordIterator = resultSet.recordIterator(); + Iterator recordIterator = resultSet.iterator(); int countVal = 0; while (recordIterator.hasNext()) { diff --git a/java/tsfile/src/main/java/org/apache/tsfile/read/query/dataset/ResultSet.java b/java/tsfile/src/main/java/org/apache/tsfile/read/query/dataset/ResultSet.java index 2e5d07061..43fbd86a9 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/read/query/dataset/ResultSet.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/read/query/dataset/ResultSet.java @@ -92,5 +92,5 @@ public interface ResultSet extends AutoCloseable { void close(); @TsFileApi - Iterator recordIterator(); + Iterator iterator(); } diff --git a/java/tsfile/src/main/java/org/apache/tsfile/read/query/dataset/TableResultSet.java b/java/tsfile/src/main/java/org/apache/tsfile/read/query/dataset/TableResultSet.java index bd908d7e3..f3017c6cd 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/read/query/dataset/TableResultSet.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/read/query/dataset/TableResultSet.java @@ -144,7 +144,7 @@ public void close() { } @Override - public Iterator recordIterator() { + public Iterator iterator() { return new RecordIterator(); } diff --git a/java/tsfile/src/main/java/org/apache/tsfile/read/query/dataset/TreeResultSet.java b/java/tsfile/src/main/java/org/apache/tsfile/read/query/dataset/TreeResultSet.java index 2b2601679..9483e1e7e 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/read/query/dataset/TreeResultSet.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/read/query/dataset/TreeResultSet.java @@ -82,7 +82,7 @@ public void close() { } @Override - public Iterator recordIterator() { + public Iterator iterator() { return new TreeResultSet.RecordIterator(); } diff --git a/java/tsfile/src/test/java/org/apache/tsfile/read/TsFileTreeReaderTest.java b/java/tsfile/src/test/java/org/apache/tsfile/read/TsFileTreeReaderTest.java index 21d953dc2..9df373270 100644 --- a/java/tsfile/src/test/java/org/apache/tsfile/read/TsFileTreeReaderTest.java +++ b/java/tsfile/src/test/java/org/apache/tsfile/read/TsFileTreeReaderTest.java @@ -95,7 +95,7 @@ public void testIteratorQuery() throws IOException { long endTime = Long.MAX_VALUE; ResultSet resultSet = treeReader.query(deviceIds, measurements, startTime, endTime); - Iterator recordIterator = resultSet.recordIterator(); + Iterator recordIterator = resultSet.iterator(); int countVal = 0; while (recordIterator.hasNext()) { diff --git a/java/tsfile/src/test/java/org/apache/tsfile/read/query/ResultSetTest.java b/java/tsfile/src/test/java/org/apache/tsfile/read/query/ResultSetTest.java index b24e6e6d8..c453a38e0 100644 --- a/java/tsfile/src/test/java/org/apache/tsfile/read/query/ResultSetTest.java +++ b/java/tsfile/src/test/java/org/apache/tsfile/read/query/ResultSetTest.java @@ -478,7 +478,7 @@ public void testQueryTableByIterator() throws Exception { try (DeviceTableModelReader tsFileReader = new DeviceTableModelReader(tsfile); ResultSet resultSet = tsFileReader.query("T1", Arrays.asList("ID1", "ID2", "S2", "S1"), 0, 2); ) { - Iterator tsRecordIterator = resultSet.recordIterator(); + Iterator tsRecordIterator = resultSet.iterator(); Assert.assertTrue(tsRecordIterator.hasNext()); TSRecord tsRecord = tsRecordIterator.next();