From 4a11bc81d4b4bbad5ce1a0991354374b47cc15db Mon Sep 17 00:00:00 2001 From: ColinLee Date: Tue, 22 Sep 2026 17:55:41 +0800 Subject: [PATCH 1/2] fix(cpp): keep aligned value columns row-aligned when a row omits measurements Every value column of an aligned chunk group has to consume exactly one row per time column row. write_record_aligned() / write_tablet_aligned() only advanced the columns that the incoming record/tablet carried, so a measurement missing from a row left its column one row short: its not-null bitmap started at the wrong row index and the values were paired with the earliest timestamps of the page on read, and the statistics recomputed after recovery described the shifted rows. - create the value chunk writer when the measurement is registered on an aligned device, so every registered measurement takes part from row 0 - write NULL rows for the measurements a record/tablet does not carry (Java: AlignedChunkGroupWriterImpl#write -> writeEmptyDataInOneRow) - reject registering a new measurement once rows have been written, which would need backfilled rows/pages; Java does not allow expanding an aligned device either - ValuePageWriter::write_null_rows() / ValueChunkWriter::write_null_batch() advance the column by NULL rows and keep page boundaries in step with the time column - a record repeating a measurement now advances that column once (last point wins) instead of running ahead of the time column Tests: TsFileWriterTest.AlignedRecordMissingMeasurementsStayRowAligned, AlignedRecordMissingMeasurementsAcrossPages, AlignedTabletMissingColumnStaysRowAligned, AlignedRecordDuplicateMeasurementWritesOneRow, AlignedRegisterAfterWriteIsRejected and RestorableTsFileIOWriterTest.AlignedTimeseriesRecoverAndWriteNullValue. --- cpp/src/writer/tsfile_writer.cc | 213 +++++++--- cpp/src/writer/tsfile_writer.h | 8 + cpp/src/writer/value_chunk_writer.h | 35 ++ cpp/src/writer/value_page_writer.h | 20 + .../file/restorable_tsfile_io_writer_test.cc | 126 ++++++ cpp/test/writer/tsfile_writer_test.cc | 385 ++++++++++++++++++ 6 files changed, 740 insertions(+), 47 deletions(-) diff --git a/cpp/src/writer/tsfile_writer.cc b/cpp/src/writer/tsfile_writer.cc index 41a485c00..6c0ebacf6 100644 --- a/cpp/src/writer/tsfile_writer.cc +++ b/cpp/src/writer/tsfile_writer.cc @@ -21,6 +21,8 @@ #include #include +#include +#include #include "chunk_writer.h" #include "common/config/config.h" @@ -334,8 +336,30 @@ int TsFileWriter::register_timeseries(const std::string& device_path, std::make_shared(device_path); DeviceSchemasMapIter device_iter = schemas_.find(device_id); if (device_iter != schemas_.end()) { - MeasurementSchemaMap& msm = - device_iter->second->measurement_schema_map_; + MeasurementSchemaGroup* device_schema = device_iter->second; + MeasurementSchemaMap& msm = device_schema->measurement_schema_map_; + if (msm.find(measurement_schema->measurement_name_) != msm.end()) { + return E_ALREADY_EXIST; + } + if (device_schema->is_aligned_ && + device_schema->time_chunk_writer_ != nullptr && + device_schema->time_chunk_writer_->hasData()) { + // A column added now would have to be padded with the rows (and + // pages) that were already written, which the current page writer + // cannot express. Java does not allow an aligned device to be + // expanded at all; refuse loudly instead of silently writing a + // chunk group whose value column row counts diverge from the time + // column. + return E_INVALID_ARG; + } + // Aligned devices advance every registered measurement on every row, + // so the value chunk writer has to exist before the first row. + if (device_schema->is_aligned_) { + int ret = ensure_aligned_value_chunk_writer(measurement_schema); + if (RET_FAIL(ret)) { + return ret; + } + } MeasurementSchemaMapInsertResult ins_res = msm.insert(std::make_pair( measurement_schema->measurement_name_, measurement_schema)); if (UNLIKELY(!ins_res.second)) { @@ -344,6 +368,13 @@ int TsFileWriter::register_timeseries(const std::string& device_path, } else { MeasurementSchemaGroup* ms_group = new MeasurementSchemaGroup; ms_group->is_aligned_ = is_aligned; + if (is_aligned) { + int ret = ensure_aligned_value_chunk_writer(measurement_schema); + if (RET_FAIL(ret)) { + delete ms_group; + return ret; + } + } ms_group->measurement_schema_map_.insert(std::make_pair( measurement_schema->measurement_name_, measurement_schema)); schemas_.insert(std::make_pair(device_id, ms_group)); @@ -516,6 +547,26 @@ int TsFileWriter::do_check_schema( return ret; } +int TsFileWriter::ensure_aligned_value_chunk_writer( + MeasurementSchema* measurement_schema) { + if (measurement_schema->value_chunk_writer_ != nullptr) { + return E_OK; + } + ValueChunkWriter* value_chunk_writer = new ValueChunkWriter; + if (IS_NULL(value_chunk_writer)) { + return E_OOM; + } + int ret = value_chunk_writer->init( + measurement_schema->measurement_name_, measurement_schema->data_type_, + measurement_schema->encoding_, measurement_schema->compression_type_); + if (RET_FAIL(ret)) { + delete value_chunk_writer; + return (ret == E_OOM) ? ret : common::E_INVALID_ARG; + } + measurement_schema->value_chunk_writer_ = value_chunk_writer; + return E_OK; +} + template int TsFileWriter::do_check_schema_aligned( std::shared_ptr device_id, @@ -548,28 +599,11 @@ int TsFileWriter::do_check_schema_aligned( // 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_); + if (RET_FAIL(ensure_aligned_value_chunk_writer(ms))) { + value_chunk_writers.push_back(NULL); + return common::E_INVALID_ARG; } + value_chunk_writers.push_back(ms->value_chunk_writer_); data_types.push_back(ms->data_type_); } } @@ -808,15 +842,50 @@ int TsFileWriter::write_record_aligned(const TsRecord& record) { if (value_chunk_writers.size() != record.points_.size()) { return E_INVALID_ARG; } + DeviceSchemasMapIter dev_it = schemas_.find(device_id); + if (UNLIKELY(dev_it == schemas_.end()) || IS_NULL(dev_it->second)) { + return E_DEVICE_NOT_EXIST; + } + MeasurementSchemaGroup* device_schema = dev_it->second; + // Index of the point that carries each measurement of this row. A + // duplicate point for the same measurement keeps the last one, so that + // every value column advances exactly one row per timestamp. + std::map row_point_index; + for (uint32_t c = 0; c < record.points_.size(); c++) { + row_point_index[record.points_[c].measurement_name_] = c; + } + // A record may only carry a subset of the device's measurements. The + // measurements it does not mention still have to advance one row (as a + // NULL) so that every value column's not-null bitmap stays aligned with + // the time column; otherwise the values of that column end up paired with + // the wrong timestamps on read. Java behaves the same way + // (AlignedChunkGroupWriterImpl#write -> writeEmptyDataInOneRow). + SimpleVector absent_schemas; + SimpleVector row_writers; + for (uint32_t c = 0; c < value_chunk_writers.size(); c++) { + if (!IS_NULL(value_chunk_writers[c])) { + row_writers.push_back(value_chunk_writers[c]); + } + } + MeasurementSchemaMap& msm = device_schema->measurement_schema_map_; + for (MeasurementSchemaMapIter ms_iter = msm.begin(); ms_iter != msm.end(); + ms_iter++) { + if (row_point_index.find(ms_iter->first) != row_point_index.end()) { + continue; + } + MeasurementSchema* ms = ms_iter->second; + if (RET_FAIL(ensure_aligned_value_chunk_writer(ms))) { + return ret; + } + absent_schemas.push_back(ms); + row_writers.push_back(ms->value_chunk_writer_); + } // Snapshot page counters before the write so we can detect any column // that crossed a page boundary and seal the rest in lockstep. int32_t time_pages_before = time_chunk_writer->num_of_pages(); - std::vector value_pages_before(value_chunk_writers.size(), 0); - for (uint32_t c = 0; c < value_chunk_writers.size(); c++) { - ValueChunkWriter* value_chunk_writer = value_chunk_writers[c]; - if (!IS_NULL(value_chunk_writer)) { - value_pages_before[c] = value_chunk_writer->num_of_pages(); - } + std::vector value_pages_before(row_writers.size(), 0); + for (uint32_t c = 0; c < row_writers.size(); c++) { + value_pages_before[c] = row_writers[c]->num_of_pages(); } // Time first: a rejected timestamp (E_OUT_OF_ORDER, OOM, etc.) must // not silently advance the value writers — that would leave the time @@ -829,8 +898,14 @@ int TsFileWriter::write_record_aligned(const TsRecord& record) { if (IS_NULL(value_chunk_writer)) { continue; } + const DataPoint& point = record.points_[c]; + if (row_point_index[point.measurement_name_] != c) { + // Duplicate point for this measurement: skipped, the last one is + // written below (one row per column per timestamp). + continue; + } if (RET_FAIL(write_point_aligned(value_chunk_writer, record.timestamp_, - data_types[c], record.points_[c]))) { + data_types[c], point))) { // Time wrote the row but at least one value column failed // mid-record; the per-column row counts no longer agree. // Mark the writer unrecoverable so flush/close refuses to @@ -839,8 +914,15 @@ int TsFileWriter::write_record_aligned(const TsRecord& record) { return ret; } } + for (uint32_t c = 0; c < absent_schemas.size(); c++) { + if (RET_FAIL( + absent_schemas[c]->value_chunk_writer_->write_null_batch(1))) { + unrecoverable_ = true; + return ret; + } + } if (RET_FAIL(maybe_seal_aligned_pages_together( - time_chunk_writer, value_chunk_writers, time_pages_before, + time_chunk_writer, row_writers, time_pages_before, value_pages_before))) { unrecoverable_ = true; return ret; @@ -984,15 +1066,50 @@ int TsFileWriter::write_tablet_aligned(const Tablet& tablet) { return E_TYPE_NOT_MATCH; } } + DeviceSchemasMapIter dev_it = schemas_.find(device_id); + if (UNLIKELY(dev_it == schemas_.end()) || IS_NULL(dev_it->second)) { + return E_DEVICE_NOT_EXIST; + } + MeasurementSchemaGroup* device_schema = dev_it->second; + // Measurements of this aligned device that the tablet does not carry still + // have to advance one row per tablet row (as NULLs), otherwise their + // not-null bitmap would start at the wrong row index and their values + // would be paired with the wrong timestamps on read. Java fills those + // rows in AlignedChunkGroupWriterImpl#write(Tablet). + SimpleVector absent_writers; + std::set tablet_measurements; + for (size_t c = 0; c < tablet.get_column_count(); c++) { + tablet_measurements.insert(tablet.schema_vec_->at(c).measurement_name_); + } + MeasurementSchemaMap& msm = device_schema->measurement_schema_map_; + for (MeasurementSchemaMapIter ms_iter = msm.begin(); ms_iter != msm.end(); + ms_iter++) { + if (tablet_measurements.find(ms_iter->first) != + tablet_measurements.end()) { + continue; + } + if (RET_FAIL(ensure_aligned_value_chunk_writer(ms_iter->second))) { + return ret; + } + absent_writers.push_back(ms_iter->second->value_chunk_writer_); + } + // Every column that takes part in this batch: the tablet's own columns + // plus the ones that only advance with NULL rows. + SimpleVector row_writers; + for (uint32_t c = 0; c < value_chunk_writers.size(); c++) { + if (!IS_NULL(value_chunk_writers[c])) { + row_writers.push_back(value_chunk_writers[c]); + } + } + for (uint32_t c = 0; c < absent_writers.size(); c++) { + row_writers.push_back(absent_writers[c]); + } // Snapshot page counters before the batch so we can detect any column // that crossed a page boundary mid-tablet and seal the rest in lockstep. int32_t time_pages_before = time_chunk_writer->num_of_pages(); - std::vector value_pages_before(value_chunk_writers.size(), 0); - for (uint32_t c = 0; c < value_chunk_writers.size(); c++) { - ValueChunkWriter* value_chunk_writer = value_chunk_writers[c]; - if (!IS_NULL(value_chunk_writer)) { - value_pages_before[c] = value_chunk_writer->num_of_pages(); - } + std::vector value_pages_before(row_writers.size(), 0); + for (uint32_t c = 0; c < row_writers.size(); c++) { + value_pages_before[c] = row_writers[c]->num_of_pages(); } // Suppress memory-driven page sealing on every column for the duration of // the batch. The count-driven seals inside write_batch still fire at the @@ -1004,18 +1121,13 @@ int TsFileWriter::write_tablet_aligned(const Tablet& tablet) { // (e.g. when a sealed value column ended a page that the time column did // not). time_chunk_writer->set_enable_page_seal_if_full(false); - for (uint32_t c = 0; c < value_chunk_writers.size(); c++) { - ValueChunkWriter* value_chunk_writer = value_chunk_writers[c]; - if (!IS_NULL(value_chunk_writer)) { - value_chunk_writer->set_enable_page_seal_if_full(false); - } + for (uint32_t c = 0; c < row_writers.size(); c++) { + row_writers[c]->set_enable_page_seal_if_full(false); } auto restore_seal = [&]() { time_chunk_writer->set_enable_page_seal_if_full(true); - for (uint32_t k = 0; k < value_chunk_writers.size(); k++) { - if (!IS_NULL(value_chunk_writers[k])) { - value_chunk_writers[k]->set_enable_page_seal_if_full(true); - } + for (uint32_t k = 0; k < row_writers.size(); k++) { + row_writers[k]->set_enable_page_seal_if_full(true); } }; // Any failure (out-of-order timestamps, OOM, etc.) must abort before we @@ -1043,9 +1155,16 @@ int TsFileWriter::write_tablet_aligned(const Tablet& tablet) { return ret; } } + for (uint32_t c = 0; c < absent_writers.size(); c++) { + if (RET_FAIL(absent_writers[c]->write_null_batch(total_rows))) { + restore_seal(); + unrecoverable_ = true; + return ret; + } + } restore_seal(); if (RET_FAIL(maybe_seal_aligned_pages_together( - time_chunk_writer, value_chunk_writers, time_pages_before, + time_chunk_writer, row_writers, time_pages_before, value_pages_before))) { unrecoverable_ = true; return ret; diff --git a/cpp/src/writer/tsfile_writer.h b/cpp/src/writer/tsfile_writer.h index 55e9e7f3a..0737eb065 100644 --- a/cpp/src/writer/tsfile_writer.h +++ b/cpp/src/writer/tsfile_writer.h @@ -128,6 +128,14 @@ class TsFileWriter { int write_point_aligned(ValueChunkWriter* value_chunk_writer, int64_t timestamp, common::TSDataType data_type, const DataPoint& point); + /* + * Create (once) the value chunk writer that carries one measurement of an + * aligned device. Aligned devices keep every registered measurement in + * lock-step with the time column, so the writer has to exist before the + * first row of that device is written. + */ + int ensure_aligned_value_chunk_writer( + storage::MeasurementSchema* measurement_schema); int maybe_seal_aligned_pages_together( TimeChunkWriter* time_chunk_writer, common::SimpleVector& value_chunk_writers, diff --git a/cpp/src/writer/value_chunk_writer.h b/cpp/src/writer/value_chunk_writer.h index cd7c75a54..88871c1a9 100644 --- a/cpp/src/writer/value_chunk_writer.h +++ b/cpp/src/writer/value_chunk_writer.h @@ -174,6 +174,41 @@ class ValueChunkWriter { return ret; } + /** + * Advance this column by `count` all-NULL rows. + * + * Aligned chunk groups keep every value column row-aligned with the time + * column, so a column whose row has no value still has to consume one row + * (with a null bit) for that timestamp. Page boundaries follow the same + * `page_writer_max_point_num_` rule as write_batch() so the page lists of + * the time column and of every value column stay in step. + */ + int write_null_batch(uint32_t count) { + int ret = common::E_OK; + uint32_t offset = 0; + const uint32_t page_cap = + common::g_config_value_.page_writer_max_point_num_; + while (offset < count) { + uint32_t cur_points = value_page_writer_.get_point_numer(); + if (cur_points >= page_cap) { + if (RET_FAIL(seal_cur_page(false))) { + return ret; + } + cur_points = 0; + } + uint32_t batch_size = + std::min(count - offset, page_cap - cur_points); + if (RET_FAIL(value_page_writer_.write_null_rows(batch_size))) { + return ret; + } + offset += batch_size; + if (RET_FAIL(seal_cur_page_if_full())) { + return ret; + } + } + return ret; + } + int end_encode_chunk(); common::ByteStream& get_chunk_data() { return chunk_data_; } Statistic* get_chunk_statistic() { return chunk_statistic_; } diff --git a/cpp/src/writer/value_page_writer.h b/cpp/src/writer/value_page_writer.h index 92c39b9b2..61bf59564 100644 --- a/cpp/src/writer/value_page_writer.h +++ b/cpp/src/writer/value_page_writer.h @@ -335,6 +335,26 @@ class ValuePageWriter { value_out_stream_.allocated_bytes()) + value_encoder_->get_max_byte_size(); } + /** + * Append `count` all-NULL rows to the current page: one zero bit per row + * in the not-null bitmap, no value bytes, no statistic update. + * + * Used to keep a value column row-aligned with the time column when a row + * carries no value for that column. The null bit has to be recorded for + * that row, otherwise the column's bitmap would start at the wrong row + * index and every later value would be paired with the wrong timestamp on + * read. + */ + int write_null_rows(uint32_t count) { + for (uint32_t i = 0; i < count; i++) { + if ((size_ / 8) + 1 > col_notnull_bitmap_.size()) { + col_notnull_bitmap_.push_back(0); + } + size_++; + } + return common::E_OK; + } + int write_to_chunk(common::ByteStream& pages_data, bool write_header, bool write_statistic, bool write_data_to_chunk_data); FORCE_INLINE common::ByteStream& get_col_notnull_bitmap_data() { diff --git a/cpp/test/file/restorable_tsfile_io_writer_test.cc b/cpp/test/file/restorable_tsfile_io_writer_test.cc index c60a855c5..137f24f08 100644 --- a/cpp/test/file/restorable_tsfile_io_writer_test.cc +++ b/cpp/test/file/restorable_tsfile_io_writer_test.cc @@ -1061,3 +1061,129 @@ TEST_F(RestorableTsFileIOWriterTest, RecoveryAlignedSparseStatRespectsBitmap) { } EXPECT_TRUE(found_value_chunk); } + +// Sparse aligned records (a row only carries a subset of the device's +// measurements, without explicit NULL DataPoints) must survive a crash + +// recovery: the recovered chunk statistics have to describe the rows that +// really carry a value, and continued sparse writes have to stay row-aligned +// with the time column. +TEST_F(RestorableTsFileIOWriterTest, + AlignedTimeseriesRecoverAndWriteNullValue) { + using namespace std; + const string device = "d1"; + // even rows carry s1..s3, odd rows carry s4..s6 + vector even_names = {"s1", "s2", "s3"}; + vector odd_names = {"s4", "s5", "s6"}; + { + TsFileWriter tw; + ASSERT_EQ(tw.open(file_name_, GetWriteCreateFlags(), 0666), E_OK); + std::vector schemas; + schemas.push_back(new MeasurementSchema("s1", BOOLEAN)); + schemas.push_back(new MeasurementSchema("s2", INT32)); + schemas.push_back(new MeasurementSchema("s3", TEXT)); + schemas.push_back(new MeasurementSchema("s4", INT64)); + schemas.push_back(new MeasurementSchema("s5", FLOAT)); + schemas.push_back(new MeasurementSchema("s6", STRING)); + ASSERT_EQ(tw.register_aligned_timeseries(device, schemas), E_OK); + for (int i = 0; i < 10; i++) { + TsRecord record(i, device); + if (i % 2 == 0) { + record.add_point(even_names[0], true); + record.add_point(even_names[1], static_cast(i)); + record.add_point(even_names[2], "even"); + } else { + record.add_point(odd_names[0], static_cast(i)); + record.add_point(odd_names[1], static_cast(i)); + record.add_point(odd_names[2], "odd"); + } + ASSERT_EQ(tw.write_record_aligned(record), E_OK); + } + ASSERT_EQ(tw.flush(), E_OK); + ASSERT_EQ(tw.close(), E_OK); + } + + CorruptCurrentFileTail(3); + + RestorableTsFileIOWriter rw; + ASSERT_EQ(rw.open(file_name_, true), E_OK); + ASSERT_TRUE(rw.can_write()); + { + // Keep writing sparse rows after the recovery point. + TsFileTreeWriter tw2(&rw); + for (int i = 10; i < 20; i++) { + TsRecord record(i, device); + if (i % 2 == 0) { + record.add_point(even_names[0], true); + record.add_point(even_names[1], static_cast(i)); + record.add_point(even_names[2], "even"); + } else { + record.add_point(odd_names[0], static_cast(i)); + record.add_point(odd_names[1], static_cast(i)); + record.add_point(odd_names[2], "odd"); + } + ASSERT_EQ(tw2.write(record), E_OK); + } + ASSERT_EQ(tw2.flush(), E_OK); + ASSERT_EQ(tw2.close(), E_OK); + } + + TsFileTreeReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + DeviceTimeseriesMetadataMap metadata = reader.get_timeseries_metadata(); + std::map meta_by_name; + for (auto& entry : metadata) { + for (auto& ts_idx : entry.second) { + meta_by_name[ts_idx->get_measurement_name().to_std_string()] = + ts_idx.get(); + } + } + ASSERT_EQ(meta_by_name.size(), 6u); + // Columns carried by the even rows: 5 points before the crash + 5 after, + // spanning timestamp 0..18. + for (auto& name : even_names) { + EXPECT_EQ(meta_by_name[name]->get_statistic()->count_, 10); + EXPECT_EQ(meta_by_name[name]->get_statistic()->start_time_, 0); + EXPECT_EQ(meta_by_name[name]->get_statistic()->end_time_, 18); + } + // Columns carried by the odd rows: first value at timestamp 1, last at 19. + for (auto& name : odd_names) { + EXPECT_EQ(meta_by_name[name]->get_statistic()->count_, 10); + EXPECT_EQ(meta_by_name[name]->get_statistic()->start_time_, 1); + EXPECT_EQ(meta_by_name[name]->get_statistic()->end_time_, 19); + } + + vector measurement_names = {"s1", "s2", "s3", "s4", "s5", "s6"}; + ASSERT_EQ(CountTreeReaderRows(reader, measurement_names), 20); + + ResultSet* result_set = nullptr; + vector device_ids = {device}; + ASSERT_EQ(reader.query(device_ids, measurement_names, 0, 100, result_set), + E_OK); + auto it = result_set->iterator(); + int row = 0; + while (it.hasNext()) { + RowRecord* rec = it.next(); + ASSERT_NE(rec, nullptr); + EXPECT_EQ(rec->get_timestamp(), row); + for (int c = 0; c < 3; c++) { + Field* field = rec->get_field(c + 1); + if (row % 2 == 0) { + EXPECT_NE(field->type_, common::NULL_TYPE); + } else { + EXPECT_EQ(field->type_, common::NULL_TYPE); + } + } + for (int c = 3; c < 6; c++) { + Field* field = rec->get_field(c + 1); + if (row % 2 == 0) { + EXPECT_EQ(field->type_, common::NULL_TYPE); + } else { + EXPECT_NE(field->type_, common::NULL_TYPE); + } + } + row++; + } + EXPECT_EQ(row, 20); + reader.destroy_query_data_set(result_set); + reader.close(); +} diff --git a/cpp/test/writer/tsfile_writer_test.cc b/cpp/test/writer/tsfile_writer_test.cc index 3b9dae92a..ef9527ff9 100644 --- a/cpp/test/writer/tsfile_writer_test.cc +++ b/cpp/test/writer/tsfile_writer_test.cc @@ -1737,3 +1737,388 @@ TEST_F(TsFileWriterTest, WriterReuseAfterDestroyProducesValidSecondFile) { delete wf; remove(second_path.c_str()); } + +// --------------------------------------------------------------------------- +// Aligned writes: a row that does not carry every measurement of the device +// must not shift the value columns. +// +// Records and tablets may legitimately carry only a subset of an aligned +// device's measurements. A measurement missing from the row still has to +// consume one row (as NULL) so that every value column's not-null bitmap stays +// aligned with the time column; otherwise the values of that column get paired +// with the earliest timestamps of the page on read (Java behaves the same way: +// AlignedChunkGroupWriterImpl#write -> writeEmptyDataInOneRow). +// --------------------------------------------------------------------------- + +TEST_F(TsFileWriterTest, AlignedRecordMissingMeasurementsStayRowAligned) { + std::string device_name = "device_missing_m"; + std::vector mnames = {"s0", "s1", "s2"}; + std::vector schemas; + for (auto& n : mnames) { + schemas.push_back(new MeasurementSchema(n, INT64, PLAIN, UNCOMPRESSED)); + } + tsfile_writer_->register_aligned_timeseries(device_name, schemas); + + const int row_num = 10; + for (int i = 0; i < row_num; i++) { + TsRecord record(1622505600000 + i, device_name); + if (i % 2 == 0) { + // Only s0 is carried by even rows, only s1 by odd rows; s2 is + // never written at all. + record.add_point(mnames[0], static_cast(100 + i)); + } else { + record.add_point(mnames[1], static_cast(200 + i)); + } + ASSERT_EQ(tsfile_writer_->write_record_aligned(record), E_OK); + } + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + // Statistics must describe the rows that really carry a value. + std::vector> devices = { + std::make_shared(device_name)}; + TsFileReader meta_reader; + ASSERT_EQ(meta_reader.open(file_name_), E_OK); + auto meta_map = meta_reader.get_timeseries_metadata(devices); + std::map meta_by_name; + for (auto& ts_idx : meta_map.at(devices[0])) { + meta_by_name[ts_idx->get_measurement_name().to_std_string()] = + ts_idx.get(); + } + // s2 is registered but never carries a value: it advances with NULL rows + // (like an all-null column of an aligned tablet), so it is present with + // count 0. + ASSERT_EQ(meta_by_name.size(), 3u); + EXPECT_EQ(meta_by_name[mnames[0]]->get_statistic()->count_, 5); + EXPECT_EQ(meta_by_name[mnames[0]]->get_statistic()->start_time_, + 1622505600000); + EXPECT_EQ(meta_by_name[mnames[0]]->get_statistic()->end_time_, + 1622505600008); + EXPECT_EQ(meta_by_name[mnames[1]]->get_statistic()->count_, 5); + EXPECT_EQ(meta_by_name[mnames[1]]->get_statistic()->start_time_, + 1622505600001); + EXPECT_EQ(meta_by_name[mnames[1]]->get_statistic()->end_time_, + 1622505600009); + EXPECT_EQ(meta_by_name[mnames[2]]->get_statistic()->count_, 0); + ASSERT_EQ(meta_reader.close(), E_OK); + + std::vector select_list; + for (auto& n : mnames) { + select_list.emplace_back(device_name, n); + } + storage::QueryExpression* qe = + storage::QueryExpression::create(select_list, nullptr); + storage::TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + storage::ResultSet* tmp_qds = nullptr; + ASSERT_EQ(reader.query(qe, tmp_qds), E_OK); + auto* qds = (QDSWithoutTimeGenerator*)tmp_qds; + + bool has_next = false; + int64_t cur_row = 0; + while (IS_SUCC(qds->next(has_next)) && has_next) { + auto* rec = qds->get_row_record(); + ASSERT_NE(rec, nullptr); + EXPECT_EQ(rec->get_timestamp(), 1622505600000 + cur_row); + const std::string s0 = field_to_string(rec->get_field(1)); + const std::string s1 = field_to_string(rec->get_field(2)); + const std::string s2 = field_to_string(rec->get_field(3)); + if (cur_row % 2 == 0) { + EXPECT_EQ(s0, std::to_string(100 + cur_row)); + EXPECT_EQ(s1, "NULL"); + } else { + EXPECT_EQ(s0, "NULL"); + EXPECT_EQ(s1, std::to_string(200 + cur_row)); + } + EXPECT_EQ(s2, "NULL"); + cur_row++; + } + EXPECT_EQ(cur_row, row_num); + reader.destroy_query_data_set(qds); + ASSERT_EQ(reader.close(), E_OK); +} + +TEST_F(TsFileWriterTest, AlignedTabletMissingColumnStaysRowAligned) { + std::string device_name = "device_tablet_missing"; + std::vector schema_vec; + schema_vec.emplace_back("s0", INT64, PLAIN, UNCOMPRESSED); + schema_vec.emplace_back("s1", INT64, PLAIN, UNCOMPRESSED); + { + std::vector reg; + for (auto& s : schema_vec) { + reg.push_back(new MeasurementSchema(s)); + } + tsfile_writer_->register_aligned_timeseries(device_name, reg); + } + { + // First tablet only carries s0: s1 must still advance with NULLs. + std::vector cols; + cols.push_back(schema_vec[0]); + Tablet tablet(device_name, + std::make_shared>(cols), + 5); + for (int i = 0; i < 5; i++) { + tablet.add_timestamp(i, 1000 + i); + tablet.add_value(i, 0u, static_cast(100 + i)); + } + ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK); + } + { + Tablet tablet( + device_name, + std::make_shared>(schema_vec), 5); + for (int i = 0; i < 5; i++) { + tablet.add_timestamp(i, 1005 + i); + tablet.add_value(i, 0u, static_cast(105 + i)); + tablet.add_value(i, 1u, static_cast(200 + i)); + } + 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::string s0_name("s0"), s1_name("s1"); + std::vector select_list; + select_list.emplace_back(device_name, s0_name); + select_list.emplace_back(device_name, s1_name); + storage::QueryExpression* qe = + storage::QueryExpression::create(select_list, nullptr); + storage::TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + storage::ResultSet* tmp_qds = nullptr; + ASSERT_EQ(reader.query(qe, tmp_qds), E_OK); + auto* qds = (QDSWithoutTimeGenerator*)tmp_qds; + + bool has_next = false; + int64_t cur_row = 0; + while (IS_SUCC(qds->next(has_next)) && has_next) { + auto* rec = qds->get_row_record(); + ASSERT_NE(rec, nullptr); + EXPECT_EQ(rec->get_timestamp(), 1000 + cur_row); + EXPECT_EQ(field_to_string(rec->get_field(1)), + std::to_string(100 + cur_row)); + if (cur_row < 5) { + // The rows written before s1 showed up are NULL for s1. + EXPECT_EQ(field_to_string(rec->get_field(2)), "NULL"); + } else { + EXPECT_EQ(field_to_string(rec->get_field(2)), + std::to_string(200 + cur_row - 5)); + } + cur_row++; + } + EXPECT_EQ(cur_row, 10); + reader.destroy_query_data_set(qds); + ASSERT_EQ(reader.close(), E_OK); +} + +// A record that repeats the same measurement must still advance that column a +// single time, otherwise the column would run ahead of the time column. +TEST_F(TsFileWriterTest, AlignedRecordDuplicateMeasurementWritesOneRow) { + std::string device_name = "device_dup_m"; + std::vector schemas; + schemas.push_back(new MeasurementSchema("s0", INT64, PLAIN, UNCOMPRESSED)); + schemas.push_back(new MeasurementSchema("s1", INT64, PLAIN, UNCOMPRESSED)); + tsfile_writer_->register_aligned_timeseries(device_name, schemas); + + TsRecord record(7, device_name); + record.add_point("s0", static_cast(1)); + record.add_point("s0", static_cast(2)); + record.add_point("s1", static_cast(3)); + ASSERT_EQ(tsfile_writer_->write_record_aligned(record), E_OK); + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + std::string s0_name("s0"), s1_name("s1"); + std::vector select_list; + select_list.emplace_back(device_name, s0_name); + select_list.emplace_back(device_name, s1_name); + storage::QueryExpression* qe = + storage::QueryExpression::create(select_list, nullptr); + storage::TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + storage::ResultSet* tmp_qds = nullptr; + ASSERT_EQ(reader.query(qe, tmp_qds), E_OK); + auto* qds = (QDSWithoutTimeGenerator*)tmp_qds; + + bool has_next = false; + int rows = 0; + while (IS_SUCC(qds->next(has_next)) && has_next) { + auto* rec = qds->get_row_record(); + ASSERT_NE(rec, nullptr); + EXPECT_EQ(rec->get_timestamp(), 7); + // The last point of the duplicated measurement wins. + EXPECT_EQ(field_to_string(rec->get_field(1)), "2"); + EXPECT_EQ(field_to_string(rec->get_field(2)), "3"); + rows++; + } + EXPECT_EQ(rows, 1); + reader.destroy_query_data_set(qds); + ASSERT_EQ(reader.close(), E_OK); +} + +// New columns have to be padded with the rows (and pages) that were already +// written, which the aligned writer cannot express: reject the registration +// instead of writing a chunk group whose value column is shifted. +TEST_F(TsFileWriterTest, AlignedRegisterAfterWriteIsRejected) { + std::string device_name = "device_late_m"; + std::vector schemas; + schemas.push_back(new MeasurementSchema("s0", INT64, PLAIN, UNCOMPRESSED)); + tsfile_writer_->register_aligned_timeseries(device_name, schemas); + + TsRecord record(1, device_name); + record.add_point("s0", static_cast(1)); + ASSERT_EQ(tsfile_writer_->write_record_aligned(record), E_OK); + + std::vector extra; + extra.push_back(new MeasurementSchema("s1", INT64, PLAIN, UNCOMPRESSED)); + EXPECT_EQ(tsfile_writer_->register_aligned_timeseries(device_name, extra), + E_INVALID_ARG); + + // A second registration of the same measurement is still reported as a + // duplicate, not as a late registration. + std::vector dup; + dup.push_back(new MeasurementSchema("s0", INT64, PLAIN, UNCOMPRESSED)); + EXPECT_EQ(tsfile_writer_->register_aligned_timeseries(device_name, dup), + E_ALREADY_EXIST); + + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); +} + +// Same as above, but with rows spread over several pages: the NULL padding of +// a missing measurement has to keep the page lists of every value column in +// step with the time column. +TEST_F(TsFileWriterTest, AlignedRecordMissingMeasurementsAcrossPages) { + uint32_t prev_pt = g_config_value_.page_writer_max_point_num_; + uint32_t prev_mem = g_config_value_.page_writer_max_memory_bytes_; + struct Guard { + uint32_t pt, mem; + ~Guard() { + g_config_value_.page_writer_max_point_num_ = pt; + g_config_value_.page_writer_max_memory_bytes_ = mem; + } + } guard{prev_pt, prev_mem}; + g_config_value_.page_writer_max_point_num_ = 7; + g_config_value_.page_writer_max_memory_bytes_ = 1024 * 1024; + + std::string device_name = "device_missing_pages"; + std::vector mnames = {"s0", "s1", "s2"}; + std::vector schemas; + for (auto& n : mnames) { + schemas.push_back(new MeasurementSchema(n, INT64, PLAIN, UNCOMPRESSED)); + } + tsfile_writer_->register_aligned_timeseries(device_name, schemas); + + const int row_num = 20; + for (int i = 0; i < row_num; i++) { + TsRecord record(1000 + i, device_name); + // s0: every row, s1: every third row, s2: only the last row. + record.add_point(mnames[0], static_cast(i)); + if (i % 3 == 0) { + record.add_point(mnames[1], static_cast(100 + i)); + } + if (i == row_num - 1) { + record.add_point(mnames[2], static_cast(999)); + } + ASSERT_EQ(tsfile_writer_->write_record_aligned(record), E_OK); + } + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + std::vector select_list; + for (auto& n : mnames) { + select_list.emplace_back(device_name, n); + } + storage::QueryExpression* qe = + storage::QueryExpression::create(select_list, nullptr); + storage::TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + storage::ResultSet* tmp_qds = nullptr; + ASSERT_EQ(reader.query(qe, tmp_qds), E_OK); + auto* qds = (QDSWithoutTimeGenerator*)tmp_qds; + + bool has_next = false; + int64_t cur_row = 0; + while (IS_SUCC(qds->next(has_next)) && has_next) { + auto* rec = qds->get_row_record(); + ASSERT_NE(rec, nullptr); + EXPECT_EQ(rec->get_timestamp(), 1000 + cur_row); + EXPECT_EQ(field_to_string(rec->get_field(1)), std::to_string(cur_row)); + if (cur_row % 3 == 0) { + EXPECT_EQ(field_to_string(rec->get_field(2)), + std::to_string(100 + cur_row)); + } else { + EXPECT_EQ(field_to_string(rec->get_field(2)), "NULL"); + } + if (cur_row == row_num - 1) { + EXPECT_EQ(field_to_string(rec->get_field(3)), "999"); + } else { + EXPECT_EQ(field_to_string(rec->get_field(3)), "NULL"); + } + cur_row++; + } + EXPECT_EQ(cur_row, row_num); + reader.destroy_query_data_set(qds); + ASSERT_EQ(reader.close(), E_OK); +} + +// A value column is identified by measurement name, not by the position of the +// point inside the record: records may add their points in any order (and in a +// different order from row to row) without disturbing the row alignment. +TEST_F(TsFileWriterTest, AlignedRecordPointOrderDoesNotMatter) { + std::string device_name = "device_point_order"; + std::vector mnames = {"s0", "s1", "s2"}; + std::vector schemas; + for (auto& n : mnames) { + schemas.push_back(new MeasurementSchema(n, INT64, PLAIN, UNCOMPRESSED)); + } + tsfile_writer_->register_aligned_timeseries(device_name, schemas); + + const int row_num = 6; + for (int i = 0; i < row_num; i++) { + TsRecord record(2000 + i, device_name); + // Reverse order of the previous row, and drop s1 on odd rows. + for (int k = 2; k >= 0; k--) { + int idx = (i + k) % 3; + if (idx == 1 && i % 2 == 1) { + continue; + } + record.add_point(mnames[idx], static_cast(idx * 100 + i)); + } + ASSERT_EQ(tsfile_writer_->write_record_aligned(record), E_OK); + } + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + std::vector select_list; + for (auto& n : mnames) { + select_list.emplace_back(device_name, n); + } + storage::QueryExpression* qe = + storage::QueryExpression::create(select_list, nullptr); + storage::TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + storage::ResultSet* tmp_qds = nullptr; + ASSERT_EQ(reader.query(qe, tmp_qds), E_OK); + auto* qds = (QDSWithoutTimeGenerator*)tmp_qds; + + bool has_next = false; + int64_t cur_row = 0; + while (IS_SUCC(qds->next(has_next)) && has_next) { + auto* rec = qds->get_row_record(); + ASSERT_NE(rec, nullptr); + EXPECT_EQ(rec->get_timestamp(), 2000 + cur_row); + for (int c = 0; c < 3; c++) { + if (c == 1 && cur_row % 2 == 1) { + EXPECT_EQ(field_to_string(rec->get_field(c + 1)), "NULL"); + } else { + EXPECT_EQ(field_to_string(rec->get_field(c + 1)), + std::to_string(c * 100 + cur_row)); + } + } + cur_row++; + } + EXPECT_EQ(cur_row, row_num); + reader.destroy_query_data_set(qds); + ASSERT_EQ(reader.close(), E_OK); +} From e5425e4fe1c3a8ed25df582e36a433783c32f90b Mon Sep 17 00:00:00 2001 From: ColinLee Date: Wed, 23 Sep 2026 10:58:04 +0800 Subject: [PATCH 2/2] test(cpp): release schemas whose aligned registration was rejected The writer only takes ownership of a MeasurementSchema when the registration succeeds, so the two schemas used to provoke E_INVALID_ARG / E_ALREADY_EXIST have to be released by the test. LeakSanitizer flagged them in the ASan jobs of PR #968 (208 bytes in 2 allocations). --- cpp/test/writer/tsfile_writer_test.cc | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/cpp/test/writer/tsfile_writer_test.cc b/cpp/test/writer/tsfile_writer_test.cc index ef9527ff9..b89ab7059 100644 --- a/cpp/test/writer/tsfile_writer_test.cc +++ b/cpp/test/writer/tsfile_writer_test.cc @@ -1969,17 +1969,23 @@ TEST_F(TsFileWriterTest, AlignedRegisterAfterWriteIsRejected) { record.add_point("s0", static_cast(1)); ASSERT_EQ(tsfile_writer_->write_record_aligned(record), E_OK); - std::vector extra; - extra.push_back(new MeasurementSchema("s1", INT64, PLAIN, UNCOMPRESSED)); + // The writer only takes ownership of a schema when the registration + // succeeds, so a rejected one has to be released by the caller. + MeasurementSchema* extra_schema = + new MeasurementSchema("s1", INT64, PLAIN, UNCOMPRESSED); + std::vector extra{extra_schema}; EXPECT_EQ(tsfile_writer_->register_aligned_timeseries(device_name, extra), E_INVALID_ARG); + delete extra_schema; // A second registration of the same measurement is still reported as a // duplicate, not as a late registration. - std::vector dup; - dup.push_back(new MeasurementSchema("s0", INT64, PLAIN, UNCOMPRESSED)); + MeasurementSchema* dup_schema = + new MeasurementSchema("s0", INT64, PLAIN, UNCOMPRESSED); + std::vector dup{dup_schema}; EXPECT_EQ(tsfile_writer_->register_aligned_timeseries(device_name, dup), E_ALREADY_EXIST); + delete dup_schema; ASSERT_EQ(tsfile_writer_->flush(), E_OK); ASSERT_EQ(tsfile_writer_->close(), E_OK);