| /* |
| * 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 "reader/tsfile_reader.h" |
| |
| #include <gtest/gtest.h> |
| #include <sys/stat.h> |
| |
| #include <map> |
| #include <random> |
| #include <unordered_map> |
| #include <vector> |
| |
| #include "common/record.h" |
| #include "common/schema.h" |
| #include "common/tablet.h" |
| #include "common/tsblock/tsblock.h" |
| #include "file/tsfile_io_reader.h" |
| #include "file/tsfile_io_writer.h" |
| #include "file/write_file.h" |
| #include "reader/block/single_device_tsblock_reader.h" |
| #include "reader/filter/time_operator.h" |
| #include "reader/qds_without_timegenerator.h" |
| #include "reader/tsfile_series_scan_iterator.h" |
| #include "writer/tsfile_writer.h" |
| |
| using namespace storage; |
| using namespace common; |
| |
| TEST(TsFileSeriesScanIteratorTest, ConsumeRowOffsetSaturates) { |
| storage::TsFileSeriesScanIterator ssi; |
| ssi.set_row_range(/*offset=*/10, /*limit=*/-1); |
| |
| ssi.consume_row_offset(-1); |
| EXPECT_EQ(ssi.get_row_offset(), 10); |
| ssi.consume_row_offset(std::numeric_limits<int>::min()); |
| EXPECT_EQ(ssi.get_row_offset(), 10); |
| |
| ssi.consume_row_offset(4); |
| EXPECT_EQ(ssi.get_row_offset(), 6); |
| ssi.consume_row_offset(6); |
| EXPECT_EQ(ssi.get_row_offset(), 0); |
| ssi.consume_row_offset(1); |
| EXPECT_EQ(ssi.get_row_offset(), 0); |
| } |
| |
| class TsFileReaderTest : public ::testing::Test { |
| protected: |
| void SetUp() override { |
| tsfile_writer_ = new TsFileWriter(); |
| libtsfile_init(); |
| file_name_ = std::string("tsfile_writer_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; |
| EXPECT_EQ(tsfile_writer_->open(file_name_, flags, mode), common::E_OK); |
| } |
| |
| void TearDown() override { |
| delete tsfile_writer_; |
| // remove(file_name_.c_str()); |
| libtsfile_destroy(); |
| } |
| |
| std::string file_name_; |
| TsFileWriter* tsfile_writer_ = nullptr; |
| |
| public: |
| static std::string generate_random_string(int length) { |
| std::mt19937 gen(static_cast<unsigned int>( |
| std::chrono::system_clock::now().time_since_epoch().count())); |
| 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; |
| } |
| |
| static std::string field_to_string(storage::Field* value) { |
| if (value->type_ == common::TEXT) { |
| return std::string(value->value_.sval_); |
| } else { |
| std::stringstream ss; |
| switch (value->type_) { |
| case common::BOOLEAN: |
| ss << (value->value_.bval_ ? "true" : "false"); |
| break; |
| case common::INT32: |
| ss << value->value_.ival_; |
| break; |
| case common::INT64: |
| ss << value->value_.lval_; |
| break; |
| case common::FLOAT: |
| ss << value->value_.fval_; |
| break; |
| case common::DOUBLE: |
| ss << value->value_.dval_; |
| break; |
| case common::NULL_TYPE: |
| ss << "NULL"; |
| break; |
| default: |
| ASSERT(false); |
| break; |
| } |
| return ss.str(); |
| } |
| } |
| }; |
| |
| TEST_F(TsFileReaderTest, ResultSetMetadata) { |
| std::string device_path = "device1"; |
| std::string measurement_name = "temperature"; |
| common::TSDataType data_type = common::TSDataType::INT32; |
| common::TSEncoding encoding = common::TSEncoding::PLAIN; |
| common::CompressionType compression_type = |
| common::CompressionType::UNCOMPRESSED; |
| tsfile_writer_->register_timeseries( |
| device_path, storage::MeasurementSchema(measurement_name, data_type, |
| encoding, compression_type)); |
| |
| for (int i = 0; i < 50000; ++i) { |
| TsRecord record(1622505600000 + i * 1000, device_path); |
| record.add_point(measurement_name, (int32_t)i); |
| ASSERT_EQ(tsfile_writer_->write_record(record), E_OK); |
| ASSERT_EQ(tsfile_writer_->flush(), E_OK); |
| } |
| ASSERT_EQ(tsfile_writer_->close(), E_OK); |
| |
| std::vector<std::string> select_list = {"device1.temperature"}; |
| |
| storage::TsFileReader reader; |
| int ret = reader.open(file_name_); |
| ASSERT_EQ(ret, common::E_OK); |
| storage::ResultSet* tmp_qds = nullptr; |
| |
| ret = reader.query(select_list, 1622505600000, 1622505600000 + 50000 * 1000, |
| tmp_qds); |
| auto* qds = (QDSWithoutTimeGenerator*)tmp_qds; |
| |
| std::shared_ptr<ResultSetMetadata> result_set_metadata = |
| qds->get_metadata(); |
| ASSERT_EQ(result_set_metadata->get_column_type(1), INT64); |
| ASSERT_EQ(result_set_metadata->get_column_name(1), "time"); |
| ASSERT_EQ(result_set_metadata->get_column_type(2), data_type); |
| ASSERT_EQ(result_set_metadata->get_column_name(2), |
| device_path + "." + measurement_name); |
| reader.destroy_query_data_set(qds); |
| reader.close(); |
| } |
| |
| TEST_F(TsFileReaderTest, GetAllDevice) { |
| std::string measurement_name = "temperature"; |
| common::TSDataType data_type = common::TSDataType::INT32; |
| common::TSEncoding encoding = common::TSEncoding::PLAIN; |
| common::CompressionType compression_type = |
| common::CompressionType::UNCOMPRESSED; |
| |
| for (size_t i = 0; i < 1024; i++) { |
| tsfile_writer_->register_timeseries( |
| "device.ln" + std::to_string(i), |
| storage::MeasurementSchema(measurement_name, data_type, encoding, |
| compression_type)); |
| } |
| |
| for (size_t i = 0; i < 1024; i++) { |
| TsRecord record(1622505600000, "device.ln" + std::to_string(i)); |
| record.add_point(measurement_name, (int32_t)0); |
| ASSERT_EQ(tsfile_writer_->write_record(record), E_OK); |
| } |
| ASSERT_EQ(tsfile_writer_->flush(), E_OK); |
| ASSERT_EQ(tsfile_writer_->close(), E_OK); |
| |
| storage::TsFileReader reader; |
| int ret = reader.open(file_name_); |
| ASSERT_EQ(ret, common::E_OK); |
| auto devices = reader.get_all_devices("device"); |
| ASSERT_EQ(devices.size(), 1024); |
| std::vector<std::shared_ptr<IDeviceID>> devices_name_expected; |
| for (size_t i = 0; i < 1024; i++) { |
| devices_name_expected.push_back(std::make_shared<StringArrayDeviceID>( |
| "device.ln" + std::to_string(i))); |
| } |
| std::sort(devices_name_expected.begin(), devices_name_expected.end(), |
| [](const std::shared_ptr<IDeviceID>& left_str, |
| const std::shared_ptr<IDeviceID>& right_str) { |
| return left_str->operator<(*right_str); |
| }); |
| |
| for (size_t i = 0; i < devices.size(); i++) { |
| ASSERT_TRUE(devices[i]->operator==(*devices_name_expected[i])); |
| } |
| } |
| |
| TEST_F(TsFileReaderTest, GetTimeseriesSchema) { |
| std::vector<std::string> device_path = {"device.ln1", "device.ln2 "}; |
| std::vector<std::string> measurement_name = {"temperature", "humidity"}; |
| common::TSDataType data_type = common::TSDataType::INT32; |
| common::TSEncoding encoding = common::TSEncoding::PLAIN; |
| common::CompressionType compression_type = |
| common::CompressionType::UNCOMPRESSED; |
| tsfile_writer_->register_timeseries( |
| device_path[0], |
| storage::MeasurementSchema(measurement_name[0], data_type, encoding, |
| compression_type)); |
| tsfile_writer_->register_timeseries( |
| device_path[1], |
| storage::MeasurementSchema(measurement_name[1], data_type, encoding, |
| compression_type)); |
| TsRecord record_0(1622505600000, device_path[0]); |
| record_0.add_point(measurement_name[0], (int32_t)0); |
| TsRecord record_1(1622505600000, device_path[1]); |
| record_1.add_point(measurement_name[1], (int32_t)1); |
| ASSERT_EQ(tsfile_writer_->write_record(record_0), E_OK); |
| ASSERT_EQ(tsfile_writer_->write_record(record_1), E_OK); |
| ASSERT_EQ(tsfile_writer_->flush(), E_OK); |
| ASSERT_EQ(tsfile_writer_->close(), E_OK); |
| |
| storage::TsFileReader reader; |
| int ret = reader.open(file_name_); |
| ASSERT_EQ(ret, common::E_OK); |
| std::vector<MeasurementSchema> measurement_schemas; |
| reader.get_timeseries_schema( |
| std::make_shared<StringArrayDeviceID>(device_path[0]), |
| measurement_schemas); |
| ASSERT_EQ(measurement_schemas[0].measurement_name_, measurement_name[0]); |
| ASSERT_EQ(measurement_schemas[0].data_type_, TSDataType::INT32); |
| |
| reader.get_timeseries_schema( |
| std::make_shared<StringArrayDeviceID>(device_path[1]), |
| measurement_schemas); |
| ASSERT_EQ(measurement_schemas[1].measurement_name_, measurement_name[1]); |
| ASSERT_EQ(measurement_schemas[1].data_type_, TSDataType::INT32); |
| |
| std::vector<std::shared_ptr<IDeviceID>> one_device = { |
| std::make_shared<StringArrayDeviceID>(device_path[0])}; |
| auto one_meta = reader.get_timeseries_metadata(one_device); |
| ASSERT_EQ(one_meta.size(), 1u); |
| auto timeseries_list = one_meta.begin()->second; |
| ASSERT_EQ(timeseries_list.size(), 1u); |
| ASSERT_EQ(timeseries_list[0]->get_measurement_name().to_std_string(), |
| measurement_name[0]); |
| ASSERT_EQ(timeseries_list[0]->get_statistic()->start_time_, 1622505600000); |
| ASSERT_EQ(timeseries_list[0]->get_statistic()->end_time_, 1622505600000); |
| ASSERT_EQ(timeseries_list[0]->get_statistic()->count_, 1); |
| |
| auto device_timeseries_map = reader.get_timeseries_metadata(); |
| ASSERT_EQ(device_timeseries_map.size(), 2u); |
| auto device_timeseries_1 = device_timeseries_map.at( |
| std::make_shared<StringArrayDeviceID>(device_path[1])); |
| ASSERT_EQ(device_timeseries_1.size(), 1u); |
| ASSERT_EQ(device_timeseries_1[0]->get_measurement_name().to_std_string(), |
| measurement_name[1]); |
| ASSERT_EQ(device_timeseries_1[0]->get_statistic()->start_time_, |
| 1622505600000); |
| ASSERT_EQ(device_timeseries_1[0]->get_statistic()->end_time_, |
| 1622505600000); |
| ASSERT_EQ(device_timeseries_1[0]->get_statistic()->count_, 1); |
| reader.close(); |
| } |
| |
| TEST_F(TsFileReaderTest, GetTimeseriesMetadataTableModelTypeAndDeviceFilter) { |
| std::vector<MeasurementSchema*> measurement_schemas = { |
| new MeasurementSchema("deviceid1", TSDataType::STRING), |
| new MeasurementSchema("deviceid2", TSDataType::STRING), |
| new MeasurementSchema("temperature", TSDataType::FLOAT), |
| new MeasurementSchema("pressure", TSDataType::DOUBLE), |
| new MeasurementSchema("humidity", TSDataType::INT32)}; |
| std::vector<ColumnCategory> column_categories = { |
| ColumnCategory::TAG, ColumnCategory::TAG, ColumnCategory::FIELD, |
| ColumnCategory::FIELD, ColumnCategory::FIELD}; |
| auto table_schema = std::make_shared<TableSchema>( |
| "testtable", measurement_schemas, column_categories); |
| |
| ASSERT_EQ(tsfile_writer_->register_table(table_schema), E_OK); |
| |
| Tablet tablet(table_schema->get_table_name(), |
| table_schema->get_measurement_names(), |
| table_schema->get_data_types(), |
| table_schema->get_column_categories(), 10); |
| for (int row = 0; row < 5; row++) { |
| ASSERT_EQ(tablet.add_timestamp(row, row), E_OK); |
| ASSERT_EQ(tablet.add_value(row, "deviceid1", "device_a"), E_OK); |
| ASSERT_EQ(tablet.add_value(row, "deviceid2", "device_b"), E_OK); |
| ASSERT_EQ(tablet.add_value(row, "temperature", static_cast<float>(row)), |
| E_OK); |
| ASSERT_EQ(tablet.add_value(row, "pressure", static_cast<double>(row)), |
| E_OK); |
| ASSERT_EQ(tablet.add_value(row, "humidity", static_cast<int32_t>(row)), |
| E_OK); |
| } |
| for (int row = 5; row < 10; row++) { |
| ASSERT_EQ(tablet.add_timestamp(row, row), E_OK); |
| ASSERT_EQ(tablet.add_value(row, "deviceid1", "device_b"), E_OK); |
| ASSERT_EQ(tablet.add_value(row, "deviceid2", "device_a"), E_OK); |
| ASSERT_EQ(tablet.add_value(row, "temperature", static_cast<float>(row)), |
| E_OK); |
| ASSERT_EQ(tablet.add_value(row, "pressure", static_cast<double>(row)), |
| E_OK); |
| ASSERT_EQ(tablet.add_value(row, "humidity", static_cast<int32_t>(row)), |
| E_OK); |
| } |
| |
| // Append one row whose middle TAG segment is null. |
| Tablet null_tag_tablet(table_schema->get_table_name(), |
| table_schema->get_measurement_names(), |
| table_schema->get_data_types(), |
| table_schema->get_column_categories(), 1); |
| int64_t null_tag_ts[1] = {10}; |
| int32_t null_tag_humidity[1] = {10}; |
| float null_tag_temperature[1] = {10.0F}; |
| double null_tag_pressure[1] = {10.0}; |
| // deviceid1 = null |
| int32_t id1_offsets[2] = {0, 0}; |
| uint8_t id1_bitmap[1] = {0x01}; // row0 is null |
| // deviceid2 = "device_b" |
| int32_t id2_offsets[2] = {0, 8}; |
| const char id2_data[] = "device_b"; |
| ASSERT_EQ(null_tag_tablet.set_timestamps(null_tag_ts, 1), E_OK); |
| ASSERT_EQ(null_tag_tablet.set_column_string_values(0, id1_offsets, "", |
| id1_bitmap, 1), |
| E_OK); |
| ASSERT_EQ(null_tag_tablet.set_column_string_values(1, id2_offsets, id2_data, |
| nullptr, 1), |
| E_OK); |
| ASSERT_EQ( |
| null_tag_tablet.set_column_values(2, null_tag_temperature, nullptr, 1), |
| E_OK); |
| ASSERT_EQ( |
| null_tag_tablet.set_column_values(3, null_tag_pressure, nullptr, 1), |
| E_OK); |
| ASSERT_EQ( |
| null_tag_tablet.set_column_values(4, null_tag_humidity, nullptr, 1), |
| E_OK); |
| |
| ASSERT_EQ(tsfile_writer_->write_table(tablet), E_OK); |
| ASSERT_EQ(tsfile_writer_->write_table(null_tag_tablet), E_OK); |
| ASSERT_EQ(tsfile_writer_->flush(), E_OK); |
| ASSERT_EQ(tsfile_writer_->close(), E_OK); |
| |
| storage::TsFileReader reader; |
| ASSERT_EQ(reader.open(file_name_), common::E_OK); |
| |
| auto all_meta = reader.get_timeseries_metadata(); |
| ASSERT_EQ(all_meta.size(), 3u); |
| |
| std::vector<std::string> selected_device_segments = { |
| "testtable", "device_a", "device_b"}; |
| std::vector<std::shared_ptr<IDeviceID>> selected_devices = { |
| std::make_shared<StringArrayDeviceID>(selected_device_segments)}; |
| auto selected_meta = reader.get_timeseries_metadata(selected_devices); |
| ASSERT_EQ(selected_meta.size(), 1u); |
| |
| auto selected_list = selected_meta.begin()->second; |
| std::unordered_map<std::string, TSDataType> type_by_measurement; |
| for (const auto& index : selected_list) { |
| type_by_measurement[index->get_measurement_name().to_std_string()] = |
| index->get_data_type(); |
| } |
| ASSERT_EQ(type_by_measurement.at("temperature"), TSDataType::FLOAT); |
| ASSERT_EQ(type_by_measurement.at("pressure"), TSDataType::DOUBLE); |
| ASSERT_EQ(type_by_measurement.at("humidity"), TSDataType::INT32); |
| |
| // Query metadata for the device with null middle TAG segment. |
| std::vector<std::string*> null_seg_device = { |
| new std::string("testtable"), nullptr, new std::string("device_b")}; |
| std::vector<std::shared_ptr<IDeviceID>> null_seg_devices = { |
| std::make_shared<StringArrayDeviceID>(null_seg_device)}; |
| for (auto* seg : null_seg_device) { |
| if (seg != nullptr) { |
| delete seg; |
| } |
| } |
| auto null_seg_meta = reader.get_timeseries_metadata(null_seg_devices); |
| ASSERT_EQ(null_seg_meta.size(), 1u); |
| auto null_seg_list = null_seg_meta.begin()->second; |
| ASSERT_EQ(null_seg_list.size(), 3u); |
| std::unordered_map<std::string, TSDataType> null_seg_type_by_measurement; |
| for (const auto& index : null_seg_list) { |
| null_seg_type_by_measurement[index->get_measurement_name() |
| .to_std_string()] = |
| index->get_data_type(); |
| } |
| ASSERT_EQ(null_seg_type_by_measurement.at("temperature"), |
| TSDataType::FLOAT); |
| ASSERT_EQ(null_seg_type_by_measurement.at("pressure"), TSDataType::DOUBLE); |
| ASSERT_EQ(null_seg_type_by_measurement.at("humidity"), TSDataType::INT32); |
| |
| reader.close(); |
| } |
| |
| static const int64_t kLargeFileNumRecords = 300000000; |
| static const int64_t kLargeFileFlushBatch = 100000; |
| |
| TEST_F(TsFileReaderTest, |
| DISABLED_LargeFileNoEncodingNoCompression_WriteAndRead) { |
| std::string device_path = "device1"; |
| std::string measurement_name = "temperature"; |
| common::TSDataType data_type = common::TSDataType::INT64; |
| common::TSEncoding encoding = common::TSEncoding::PLAIN; |
| common::CompressionType compression_type = |
| common::CompressionType::UNCOMPRESSED; |
| |
| tsfile_writer_->register_timeseries( |
| device_path, storage::MeasurementSchema(measurement_name, data_type, |
| encoding, compression_type)); |
| |
| const int64_t start_time = 1622505600000LL; |
| for (int64_t i = 0; i < kLargeFileNumRecords; ++i) { |
| TsRecord record(start_time + i * 1000, device_path); |
| record.add_point(measurement_name, static_cast<int64_t>(i)); |
| ASSERT_EQ(tsfile_writer_->write_record(record), E_OK); |
| if ((i + 1) % kLargeFileFlushBatch == 0) { |
| ASSERT_EQ(tsfile_writer_->flush(), E_OK); |
| } |
| } |
| ASSERT_EQ(tsfile_writer_->flush(), E_OK); |
| ASSERT_EQ(tsfile_writer_->close(), E_OK); |
| |
| std::vector<std::string> select_list = {"device1.temperature"}; |
| const int64_t end_time = start_time + (kLargeFileNumRecords - 1) * 1000 + 1; |
| |
| storage::TsFileReader reader; |
| int ret = reader.open(file_name_); |
| ASSERT_EQ(ret, common::E_OK); |
| |
| storage::ResultSet* tmp_qds = nullptr; |
| ret = reader.query(select_list, start_time, end_time, tmp_qds); |
| ASSERT_EQ(ret, common::E_OK); |
| ASSERT_NE(tmp_qds, nullptr); |
| |
| auto* qds = static_cast<QDSWithoutTimeGenerator*>(tmp_qds); |
| std::shared_ptr<ResultSetMetadata> meta = qds->get_metadata(); |
| ASSERT_NE(meta, nullptr); |
| ASSERT_EQ(meta->get_column_type(1), INT64); |
| ASSERT_EQ(meta->get_column_type(2), INT64); |
| |
| int64_t row_count = 0; |
| bool has_next = false; |
| |
| while (true) { |
| ret = qds->next(has_next); |
| ASSERT_EQ(ret, common::E_OK); |
| if (!has_next) break; |
| row_count++; |
| } |
| |
| ASSERT_EQ(row_count, kLargeFileNumRecords); |
| |
| reader.destroy_query_data_set(qds); |
| reader.close(); |
| } |
| |
| // The multi-value page plan consumes row_offset before value decoding. When |
| // the offset lands inside a page, the returned TsBlock starts at the first row |
| // after the offset instead of exposing a residual to the row reader. |
| TEST_F(TsFileReaderTest, MultiValueAlignedRowOffsetTrimsPartialPage) { |
| const std::string device = "root.dev_multi_offset"; |
| std::vector<MeasurementSchema> schema_vec; |
| schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED); |
| schema_vec.emplace_back("v1", INT64, PLAIN, UNCOMPRESSED); |
| { |
| std::vector<MeasurementSchema*> reg; |
| for (auto& s : schema_vec) reg.push_back(new MeasurementSchema(s)); |
| ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device, reg), |
| E_OK); |
| } |
| const int N = 32; |
| Tablet tablet(device, |
| std::make_shared<std::vector<MeasurementSchema>>(schema_vec), |
| N); |
| for (int i = 0; i < N; ++i) { |
| ASSERT_EQ(tablet.add_timestamp(i, 1000 + i), E_OK); |
| ASSERT_EQ(tablet.add_value(i, 0u, static_cast<int64_t>(i)), E_OK); |
| ASSERT_EQ(tablet.add_value(i, 1u, static_cast<int64_t>(i * 2)), E_OK); |
| } |
| ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK); |
| ASSERT_EQ(tsfile_writer_->flush(), E_OK); |
| ASSERT_EQ(tsfile_writer_->close(), E_OK); |
| |
| storage::TsFileIOReader io_reader; |
| ASSERT_EQ(io_reader.init(file_name_), E_OK); |
| |
| auto device_id = std::make_shared<StringArrayDeviceID>(device); |
| std::vector<std::string> measurements = {"v0", "v1"}; |
| storage::TsFileSeriesScanIterator* ssi = nullptr; |
| common::PageArena pa; |
| pa.init(512, common::MOD_DEFAULT); |
| ASSERT_EQ(io_reader.alloc_multi_ssi(device_id, measurements, ssi, pa, |
| /*time_filter=*/nullptr), |
| E_OK); |
| ASSERT_NE(ssi, nullptr); |
| |
| ssi->set_row_range(/*offset=*/5, /*limit=*/-1); |
| common::TsBlock* block = nullptr; |
| EXPECT_EQ(ssi->get_next(block, /*alloc_tsblock=*/true), common::E_OK); |
| ASSERT_NE(block, nullptr); |
| EXPECT_EQ(block->get_row_count(), static_cast<uint32_t>(N - 5)); |
| EXPECT_EQ(ssi->get_row_offset(), 0); |
| { |
| common::ColIterator time_iter(0, block); |
| common::ColIterator v0_iter(1, block); |
| common::ColIterator v1_iter(2, block); |
| uint32_t len = 0; |
| EXPECT_EQ(*reinterpret_cast<int64_t*>(time_iter.read(&len)), 1005); |
| EXPECT_EQ(*reinterpret_cast<int64_t*>(v0_iter.read(&len)), 5); |
| EXPECT_EQ(*reinterpret_cast<int64_t*>(v1_iter.read(&len)), 10); |
| } |
| |
| if (block != nullptr) { |
| ssi->revert_tsblock(); |
| } |
| io_reader.revert_ssi(ssi); |
| // RAII handles io_reader teardown — explicit reset() would destroy the |
| // tsfile_meta page arena while tsfile_meta_ still holds shared_ptrs into |
| // it, then ~TsFileMeta would call self_deleter on freed memory. |
| } |
| |
| TEST_F(TsFileReaderTest, MultiValueAlignedRowOffsetSkipsWholePage) { |
| struct PagePointGuard { |
| explicit PagePointGuard(uint32_t page_points) |
| : saved_(common::g_config_value_.page_writer_max_point_num_) { |
| common::g_config_value_.page_writer_max_point_num_ = page_points; |
| } |
| ~PagePointGuard() { |
| common::g_config_value_.page_writer_max_point_num_ = saved_; |
| } |
| uint32_t saved_; |
| } page_point_guard(16); |
| |
| const std::string device = "root.dev_multi_page_offset"; |
| std::vector<MeasurementSchema> schema_vec; |
| schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED); |
| schema_vec.emplace_back("v1", INT64, PLAIN, UNCOMPRESSED); |
| { |
| std::vector<MeasurementSchema*> reg; |
| for (auto& s : schema_vec) reg.push_back(new MeasurementSchema(s)); |
| ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device, reg), |
| E_OK); |
| } |
| |
| const int rows_per_page = 16; |
| const int N = rows_per_page * 4; |
| Tablet tablet(device, |
| std::make_shared<std::vector<MeasurementSchema>>(schema_vec), |
| N); |
| for (int i = 0; i < N; ++i) { |
| ASSERT_EQ(tablet.add_timestamp(i, static_cast<int64_t>(i)), E_OK); |
| ASSERT_EQ(tablet.add_value(i, 0u, static_cast<int64_t>(i)), E_OK); |
| ASSERT_EQ(tablet.add_value(i, 1u, static_cast<int64_t>(i * 2)), E_OK); |
| } |
| ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK); |
| ASSERT_EQ(tsfile_writer_->flush(), E_OK); |
| ASSERT_EQ(tsfile_writer_->close(), E_OK); |
| |
| storage::TsFileIOReader io_reader; |
| ASSERT_EQ(io_reader.init(file_name_), E_OK); |
| |
| auto device_id = std::make_shared<StringArrayDeviceID>(device); |
| std::vector<std::string> measurements = {"v0", "v1"}; |
| storage::TsFileSeriesScanIterator* ssi = nullptr; |
| common::PageArena pa; |
| pa.init(512, common::MOD_TSFILE_READER); |
| ASSERT_EQ(io_reader.alloc_multi_ssi(device_id, measurements, ssi, pa, |
| /*time_filter=*/nullptr), |
| E_OK); |
| ASSERT_NE(ssi, nullptr); |
| |
| // The first chunk is preloaded before set_row_range(). The page plan skips |
| // page 0 and trims the first four rows from page 1 before value decoding. |
| ssi->set_row_range(/*offset=*/20, /*limit=*/-1); |
| common::TsBlock* block = nullptr; |
| ASSERT_EQ(ssi->get_next(block, /*alloc_tsblock=*/true), common::E_OK); |
| ASSERT_NE(block, nullptr); |
| ASSERT_EQ(block->get_row_count(), static_cast<uint32_t>(N - 20)); |
| EXPECT_EQ(ssi->get_row_offset(), 0); |
| |
| { |
| common::ColIterator time_iter(0, block); |
| common::ColIterator v0_iter(1, block); |
| common::ColIterator v1_iter(2, block); |
| uint32_t len = 0; |
| EXPECT_EQ(*reinterpret_cast<int64_t*>(time_iter.read(&len)), 20); |
| EXPECT_EQ(*reinterpret_cast<int64_t*>(v0_iter.read(&len)), 20); |
| EXPECT_EQ(*reinterpret_cast<int64_t*>(v1_iter.read(&len)), 40); |
| } |
| |
| ssi->revert_tsblock(); |
| io_reader.revert_ssi(ssi); |
| } |
| |
| TEST_F(TsFileReaderTest, MultiValueAlignedRowOffsetSkipsWholeChunk) { |
| const std::string device = "root.dev_multi_chunk_offset"; |
| std::vector<MeasurementSchema> schema_vec; |
| schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED); |
| schema_vec.emplace_back("v1", INT64, PLAIN, UNCOMPRESSED); |
| { |
| std::vector<MeasurementSchema*> reg; |
| for (auto& s : schema_vec) reg.push_back(new MeasurementSchema(s)); |
| ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device, reg), |
| E_OK); |
| } |
| |
| const int rows_per_chunk = 64; |
| const int chunk_count = 4; |
| for (int chunk = 0; chunk < chunk_count; ++chunk) { |
| Tablet tablet( |
| device, |
| std::make_shared<std::vector<MeasurementSchema>>(schema_vec), |
| rows_per_chunk); |
| const int base = chunk * rows_per_chunk; |
| for (int i = 0; i < rows_per_chunk; ++i) { |
| const int row = base + i; |
| ASSERT_EQ(tablet.add_timestamp(i, static_cast<int64_t>(row)), E_OK); |
| ASSERT_EQ(tablet.add_value(i, 0u, static_cast<int64_t>(row)), E_OK); |
| ASSERT_EQ(tablet.add_value(i, 1u, static_cast<int64_t>(row * 2)), |
| E_OK); |
| } |
| ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK); |
| ASSERT_EQ(tsfile_writer_->flush(), E_OK); |
| } |
| ASSERT_EQ(tsfile_writer_->close(), E_OK); |
| |
| storage::TsFileIOReader io_reader; |
| ASSERT_EQ(io_reader.init(file_name_), E_OK); |
| |
| auto device_id = std::make_shared<StringArrayDeviceID>(device); |
| std::vector<std::string> measurements = {"v0", "v1"}; |
| storage::TsFileSeriesScanIterator* ssi = nullptr; |
| common::PageArena pa; |
| pa.init(512, common::MOD_TSFILE_READER); |
| ASSERT_EQ(io_reader.alloc_multi_ssi(device_id, measurements, ssi, pa, |
| /*time_filter=*/nullptr), |
| E_OK); |
| ASSERT_NE(ssi, nullptr); |
| |
| // alloc_multi_ssi() preloads chunk 0 before row range is set. Read it |
| // normally first, then set an offset that skips the next whole chunk by |
| // chunk metadata count and trims the first 32 rows from chunk 2's plan. |
| common::TsBlock* block = nullptr; |
| ASSERT_EQ(ssi->get_next(block, /*alloc_tsblock=*/true), common::E_OK); |
| ASSERT_NE(block, nullptr); |
| ASSERT_EQ(block->get_row_count(), static_cast<uint32_t>(rows_per_chunk)); |
| { |
| common::ColIterator time_iter(0, block); |
| common::ColIterator v0_iter(1, block); |
| common::ColIterator v1_iter(2, block); |
| uint32_t len = 0; |
| EXPECT_EQ(*reinterpret_cast<int64_t*>(time_iter.read(&len)), 0); |
| EXPECT_EQ(*reinterpret_cast<int64_t*>(v0_iter.read(&len)), 0); |
| EXPECT_EQ(*reinterpret_cast<int64_t*>(v1_iter.read(&len)), 0); |
| } |
| ssi->revert_tsblock(); |
| |
| ssi->set_row_range(/*offset=*/96, /*limit=*/-1); |
| block = nullptr; |
| ASSERT_EQ(ssi->get_next(block, /*alloc_tsblock=*/true), common::E_OK); |
| ASSERT_NE(block, nullptr); |
| ASSERT_EQ(block->get_row_count(), |
| static_cast<uint32_t>(rows_per_chunk / 2)); |
| EXPECT_EQ(ssi->get_row_offset(), 0); |
| |
| { |
| common::ColIterator time_iter(0, block); |
| common::ColIterator v0_iter(1, block); |
| common::ColIterator v1_iter(2, block); |
| uint32_t len = 0; |
| EXPECT_EQ(*reinterpret_cast<int64_t*>(time_iter.read(&len)), 160); |
| EXPECT_EQ(*reinterpret_cast<int64_t*>(v0_iter.read(&len)), 160); |
| EXPECT_EQ(*reinterpret_cast<int64_t*>(v1_iter.read(&len)), 320); |
| } |
| |
| ssi->revert_tsblock(); |
| io_reader.revert_ssi(ssi); |
| } |
| |
| // Offset is defined over the rows left after filtering. A partially matching |
| // chunk therefore cannot consume its full time-row count from the offset. |
| TEST_F(TsFileReaderTest, MultiValueAlignedRowOffsetCountsOnlyFilteredRows) { |
| const std::string device = "root.dev_multi_filtered_offset"; |
| std::vector<MeasurementSchema> schema_vec; |
| schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED); |
| schema_vec.emplace_back("v1", INT64, PLAIN, UNCOMPRESSED); |
| { |
| std::vector<MeasurementSchema*> reg; |
| for (auto& s : schema_vec) reg.push_back(new MeasurementSchema(s)); |
| ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device, reg), |
| E_OK); |
| } |
| |
| const int rows_per_chunk = 64; |
| const int chunk_count = 4; |
| for (int chunk = 0; chunk < chunk_count; ++chunk) { |
| Tablet tablet( |
| device, |
| std::make_shared<std::vector<MeasurementSchema>>(schema_vec), |
| rows_per_chunk); |
| const int base = chunk * rows_per_chunk; |
| for (int i = 0; i < rows_per_chunk; ++i) { |
| const int row = base + i; |
| ASSERT_EQ(tablet.add_timestamp(i, static_cast<int64_t>(row)), E_OK); |
| ASSERT_EQ(tablet.add_value(i, 0u, static_cast<int64_t>(row)), E_OK); |
| ASSERT_EQ(tablet.add_value(i, 1u, static_cast<int64_t>(row * 2)), |
| E_OK); |
| } |
| ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK); |
| ASSERT_EQ(tsfile_writer_->flush(), E_OK); |
| } |
| ASSERT_EQ(tsfile_writer_->close(), E_OK); |
| |
| storage::TsFileIOReader io_reader; |
| ASSERT_EQ(io_reader.init(file_name_), E_OK); |
| |
| auto device_id = std::make_shared<StringArrayDeviceID>(device); |
| std::vector<std::string> measurements = {"v0", "v1"}; |
| storage::TsFileSeriesScanIterator* ssi = nullptr; |
| common::PageArena pa; |
| pa.init(512, common::MOD_TSFILE_READER); |
| |
| // Keep all 64 rows from chunk 0, only 10 rows from chunk 1, and all 128 |
| // rows from chunks 2 and 3. The filtered stream has 202 rows, so offset |
| // 200 must leave only timestamps 254 and 255. |
| std::vector<int64_t> selected_times; |
| for (int64_t t = 0; t < 64; ++t) selected_times.push_back(t); |
| for (int64_t t = 64; t < 74; ++t) selected_times.push_back(t); |
| for (int64_t t = 128; t < 256; ++t) selected_times.push_back(t); |
| storage::TimeIn time_filter(selected_times, /*not_in=*/false); |
| |
| ASSERT_EQ(io_reader.alloc_multi_ssi(device_id, measurements, ssi, pa, |
| &time_filter), |
| E_OK); |
| ASSERT_NE(ssi, nullptr); |
| ssi->set_row_range(/*offset=*/200, /*limit=*/-1); |
| |
| std::vector<int64_t> actual_times; |
| while (true) { |
| common::TsBlock* block = nullptr; |
| int ret = ssi->get_next(block, /*alloc_tsblock=*/true, &time_filter); |
| if (ret == common::E_NO_MORE_DATA) break; |
| ASSERT_EQ(ret, common::E_OK); |
| ASSERT_NE(block, nullptr); |
| |
| const uint32_t rows = block->get_row_count(); |
| const int remaining_offset = ssi->get_row_offset(); |
| const uint32_t rows_to_skip = std::min( |
| rows, static_cast<uint32_t>(std::max(remaining_offset, 0))); |
| { |
| common::ColIterator time_iter(0, block); |
| for (uint32_t row = 0; row < rows; ++row) { |
| uint32_t len = 0; |
| int64_t time = |
| *reinterpret_cast<int64_t*>(time_iter.read(&len)); |
| if (row >= rows_to_skip) actual_times.push_back(time); |
| time_iter.next(); |
| } |
| } |
| ssi->set_row_range(remaining_offset - static_cast<int>(rows_to_skip), |
| /*limit=*/-1); |
| ssi->revert_tsblock(); |
| } |
| |
| ASSERT_EQ(actual_times.size(), 2u); |
| EXPECT_EQ(actual_times[0], 254); |
| EXPECT_EQ(actual_times[1], 255); |
| |
| io_reader.revert_ssi(ssi); |
| } |
| |
| // A boundary page may contain holes between matching timestamps. Offset must |
| // consume only matching rows and keep every value column aligned with the |
| // first retained match. |
| TEST_F(TsFileReaderTest, MultiValueAlignedPagePlanOffsetSkipsBoundaryMatches) { |
| const std::string device = "root.dev_multi_boundary_offset"; |
| std::vector<MeasurementSchema> schema_vec; |
| schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED); |
| schema_vec.emplace_back("v1", INT64, PLAIN, UNCOMPRESSED); |
| { |
| std::vector<MeasurementSchema*> reg; |
| for (auto& s : schema_vec) reg.push_back(new MeasurementSchema(s)); |
| ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device, reg), |
| E_OK); |
| } |
| |
| constexpr int kRowCount = 16; |
| Tablet tablet(device, |
| std::make_shared<std::vector<MeasurementSchema>>(schema_vec), |
| kRowCount); |
| for (int i = 0; i < kRowCount; ++i) { |
| ASSERT_EQ(tablet.add_timestamp(i, i), E_OK); |
| ASSERT_EQ(tablet.add_value(i, 0u, static_cast<int64_t>(i)), E_OK); |
| ASSERT_EQ(tablet.add_value(i, 1u, static_cast<int64_t>(i * 10)), E_OK); |
| } |
| ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK); |
| ASSERT_EQ(tsfile_writer_->flush(), E_OK); |
| ASSERT_EQ(tsfile_writer_->close(), E_OK); |
| |
| storage::TsFileIOReader io_reader; |
| ASSERT_EQ(io_reader.init(file_name_), E_OK); |
| |
| auto device_id = std::make_shared<StringArrayDeviceID>(device); |
| std::vector<std::string> measurements = {"v0", "v1"}; |
| storage::TsFileSeriesScanIterator* ssi = nullptr; |
| common::PageArena pa; |
| pa.init(512, common::MOD_TSFILE_READER); |
| const std::vector<int64_t> selected_times = {1, 3, 5, 7, 9}; |
| storage::TimeIn time_filter(selected_times, /*not_in=*/false); |
| ASSERT_EQ(io_reader.alloc_multi_ssi(device_id, measurements, ssi, pa, |
| &time_filter), |
| E_OK); |
| ASSERT_NE(ssi, nullptr); |
| |
| ssi->set_row_range(/*offset=*/2, /*limit=*/-1); |
| common::TsBlock* block = nullptr; |
| ASSERT_EQ(ssi->get_next(block, /*alloc_tsblock=*/true, &time_filter), E_OK); |
| ASSERT_NE(block, nullptr); |
| EXPECT_EQ(ssi->get_row_offset(), 0); |
| ASSERT_EQ(block->get_row_count(), 3u); |
| |
| const std::vector<int64_t> expected_times = {5, 7, 9}; |
| { |
| common::ColIterator time_iter(0, block); |
| common::ColIterator v0_iter(1, block); |
| common::ColIterator v1_iter(2, block); |
| for (int64_t expected : expected_times) { |
| uint32_t len = 0; |
| EXPECT_EQ(*reinterpret_cast<int64_t*>(time_iter.read(&len)), |
| expected); |
| EXPECT_EQ(*reinterpret_cast<int64_t*>(v0_iter.read(&len)), |
| expected); |
| EXPECT_EQ(*reinterpret_cast<int64_t*>(v1_iter.read(&len)), |
| expected * 10); |
| time_iter.next(); |
| v0_iter.next(); |
| v1_iter.next(); |
| } |
| } |
| |
| ssi->revert_tsblock(); |
| io_reader.revert_ssi(ssi); |
| } |
| |
| // The single-value SSI paths must apply offset to the filtered stream too. |
| // Cover both a non-aligned series (ChunkReader) and a single-value aligned |
| // series (AlignedChunkReader), including their chunk- and page-level skips. |
| TEST_F(TsFileReaderTest, SingleValueRowOffsetCountsOnlyFilteredRows) { |
| const std::string non_aligned_device = "root.dev_filtered_offset"; |
| const std::string aligned_device = "root.dev_aligned_filtered_offset"; |
| MeasurementSchema schema("v0", INT64, PLAIN, UNCOMPRESSED); |
| ASSERT_EQ(tsfile_writer_->register_timeseries(non_aligned_device, schema), |
| E_OK); |
| ASSERT_EQ( |
| tsfile_writer_->register_aligned_timeseries(aligned_device, schema), |
| E_OK); |
| |
| const int rows_per_chunk = 64; |
| const int chunk_count = 4; |
| auto schemas = std::make_shared<std::vector<MeasurementSchema>>(1, schema); |
| for (int chunk = 0; chunk < chunk_count; ++chunk) { |
| const int base = chunk * rows_per_chunk; |
| Tablet non_aligned_tablet(non_aligned_device, schemas, rows_per_chunk); |
| Tablet aligned_tablet(aligned_device, schemas, rows_per_chunk); |
| for (int i = 0; i < rows_per_chunk; ++i) { |
| const int row = base + i; |
| ASSERT_EQ(non_aligned_tablet.add_timestamp(i, row), E_OK); |
| ASSERT_EQ( |
| non_aligned_tablet.add_value(i, 0u, static_cast<int64_t>(row)), |
| E_OK); |
| ASSERT_EQ(aligned_tablet.add_timestamp(i, row), E_OK); |
| ASSERT_EQ( |
| aligned_tablet.add_value(i, 0u, static_cast<int64_t>(row)), |
| E_OK); |
| } |
| ASSERT_EQ(tsfile_writer_->write_tablet(non_aligned_tablet), E_OK); |
| ASSERT_EQ(tsfile_writer_->write_tablet_aligned(aligned_tablet), E_OK); |
| ASSERT_EQ(tsfile_writer_->flush(), E_OK); |
| } |
| ASSERT_EQ(tsfile_writer_->close(), E_OK); |
| |
| std::vector<int64_t> selected_times; |
| for (int64_t t = 0; t < 64; ++t) selected_times.push_back(t); |
| for (int64_t t = 64; t < 74; ++t) selected_times.push_back(t); |
| for (int64_t t = 128; t < 256; ++t) selected_times.push_back(t); |
| storage::TimeIn time_filter(selected_times, /*not_in=*/false); |
| |
| storage::TsFileIOReader io_reader; |
| ASSERT_EQ(io_reader.init(file_name_), E_OK); |
| for (const std::string& device : {non_aligned_device, aligned_device}) { |
| SCOPED_TRACE(device); |
| auto device_id = std::make_shared<StringArrayDeviceID>(device); |
| storage::TsFileSeriesScanIterator* ssi = nullptr; |
| common::PageArena pa; |
| pa.init(512, common::MOD_TSFILE_READER); |
| ASSERT_EQ(io_reader.alloc_ssi(device_id, "v0", ssi, pa, &time_filter), |
| E_OK); |
| ASSERT_NE(ssi, nullptr); |
| ssi->set_row_range(/*offset=*/200, /*limit=*/-1); |
| |
| std::vector<int64_t> actual_times; |
| while (true) { |
| common::TsBlock* block = nullptr; |
| int ret = |
| ssi->get_next(block, /*alloc_tsblock=*/true, &time_filter); |
| if (ret == common::E_NO_MORE_DATA) break; |
| ASSERT_EQ(ret, common::E_OK); |
| ASSERT_NE(block, nullptr); |
| |
| const uint32_t rows = block->get_row_count(); |
| const int remaining_offset = ssi->get_row_offset(); |
| const uint32_t rows_to_skip = std::min( |
| rows, static_cast<uint32_t>(std::max(remaining_offset, 0))); |
| { |
| common::ColIterator time_iter(0, block); |
| for (uint32_t row = 0; row < rows; ++row) { |
| uint32_t len = 0; |
| int64_t time = |
| *reinterpret_cast<int64_t*>(time_iter.read(&len)); |
| if (row >= rows_to_skip) actual_times.push_back(time); |
| time_iter.next(); |
| } |
| } |
| ssi->set_row_range( |
| remaining_offset - static_cast<int>(rows_to_skip), |
| /*limit=*/-1); |
| ssi->revert_tsblock(); |
| } |
| |
| EXPECT_EQ(actual_times.size(), 2u); |
| if (actual_times.size() == 2u) { |
| EXPECT_EQ(actual_times[0], 254); |
| EXPECT_EQ(actual_times[1], 255); |
| } |
| io_reader.revert_ssi(ssi); |
| } |
| } |
| |
| namespace storage { |
| // Subclass that lets the test (a) inject an error from the next-tsblock load |
| // and (b) wire a manually constructed TsBlock into the inherited iterator |
| // fields, so we can exercise the end-of-block branch of skip_rows() |
| // deterministically. The base destructor calls revert_ssi(nullptr), which |
| // short-circuits safely; we hand it a default-constructed (never-init'd) |
| // TsFileIOReader purely to satisfy the constructor. |
| class FaultySingleMeasurementColumnContext |
| : public SingleMeasurementColumnContext { |
| public: |
| using SingleMeasurementColumnContext::SingleMeasurementColumnContext; |
| int get_next_tsblock_ret_ = common::E_OK; |
| int get_next_tsblock_calls_ = 0; |
| int get_next_tsblock(bool /*alloc_mem*/) override { |
| ++get_next_tsblock_calls_; |
| return get_next_tsblock_ret_; |
| } |
| void prime_iters_for_block(common::TsBlock* tsb) { |
| tsblock_ = tsb; |
| time_iter_ = new common::ColIterator(0, tsb); |
| value_iter_ = new common::ColIterator(1, tsb); |
| } |
| }; |
| } // namespace storage |
| |
| // Regression: skip_rows() used to be a void method that called |
| // get_next_tsblock(false) for its side effects when the current block ran |
| // out. An IO/decode error from that call was silently swallowed and the |
| // outer reader treated the source as exhausted, returning fewer rows than |
| // requested with no error indication. skip_rows() now returns int and must |
| // surface hard errors (E_NO_MORE_DATA is the legitimate EOF and stays |
| // suppressed). |
| TEST_F(TsFileReaderTest, |
| SingleMeasurementSkipRowsPropagatesGetNextTsBlockError) { |
| common::TupleDesc desc; |
| desc.push_back(common::ColumnSchema("time", common::INT64, |
| common::UNCOMPRESSED, common::PLAIN)); |
| desc.push_back(common::ColumnSchema("v0", common::INT64, |
| common::UNCOMPRESSED, common::PLAIN)); |
| common::TsBlock tsb(&desc, 4); |
| ASSERT_EQ(tsb.init(), common::E_OK); |
| common::RowAppender ra(&tsb); |
| for (int i = 0; i < 2; i++) { |
| ASSERT_TRUE(ra.add_row()); |
| int64_t t = 1000 + i; |
| int64_t v = i; |
| ra.append(0, reinterpret_cast<const char*>(&t), sizeof(int64_t)); |
| ra.append(1, reinterpret_cast<const char*>(&v), sizeof(int64_t)); |
| } |
| |
| storage::TsFileIOReader io_reader_stub; |
| storage::FaultySingleMeasurementColumnContext ctx(&io_reader_stub); |
| ctx.prime_iters_for_block(&tsb); |
| |
| // Hard error: skip_rows must propagate. |
| ctx.get_next_tsblock_ret_ = common::E_INVALID_ARG; |
| EXPECT_EQ(ctx.skip_rows(2), common::E_INVALID_ARG); |
| EXPECT_EQ(ctx.get_next_tsblock_calls_, 1); |
| } |
| |
| TEST_F(TsFileReaderTest, SingleMeasurementSkipRowsSwallowsEndOfStream) { |
| common::TupleDesc desc; |
| desc.push_back(common::ColumnSchema("time", common::INT64, |
| common::UNCOMPRESSED, common::PLAIN)); |
| desc.push_back(common::ColumnSchema("v0", common::INT64, |
| common::UNCOMPRESSED, common::PLAIN)); |
| common::TsBlock tsb(&desc, 4); |
| ASSERT_EQ(tsb.init(), common::E_OK); |
| common::RowAppender ra(&tsb); |
| for (int i = 0; i < 2; i++) { |
| ASSERT_TRUE(ra.add_row()); |
| int64_t t = 1000 + i; |
| int64_t v = i; |
| ra.append(0, reinterpret_cast<const char*>(&t), sizeof(int64_t)); |
| ra.append(1, reinterpret_cast<const char*>(&v), sizeof(int64_t)); |
| } |
| |
| storage::TsFileIOReader io_reader_stub; |
| storage::FaultySingleMeasurementColumnContext ctx(&io_reader_stub); |
| ctx.prime_iters_for_block(&tsb); |
| |
| // EOF: skip_rows must squash to E_OK so the outer loop notices via |
| // available_rows() instead of bubbling the EOF up as a query failure. |
| ctx.get_next_tsblock_ret_ = common::E_NO_MORE_DATA; |
| EXPECT_EQ(ctx.skip_rows(2), common::E_OK); |
| EXPECT_EQ(ctx.get_next_tsblock_calls_, 1); |
| } |
| |
| // Regression: the multi-value aligned batch loop required the destination |
| // TsBlock to have >= BATCH (=129) rows of free capacity, otherwise it |
| // returned E_OVERFLOW immediately and the SSI surfaced that error to the |
| // caller. When tsblock_max_memory_ is small enough to land max_row_count_ |
| // below 129 (e.g. very small per-block memory in low-RAM configs) no rows |
| // could ever be decoded. The fix caps the batch by remaining capacity, |
| // matching ChunkReader's per-type batch loops. |
| TEST_F(TsFileReaderTest, MultiValueAlignedProgressesWithSmallTsBlock) { |
| const std::string device = "root.dev_multi_small_block"; |
| std::vector<MeasurementSchema> schema_vec; |
| schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED); |
| schema_vec.emplace_back("v1", INT64, PLAIN, UNCOMPRESSED); |
| { |
| std::vector<MeasurementSchema*> reg; |
| for (auto& s : schema_vec) reg.push_back(new MeasurementSchema(s)); |
| ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device, reg), |
| E_OK); |
| } |
| const int N = 200; // > BATCH (129) so the batch loop iterates twice |
| Tablet tablet(device, |
| std::make_shared<std::vector<MeasurementSchema>>(schema_vec), |
| N); |
| for (int i = 0; i < N; ++i) { |
| ASSERT_EQ(tablet.add_timestamp(i, 1000 + i), E_OK); |
| ASSERT_EQ(tablet.add_value(i, 0u, static_cast<int64_t>(i)), E_OK); |
| ASSERT_EQ(tablet.add_value(i, 1u, static_cast<int64_t>(i * 2)), E_OK); |
| } |
| ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK); |
| ASSERT_EQ(tsfile_writer_->flush(), E_OK); |
| ASSERT_EQ(tsfile_writer_->close(), E_OK); |
| |
| // Force max_row_count_ below BATCH: ~2 KB / 24 B per row → ~85 rows. |
| // Also force the multi_DECODE_TV_BATCH path by disabling parallel reads: |
| // with a thread pool the chunk-level pre-decode shortcut would otherwise |
| // run for any multi-column query (no upper column-count cutoff anymore). |
| uint32_t prev_capacity = common::g_config_value_.tsblock_max_memory_; |
| bool prev_parallel = common::g_config_value_.parallel_read_enabled_; |
| struct Guard { |
| uint32_t cap; |
| bool par; |
| ~Guard() { |
| common::g_config_value_.tsblock_max_memory_ = cap; |
| common::g_config_value_.parallel_read_enabled_ = par; |
| } |
| } guard{prev_capacity, prev_parallel}; |
| common::g_config_value_.tsblock_max_memory_ = 2048; |
| common::g_config_value_.parallel_read_enabled_ = false; |
| |
| storage::TsFileIOReader io_reader; |
| ASSERT_EQ(io_reader.init(file_name_), E_OK); |
| |
| auto device_id = std::make_shared<StringArrayDeviceID>(device); |
| std::vector<std::string> measurements = {"v0", "v1"}; |
| storage::TsFileSeriesScanIterator* ssi = nullptr; |
| common::PageArena pa; |
| pa.init(512, common::MOD_TSFILE_READER); |
| ASSERT_EQ(io_reader.alloc_multi_ssi(device_id, measurements, ssi, pa, |
| /*time_filter=*/nullptr), |
| E_OK); |
| ASSERT_NE(ssi, nullptr); |
| |
| int collected = 0; |
| while (true) { |
| common::TsBlock* block = nullptr; |
| int ret = ssi->get_next(block, /*alloc_tsblock=*/true); |
| if (ret == common::E_NO_MORE_DATA) break; |
| ASSERT_EQ(ret, common::E_OK); |
| ASSERT_NE(block, nullptr); |
| ASSERT_GT(block->get_max_row_count(), 0u); |
| ASSERT_LT(block->get_max_row_count(), 129u); |
| collected += static_cast<int>(block->get_row_count()); |
| ssi->revert_tsblock(); |
| } |
| EXPECT_EQ(collected, N); |
| |
| io_reader.revert_ssi(ssi); |
| } |
| |
| // Regression: when a whole batch is filtered out, multi_DECODE_TV_BATCH skips |
| // the non-null value bytes for each column. The old code ignored the skip |
| // return code and the `skipped` count, so a short/truncated page could leave |
| // the decoder mid-value; subsequent batches would then read garbage bytes as |
| // values. This test exercises an intact page: the filter rejects rows |
| // 0..127 (one full batch worth), then the rows after must come back with |
| // their *correct* values — proving the decoder advanced exactly nonnull_count |
| // values, not some smaller number that would shift the value alignment. |
| TEST_F(TsFileReaderTest, MultiValueAlignedSkipsBatchPreservesValueAlignment) { |
| const std::string device = "root.dev_multi_skip_align"; |
| std::vector<MeasurementSchema> schema_vec; |
| schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED); |
| schema_vec.emplace_back("v1", INT64, PLAIN, UNCOMPRESSED); |
| { |
| std::vector<MeasurementSchema*> reg; |
| for (auto& s : schema_vec) reg.push_back(new MeasurementSchema(s)); |
| ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device, reg), |
| E_OK); |
| } |
| // Two batches' worth of rows so the filter skips the first batch entirely |
| // and decodes the second. |
| const int N = 200; |
| Tablet tablet(device, |
| std::make_shared<std::vector<MeasurementSchema>>(schema_vec), |
| N); |
| for (int i = 0; i < N; ++i) { |
| // Distinctive value pattern: i and 1000000 + i. If skip |
| // mis-advances the decoder by even one value, the v0/v1 read after |
| // the skip will land on the wrong row's bytes. |
| ASSERT_EQ(tablet.add_timestamp(i, static_cast<int64_t>(i)), E_OK); |
| ASSERT_EQ(tablet.add_value(i, 0u, static_cast<int64_t>(i)), E_OK); |
| ASSERT_EQ(tablet.add_value(i, 1u, static_cast<int64_t>(1000000 + i)), |
| E_OK); |
| } |
| ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK); |
| ASSERT_EQ(tsfile_writer_->flush(), E_OK); |
| ASSERT_EQ(tsfile_writer_->close(), E_OK); |
| |
| bool prev_parallel = common::g_config_value_.parallel_read_enabled_; |
| struct Guard { |
| bool par; |
| ~Guard() { common::g_config_value_.parallel_read_enabled_ = par; } |
| } guard{prev_parallel}; |
| // Force the multi_DECODE_TV_BATCH path (the chunk-level shortcut would |
| // bypass the skip branch we want to exercise). |
| common::g_config_value_.parallel_read_enabled_ = false; |
| |
| storage::TsFileIOReader io_reader; |
| ASSERT_EQ(io_reader.init(file_name_), E_OK); |
| |
| auto device_id = std::make_shared<StringArrayDeviceID>(device); |
| std::vector<std::string> measurements = {"v0", "v1"}; |
| storage::TsFileSeriesScanIterator* ssi = nullptr; |
| common::PageArena pa; |
| pa.init(512, common::MOD_TSFILE_READER); |
| |
| // TimeIn filter selecting only rows 130..139 — entirely past the first |
| // 129-row batch, so the first batch hits the pass_count==0 skip branch |
| // for both value columns. |
| std::vector<int64_t> want; |
| for (int i = 130; i < 140; ++i) want.push_back(i); |
| storage::TimeIn time_filter(want, /*not_in=*/false); |
| |
| ASSERT_EQ(io_reader.alloc_multi_ssi(device_id, measurements, ssi, pa, |
| &time_filter), |
| E_OK); |
| ASSERT_NE(ssi, nullptr); |
| |
| std::vector<std::pair<int64_t, int64_t>> got; |
| while (true) { |
| common::TsBlock* block = nullptr; |
| int ret = ssi->get_next(block, /*alloc_tsblock=*/true, &time_filter); |
| if (ret == common::E_NO_MORE_DATA) break; |
| ASSERT_EQ(ret, common::E_OK); |
| ASSERT_NE(block, nullptr); |
| // Scope the ColIterators so they are destroyed *before* |
| // revert_tsblock() frees the block. ~ColIterator() writes back to its |
| // vector (reset_offset()), so reverting while an iterator is still in |
| // scope would touch freed memory. |
| { |
| // Columns: time, v0, v1. |
| common::ColIterator t_iter(0, block); |
| common::ColIterator v0_iter(1, block); |
| common::ColIterator v1_iter(2, block); |
| const uint32_t rows = block->get_row_count(); |
| for (uint32_t r = 0; r < rows; ++r) { |
| uint32_t len = 0; |
| int64_t t = *reinterpret_cast<int64_t*>(t_iter.read(&len)); |
| int64_t v0 = *reinterpret_cast<int64_t*>(v0_iter.read(&len)); |
| int64_t v1 = *reinterpret_cast<int64_t*>(v1_iter.read(&len)); |
| got.push_back({t, v0}); |
| // The decoder must have advanced exactly nonnull_count values |
| // when it skipped batch #1. If it under-advanced (the latent |
| // bug), v1 would land on the wrong row's bytes here. |
| EXPECT_EQ(v1, 1000000 + t); |
| EXPECT_EQ(v0, t); |
| t_iter.next(); |
| v0_iter.next(); |
| v1_iter.next(); |
| } |
| } |
| ssi->revert_tsblock(); |
| } |
| |
| ASSERT_EQ(got.size(), want.size()); |
| for (size_t i = 0; i < got.size(); ++i) { |
| EXPECT_EQ(got[i].first, want[i]); |
| EXPECT_EQ(got[i].second, want[i]); |
| } |
| |
| io_reader.revert_ssi(ssi); |
| } |
| |
| // Coverage: an aligned read with > 6 value columns now takes the chunk-level |
| // parallel decode path (decode_all_planned_pages) exactly like the 2..6 column |
| // case — the old "<= 6 columns" dispatch cutoff that sent wide chunks down the |
| // per-page serial path is gone. With libtsfile_init() having built the global |
| // pool and parallel_read_enabled_ on by default, an 8-column query exercises |
| // that path end-to-end; each column carries a disjoint value range so any |
| // cross-column misalignment in the wide chunk-level decode would be caught. |
| TEST_F(TsFileReaderTest, MultiValueAlignedWideChunkParallelDecode) { |
| const std::string device = "root.dev_multi_wide"; |
| const uint32_t kCols = 8; // > 6: previously bypassed the chunk-level path |
| std::vector<MeasurementSchema> schema_vec; |
| for (uint32_t c = 0; c < kCols; ++c) { |
| schema_vec.emplace_back("v" + std::to_string(c), INT64, PLAIN, |
| UNCOMPRESSED); |
| } |
| { |
| std::vector<MeasurementSchema*> reg; |
| for (auto& s : schema_vec) reg.push_back(new MeasurementSchema(s)); |
| ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device, reg), |
| E_OK); |
| } |
| const int N = 200; // > BATCH (129) so the decode loop iterates more once |
| Tablet tablet(device, |
| std::make_shared<std::vector<MeasurementSchema>>(schema_vec), |
| N); |
| // Row i, column c carries c * 1000000 + i so each column's values occupy a |
| // disjoint range; a wide-chunk decode that crossed column boundaries would |
| // surface as a value landing in the wrong column's range. |
| for (int i = 0; i < N; ++i) { |
| ASSERT_EQ(tablet.add_timestamp(i, 1000 + i), E_OK); |
| for (uint32_t c = 0; c < kCols; ++c) { |
| ASSERT_EQ( |
| tablet.add_value(i, c, static_cast<int64_t>(c * 1000000 + i)), |
| E_OK); |
| } |
| } |
| ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK); |
| ASSERT_EQ(tsfile_writer_->flush(), E_OK); |
| ASSERT_EQ(tsfile_writer_->close(), E_OK); |
| |
| // parallel_read_enabled_ defaults to true and SetUp() ran libtsfile_init(), |
| // so the SSI hands the AlignedChunkReader the global pool; with 8 value |
| // columns (> 1) the reader takes the chunk-level decode path. |
| ASSERT_TRUE(common::g_config_value_.parallel_read_enabled_); |
| |
| storage::TsFileIOReader io_reader; |
| ASSERT_EQ(io_reader.init(file_name_), E_OK); |
| |
| auto device_id = std::make_shared<StringArrayDeviceID>(device); |
| std::vector<std::string> measurements; |
| for (uint32_t c = 0; c < kCols; ++c) |
| measurements.push_back("v" + std::to_string(c)); |
| storage::TsFileSeriesScanIterator* ssi = nullptr; |
| common::PageArena pa; |
| pa.init(512, common::MOD_TSFILE_READER); |
| ASSERT_EQ(io_reader.alloc_multi_ssi(device_id, measurements, ssi, pa, |
| /*time_filter=*/nullptr), |
| E_OK); |
| ASSERT_NE(ssi, nullptr); |
| |
| int collected = 0; |
| while (true) { |
| common::TsBlock* block = nullptr; |
| int ret = ssi->get_next(block, /*alloc_tsblock=*/true); |
| if (ret == common::E_NO_MORE_DATA) break; |
| ASSERT_EQ(ret, common::E_OK); |
| ASSERT_NE(block, nullptr); |
| const uint32_t rows = block->get_row_count(); |
| |
| // Scope all ColIterators so they are destroyed *before* |
| // revert_tsblock() frees the block — ~ColIterator() writes back to its |
| // vector (reset_offset()), which would be use-after-free otherwise. |
| { |
| common::ColIterator t_iter(0, block); |
| std::vector<int64_t> times; |
| times.reserve(rows); |
| for (uint32_t r = 0; r < rows; ++r) { |
| uint32_t len = 0; |
| times.push_back(*reinterpret_cast<int64_t*>(t_iter.read(&len))); |
| t_iter.next(); |
| } |
| // One independent iterator per value column so we never rely on |
| // vector<ColIterator> being movable. |
| for (uint32_t c = 0; c < kCols; ++c) { |
| common::ColIterator it(c + 1, block); |
| for (uint32_t r = 0; r < rows; ++r) { |
| uint32_t len = 0; |
| int64_t v = *reinterpret_cast<int64_t*>(it.read(&len)); |
| int64_t i = times[r] - 1000; // timestamp == 1000 + i |
| EXPECT_EQ(v, static_cast<int64_t>(c) * 1000000 + i); |
| it.next(); |
| } |
| } |
| } |
| collected += static_cast<int>(rows); |
| ssi->revert_tsblock(); |
| } |
| EXPECT_EQ(collected, N); |
| |
| io_reader.revert_ssi(ssi); |
| } |
| |
| // Regression: AlignedTimeseriesIndex::get_data_type() returns the time column |
| // type (VECTOR), which the schema accessor used to surface verbatim — every |
| // aligned column came back as VECTOR instead of its real INT32/FLOAT/etc. |
| // type. get_timeseries_schema() now unwraps AlignedTimeseriesIndex to read |
| // value_ts_idx_->get_data_type() like the develop branch did. |
| TEST_F(TsFileReaderTest, AlignedSchemaReportsValueDataType) { |
| const std::string device = "root.dev_aligned_schema"; |
| std::vector<MeasurementSchema> schema_vec; |
| schema_vec.emplace_back("v_i32", INT32, PLAIN, UNCOMPRESSED); |
| schema_vec.emplace_back("v_dbl", DOUBLE, PLAIN, UNCOMPRESSED); |
| { |
| std::vector<MeasurementSchema*> reg; |
| for (auto& s : schema_vec) reg.push_back(new MeasurementSchema(s)); |
| ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device, reg), |
| E_OK); |
| } |
| const int N = 8; |
| Tablet tablet(device, |
| std::make_shared<std::vector<MeasurementSchema>>(schema_vec), |
| N); |
| for (int i = 0; i < N; ++i) { |
| ASSERT_EQ(tablet.add_timestamp(i, 1000 + i), E_OK); |
| ASSERT_EQ(tablet.add_value(i, 0u, static_cast<int32_t>(i)), E_OK); |
| ASSERT_EQ(tablet.add_value(i, 1u, static_cast<double>(i) * 0.5), E_OK); |
| } |
| ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK); |
| ASSERT_EQ(tsfile_writer_->flush(), E_OK); |
| ASSERT_EQ(tsfile_writer_->close(), E_OK); |
| |
| storage::TsFileReader reader; |
| ASSERT_EQ(reader.open(file_name_), E_OK); |
| |
| auto device_id = std::make_shared<StringArrayDeviceID>(device); |
| std::vector<MeasurementSchema> schemas; |
| ASSERT_EQ(reader.get_timeseries_schema(device_id, schemas), E_OK); |
| ASSERT_EQ(schemas.size(), 2u); |
| |
| // Match by name — IO reader iteration order isn't part of the contract. |
| common::TSDataType i32_type = common::INVALID_DATATYPE; |
| common::TSDataType dbl_type = common::INVALID_DATATYPE; |
| for (const auto& s : schemas) { |
| if (s.measurement_name_ == "v_i32") i32_type = s.data_type_; |
| if (s.measurement_name_ == "v_dbl") dbl_type = s.data_type_; |
| } |
| EXPECT_EQ(i32_type, INT32); |
| EXPECT_EQ(dbl_type, DOUBLE); |
| reader.close(); |
| } |
| |
| namespace storage { |
| class TsFileReaderMetaArenaTest { |
| public: |
| static int64_t arena_used(const storage::TsFileReader& r) { |
| return r.tsfile_reader_meta_pa_.get_total_used_bytes(); |
| } |
| }; |
| } // namespace storage |
| |
| // Regression: tsfile_reader_meta_pa_ used to be re-initialised at the start |
| // of each get_timeseries_metadata() call. When that reset was removed, |
| // every call accumulated another copy of the per-device meta into the same |
| // arena, so a long-lived reader that polled metadata kept growing memory |
| // without bound. Re-init now happens at the top of both overloads; verify |
| // arena usage stays flat across repeated calls instead of growing linearly. |
| TEST_F(TsFileReaderTest, RepeatedGetTimeseriesMetadataDoesNotLeakArena) { |
| const std::string device = "root.dev_arena_growth"; |
| { |
| std::vector<MeasurementSchema*> reg; |
| reg.push_back(new MeasurementSchema("v0", INT64, PLAIN, UNCOMPRESSED)); |
| ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device, reg), |
| E_OK); |
| } |
| TsRecord r(1000, device); |
| r.points_.emplace_back("v0", static_cast<int64_t>(0)); |
| ASSERT_EQ(tsfile_writer_->write_record_aligned(r), E_OK); |
| ASSERT_EQ(tsfile_writer_->flush(), E_OK); |
| ASSERT_EQ(tsfile_writer_->close(), E_OK); |
| |
| storage::TsFileReader reader; |
| ASSERT_EQ(reader.open(file_name_), E_OK); |
| std::vector<std::shared_ptr<IDeviceID>> ids = { |
| std::make_shared<StringArrayDeviceID>(device)}; |
| |
| // Prime the arena and capture the steady-state size. |
| (void)reader.get_timeseries_metadata(ids); |
| const int64_t after_one = |
| storage::TsFileReaderMetaArenaTest::arena_used(reader); |
| ASSERT_GT(after_one, 0); |
| |
| for (int i = 0; i < 10; ++i) { |
| (void)reader.get_timeseries_metadata(ids); |
| } |
| const int64_t after_eleven = |
| storage::TsFileReaderMetaArenaTest::arena_used(reader); |
| // Without the fix, after_eleven ≈ 11 × after_one. With the fix it |
| // should equal after_one (arena reset before each call). Allow a small |
| // slack for arena page rounding, but reject anything close to 2× growth. |
| EXPECT_LT(after_eleven, after_one * 2) |
| << "arena grew from " << after_one << " to " << after_eleven |
| << " across 11 calls — reset on entry is missing"; |
| reader.close(); |
| } |
