From 9dc1c778dfc56a02f366c398b8834c80ac77e7c7 Mon Sep 17 00:00:00 2001 From: gx Date: Wed, 2 Sep 2026 09:06:21 +0800 Subject: [PATCH 1/3] perf(cpp): cache resolved chunk writers in do_check_schema Repeated tablet/record writes with a fixed device schema re-resolve every measurement name against measurement_schema_map_ on each write, which is a CPU hotspot for wide schemas (#885). Cache the resolved chunk writers and data types per device in MeasurementSchemaGroup, keyed by the measurement NAME SEQUENCE (column count alone is not a safe key: entries are reused by position, so a same-count tablet with a different name order would write values into the wrong column with the wrong data type). A mismatch drops the stale cache and re-resolves. The plain and aligned paths keep separate caches. Only fully-resolved results are cached: a NULL chunk writer for a not-yet-registered measurement must not be pinned, or the column would stay masked even after it is registered. --- cpp/src/common/schema.h | 25 +++++++ cpp/src/writer/tsfile_writer.cc | 114 +++++++++++++++++++++++++++++++- 2 files changed, 136 insertions(+), 3 deletions(-) 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..7ee6b1250 100644 --- a/cpp/src/writer/tsfile_writer.cc +++ b/cpp/src/writer/tsfile_writer.cc @@ -468,10 +468,48 @@ int TsFileWriter::do_check_schema( } MeasurementSchemaMap& msm = device_schema->measurement_schema_map_; uint32_t measurement_count = measurement_names.get_count(); - // chunk_writers.reserve(measurement_count); + // The getter is single-pass (next() advances an index), so buffer the + // name sequence up front. next() returns a reference, so this buffers + // pointers rather than copies, which lets the cache check below compare + // the real names and still fall through to the full lookup on mismatch. + static thread_local std::vector names; + names.clear(); + names.reserve(measurement_count); for (uint32_t i = 0; i < measurement_count; i++) { - auto ms_iter = msm.find(measurement_names.next()); + names.push_back(&measurement_names.next()); + } + // Column count alone is not a safe cache key: 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 with the + // wrong data type. + if (device_schema->schema_check_cached_ && + device_schema->cached_measurement_names_.size() == measurement_count) { + bool same_schema = true; + for (uint32_t i = 0; i < measurement_count; i++) { + if (device_schema->cached_measurement_names_[i] != *names[i]) { + same_schema = false; + break; + } + } + if (same_schema) { + for (uint32_t i = 0; i < measurement_count; i++) { + chunk_writers.push_back( + device_schema->cached_chunk_writers_[i]); + data_types.push_back(device_schema->cached_data_types_[i]); + } + return E_OK; + } + // Schema changed for this device: drop the stale cache and re-resolve. + device_schema->cached_chunk_writers_.clear(); + device_schema->cached_data_types_.clear(); + device_schema->cached_measurement_names_.clear(); + device_schema->schema_check_cached_ = false; + } + bool all_resolved = true; + for (uint32_t i = 0; i < measurement_count; 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 { @@ -502,6 +540,22 @@ int TsFileWriter::do_check_schema( data_types.push_back(ms->data_type_); } } + // Cache only fully-resolved results: a NULL entry for a measurement that + // is not registered yet would keep masking the column even after it is + // registered later. + if (IS_SUCC(ret) && all_resolved) { + device_schema->cached_chunk_writers_.reserve(measurement_count); + device_schema->cached_data_types_.reserve(measurement_count); + for (uint32_t i = 0; i < measurement_count; i++) { + device_schema->cached_chunk_writers_.push_back(chunk_writers[i]); + device_schema->cached_data_types_.push_back(data_types[i]); + } + device_schema->cached_measurement_names_.reserve(measurement_count); + for (uint32_t i = 0; i < measurement_count; i++) { + device_schema->cached_measurement_names_.push_back(*names[i]); + } + device_schema->schema_check_cached_ = true; + } return ret; } @@ -528,9 +582,46 @@ int TsFileWriter::do_check_schema_aligned( time_chunk_writer = device_schema->time_chunk_writer_; MeasurementSchemaMap& msm = device_schema->measurement_schema_map_; uint32_t measurement_count = measurement_names.get_count(); + // Same single-pass buffering + name-sequence guard as do_check_schema; + // see the comment there and on MeasurementSchemaGroup. This path keeps + // its own cache because sharing one flag with the plain path locked + // whichever ran second out. + static thread_local std::vector names; + names.clear(); + names.reserve(measurement_count); for (uint32_t i = 0; i < measurement_count; i++) { - auto ms_iter = msm.find(measurement_names.next()); + names.push_back(&measurement_names.next()); + } + if (device_schema->schema_check_aligned_cached_ && + device_schema->cached_aligned_measurement_names_.size() == + measurement_count) { + bool same_schema = true; + for (uint32_t i = 0; i < measurement_count; i++) { + if (device_schema->cached_aligned_measurement_names_[i] != + *names[i]) { + same_schema = false; + break; + } + } + if (same_schema) { + for (uint32_t i = 0; i < measurement_count; i++) { + value_chunk_writers.push_back( + device_schema->cached_value_chunk_writers_[i]); + data_types.push_back( + device_schema->cached_aligned_data_types_[i]); + } + return E_OK; + } + device_schema->cached_value_chunk_writers_.clear(); + device_schema->cached_aligned_data_types_.clear(); + device_schema->cached_aligned_measurement_names_.clear(); + device_schema->schema_check_aligned_cached_ = false; + } + bool all_resolved = true; + for (uint32_t i = 0; i < measurement_count; 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 { @@ -562,6 +653,23 @@ int TsFileWriter::do_check_schema_aligned( data_types.push_back(ms->data_type_); } } + // See do_check_schema: never cache a result with unresolved columns. + if (IS_SUCC(ret) && all_resolved) { + device_schema->cached_value_chunk_writers_.reserve(measurement_count); + device_schema->cached_aligned_data_types_.reserve(measurement_count); + for (uint32_t i = 0; i < measurement_count; i++) { + device_schema->cached_value_chunk_writers_.push_back( + value_chunk_writers[i]); + device_schema->cached_aligned_data_types_.push_back(data_types[i]); + } + device_schema->cached_aligned_measurement_names_.reserve( + measurement_count); + for (uint32_t i = 0; i < measurement_count; i++) { + device_schema->cached_aligned_measurement_names_.push_back( + *names[i]); + } + device_schema->schema_check_aligned_cached_ = true; + } return ret; } From 2d3bf750b1763f7c1452aad179cd43a4ad25dc88 Mon Sep 17 00:00:00 2001 From: gx Date: Wed, 2 Sep 2026 09:06:30 +0800 Subject: [PATCH 2/3] test(cpp): cover the per-device schema-check cache Five cases pinning the behaviors the cache must preserve: - repeated same-schema writes round-trip every row (hit path, incl. after a flush seals and resets the chunk writers); - a same-column-count tablet with different names/order re-resolves and writes each value into the column its name says; - a column unregistered at first write is not masked by a cached NULL after it is registered (only fully-resolved results are cached); - the aligned path keeps its own cache with the same guarantees; - per-device caches never cross-wire two devices. --- .../writer/tsfile_writer_schema_cache_test.cc | 437 ++++++++++++++++++ 1 file changed, 437 insertions(+) create mode 100644 cpp/test/writer/tsfile_writer_schema_cache_test.cc 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..c163daef9 --- /dev/null +++ b/cpp/test/writer/tsfile_writer_schema_cache_test.cc @@ -0,0 +1,437 @@ +/* + * 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. + */ + +// Tests for the per-device schema-check cache in do_check_schema / +// do_check_schema_aligned (issue #885). The cache resolves chunk writers and +// data types once per device and reuses them while the tablet's measurement +// NAME SEQUENCE is unchanged. These tests pin the behaviors the cache must +// preserve: +// 1. repeated same-schema writes round-trip every row (cache hit path); +// 2. a same-column-count tablet with different names/order re-resolves and +// writes each value into the right column (cache invalidation); +// 3. a column that was unregistered at first write is NOT masked by a cached +// NULL after it is registered (only fully-resolved results are cached); +// 4. the aligned path keeps its own cache with the same guarantees; +// 5. per-device caches never cross-wire two devices. +#include + +#include "writer/tsfile_writer.h" + +#ifdef _WIN32 +#include +#else +#include +#endif + +#include +#include +#include +#include +#include +#include + +#include "common/path.h" +#include "common/record.h" +#include "common/schema.h" +#include "common/tablet.h" +#include "common/tsfile_common.h" +#include "reader/qds_without_timegenerator.h" +#include "reader/tsfile_reader.h" + +using namespace storage; +using namespace common; + +namespace { + +class SchemaCheckCacheTest : public ::testing::Test { + protected: + void SetUp() override { + libtsfile_init(); + tsfile_writer_ = new TsFileWriter(); + file_name_ = std::string("tsfile_schema_cache_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 + ASSERT_EQ(tsfile_writer_->open(file_name_, flags, 0666), common::E_OK); + } + void TearDown() override { + delete tsfile_writer_; + ASSERT_EQ(0, remove(file_name_.c_str())); + libtsfile_destroy(); + } + + std::string file_name_; + TsFileWriter* tsfile_writer_ = nullptr; + + public: + static std::string generate_random_string(int length) { + static std::atomic counter{0}; + std::mt19937 gen(static_cast( + 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)]; + } +#ifdef _WIN32 + const auto process_id = static_cast(_getpid()); +#else + const auto process_id = static_cast(getpid()); +#endif + random_string += "_" + std::to_string(process_id) + "_" + + std::to_string(counter.fetch_add(1)); + return random_string; + } + + // Reads back (device, measurement) pairs and returns one row per + // timestamp: {timestamp, value-string per series}. Row count is asserted + // by the caller so dropped rows cannot pass silently. + std::vector> query_all( + const std::vector& select_list) { + storage::TsFileReader reader; + EXPECT_EQ(reader.open(file_name_), E_OK); + QueryExpression* query_expr = + QueryExpression::create(select_list, nullptr); + ResultSet* tmp_qds = nullptr; + EXPECT_EQ(reader.query(query_expr, tmp_qds), E_OK); + auto* qds = (QDSWithoutTimeGenerator*)tmp_qds; + + std::vector> rows; + bool has_next = false; + while (IS_SUCC(qds->next(has_next)) && has_next) { + RowRecord* record = qds->get_row_record(); + std::vector row; + row.push_back(std::to_string(record->get_timestamp())); + // field(0) is the timestamp; value fields start at 1. + 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(qds); + return rows; + } + + MeasurementSchema int32_schema(const std::string& name) { + return MeasurementSchema(name, TSDataType::INT32, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED); + } + + static std::string field_to_string(storage::Field* value) { + if (value->type_ == common::TEXT || value->type_ == STRING || + value->type_ == BLOB) { + return std::string(value->value_.sval_); + } + 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: + case common::TIMESTAMP: + 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(); + } + + // Path's two-part ctor takes non-const std::string&, so route every + // construction through copies. + Path make_path(const std::string& device, const std::string& measurement) { + std::string dev = device; + std::string meas = measurement; + return Path(dev, meas); + } +}; + +// 1. Cache hit: the same tablet schema written repeatedly (with a flush in +// between, so chunk writers survive a seal and are re-resolved from the +// cache) must round-trip every row of every column. +TEST_F(SchemaCheckCacheTest, RepeatedSameSchemaRoundTrip) { + const std::string device = "root.cache_hit"; + const std::vector names = {"s0", "s1", "s2"}; + for (const auto& name : names) { + ASSERT_EQ( + tsfile_writer_->register_timeseries(device, int32_schema(name)), + E_OK); + } + + const int num_tablets = 5; + for (int t = 0; t < num_tablets; t++) { + std::vector schema_vec; + for (const auto& name : names) schema_vec.push_back(int32_schema(name)); + Tablet tablet( + device, + std::make_shared>(schema_vec), 1); + ASSERT_EQ(tablet.add_timestamp(0, 1000 + t), E_OK); + for (uint32_t j = 0; j < names.size(); j++) { + ASSERT_EQ(tablet.add_value(0, j, t * 100 + (int32_t)j), E_OK); + } + ASSERT_EQ(tsfile_writer_->write_tablet(tablet), E_OK); + if (t == 2) { + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + } + } + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + std::vector select_list; + for (const auto& name : names) + select_list.push_back(make_path(device, name)); + auto rows = query_all(select_list); + ASSERT_EQ(rows.size(), (size_t)num_tablets); + for (int t = 0; t < num_tablets; t++) { + ASSERT_EQ(rows[t][0], std::to_string(1000 + t)); + for (uint32_t j = 0; j < names.size(); j++) { + ASSERT_EQ(rows[t][j + 1], std::to_string(t * 100 + j)) + << "row " << t << " column " << j; + } + } +} + +// 2. Invalidation by name sequence: same column count, different names and +// order. Values must land in the column their NAME says, not the position +// the previous tablet used. +TEST_F(SchemaCheckCacheTest, SameCountDifferentNamesAndOrder) { + const std::string device = "root.cache_inval"; + for (const auto& name : {"s0", "s1", "s2"}) { + ASSERT_EQ( + tsfile_writer_->register_timeseries(device, int32_schema(name)), + E_OK); + } + + // Tablet 1: [s0, s1] at t=0. + { + std::vector schema_vec = {int32_schema("s0"), + int32_schema("s1")}; + Tablet tablet( + device, + std::make_shared>(schema_vec), 1); + ASSERT_EQ(tablet.add_timestamp(0, 0), E_OK); + ASSERT_EQ(tablet.add_value(0, 0, 10), E_OK); // s0 = 10 + ASSERT_EQ(tablet.add_value(0, 1, 11), E_OK); // s1 = 11 + ASSERT_EQ(tsfile_writer_->write_tablet(tablet), E_OK); + } + // Tablet 2: same count, REVERSED order, at t=1. + { + std::vector schema_vec = {int32_schema("s1"), + int32_schema("s0")}; + Tablet tablet( + device, + std::make_shared>(schema_vec), 1); + ASSERT_EQ(tablet.add_timestamp(0, 1), E_OK); + ASSERT_EQ(tablet.add_value(0, 0, 21), E_OK); // s1 = 21 + ASSERT_EQ(tablet.add_value(0, 1, 20), E_OK); // s0 = 20 + ASSERT_EQ(tsfile_writer_->write_tablet(tablet), E_OK); + } + // Tablet 3: same count, one column swapped for an unseen name, at t=2. + { + std::vector schema_vec = {int32_schema("s0"), + int32_schema("s2")}; + Tablet tablet( + device, + std::make_shared>(schema_vec), 1); + ASSERT_EQ(tablet.add_timestamp(0, 2), E_OK); + ASSERT_EQ(tablet.add_value(0, 0, 30), E_OK); // s0 = 30 + ASSERT_EQ(tablet.add_value(0, 1, 32), E_OK); // s2 = 32 + ASSERT_EQ(tsfile_writer_->write_tablet(tablet), E_OK); + } + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + std::vector select_list; + for (const auto& name : {"s0", "s1", "s2"}) { + select_list.push_back(make_path(device, name)); + } + auto rows = query_all(select_list); + ASSERT_EQ(rows.size(), (size_t)3); + // t=0 + EXPECT_EQ(rows[0][0], "0"); + EXPECT_EQ(rows[0][1], "10"); + EXPECT_EQ(rows[0][2], "11"); + // t=1: swapped order must not swap values + EXPECT_EQ(rows[1][0], "1"); + EXPECT_EQ(rows[1][1], "20"); + EXPECT_EQ(rows[1][2], "21"); + // t=2: s1 has no point at t=2 + EXPECT_EQ(rows[2][0], "2"); + EXPECT_EQ(rows[2][1], "30"); + EXPECT_EQ(rows[2][3], "32"); +} + +// 3. A measurement missing at first write resolves to a NULL chunk writer +// (column skipped). After it is registered, the same tablet schema must +// write that column: the cache must not pin the stale NULL. +TEST_F(SchemaCheckCacheTest, ColumnRegisteredAfterFirstWriteIsNotMasked) { + const std::string device = "root.cache_late_register"; + ASSERT_EQ(tsfile_writer_->register_timeseries(device, int32_schema("s0")), + E_OK); + // Deliberately NOT registering s1 yet. + + // First write: s1 unresolved -> NULL chunk writer, column skipped. + { + std::vector schema_vec = {int32_schema("s0"), + int32_schema("s1")}; + Tablet tablet( + device, + std::make_shared>(schema_vec), 1); + ASSERT_EQ(tablet.add_timestamp(0, 0), E_OK); + ASSERT_EQ(tablet.add_value(0, 0, 100), E_OK); + ASSERT_EQ(tablet.add_value(0, 1, 101), E_OK); + ASSERT_EQ(tsfile_writer_->write_tablet(tablet), E_OK); + } + // Now register s1 and write the same schema again. + ASSERT_EQ(tsfile_writer_->register_timeseries(device, int32_schema("s1")), + E_OK); + { + std::vector schema_vec = {int32_schema("s0"), + int32_schema("s1")}; + Tablet tablet( + device, + std::make_shared>(schema_vec), 1); + ASSERT_EQ(tablet.add_timestamp(0, 1), E_OK); + ASSERT_EQ(tablet.add_value(0, 0, 200), E_OK); + ASSERT_EQ(tablet.add_value(0, 1, 201), E_OK); + ASSERT_EQ(tsfile_writer_->write_tablet(tablet), E_OK); + } + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + auto rows = query_all({make_path(device, "s1")}); + // Without the fully-resolved guard the cached NULL would drop this + // column forever and this query would return zero rows. + ASSERT_EQ(rows.size(), (size_t)1); + EXPECT_EQ(rows[0][0], "1"); + EXPECT_EQ(rows[0][1], "201"); +} + +// 4. Aligned path: same-schema repeated writes round-trip, and a reordered +// tablet re-resolves instead of reusing positions. +TEST_F(SchemaCheckCacheTest, AlignedRepeatedAndReordered) { + const std::string device = "root.cache_aligned"; + for (const auto& name : {"a0", "a1"}) { + ASSERT_EQ(tsfile_writer_->register_aligned_timeseries( + device, int32_schema(name)), + E_OK); + } + + const int num_tablets = 4; + for (int t = 0; t < num_tablets; t++) { + // Last tablet reverses the column order. + std::vector schema_vec; + if (t < num_tablets - 1) { + schema_vec = {int32_schema("a0"), int32_schema("a1")}; + } else { + schema_vec = {int32_schema("a1"), int32_schema("a0")}; + } + Tablet tablet( + device, + std::make_shared>(schema_vec), 1); + ASSERT_EQ(tablet.add_timestamp(0, 500 + t), E_OK); + if (t < num_tablets - 1) { + ASSERT_EQ(tablet.add_value(0, 0, t), E_OK); // a0 + ASSERT_EQ(tablet.add_value(0, 1, 10 + t), E_OK); // a1 + } else { + ASSERT_EQ(tablet.add_value(0, 0, 19), E_OK); // a1 + ASSERT_EQ(tablet.add_value(0, 1, 9), E_OK); // a0 + } + ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK); + } + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + std::vector select_list; + for (const auto& name : {"a0", "a1"}) { + select_list.push_back(make_path(device, name)); + } + auto rows = query_all(select_list); + ASSERT_EQ(rows.size(), (size_t)num_tablets); + for (int t = 0; t < num_tablets - 1; t++) { + EXPECT_EQ(rows[t][1], std::to_string(t)); + EXPECT_EQ(rows[t][2], std::to_string(10 + t)); + } + // Reordered final tablet: a0=9, a1=19. + EXPECT_EQ(rows[num_tablets - 1][1], "9"); + EXPECT_EQ(rows[num_tablets - 1][2], "19"); +} + +// 5. Per-device caches are independent: two devices with identical +// measurement names, interleaved writes, different values. +TEST_F(SchemaCheckCacheTest, MultiDeviceCachesIndependent) { + const std::string devices[2] = {"root.cache_dev0", "root.cache_dev1"}; + for (const auto& device : devices) { + for (const auto& name : {"m0", "m1"}) { + ASSERT_EQ( + tsfile_writer_->register_timeseries(device, int32_schema(name)), + E_OK); + } + } + + for (int t = 0; t < 3; t++) { + for (int d = 0; d < 2; d++) { + std::vector schema_vec = {int32_schema("m0"), + int32_schema("m1")}; + Tablet tablet( + devices[d], + std::make_shared>(schema_vec), + 1); + ASSERT_EQ(tablet.add_timestamp(0, 700 + t), E_OK); + // d*1000 separates the two devices' value spaces. + ASSERT_EQ(tablet.add_value(0, 0, d * 1000 + t), E_OK); + ASSERT_EQ(tablet.add_value(0, 1, d * 1000 + 10 + t), E_OK); + ASSERT_EQ(tsfile_writer_->write_tablet(tablet), E_OK); + } + } + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + for (int d = 0; d < 2; d++) { + auto rows = query_all( + {make_path(devices[d], "m0"), make_path(devices[d], "m1")}); + ASSERT_EQ(rows.size(), (size_t)3); + for (int t = 0; t < 3; t++) { + EXPECT_EQ(rows[t][1], std::to_string(d * 1000 + t)); + EXPECT_EQ(rows[t][2], std::to_string(d * 1000 + 10 + t)); + } + } +} + +} // namespace From 61bc506a1b1eb6f5eb50f6d8360b84b76fc603df Mon Sep 17 00:00:00 2001 From: gx Date: Mon, 7 Sep 2026 14:07:35 +0800 Subject: [PATCH 3/3] perf(cpp): bound schema-check caches at writer level MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move schema-check cache state out of every device schema group and keep aligned and non-aligned caches independently bounded. Add regression coverage for cache invalidation, ordering, interleaved devices, and record write paths. 🤖 Generated with Codebuff Co-Authored-By: Codebuff EOF ) --- cpp/src/writer/tsfile_writer.cc | 276 ++++----- cpp/src/writer/tsfile_writer.h | 45 ++ .../writer/tsfile_writer_schema_cache_test.cc | 548 +++++++++--------- 3 files changed, 435 insertions(+), 434 deletions(-) diff --git a/cpp/src/writer/tsfile_writer.cc b/cpp/src/writer/tsfile_writer.cc index 7ee6b1250..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,95 +503,59 @@ 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(); - // The getter is single-pass (next() advances an index), so buffer the - // name sequence up front. next() returns a reference, so this buffers - // pointers rather than copies, which lets the cache check below compare - // the real names and still fall through to the full lookup on mismatch. - static thread_local std::vector names; + static thread_local std::vector names; names.clear(); - names.reserve(measurement_count); - for (uint32_t i = 0; i < measurement_count; i++) { - names.push_back(&measurement_names.next()); - } - // Column count alone is not a safe cache key: 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 with the - // wrong data type. - if (device_schema->schema_check_cached_ && - device_schema->cached_measurement_names_.size() == measurement_count) { - bool same_schema = true; - for (uint32_t i = 0; i < measurement_count; i++) { - if (device_schema->cached_measurement_names_[i] != *names[i]) { - same_schema = false; - break; - } - } - if (same_schema) { - for (uint32_t i = 0; i < measurement_count; i++) { - chunk_writers.push_back( - device_schema->cached_chunk_writers_[i]); - data_types.push_back(device_schema->cached_data_types_[i]); - } - return E_OK; + 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]); } - // Schema changed for this device: drop the stale cache and re-resolve. - device_schema->cached_chunk_writers_.clear(); - device_schema->cached_data_types_.clear(); - device_schema->cached_measurement_names_.clear(); - device_schema->schema_check_cached_ = false; + return E_OK; } + bool all_resolved = true; - for (uint32_t i = 0; i < measurement_count; i++) { - auto ms_iter = msm.find(*names[i]); + 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_); } - // Cache only fully-resolved results: a NULL entry for a measurement that - // is not registered yet would keep masking the column even after it is - // registered later. + if (IS_SUCC(ret) && all_resolved) { - device_schema->cached_chunk_writers_.reserve(measurement_count); - device_schema->cached_data_types_.reserve(measurement_count); - for (uint32_t i = 0; i < measurement_count; i++) { - device_schema->cached_chunk_writers_.push_back(chunk_writers[i]); - device_schema->cached_data_types_.push_back(data_types[i]); - } - device_schema->cached_measurement_names_.reserve(measurement_count); - for (uint32_t i = 0; i < measurement_count; i++) { - device_schema->cached_measurement_names_.push_back(*names[i]); + 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]); } - device_schema->schema_check_cached_ = true; } return ret; } @@ -580,95 +581,60 @@ 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(); - // Same single-pass buffering + name-sequence guard as do_check_schema; - // see the comment there and on MeasurementSchemaGroup. This path keeps - // its own cache because sharing one flag with the plain path locked - // whichever ran second out. - static thread_local std::vector names; + static thread_local std::vector names; names.clear(); - names.reserve(measurement_count); - for (uint32_t i = 0; i < measurement_count; i++) { - names.push_back(&measurement_names.next()); - } - if (device_schema->schema_check_aligned_cached_ && - device_schema->cached_aligned_measurement_names_.size() == - measurement_count) { - bool same_schema = true; - for (uint32_t i = 0; i < measurement_count; i++) { - if (device_schema->cached_aligned_measurement_names_[i] != - *names[i]) { - same_schema = false; - break; - } - } - if (same_schema) { - for (uint32_t i = 0; i < measurement_count; i++) { - value_chunk_writers.push_back( - device_schema->cached_value_chunk_writers_[i]); - data_types.push_back( - device_schema->cached_aligned_data_types_[i]); - } - return E_OK; + 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]); } - device_schema->cached_value_chunk_writers_.clear(); - device_schema->cached_aligned_data_types_.clear(); - device_schema->cached_aligned_measurement_names_.clear(); - device_schema->schema_check_aligned_cached_ = false; + return E_OK; } + bool all_resolved = true; - for (uint32_t i = 0; i < measurement_count; i++) { - auto ms_iter = msm.find(*names[i]); + 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_); } - // See do_check_schema: never cache a result with unresolved columns. + if (IS_SUCC(ret) && all_resolved) { - device_schema->cached_value_chunk_writers_.reserve(measurement_count); - device_schema->cached_aligned_data_types_.reserve(measurement_count); - for (uint32_t i = 0; i < measurement_count; i++) { - device_schema->cached_value_chunk_writers_.push_back( + 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]); - device_schema->cached_aligned_data_types_.push_back(data_types[i]); - } - device_schema->cached_aligned_measurement_names_.reserve( - measurement_count); - for (uint32_t i = 0; i < measurement_count; i++) { - device_schema->cached_aligned_measurement_names_.push_back( - *names[i]); + aligned_schema_check_cache_.data_types_.push_back(data_types[i]); } - device_schema->schema_check_aligned_cached_ = true; } 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 index c163daef9..638dc5ccf 100644 --- a/cpp/test/writer/tsfile_writer_schema_cache_test.cc +++ b/cpp/test/writer/tsfile_writer_schema_cache_test.cc @@ -17,18 +17,6 @@ * under the License. */ -// Tests for the per-device schema-check cache in do_check_schema / -// do_check_schema_aligned (issue #885). The cache resolves chunk writers and -// data types once per device and reuses them while the tablet's measurement -// NAME SEQUENCE is unchanged. These tests pin the behaviors the cache must -// preserve: -// 1. repeated same-schema writes round-trip every row (cache hit path); -// 2. a same-column-count tablet with different names/order re-resolves and -// writes each value into the right column (cache invalidation); -// 3. a column that was unregistered at first write is NOT masked by a cached -// NULL after it is registered (only fully-resolved results are cached); -// 4. the aligned path keeps its own cache with the same guarantees; -// 5. per-device caches never cross-wire two devices. #include #include "writer/tsfile_writer.h" @@ -40,6 +28,7 @@ #endif #include +#include #include #include #include @@ -50,388 +39,389 @@ #include "common/record.h" #include "common/schema.h" #include "common/tablet.h" -#include "common/tsfile_common.h" #include "reader/qds_without_timegenerator.h" #include "reader/tsfile_reader.h" -using namespace storage; using namespace common; +using namespace storage; namespace { class SchemaCheckCacheTest : public ::testing::Test { protected: void SetUp() override { - libtsfile_init(); - tsfile_writer_ = new TsFileWriter(); - file_name_ = std::string("tsfile_schema_cache_test_") + - generate_random_string(10) + std::string(".tsfile"); + 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(tsfile_writer_->open(file_name_, flags, 0666), common::E_OK); + ASSERT_EQ(writer_->open(file_name_, flags, 0666), E_OK); } + void TearDown() override { - delete tsfile_writer_; - ASSERT_EQ(0, remove(file_name_.c_str())); + delete writer_; + ASSERT_EQ(remove(file_name_.c_str()), 0); libtsfile_destroy(); } - std::string file_name_; - TsFileWriter* tsfile_writer_ = nullptr; - - public: - static std::string generate_random_string(int length) { + static std::string unique_suffix() { static std::atomic counter{0}; - std::mt19937 gen(static_cast( - 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)]; - } #ifdef _WIN32 const auto process_id = static_cast(_getpid()); #else const auto process_id = static_cast(getpid()); #endif - random_string += "_" + std::to_string(process_id) + "_" + - std::to_string(counter.fetch_add(1)); - return random_string; - } - - // Reads back (device, measurement) pairs and returns one row per - // timestamp: {timestamp, value-string per series}. Row count is asserted - // by the caller so dropped rows cannot pass silently. - std::vector> query_all( - const std::vector& select_list) { - storage::TsFileReader reader; - EXPECT_EQ(reader.open(file_name_), E_OK); - QueryExpression* query_expr = - QueryExpression::create(select_list, nullptr); - ResultSet* tmp_qds = nullptr; - EXPECT_EQ(reader.query(query_expr, tmp_qds), E_OK); - auto* qds = (QDSWithoutTimeGenerator*)tmp_qds; - - std::vector> rows; - bool has_next = false; - while (IS_SUCC(qds->next(has_next)) && has_next) { - RowRecord* record = qds->get_row_record(); - std::vector row; - row.push_back(std::to_string(record->get_timestamp())); - // field(0) is the timestamp; value fields start at 1. - 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(qds); - return rows; + return std::to_string(process_id) + "_" + + std::to_string(counter.fetch_add(1)); } - MeasurementSchema int32_schema(const std::string& name) { + static MeasurementSchema schema(const std::string& name) { return MeasurementSchema(name, TSDataType::INT32, TSEncoding::PLAIN, CompressionType::UNCOMPRESSED); } - static std::string field_to_string(storage::Field* value) { - if (value->type_ == common::TEXT || value->type_ == STRING || - value->type_ == BLOB) { - return std::string(value->value_.sval_); + static std::vector schemas( + const std::vector& names) { + std::vector result; + for (const auto& name : names) { + result.push_back(schema(name)); } - std::stringstream ss; - switch (value->type_) { - case common::BOOLEAN: - ss << (value->value_.bval_ ? "true" : "false"); + 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 common::INT32: - ss << value->value_.ival_; + case INT32: + case DATE: + stream << field->value_.ival_; break; - case common::INT64: - case common::TIMESTAMP: - ss << value->value_.lval_; + case INT64: + case TIMESTAMP: + stream << field->value_.lval_; break; - case common::FLOAT: - ss << value->value_.fval_; + case FLOAT: + stream << field->value_.fval_; break; - case common::DOUBLE: - ss << value->value_.dval_; + case DOUBLE: + stream << field->value_.dval_; break; - case common::NULL_TYPE: - ss << "NULL"; + case NULL_TYPE: + stream << "NULL"; break; default: - ASSERT(false); - break; + ADD_FAILURE() << "Unexpected field type: " << field->type_; } - return ss.str(); + return stream.str(); } - // Path's two-part ctor takes non-const std::string&, so route every - // construction through copies. - Path make_path(const std::string& device, const std::string& measurement) { - std::string dev = device; - std::string meas = measurement; - return Path(dev, meas); + 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_; }; -// 1. Cache hit: the same tablet schema written repeatedly (with a flush in -// between, so chunk writers survive a seal and are re-resolved from the -// cache) must round-trip every row of every column. -TEST_F(SchemaCheckCacheTest, RepeatedSameSchemaRoundTrip) { +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( - tsfile_writer_->register_timeseries(device, int32_schema(name)), - E_OK); + ASSERT_EQ(writer_->register_timeseries(device, schema(name)), E_OK); } - const int num_tablets = 5; - for (int t = 0; t < num_tablets; t++) { - std::vector schema_vec; - for (const auto& name : names) schema_vec.push_back(int32_schema(name)); + for (int row = 0; row < 5; ++row) { + auto tablet_schema = schemas(names); Tablet tablet( device, - std::make_shared>(schema_vec), 1); - ASSERT_EQ(tablet.add_timestamp(0, 1000 + t), E_OK); - for (uint32_t j = 0; j < names.size(); j++) { - ASSERT_EQ(tablet.add_value(0, j, t * 100 + (int32_t)j), E_OK); + 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(tsfile_writer_->write_tablet(tablet), E_OK); - if (t == 2) { - ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(writer_->write_tablet(tablet), E_OK); + if (row == 2) { + ASSERT_EQ(writer_->flush(), E_OK); } } - ASSERT_EQ(tsfile_writer_->flush(), E_OK); - ASSERT_EQ(tsfile_writer_->close(), E_OK); - - std::vector select_list; - for (const auto& name : names) - select_list.push_back(make_path(device, name)); - auto rows = query_all(select_list); - ASSERT_EQ(rows.size(), (size_t)num_tablets); - for (int t = 0; t < num_tablets; t++) { - ASSERT_EQ(rows[t][0], std::to_string(1000 + t)); - for (uint32_t j = 0; j < names.size(); j++) { - ASSERT_EQ(rows[t][j + 1], std::to_string(t * 100 + j)) - << "row " << t << " column " << j; + 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)); } } } -// 2. Invalidation by name sequence: same column count, different names and -// order. Values must land in the column their NAME says, not the position -// the previous tablet used. -TEST_F(SchemaCheckCacheTest, SameCountDifferentNamesAndOrder) { - const std::string device = "root.cache_inval"; +TEST_F(SchemaCheckCacheTest, SameCountDifferentNameOrderDoesNotCrossWire) { + const std::string device = "root.cache_reorder"; for (const auto& name : {"s0", "s1", "s2"}) { - ASSERT_EQ( - tsfile_writer_->register_timeseries(device, int32_schema(name)), - E_OK); + ASSERT_EQ(writer_->register_timeseries(device, schema(name)), E_OK); } - // Tablet 1: [s0, s1] at t=0. { - std::vector schema_vec = {int32_schema("s0"), - int32_schema("s1")}; + auto tablet_schema = schemas({"s0", "s1"}); Tablet tablet( device, - std::make_shared>(schema_vec), 1); + std::make_shared>(tablet_schema), 1); ASSERT_EQ(tablet.add_timestamp(0, 0), E_OK); - ASSERT_EQ(tablet.add_value(0, 0, 10), E_OK); // s0 = 10 - ASSERT_EQ(tablet.add_value(0, 1, 11), E_OK); // s1 = 11 - ASSERT_EQ(tsfile_writer_->write_tablet(tablet), 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); } - // Tablet 2: same count, REVERSED order, at t=1. { - std::vector schema_vec = {int32_schema("s1"), - int32_schema("s0")}; + auto tablet_schema = schemas({"s1", "s0"}); Tablet tablet( device, - std::make_shared>(schema_vec), 1); + std::make_shared>(tablet_schema), 1); ASSERT_EQ(tablet.add_timestamp(0, 1), E_OK); - ASSERT_EQ(tablet.add_value(0, 0, 21), E_OK); // s1 = 21 - ASSERT_EQ(tablet.add_value(0, 1, 20), E_OK); // s0 = 20 - ASSERT_EQ(tsfile_writer_->write_tablet(tablet), 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); } - // Tablet 3: same count, one column swapped for an unseen name, at t=2. { - std::vector schema_vec = {int32_schema("s0"), - int32_schema("s2")}; + auto tablet_schema = schemas({"s0", "s2"}); Tablet tablet( device, - std::make_shared>(schema_vec), 1); + std::make_shared>(tablet_schema), 1); ASSERT_EQ(tablet.add_timestamp(0, 2), E_OK); - ASSERT_EQ(tablet.add_value(0, 0, 30), E_OK); // s0 = 30 - ASSERT_EQ(tablet.add_value(0, 1, 32), E_OK); // s2 = 32 - ASSERT_EQ(tsfile_writer_->write_tablet(tablet), 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(tsfile_writer_->flush(), E_OK); - ASSERT_EQ(tsfile_writer_->close(), E_OK); + ASSERT_EQ(writer_->flush(), E_OK); + ASSERT_EQ(writer_->close(), E_OK); - std::vector select_list; - for (const auto& name : {"s0", "s1", "s2"}) { - select_list.push_back(make_path(device, name)); - } - auto rows = query_all(select_list); - ASSERT_EQ(rows.size(), (size_t)3); - // t=0 - EXPECT_EQ(rows[0][0], "0"); + 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"); - // t=1: swapped order must not swap values - EXPECT_EQ(rows[1][0], "1"); EXPECT_EQ(rows[1][1], "20"); EXPECT_EQ(rows[1][2], "21"); - // t=2: s1 has no point at t=2 - EXPECT_EQ(rows[2][0], "2"); EXPECT_EQ(rows[2][1], "30"); EXPECT_EQ(rows[2][3], "32"); } -// 3. A measurement missing at first write resolves to a NULL chunk writer -// (column skipped). After it is registered, the same tablet schema must -// write that column: the cache must not pin the stale NULL. -TEST_F(SchemaCheckCacheTest, ColumnRegisteredAfterFirstWriteIsNotMasked) { +TEST_F(SchemaCheckCacheTest, LateRegistrationIsNotMaskedByCache) { const std::string device = "root.cache_late_register"; - ASSERT_EQ(tsfile_writer_->register_timeseries(device, int32_schema("s0")), - E_OK); - // Deliberately NOT registering s1 yet. + ASSERT_EQ(writer_->register_timeseries(device, schema("s0")), E_OK); - // First write: s1 unresolved -> NULL chunk writer, column skipped. - { - std::vector schema_vec = {int32_schema("s0"), - int32_schema("s1")}; - Tablet tablet( - device, - std::make_shared>(schema_vec), 1); - ASSERT_EQ(tablet.add_timestamp(0, 0), E_OK); - ASSERT_EQ(tablet.add_value(0, 0, 100), E_OK); - ASSERT_EQ(tablet.add_value(0, 1, 101), E_OK); - ASSERT_EQ(tsfile_writer_->write_tablet(tablet), E_OK); - } - // Now register s1 and write the same schema again. - ASSERT_EQ(tsfile_writer_->register_timeseries(device, int32_schema("s1")), - E_OK); - { - std::vector schema_vec = {int32_schema("s0"), - int32_schema("s1")}; - Tablet tablet( - device, - std::make_shared>(schema_vec), 1); - ASSERT_EQ(tablet.add_timestamp(0, 1), E_OK); - ASSERT_EQ(tablet.add_value(0, 0, 200), E_OK); - ASSERT_EQ(tablet.add_value(0, 1, 201), E_OK); - ASSERT_EQ(tsfile_writer_->write_tablet(tablet), E_OK); - } - ASSERT_EQ(tsfile_writer_->flush(), E_OK); - ASSERT_EQ(tsfile_writer_->close(), 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); - auto rows = query_all({make_path(device, "s1")}); - // Without the fully-resolved guard the cached NULL would drop this - // column forever and this query would return zero rows. - ASSERT_EQ(rows.size(), (size_t)1); + 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"); } -// 4. Aligned path: same-schema repeated writes round-trip, and a reordered -// tablet re-resolves instead of reusing positions. -TEST_F(SchemaCheckCacheTest, AlignedRepeatedAndReordered) { +TEST_F(SchemaCheckCacheTest, AlignedCacheIsIndependentAndHandlesReorder) { const std::string device = "root.cache_aligned"; for (const auto& name : {"a0", "a1"}) { - ASSERT_EQ(tsfile_writer_->register_aligned_timeseries( - device, int32_schema(name)), + ASSERT_EQ(writer_->register_aligned_timeseries(device, schema(name)), E_OK); } - const int num_tablets = 4; - for (int t = 0; t < num_tablets; t++) { - // Last tablet reverses the column order. - std::vector schema_vec; - if (t < num_tablets - 1) { - schema_vec = {int32_schema("a0"), int32_schema("a1")}; - } else { - schema_vec = {int32_schema("a1"), int32_schema("a0")}; - } + 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>(schema_vec), 1); - ASSERT_EQ(tablet.add_timestamp(0, 500 + t), E_OK); - if (t < num_tablets - 1) { - ASSERT_EQ(tablet.add_value(0, 0, t), E_OK); // a0 - ASSERT_EQ(tablet.add_value(0, 1, 10 + t), E_OK); // a1 + 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, 19), E_OK); // a1 - ASSERT_EQ(tablet.add_value(0, 1, 9), E_OK); // a0 + ASSERT_EQ(tablet.add_value(0, 0, row), E_OK); + ASSERT_EQ(tablet.add_value(0, 1, 10 + row), E_OK); } - ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK); + ASSERT_EQ(writer_->write_tablet_aligned(tablet), E_OK); } - ASSERT_EQ(tsfile_writer_->flush(), E_OK); - ASSERT_EQ(tsfile_writer_->close(), E_OK); + ASSERT_EQ(writer_->flush(), E_OK); + ASSERT_EQ(writer_->close(), E_OK); - std::vector select_list; - for (const auto& name : {"a0", "a1"}) { - select_list.push_back(make_path(device, name)); - } - auto rows = query_all(select_list); - ASSERT_EQ(rows.size(), (size_t)num_tablets); - for (int t = 0; t < num_tablets - 1; t++) { - EXPECT_EQ(rows[t][1], std::to_string(t)); - EXPECT_EQ(rows[t][2], std::to_string(10 + t)); + 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)); } - // Reordered final tablet: a0=9, a1=19. - EXPECT_EQ(rows[num_tablets - 1][1], "9"); - EXPECT_EQ(rows[num_tablets - 1][2], "19"); + EXPECT_EQ(rows[3][1], "9"); + EXPECT_EQ(rows[3][2], "19"); } -// 5. Per-device caches are independent: two devices with identical -// measurement names, interleaved writes, different values. -TEST_F(SchemaCheckCacheTest, MultiDeviceCachesIndependent) { - const std::string devices[2] = {"root.cache_dev0", "root.cache_dev1"}; - for (const auto& device : devices) { - for (const auto& name : {"m0", "m1"}) { - ASSERT_EQ( - tsfile_writer_->register_timeseries(device, int32_schema(name)), - E_OK); - } +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 t = 0; t < 3; t++) { - for (int d = 0; d < 2; d++) { - std::vector schema_vec = {int32_schema("m0"), - int32_schema("m1")}; - Tablet tablet( - devices[d], - std::make_shared>(schema_vec), + 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(tablet.add_timestamp(0, 700 + t), E_OK); - // d*1000 separates the two devices' value spaces. - ASSERT_EQ(tablet.add_value(0, 0, d * 1000 + t), E_OK); - ASSERT_EQ(tablet.add_value(0, 1, d * 1000 + 10 + t), E_OK); - ASSERT_EQ(tsfile_writer_->write_tablet(tablet), E_OK); + 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(tsfile_writer_->flush(), E_OK); - ASSERT_EQ(tsfile_writer_->close(), E_OK); - - for (int d = 0; d < 2; d++) { - auto rows = query_all( - {make_path(devices[d], "m0"), make_path(devices[d], "m1")}); - ASSERT_EQ(rows.size(), (size_t)3); - for (int t = 0; t < 3; t++) { - EXPECT_EQ(rows[t][1], std::to_string(d * 1000 + t)); - EXPECT_EQ(rows[t][2], std::to_string(d * 1000 + 10 + t)); + + 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