diff --git a/cpp/src/common/schema.h b/cpp/src/common/schema.h index ed59b27b1..f9344ef4e 100644 --- a/cpp/src/common/schema.h +++ b/cpp/src/common/schema.h @@ -28,6 +28,7 @@ #include #include #include +#include #include "common/db_common.h" #include "writer/time_chunk_writer.h" @@ -176,6 +177,30 @@ struct MeasurementSchemaGroup { TimeChunkWriter* time_chunk_writer_ = nullptr; int64_t last_time_ = INT64_MIN; + // Per-device schema-check cache (do_check_schema / + // do_check_schema_aligned): the resolved chunk writers + data types for + // the last fully-resolved measurement-name sequence, so repeated writes + // with an unchanged schema skip the per-column measurement_schema_map_ + // string lookup (#885). Entries are the SAME non-owning pointers the + // uncached path returns, so the written file is byte-identical. + // + // Guarded by the measurement NAME SEQUENCE, not just the column count: + // entries are reused by position, so the same count with a different + // name order (or one column swapped) would silently write values into + // the wrong column AND with the wrong data type. + // + // The plain and aligned paths keep separate caches: sharing one flag + // would let whichever path ran first lock the other out permanently. + std::vector cached_chunk_writers_; + std::vector cached_data_types_; + std::vector cached_measurement_names_; + bool schema_check_cached_ = false; + + std::vector cached_value_chunk_writers_; + std::vector cached_aligned_data_types_; + std::vector cached_aligned_measurement_names_; + bool schema_check_aligned_cached_ = false; + ~MeasurementSchemaGroup() { if (time_chunk_writer_ != nullptr) { delete time_chunk_writer_; diff --git a/cpp/src/writer/tsfile_writer.cc b/cpp/src/writer/tsfile_writer.cc index 564d1f203..bb32a3a0c 100644 --- a/cpp/src/writer/tsfile_writer.cc +++ b/cpp/src/writer/tsfile_writer.cc @@ -88,7 +88,31 @@ TsFileWriter::TsFileWriter() TsFileWriter::~TsFileWriter() { destroy(); } +void TsFileWriter::clear_schema_check_cache() { + plain_schema_check_cache_.clear(); + aligned_schema_check_cache_.clear(); +} + +bool TsFileWriter::plain_schema_cache_matches( + const std::shared_ptr& device_id, + const std::vector& measurement_names) const { + return plain_schema_check_cache_.valid_ && + plain_schema_check_cache_.device_id_ != nullptr && + *plain_schema_check_cache_.device_id_ == *device_id && + plain_schema_check_cache_.measurement_names_ == measurement_names; +} + +bool TsFileWriter::aligned_schema_cache_matches( + const std::shared_ptr& device_id, + const std::vector& measurement_names) const { + return aligned_schema_check_cache_.valid_ && + aligned_schema_check_cache_.device_id_ != nullptr && + *aligned_schema_check_cache_.device_id_ == *device_id && + aligned_schema_check_cache_.measurement_names_ == measurement_names; +} + void TsFileWriter::destroy() { + clear_schema_check_cache(); if (write_file_created_ && write_file_ != nullptr) { delete write_file_; write_file_ = nullptr; @@ -141,6 +165,7 @@ int TsFileWriter::init(WriteFile* write_file) { // the new file and produce headerless output. enforce_recovered_last_time_order_ = false; unrecoverable_ = false; + clear_schema_check_cache(); start_file_done_ = false; record_count_since_last_flush_ = 0; io_writer_ = new TsFileIOWriter(); @@ -165,6 +190,7 @@ int TsFileWriter::init(RestorableTsFileIOWriter* rw) { // Clear any unrecoverable_ latched from a previous lifecycle so the // re-init isn't immediately poisoned. unrecoverable_ = false; + clear_schema_check_cache(); // Reject new writes whose timestamps fall back into the recovered range. enforce_recovered_last_time_order_ = true; io_writer_ = rw; @@ -213,13 +239,22 @@ int TsFileWriter::init(RestorableTsFileIOWriter* rw) { if (mname.empty()) { continue; } - if (group->measurement_schema_map_.find(mname) != - group->measurement_schema_map_.end()) { - continue; + auto schema_it = group->measurement_schema_map_.find(mname); + if (schema_it == group->measurement_schema_map_.end()) { + MeasurementSchema* ms = + new MeasurementSchema(mname, cm->data_type_, cm->encoding_, + cm->compression_type_); + group->measurement_schema_map_.insert( + std::make_pair(mname, ms)); + } else { + // A series may have different codecs in different chunks. + // Appends must use the latest chunk's codec, not the first + // recovered chunk's stale settings. + MeasurementSchema* ms = schema_it->second; + ms->data_type_ = cm->data_type_; + ms->encoding_ = cm->encoding_; + ms->compression_type_ = cm->compression_type_; } - MeasurementSchema* ms = new MeasurementSchema( - mname, cm->data_type_, cm->encoding_, cm->compression_type_); - group->measurement_schema_map_.insert(std::make_pair(mname, ms)); } } @@ -330,12 +365,14 @@ int TsFileWriter::register_timeseries(const std::string& device_path, if (UNLIKELY(!ins_res.second)) { return E_ALREADY_EXIST; } + clear_schema_check_cache(); } else { MeasurementSchemaGroup* ms_group = new MeasurementSchemaGroup; ms_group->is_aligned_ = is_aligned; ms_group->measurement_schema_map_.insert(std::make_pair( measurement_schema->measurement_name_, measurement_schema)); schemas_.insert(std::make_pair(device_id, ms_group)); + clear_schema_check_cache(); } return E_OK; } @@ -466,40 +503,58 @@ int TsFileWriter::do_check_schema( IS_NULL(device_schema = dev_it->second)) { return E_DEVICE_NOT_EXIST; } + MeasurementSchemaMap& msm = device_schema->measurement_schema_map_; - uint32_t measurement_count = measurement_names.get_count(); - // chunk_writers.reserve(measurement_count); - for (uint32_t i = 0; i < measurement_count; i++) { - auto ms_iter = msm.find(measurement_names.next()); + static thread_local std::vector names; + names.clear(); + names.reserve(measurement_names.get_count()); + for (uint32_t i = 0; i < measurement_names.get_count(); ++i) { + names.push_back(measurement_names.next()); + } + if (plain_schema_cache_matches(device_id, names)) { + for (size_t i = 0; i < plain_schema_check_cache_.chunk_writers_.size(); + ++i) { + chunk_writers.push_back( + plain_schema_check_cache_.chunk_writers_[i]); + data_types.push_back(plain_schema_check_cache_.data_types_[i]); + } + return E_OK; + } + + bool all_resolved = true; + for (uint32_t i = 0; i < names.size(); ++i) { + auto ms_iter = msm.find(names[i]); if (UNLIKELY(ms_iter == msm.end())) { + all_resolved = false; chunk_writers.push_back(NULL); data_types.push_back(common::NULL_TYPE); - } else { - // In Java we will check data_type. But in C++, no check here. - // Because checks are performed at the chunk layer and page layer - MeasurementSchema* ms = ms_iter->second; - if (IS_NULL(ms->chunk_writer_)) { - ms->chunk_writer_ = new ChunkWriter; - ret = ms->chunk_writer_->init(ms->measurement_name_, - ms->data_type_, ms->encoding_, - ms->compression_type_); - if (IS_SUCC(ret)) { - chunk_writers.push_back(ms->chunk_writer_); - } else { - for (size_t chunk_writer_idx = 0; - chunk_writer_idx < chunk_writers.size(); - chunk_writer_idx++) { - if (!chunk_writers[chunk_writer_idx]) { - delete chunk_writers[chunk_writer_idx]; - } - } - ret = common::E_INVALID_ARG; - return ret; - } - } else { - chunk_writers.push_back(ms->chunk_writer_); + continue; + } + + MeasurementSchema* ms = ms_iter->second; + if (IS_NULL(ms->chunk_writer_)) { + ms->chunk_writer_ = new ChunkWriter; + ret = ms->chunk_writer_->init(ms->measurement_name_, ms->data_type_, + ms->encoding_, ms->compression_type_); + if (IS_FAIL(ret)) { + ret = common::E_INVALID_ARG; + return ret; } - data_types.push_back(ms->data_type_); + } + chunk_writers.push_back(ms->chunk_writer_); + data_types.push_back(ms->data_type_); + } + + if (IS_SUCC(ret) && all_resolved) { + plain_schema_check_cache_.valid_ = true; + plain_schema_check_cache_.device_id_ = device_id; + plain_schema_check_cache_.measurement_names_ = names; + plain_schema_check_cache_.chunk_writers_.clear(); + plain_schema_check_cache_.data_types_.clear(); + for (uint32_t i = 0; i < names.size(); ++i) { + plain_schema_check_cache_.chunk_writers_.push_back( + chunk_writers[i]); + plain_schema_check_cache_.data_types_.push_back(data_types[i]); } } return ret; @@ -526,40 +581,59 @@ int TsFileWriter::do_check_schema_aligned( g_config_value_.time_compress_type_); } time_chunk_writer = device_schema->time_chunk_writer_; + MeasurementSchemaMap& msm = device_schema->measurement_schema_map_; - uint32_t measurement_count = measurement_names.get_count(); - for (uint32_t i = 0; i < measurement_count; i++) { - auto ms_iter = msm.find(measurement_names.next()); + static thread_local std::vector names; + names.clear(); + names.reserve(measurement_names.get_count()); + for (uint32_t i = 0; i < measurement_names.get_count(); ++i) { + names.push_back(measurement_names.next()); + } + if (aligned_schema_cache_matches(device_id, names)) { + for (size_t i = 0; + i < aligned_schema_check_cache_.value_chunk_writers_.size(); ++i) { + value_chunk_writers.push_back( + aligned_schema_check_cache_.value_chunk_writers_[i]); + data_types.push_back(aligned_schema_check_cache_.data_types_[i]); + } + return E_OK; + } + + bool all_resolved = true; + for (uint32_t i = 0; i < names.size(); ++i) { + auto ms_iter = msm.find(names[i]); if (UNLIKELY(ms_iter == msm.end())) { + all_resolved = false; value_chunk_writers.push_back(NULL); data_types.push_back(common::NULL_TYPE); - } else { - // Here we may check data_type against ms_iter. But in Java - // libtsfile, no check here. - MeasurementSchema* ms = ms_iter->second; - if (IS_NULL(ms->value_chunk_writer_)) { - ms->value_chunk_writer_ = new ValueChunkWriter; - ret = ms->value_chunk_writer_->init( - ms->measurement_name_, ms->data_type_, ms->encoding_, - ms->compression_type_); - if (IS_SUCC(ret)) { - value_chunk_writers.push_back(ms->value_chunk_writer_); - } else { - value_chunk_writers.push_back(NULL); - for (size_t chunk_writer_idx = 0; - chunk_writer_idx < value_chunk_writers.size(); - chunk_writer_idx++) { - if (!value_chunk_writers[chunk_writer_idx]) { - delete value_chunk_writers[chunk_writer_idx]; - } - } - ret = common::E_INVALID_ARG; - return ret; - } - } else { - value_chunk_writers.push_back(ms->value_chunk_writer_); + continue; + } + + MeasurementSchema* ms = ms_iter->second; + if (IS_NULL(ms->value_chunk_writer_)) { + ms->value_chunk_writer_ = new ValueChunkWriter; + ret = ms->value_chunk_writer_->init(ms->measurement_name_, + ms->data_type_, ms->encoding_, + ms->compression_type_); + if (IS_FAIL(ret)) { + ret = common::E_INVALID_ARG; + return ret; } - data_types.push_back(ms->data_type_); + } + value_chunk_writers.push_back(ms->value_chunk_writer_); + data_types.push_back(ms->data_type_); + } + + if (IS_SUCC(ret) && all_resolved) { + aligned_schema_check_cache_.valid_ = true; + aligned_schema_check_cache_.device_id_ = device_id; + aligned_schema_check_cache_.measurement_names_ = names; + aligned_schema_check_cache_.value_chunk_writers_.clear(); + aligned_schema_check_cache_.data_types_.clear(); + for (uint32_t i = 0; i < names.size(); ++i) { + aligned_schema_check_cache_.value_chunk_writers_.push_back( + value_chunk_writers[i]); + aligned_schema_check_cache_.data_types_.push_back(data_types[i]); } } return ret; diff --git a/cpp/src/writer/tsfile_writer.h b/cpp/src/writer/tsfile_writer.h index 55e9e7f3a..d76fe4929 100644 --- a/cpp/src/writer/tsfile_writer.h +++ b/cpp/src/writer/tsfile_writer.h @@ -183,6 +183,51 @@ class TsFileWriter { std::vector, int>> split_tablet_by_device(const Tablet& tablet); + struct PlainSchemaCheckCache { + bool valid_ = false; + std::shared_ptr device_id_; + std::vector measurement_names_; + std::vector chunk_writers_; + std::vector data_types_; + + void clear() { + valid_ = false; + device_id_.reset(); + measurement_names_.clear(); + chunk_writers_.clear(); + data_types_.clear(); + } + }; + + struct AlignedSchemaCheckCache { + bool valid_ = false; + std::shared_ptr device_id_; + std::vector measurement_names_; + std::vector value_chunk_writers_; + std::vector data_types_; + + void clear() { + valid_ = false; + device_id_.reset(); + measurement_names_.clear(); + value_chunk_writers_.clear(); + data_types_.clear(); + } + }; + + void clear_schema_check_cache(); + bool plain_schema_cache_matches( + const std::shared_ptr& device_id, + const std::vector& measurement_names) const; + bool aligned_schema_cache_matches( + const std::shared_ptr& device_id, + const std::vector& measurement_names) const; + + // Each cache is a one-entry MRU cache. Entries contain non-owning writer + // pointers; MeasurementSchema remains responsible for their lifetime. + PlainSchemaCheckCache plain_schema_check_cache_; + AlignedSchemaCheckCache aligned_schema_check_cache_; + private: storage::WriteFile* write_file_; storage::TsFileIOWriter* io_writer_; diff --git a/cpp/test/writer/tsfile_writer_schema_cache_test.cc b/cpp/test/writer/tsfile_writer_schema_cache_test.cc new file mode 100644 index 000000000..638dc5ccf --- /dev/null +++ b/cpp/test/writer/tsfile_writer_schema_cache_test.cc @@ -0,0 +1,427 @@ +/* + * 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 + +#include "writer/tsfile_writer.h" + +#ifdef _WIN32 +#include +#else +#include +#endif + +#include +#include +#include +#include +#include +#include +#include + +#include "common/path.h" +#include "common/record.h" +#include "common/schema.h" +#include "common/tablet.h" +#include "reader/qds_without_timegenerator.h" +#include "reader/tsfile_reader.h" + +using namespace common; +using namespace storage; + +namespace { + +class SchemaCheckCacheTest : public ::testing::Test { + protected: + void SetUp() override { + ASSERT_EQ(libtsfile_init(), E_OK); + writer_ = new TsFileWriter(); + file_name_ = "tsfile_schema_cache_test_" + unique_suffix() + ".tsfile"; + remove(file_name_.c_str()); + int flags = O_WRONLY | O_CREAT | O_TRUNC; +#ifdef _WIN32 + flags |= O_BINARY; +#endif + ASSERT_EQ(writer_->open(file_name_, flags, 0666), E_OK); + } + + void TearDown() override { + delete writer_; + ASSERT_EQ(remove(file_name_.c_str()), 0); + libtsfile_destroy(); + } + + static std::string unique_suffix() { + static std::atomic counter{0}; +#ifdef _WIN32 + const auto process_id = static_cast(_getpid()); +#else + const auto process_id = static_cast(getpid()); +#endif + return std::to_string(process_id) + "_" + + std::to_string(counter.fetch_add(1)); + } + + static MeasurementSchema schema(const std::string& name) { + return MeasurementSchema(name, TSDataType::INT32, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED); + } + + static std::vector schemas( + const std::vector& names) { + std::vector result; + for (const auto& name : names) { + result.push_back(schema(name)); + } + return result; + } + + static std::string field_to_string(Field* field) { + if (field->type_ == TEXT || field->type_ == STRING || + field->type_ == BLOB) { + return std::string(field->value_.sval_); + } + std::stringstream stream; + switch (field->type_) { + case BOOLEAN: + stream << (field->value_.bval_ ? "true" : "false"); + break; + case INT32: + case DATE: + stream << field->value_.ival_; + break; + case INT64: + case TIMESTAMP: + stream << field->value_.lval_; + break; + case FLOAT: + stream << field->value_.fval_; + break; + case DOUBLE: + stream << field->value_.dval_; + break; + case NULL_TYPE: + stream << "NULL"; + break; + default: + ADD_FAILURE() << "Unexpected field type: " << field->type_; + } + return stream.str(); + } + + std::vector> read_rows( + const std::vector& paths) const { + TsFileReader reader; + EXPECT_EQ(reader.open(file_name_), E_OK); + QueryExpression* expression = QueryExpression::create(paths, nullptr); + ResultSet* result = nullptr; + EXPECT_EQ(reader.query(expression, result), E_OK); + auto* query = static_cast(result); + + std::vector> rows; + bool has_next = false; + while (IS_SUCC(query->next(has_next)) && has_next) { + RowRecord* record = query->get_row_record(); + std::vector row; + row.push_back(std::to_string(record->get_timestamp())); + for (size_t i = 1; i < record->get_fields()->size(); ++i) { + row.push_back(field_to_string(record->get_field(i))); + } + rows.push_back(row); + } + reader.destroy_query_data_set(query); + reader.close(); + return rows; + } + + static Path path(const std::string& device, + const std::string& measurement) { + std::string device_copy = device; + std::string measurement_copy = measurement; + return Path(device_copy, measurement_copy); + } + + TsFileWriter* writer_ = nullptr; + std::string file_name_; +}; + +TEST_F(SchemaCheckCacheTest, RepeatedSameSchemaSurvivesFlush) { + const std::string device = "root.cache_hit"; + const std::vector names = {"s0", "s1", "s2"}; + for (const auto& name : names) { + ASSERT_EQ(writer_->register_timeseries(device, schema(name)), E_OK); + } + + for (int row = 0; row < 5; ++row) { + auto tablet_schema = schemas(names); + Tablet tablet( + device, + std::make_shared>(tablet_schema), 1); + ASSERT_EQ(tablet.add_timestamp(0, 1000 + row), E_OK); + for (uint32_t column = 0; column < names.size(); ++column) { + ASSERT_EQ(tablet.add_value( + 0, column, static_cast(row * 100 + column)), + E_OK); + } + ASSERT_EQ(writer_->write_tablet(tablet), E_OK); + if (row == 2) { + ASSERT_EQ(writer_->flush(), E_OK); + } + } + ASSERT_EQ(writer_->flush(), E_OK); + ASSERT_EQ(writer_->close(), E_OK); + + auto rows = + read_rows({path(device, "s0"), path(device, "s1"), path(device, "s2")}); + ASSERT_EQ(rows.size(), 5u); + for (int row = 0; row < 5; ++row) { + EXPECT_EQ(rows[row][0], std::to_string(1000 + row)); + for (int column = 0; column < 3; ++column) { + EXPECT_EQ(rows[row][column + 1], + std::to_string(row * 100 + column)); + } + } +} + +TEST_F(SchemaCheckCacheTest, SameCountDifferentNameOrderDoesNotCrossWire) { + const std::string device = "root.cache_reorder"; + for (const auto& name : {"s0", "s1", "s2"}) { + ASSERT_EQ(writer_->register_timeseries(device, schema(name)), E_OK); + } + + { + auto tablet_schema = schemas({"s0", "s1"}); + Tablet tablet( + device, + std::make_shared>(tablet_schema), 1); + ASSERT_EQ(tablet.add_timestamp(0, 0), E_OK); + ASSERT_EQ(tablet.add_value(0, 0, static_cast(10)), E_OK); + ASSERT_EQ(tablet.add_value(0, 1, static_cast(11)), E_OK); + ASSERT_EQ(writer_->write_tablet(tablet), E_OK); + } + { + auto tablet_schema = schemas({"s1", "s0"}); + Tablet tablet( + device, + std::make_shared>(tablet_schema), 1); + ASSERT_EQ(tablet.add_timestamp(0, 1), E_OK); + ASSERT_EQ(tablet.add_value(0, 0, static_cast(21)), E_OK); + ASSERT_EQ(tablet.add_value(0, 1, static_cast(20)), E_OK); + ASSERT_EQ(writer_->write_tablet(tablet), E_OK); + } + { + auto tablet_schema = schemas({"s0", "s2"}); + Tablet tablet( + device, + std::make_shared>(tablet_schema), 1); + ASSERT_EQ(tablet.add_timestamp(0, 2), E_OK); + ASSERT_EQ(tablet.add_value(0, 0, static_cast(30)), E_OK); + ASSERT_EQ(tablet.add_value(0, 1, static_cast(32)), E_OK); + ASSERT_EQ(writer_->write_tablet(tablet), E_OK); + } + ASSERT_EQ(writer_->flush(), E_OK); + ASSERT_EQ(writer_->close(), E_OK); + + auto rows = + read_rows({path(device, "s0"), path(device, "s1"), path(device, "s2")}); + ASSERT_EQ(rows.size(), 3u); + EXPECT_EQ(rows[0][1], "10"); + EXPECT_EQ(rows[0][2], "11"); + EXPECT_EQ(rows[1][1], "20"); + EXPECT_EQ(rows[1][2], "21"); + EXPECT_EQ(rows[2][1], "30"); + EXPECT_EQ(rows[2][3], "32"); +} + +TEST_F(SchemaCheckCacheTest, LateRegistrationIsNotMaskedByCache) { + const std::string device = "root.cache_late_register"; + ASSERT_EQ(writer_->register_timeseries(device, schema("s0")), E_OK); + + auto first_schema = schemas({"s0", "s1"}); + Tablet first(device, + std::make_shared>(first_schema), + 1); + ASSERT_EQ(first.add_timestamp(0, 0), E_OK); + ASSERT_EQ(first.add_value(0, 0, static_cast(100)), E_OK); + ASSERT_EQ(first.add_value(0, 1, static_cast(101)), E_OK); + ASSERT_EQ(writer_->write_tablet(first), E_OK); + + ASSERT_EQ(writer_->register_timeseries(device, schema("s1")), E_OK); + auto second_schema = schemas({"s0", "s1"}); + Tablet second( + device, std::make_shared>(second_schema), + 1); + ASSERT_EQ(second.add_timestamp(0, 1), E_OK); + ASSERT_EQ(second.add_value(0, 0, static_cast(200)), E_OK); + ASSERT_EQ(second.add_value(0, 1, static_cast(201)), E_OK); + ASSERT_EQ(writer_->write_tablet(second), E_OK); + + ASSERT_EQ(writer_->flush(), E_OK); + ASSERT_EQ(writer_->close(), E_OK); + auto rows = read_rows({path(device, "s1")}); + ASSERT_EQ(rows.size(), 1u); + EXPECT_EQ(rows[0][0], "1"); + EXPECT_EQ(rows[0][1], "201"); +} + +TEST_F(SchemaCheckCacheTest, AlignedCacheIsIndependentAndHandlesReorder) { + const std::string device = "root.cache_aligned"; + for (const auto& name : {"a0", "a1"}) { + ASSERT_EQ(writer_->register_aligned_timeseries(device, schema(name)), + E_OK); + } + + for (int row = 0; row < 4; ++row) { + const std::vector names = + row == 3 ? std::vector{"a1", "a0"} + : std::vector{"a0", "a1"}; + auto tablet_schema = schemas(names); + Tablet tablet( + device, + std::make_shared>(tablet_schema), 1); + ASSERT_EQ(tablet.add_timestamp(0, 500 + row), E_OK); + if (row == 3) { + ASSERT_EQ(tablet.add_value(0, 0, static_cast(19)), E_OK); + ASSERT_EQ(tablet.add_value(0, 1, static_cast(9)), E_OK); + } else { + ASSERT_EQ(tablet.add_value(0, 0, row), E_OK); + ASSERT_EQ(tablet.add_value(0, 1, 10 + row), E_OK); + } + ASSERT_EQ(writer_->write_tablet_aligned(tablet), E_OK); + } + ASSERT_EQ(writer_->flush(), E_OK); + ASSERT_EQ(writer_->close(), E_OK); + + auto rows = read_rows({path(device, "a0"), path(device, "a1")}); + ASSERT_EQ(rows.size(), 4u); + for (int row = 0; row < 3; ++row) { + EXPECT_EQ(rows[row][1], std::to_string(row)); + EXPECT_EQ(rows[row][2], std::to_string(10 + row)); + } + EXPECT_EQ(rows[3][1], "9"); + EXPECT_EQ(rows[3][2], "19"); +} + +TEST_F(SchemaCheckCacheTest, InterleavedDevicesDoNotCrossWire) { + const std::string plain_device0 = "root.cache_plain0"; + const std::string plain_device1 = "root.cache_plain1"; + const std::string aligned_device = "root.cache_aligned_interleaved"; + for (const auto& name : {"m0", "m1"}) { + ASSERT_EQ(writer_->register_timeseries(plain_device0, schema(name)), + E_OK); + ASSERT_EQ(writer_->register_timeseries(plain_device1, schema(name)), + E_OK); + ASSERT_EQ( + writer_->register_aligned_timeseries(aligned_device, schema(name)), + E_OK); + } + + for (int row = 0; row < 3; ++row) { + for (const auto& device : {plain_device0, plain_device1}) { + auto plain_schema = schemas({"m0", "m1"}); + Tablet plain_tablet( + device, + std::make_shared>(plain_schema), + 1); + ASSERT_EQ(plain_tablet.add_timestamp(0, 700 + row), E_OK); + const int device_offset = device == plain_device0 ? 0 : 100; + ASSERT_EQ(plain_tablet.add_value( + 0, 0, static_cast(device_offset + row)), + E_OK); + ASSERT_EQ(plain_tablet.add_value( + 0, 1, static_cast(device_offset + 10 + row)), + E_OK); + ASSERT_EQ(writer_->write_tablet(plain_tablet), E_OK); + } + + auto aligned_schema = schemas({"m0", "m1"}); + Tablet aligned_tablet( + aligned_device, + std::make_shared>(aligned_schema), + 1); + ASSERT_EQ(aligned_tablet.add_timestamp(0, 700 + row), E_OK); + ASSERT_EQ( + aligned_tablet.add_value(0, 0, static_cast(1000 + row)), + E_OK); + ASSERT_EQ( + aligned_tablet.add_value(0, 1, static_cast(1010 + row)), + E_OK); + ASSERT_EQ(writer_->write_tablet_aligned(aligned_tablet), E_OK); + } + + ASSERT_EQ(writer_->flush(), E_OK); + ASSERT_EQ(writer_->close(), E_OK); + for (const auto& device : {plain_device0, plain_device1}) { + auto rows = read_rows({path(device, "m0"), path(device, "m1")}); + ASSERT_EQ(rows.size(), 3u); + const int device_offset = device == plain_device0 ? 0 : 100; + for (int row = 0; row < 3; ++row) { + EXPECT_EQ(rows[row][1], std::to_string(device_offset + row)); + EXPECT_EQ(rows[row][2], std::to_string(device_offset + 10 + row)); + } + } + auto aligned_rows = + read_rows({path(aligned_device, "m0"), path(aligned_device, "m1")}); + ASSERT_EQ(aligned_rows.size(), 3u); + for (int row = 0; row < 3; ++row) { + EXPECT_EQ(aligned_rows[row][1], std::to_string(1000 + row)); + EXPECT_EQ(aligned_rows[row][2], std::to_string(1010 + row)); + } +} + +TEST_F(SchemaCheckCacheTest, RecordPathsUseIndependentCaches) { + const std::string plain_device = "root.cache_record_plain"; + const std::string aligned_device = "root.cache_record_aligned"; + for (const auto& name : {"m0", "m1"}) { + ASSERT_EQ(writer_->register_timeseries(plain_device, schema(name)), + E_OK); + ASSERT_EQ( + writer_->register_aligned_timeseries(aligned_device, schema(name)), + E_OK); + } + + for (int row = 0; row < 4; ++row) { + TsRecord plain_record(800 + row, plain_device); + plain_record.add_point("m0", static_cast(row)); + plain_record.add_point("m1", static_cast(10 + row)); + ASSERT_EQ(writer_->write_record(plain_record), E_OK); + + TsRecord aligned_record(800 + row, aligned_device); + aligned_record.add_point("m0", static_cast(100 + row)); + aligned_record.add_point("m1", static_cast(110 + row)); + ASSERT_EQ(writer_->write_record_aligned(aligned_record), E_OK); + } + + ASSERT_EQ(writer_->flush(), E_OK); + ASSERT_EQ(writer_->close(), E_OK); + auto plain_rows = + read_rows({path(plain_device, "m0"), path(plain_device, "m1")}); + auto aligned_rows = + read_rows({path(aligned_device, "m0"), path(aligned_device, "m1")}); + ASSERT_EQ(plain_rows.size(), 4u); + ASSERT_EQ(aligned_rows.size(), 4u); + for (int row = 0; row < 4; ++row) { + EXPECT_EQ(plain_rows[row][1], std::to_string(row)); + EXPECT_EQ(plain_rows[row][2], std::to_string(10 + row)); + EXPECT_EQ(aligned_rows[row][1], std::to_string(100 + row)); + EXPECT_EQ(aligned_rows[row][2], std::to_string(110 + row)); + } +} + +} // namespace