From 26139ba57ba4eaf1445cd7dece72f31b05d2cc8e Mon Sep 17 00:00:00 2001 From: gx Date: Tue, 18 Aug 2026 20:19:33 +0800 Subject: [PATCH 1/6] fix(cpp): never seal empty chunks and fix dangling ref in parallel tablet write Two fixes in TsFileWriter: 1. flush_chunk_group / flush_chunk_group_encoded: skip registered-but-empty measurement columns. A measurement that received no data in a window used to be sealed as an EMPTY chunk (count=0, dataSize=0). Java readers (TsFileSequenceReader self-check) treat such a file as crashed and refuse to load it. Mirror the aligned branch's existing hasData() check so empty columns never produce a chunk. 2. write_table (aligned parallel path): the submitted tasks run asynchronously on the thread pool, but the lambdas captured the loop variables (ctx, vt) by reference. Once the loop advances, every queued task reads the same / already-destroyed loop variable. Capture the per-iteration addresses by value instead. --- cpp/src/writer/tsfile_writer.cc | 32 ++++++++++++++++++++++++-------- 1 file changed, 24 insertions(+), 8 deletions(-) diff --git a/cpp/src/writer/tsfile_writer.cc b/cpp/src/writer/tsfile_writer.cc index aa0e555f8..9deed467c 100644 --- a/cpp/src/writer/tsfile_writer.cc +++ b/cpp/src/writer/tsfile_writer.cc @@ -1351,17 +1351,24 @@ int TsFileWriter::write_table(Tablet& tablet) { common::g_thread_pool_ != nullptr) { std::vector> futures; for (auto& ctx : device_ctxs) { + // Capture pointers by value: the submitted tasks run on pool + // threads asynchronously (after this loop returns), so + // capturing the loop variables by reference would dangle + // (all tasks would read the same/out-of-scope ctx). + auto* ctx_ptr = &ctx; futures.push_back(common::g_thread_pool_->submit( - [&write_time_segments, &ctx]() { - return write_time_segments(ctx.tcw, ctx.segments, - ctx.initial_page_points); + [&write_time_segments, ctx_ptr]() { + return write_time_segments(ctx_ptr->tcw, + ctx_ptr->segments, + ctx_ptr->initial_page_points); })); for (auto& vt : ctx.value_tasks) { + auto* vt_ptr = &vt; futures.push_back(common::g_thread_pool_->submit( - [&write_value_segments, &vt, &ctx]() { + [&write_value_segments, vt_ptr, ctx_ptr]() { return write_value_segments( - vt.vcw, vt.col_idx, ctx.segments, - ctx.initial_page_points); + vt_ptr->vcw, vt_ptr->col_idx, + ctx_ptr->segments, ctx_ptr->initial_page_points); })); } } @@ -1898,7 +1905,13 @@ int TsFileWriter::flush_chunk_group_encoded(MeasurementSchemaGroup* chunk_group, for (MeasurementSchemaMapIter ms_iter = map.begin(); ms_iter != map.end(); ms_iter++) { MeasurementSchema* m_schema = ms_iter->second; - if (!chunk_group->is_aligned_ && m_schema->chunk_writer_ != nullptr) { + // Skip registered-but-empty columns: a measurement that was never + // written in this window would otherwise be sealed as an EMPTY chunk + // (count=0, dataSize=0). Java readers (TsFileSequenceReader self- + // check) treat such a file as crashed. Mirror the aligned branch's + // hasData() check below. + if (!chunk_group->is_aligned_ && m_schema->chunk_writer_ != nullptr && + m_schema->chunk_writer_->hasData()) { ChunkWriter*& chunk_writer = m_schema->chunk_writer_; FLUSH_CHUNK_ENCODED( chunk_writer, io_writer_, m_schema->measurement_name_, @@ -1935,7 +1948,10 @@ int TsFileWriter::flush_chunk_group(MeasurementSchemaGroup* chunk_group, for (MeasurementSchemaMapIter ms_iter = map.begin(); ms_iter != map.end(); ms_iter++) { MeasurementSchema* m_schema = ms_iter->second; - if (!chunk_group->is_aligned_ && m_schema->chunk_writer_ != nullptr) { + // See flush_chunk_group_encoded: never seal a registered-but-empty + // column as a count=0 chunk. + if (!chunk_group->is_aligned_ && m_schema->chunk_writer_ != nullptr && + m_schema->chunk_writer_->hasData()) { ChunkWriter*& chunk_writer = m_schema->chunk_writer_; FLUSH_CHUNK(chunk_writer, io_writer_, m_schema->measurement_name_, m_schema->data_type_, m_schema->encoding_, From 6363b029585c205c8ceed9669bf95ccadabcc880 Mon Sep 17 00:00:00 2001 From: gx Date: Wed, 19 Aug 2026 21:52:15 +0800 Subject: [PATCH 2/6] style(cpp): fix spotless clang-format violations in parallel tablet write --- cpp/src/writer/tsfile_writer.cc | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/cpp/src/writer/tsfile_writer.cc b/cpp/src/writer/tsfile_writer.cc index 9deed467c..20512e68a 100644 --- a/cpp/src/writer/tsfile_writer.cc +++ b/cpp/src/writer/tsfile_writer.cc @@ -1358,17 +1358,17 @@ int TsFileWriter::write_table(Tablet& tablet) { auto* ctx_ptr = &ctx; futures.push_back(common::g_thread_pool_->submit( [&write_time_segments, ctx_ptr]() { - return write_time_segments(ctx_ptr->tcw, - ctx_ptr->segments, - ctx_ptr->initial_page_points); + return write_time_segments( + ctx_ptr->tcw, ctx_ptr->segments, + ctx_ptr->initial_page_points); })); for (auto& vt : ctx.value_tasks) { auto* vt_ptr = &vt; futures.push_back(common::g_thread_pool_->submit( [&write_value_segments, vt_ptr, ctx_ptr]() { return write_value_segments( - vt_ptr->vcw, vt_ptr->col_idx, - ctx_ptr->segments, ctx_ptr->initial_page_points); + vt_ptr->vcw, vt_ptr->col_idx, ctx_ptr->segments, + ctx_ptr->initial_page_points); })); } } From 3a4c59f0174321b59616fbafe8a7fd2f0c0ef7a4 Mon Sep 17 00:00:00 2001 From: gx Date: Tue, 25 Aug 2026 23:01:44 +0800 Subject: [PATCH 3/6] test(cpp): add regression tests for empty-chunk seal and parallel tablet write Covers the two fixes in this PR (issue #908): 1. Never seal registered-but-empty measurements as count=0 chunks (non-aligned flush path). Seven tests drive write_tablet, write_record, multi-window flush, empty-window-then-write, sibling devices, and mixed aligned/non-aligned devices, asserting via TsFileReader timeseries metadata that an unwritten measurement is absent from the file while surviving columns carry real statistics. 2. Parallel aligned tablet write capture-by-value. Five tests drive the thread-pool path in write_table() through TsFileTableWriter with multiple devices x columns x rows crossing page boundaries, and verify every cell round-trips (row completeness, per-cell values, tag correctness). Verified: with both fixes temporarily reverted, the empty-chunk tests fail (empty column sealed, meta.size()==3); with the fixes in place all 12 tests pass (10 consecutive runs) and the full TsFile_Test suite (762 tests) passes. --- ...e_parallel_tablet_write_regression_test.cc | 356 ++++++++++++ ...file_writer_empty_chunk_regression_test.cc | 529 ++++++++++++++++++ 2 files changed, 885 insertions(+) create mode 100644 cpp/test/writer/table_view/tsfile_parallel_tablet_write_regression_test.cc create mode 100644 cpp/test/writer/tsfile_writer_empty_chunk_regression_test.cc diff --git a/cpp/test/writer/table_view/tsfile_parallel_tablet_write_regression_test.cc b/cpp/test/writer/table_view/tsfile_parallel_tablet_write_regression_test.cc new file mode 100644 index 000000000..bc5417557 --- /dev/null +++ b/cpp/test/writer/table_view/tsfile_parallel_tablet_write_regression_test.cc @@ -0,0 +1,356 @@ +/* + * 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. + */ +// Regression tests for PR #909 / issue #908 bug 2: the parallel aligned +// tablet write path in TsFileWriter::write_table() submits per-device / +// per-column tasks to the global thread pool. The task lambdas used to +// capture the loop variables (ctx / vt) by reference; since the pool runs +// the tasks asynchronously after the submission loop advanced or exited, +// every task could read the same or out-of-scope state (use-after-scope). +// The fix captures the per-iteration addresses by value. These tests drive +// the parallel path (multiple devices x multiple value columns x enough +// rows to cross page boundaries) and verify every row survives round-trip. +#include + +#ifdef _WIN32 +#include +#else +#include +#endif + +#include +#include +#include +#include +#include + +#include "common/schema.h" +#include "common/tablet.h" +#include "file/write_file.h" +#include "reader/tsfile_reader.h" +#include "writer/tsfile_table_writer.h" + +using namespace storage; +using namespace common; + +namespace { + +class ParallelTabletWriteRegressionTest : public ::testing::Test { + protected: + void SetUp() override { + libtsfile_init(); + file_name_ = std::string("tsfile_parallel_write_regression_") + + 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 + write_file_.create(file_name_, flags, 0666); + } + void TearDown() override { + remove(file_name_.c_str()); + libtsfile_destroy(); + } + + std::string file_name_; + WriteFile write_file_; + + 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; + } + + // 1 TAG column (device id) + `field_col_num` INT64 field columns. + static TableSchema* gen_table_schema(int field_col_num) { + std::vector measurement_schemas; + std::vector column_categories; + measurement_schemas.emplace_back( + new MeasurementSchema("id0", TSDataType::STRING, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED)); + column_categories.emplace_back(ColumnCategory::TAG); + for (int i = 0; i < field_col_num; i++) { + measurement_schemas.emplace_back(new MeasurementSchema( + "s" + std::to_string(i), TSDataType::INT64, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED)); + column_categories.emplace_back(ColumnCategory::FIELD); + } + return new TableSchema("test_table", measurement_schemas, + column_categories); + } + + // Fill a tablet with rows for `device_num` devices. Device d covers rows + // [d * rows_per_device, (d+1) * rows_per_device). Field column c gets + // value row * 100 + c so every cell is uniquely identifiable. + static void gen_tablet(Tablet& tablet, TableSchema* table_schema, + int64_t time_base, int device_num, + int rows_per_device, int field_col_num) { + tablet.set_table_name("test_table"); + PageArena pa; + pa.init(512, MOD_DEFAULT); + for (int d = 0; d < device_num; d++) { + std::string device_str = "device_" + std::to_string(d); + String literal_str(device_str, pa); + for (int l = 0; l < rows_per_device; l++) { + int row_index = d * rows_per_device + l; + int64_t ts = time_base + row_index; + ASSERT_EQ(tablet.add_timestamp(row_index, ts), E_OK); + tablet.add_value(row_index, "id0", literal_str); + for (int c = 0; c < field_col_num; c++) { + tablet.add_value(row_index, "s" + std::to_string(c), + static_cast(row_index * 100 + c)); + } + } + } + } +}; + +// Query the whole table back and verify every cell: row_count rows, tag +// column matching the owning device, and each field column carrying +// row_index * 100 + c. +// Query the table back and verify every cell. The reader returns rows +// grouped per device in device-id order (not tablet insertion order), so the +// check is keyed on the timestamp: every row was written with +// ts = time_base + row_index, field s = row_index * 100 + c, and the tag +// column = "device_". +void VerifyTableRoundTrip(const std::string& file_name, + TableSchema* table_schema, int device_num, + int rows_per_device, int field_col_num, + int64_t time_base = 0) { + TsFileReader reader; + ASSERT_EQ(E_OK, reader.open(file_name)); + ResultSet* tmp_result_set = nullptr; + ASSERT_EQ(E_OK, + reader.query("test_table", table_schema->get_measurement_names(), + 0, INT64_MAX, tmp_result_set)); + auto* result_set = (TableResultSet*)tmp_result_set; + std::vector seen(static_cast(device_num) * rows_per_device, + 0); + bool has_next = false; + int64_t read_rows = 0; + while (IS_SUCC(result_set->next(has_next)) && has_next) { + const int64_t ts = result_set->get_value("time"); + ASSERT_GE(ts, time_base); + const int64_t row_index = ts - time_base; + ASSERT_LT(row_index, static_cast(device_num) * rows_per_device) + << "timestamp out of written range: " << ts; + ASSERT_EQ(seen[row_index], 0) << "duplicate row for timestamp " << ts; + seen[row_index] = 1; + + const int device_idx = static_cast(row_index / rows_per_device); + char literal[32]; + snprintf(literal, sizeof(literal), "device_%d", device_idx); + common::String* tag = result_set->get_value("id0"); + ASSERT_NE(tag, nullptr); + EXPECT_EQ(0, tag->compare(String(literal, strlen(literal)))) + << "tag mismatch at timestamp " << ts; + for (int c = 0; c < field_col_num; c++) { + EXPECT_EQ(result_set->get_value("s" + std::to_string(c)), + row_index * 100 + c) + << "field column s" << c << " corrupted at timestamp " << ts; + } + read_rows++; + } + const int64_t total_rows = + static_cast(device_num) * rows_per_device; + EXPECT_EQ(read_rows, total_rows) + << "row loss: parallel aligned write dropped rows"; + reader.destroy_query_data_set(result_set); + reader.close(); +} + +} // namespace + +// Core regression for the dangling-reference capture: enough devices and +// columns that the submission loop enqueues many tasks whose loop variables +// would alias (the old [&ctx] / [&vt] captures), plus enough rows to cross +// the 10000-point page boundary so page sealing also runs per task. +TEST_F(ParallelTabletWriteRegressionTest, MultiDeviceMultiColumnRoundTrip) { + const int device_num = 8; + const int rows_per_device = 1500; // > page_writer_max_point_num_ (10000/8) + const int field_col_num = 6; + auto table_schema = gen_table_schema(field_col_num); + auto writer = + std::make_shared(&write_file_, table_schema); + Tablet tablet(table_schema->get_measurement_names(), + table_schema->get_data_types(), + static_cast(device_num * rows_per_device)); + gen_tablet(tablet, table_schema, 0, device_num, rows_per_device, + field_col_num); + ASSERT_EQ(E_OK, writer->write_table(tablet)); + ASSERT_EQ(E_OK, writer->flush()); + ASSERT_EQ(E_OK, writer->close()); + VerifyTableRoundTrip(file_name_, table_schema, device_num, rows_per_device, + field_col_num); + delete table_schema; +} + +// Single device, many columns: the per-column ValueTask loop is the inner +// one whose [&vt] capture dangled; with enough columns the tasks are queued +// well past the loop's lifetime. +TEST_F(ParallelTabletWriteRegressionTest, SingleDeviceManyColumnsRoundTrip) { + const int device_num = 1; + const int rows_per_device = 12000; // crosses one page boundary + const int field_col_num = 16; + auto table_schema = gen_table_schema(field_col_num); + auto writer = + std::make_shared(&write_file_, table_schema); + Tablet tablet(table_schema->get_measurement_names(), + table_schema->get_data_types(), + static_cast(device_num * rows_per_device)); + gen_tablet(tablet, table_schema, 0, device_num, rows_per_device, + field_col_num); + ASSERT_EQ(E_OK, writer->write_table(tablet)); + ASSERT_EQ(E_OK, writer->flush()); + ASSERT_EQ(E_OK, writer->close()); + VerifyTableRoundTrip(file_name_, table_schema, device_num, rows_per_device, + field_col_num); + delete table_schema; +} + +// Many devices but few rows each: maximizes the number of DeviceWriteCtx +// entries whose vector reallocation would move the captured ctx references +// while queued tasks still hold them (the classic dangling scenario the fix +// addresses — device_ctxs grows via push_back as the loop progresses). +TEST_F(ParallelTabletWriteRegressionTest, ManyDevicesVectorReallocation) { + const int device_num = 64; + const int rows_per_device = 50; + const int field_col_num = 4; + auto table_schema = gen_table_schema(field_col_num); + auto writer = + std::make_shared(&write_file_, table_schema); + Tablet tablet(table_schema->get_measurement_names(), + table_schema->get_data_types(), + static_cast(device_num * rows_per_device)); + gen_tablet(tablet, table_schema, 0, device_num, rows_per_device, + field_col_num); + ASSERT_EQ(E_OK, writer->write_table(tablet)); + ASSERT_EQ(E_OK, writer->flush()); + ASSERT_EQ(E_OK, writer->close()); + VerifyTableRoundTrip(file_name_, table_schema, device_num, rows_per_device, + field_col_num); + delete table_schema; +} + +// Repeated write_table() calls on the same writer: each call builds a fresh +// device_ctxs vector on the stack, so previously-submitted tasks referencing +// the destroyed vector are exactly the use-after-scope hazard. Several +// sequential batches must all survive. +TEST_F(ParallelTabletWriteRegressionTest, SequentialBatchesRoundTrip) { + const int device_num = 4; + const int rows_per_device = 300; + const int field_col_num = 5; + const int batches = 6; + auto table_schema = gen_table_schema(field_col_num); + auto writer = + std::make_shared(&write_file_, table_schema); + for (int b = 0; b < batches; b++) { + Tablet tablet(table_schema->get_measurement_names(), + table_schema->get_data_types(), + static_cast(device_num * rows_per_device)); + gen_tablet(tablet, table_schema, 1000000 + b * 100000, device_num, + rows_per_device, field_col_num); + ASSERT_EQ(E_OK, writer->write_table(tablet)); + } + ASSERT_EQ(E_OK, writer->flush()); + ASSERT_EQ(E_OK, writer->close()); + + // Each batch wrote the same relative row layout with different + // timestamps; total rows = batches * device_num * rows_per_device. + TsFileReader reader; + ASSERT_EQ(E_OK, reader.open(file_name_)); + ResultSet* tmp_result_set = nullptr; + ASSERT_EQ(E_OK, + reader.query("test_table", table_schema->get_measurement_names(), + 0, INT64_MAX, tmp_result_set)); + auto* result_set = (TableResultSet*)tmp_result_set; + bool has_next = false; + int64_t row_num = 0; + while (IS_SUCC(result_set->next(has_next)) && has_next) { + row_num++; + } + EXPECT_EQ(row_num, + static_cast(batches) * device_num * rows_per_device) + << "row loss across sequential parallel batches"; + reader.destroy_query_data_set(result_set); + reader.close(); + delete table_schema; +} + +// Interleaved flushes between batches: the parallel path runs while earlier +// chunk groups are already sealed, and the value writers reused across +// batches carry non-zero initial_page_points (partial pages). The dangling +// capture could make tasks read the wrong ctx.initial_page_points and +// mis-align page boundaries; the round-trip check catches that. +TEST_F(ParallelTabletWriteRegressionTest, FlushBetweenBatchesRoundTrip) { + const int device_num = 3; + const int rows_per_device = 800; + const int field_col_num = 3; + const int batches = 4; + auto table_schema = gen_table_schema(field_col_num); + auto writer = + std::make_shared(&write_file_, table_schema); + for (int b = 0; b < batches; b++) { + Tablet tablet(table_schema->get_measurement_names(), + table_schema->get_data_types(), + static_cast(device_num * rows_per_device)); + gen_tablet(tablet, table_schema, 500000 + b * 100000, device_num, + rows_per_device, field_col_num); + ASSERT_EQ(E_OK, writer->write_table(tablet)); + ASSERT_EQ(E_OK, writer->flush()); + } + ASSERT_EQ(E_OK, writer->close()); + + TsFileReader reader; + ASSERT_EQ(E_OK, reader.open(file_name_)); + ResultSet* tmp_result_set = nullptr; + ASSERT_EQ(E_OK, + reader.query("test_table", table_schema->get_measurement_names(), + 0, INT64_MAX, tmp_result_set)); + auto* result_set = (TableResultSet*)tmp_result_set; + bool has_next = false; + int64_t row_num = 0; + while (IS_SUCC(result_set->next(has_next)) && has_next) { + row_num++; + } + EXPECT_EQ(row_num, + static_cast(batches) * device_num * rows_per_device) + << "row loss with flush between parallel batches"; + reader.destroy_query_data_set(result_set); + reader.close(); + delete table_schema; +} diff --git a/cpp/test/writer/tsfile_writer_empty_chunk_regression_test.cc b/cpp/test/writer/tsfile_writer_empty_chunk_regression_test.cc new file mode 100644 index 000000000..c729f809a --- /dev/null +++ b/cpp/test/writer/tsfile_writer_empty_chunk_regression_test.cc @@ -0,0 +1,529 @@ +/* + * 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. + */ +// Regression tests for PR #909 / issue #908: +// 1. registered-but-empty measurements must never be sealed as count=0 +// chunks (non-aligned flush path in flush_chunk_group / +// flush_chunk_group_encoded); +// 2. the parallel aligned tablet write path in write_table() must capture +// per-iteration state by value so pool threads never read dangling +// references. +#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 EmptyChunkRegressionTest : public ::testing::Test { + protected: + void SetUp() override { + libtsfile_init(); + tsfile_writer_ = new TsFileWriter(); + file_name_ = std::string("tsfile_empty_chunk_regression_") + + 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; + } + + // Collect per-measurement timeseries index pointers for one device from + // the written file. The caller must keep `reader` open while using the + // returned pointers (they reference reader-owned arenas). + std::map CollectMeasurementMeta( + storage::TsFileReader& reader, const std::string& device) { + std::map out; + std::vector> devices = { + std::make_shared(device)}; + auto meta_map = reader.get_timeseries_metadata(devices); + for (auto& dev_pair : meta_map) { + for (auto& ts_idx : dev_pair.second) { + out[ts_idx->get_measurement_name().to_std_string()] = + ts_idx.get(); + } + } + return out; + } +}; + +} // namespace + +// Regression (issue #908 bug 1): registering a measurement but never writing +// data to it used to seal an EMPTY chunk (count=0, dataSize=0) at flush time. +// Java readers (TsFileSequenceReader self-check / TsFileSketchTool) treat such +// a file as crashed and refuse to load it. The fix mirrors the aligned +// branch's hasData() guard: an empty column must produce no chunk at all, so +// the measurement must be absent from the file's metadata. +TEST_F(EmptyChunkRegressionTest, NonAlignedRegisteredButEmptyNotSealed) { + std::string device = "root.dev_empty_col"; + const int total_measurements = 3; + std::vector schemas; + for (int i = 0; i < total_measurements; i++) { + schemas.emplace_back("m" + std::to_string(i), TSDataType::INT32, + TSEncoding::PLAIN, CompressionType::UNCOMPRESSED); + ASSERT_EQ( + tsfile_writer_->register_timeseries( + device, MeasurementSchema("m" + std::to_string(i), + TSDataType::INT32, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED)), + E_OK); + } + + // Tablet carries all 3 registered columns, but only m0/m1 receive values. + // m2's column stays all-null so its chunk writer never accumulates data. + const int rows = 10; + storage::Tablet tablet( + device, std::make_shared>(schemas), + rows); + for (int r = 0; r < rows; r++) { + ASSERT_EQ(tablet.add_timestamp(r, 1000 + r), E_OK); + ASSERT_EQ(tablet.add_value(r, 0u, static_cast(r)), E_OK); + ASSERT_EQ(tablet.add_value(r, 1u, static_cast(r * 10)), E_OK); + // Column 2 (m2) intentionally never written. + } + ASSERT_EQ(tsfile_writer_->write_tablet(tablet), E_OK); + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + storage::TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + auto meta = CollectMeasurementMeta(reader, device); + + // Only measurements that received data may appear in the file. + ASSERT_EQ(meta.size(), 2u); + ASSERT_NE(meta.count("m0"), 0u); + ASSERT_NE(meta.count("m1"), 0u); + EXPECT_EQ(meta.count("m2"), 0u) + << "registered-but-empty measurement must not be sealed as a chunk"; + // Every surviving series must carry real statistics, not the count=0 / + // start=INT64_MAX / end=INT64_MIN signature of a sealed empty chunk. + EXPECT_EQ(meta["m0"]->get_statistic()->count_, rows); + EXPECT_EQ(meta["m0"]->get_statistic()->start_time_, 1000); + EXPECT_EQ(meta["m0"]->get_statistic()->end_time_, 1000 + rows - 1); + EXPECT_EQ(meta["m1"]->get_statistic()->count_, rows); + reader.close(); +} + +// The file produced by the scenario above must also be fully readable through +// a normal query: the non-empty columns return every row. +TEST_F(EmptyChunkRegressionTest, NonAlignedEmptyColumnFileIsQueryable) { + std::string device = "root.dev_empty_col_query"; + std::vector schemas; + for (int i = 0; i < 3; i++) { + schemas.emplace_back("m" + std::to_string(i), TSDataType::INT32, + TSEncoding::PLAIN, CompressionType::UNCOMPRESSED); + ASSERT_EQ( + tsfile_writer_->register_timeseries( + device, MeasurementSchema("m" + std::to_string(i), + TSDataType::INT32, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED)), + E_OK); + } + const int rows = 7; + storage::Tablet tablet( + device, std::make_shared>(schemas), + rows); + for (int r = 0; r < rows; r++) { + ASSERT_EQ(tablet.add_timestamp(r, 100 + r), E_OK); + ASSERT_EQ(tablet.add_value(r, 0u, static_cast(r)), E_OK); + ASSERT_EQ(tablet.add_value(r, 2u, static_cast(r + 5)), 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); + + storage::TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + std::vector select_list; + select_list.push_back(device + ".m0"); + select_list.push_back(device + ".m2"); + storage::ResultSet* tmp_qds = nullptr; + ASSERT_EQ(reader.query(select_list, 0, INT64_MAX, tmp_qds), E_OK); + auto* qds = (QDSWithoutTimeGenerator*)tmp_qds; + + int row_count = 0; + bool has_next = false; + while (IS_SUCC(qds->next(has_next)) && has_next) { + storage::RowRecord* rec = qds->get_row_record(); + EXPECT_EQ(rec->get_timestamp(), 100 + row_count); + // field(0) is the time column; the selected m0/m2 follow. + EXPECT_EQ(rec->get_field(1)->value_.ival_, row_count); + EXPECT_EQ(rec->get_field(2)->value_.ival_, row_count + 5); + row_count++; + } + EXPECT_EQ(row_count, rows); + reader.destroy_query_data_set(qds); + reader.close(); +} + +// Variant driven through write_record (non-aligned record path): the same +// flush_chunk_group code seals the chunk group, so a measurement that never +// appears in any record must not get an empty chunk either. +TEST_F(EmptyChunkRegressionTest, NonAlignedEmptyMeasurementRecordPath) { + std::string device = "root.dev_empty_col_rec"; + std::vector names = {"s0", "s1"}; + for (const auto& name : names) { + ASSERT_EQ(tsfile_writer_->register_timeseries( + device, MeasurementSchema(name, TSDataType::INT64, + TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED)), + E_OK); + } + for (int i = 0; i < 5; i++) { + TsRecord record(1622505600000 + i, device); + record.add_point(names[0], static_cast(i)); + // s1 never appears in any record. + ASSERT_EQ(tsfile_writer_->write_record(record), E_OK); + } + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + storage::TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + auto meta = CollectMeasurementMeta(reader, device); + ASSERT_EQ(meta.size(), 1u); + EXPECT_EQ(meta.count("s0"), 1u); + EXPECT_EQ(meta.count("s1"), 0u); + EXPECT_EQ(meta["s0"]->get_statistic()->count_, 5); + reader.close(); +} + +// Multi-flush variant: every flush window must skip empty columns, and a +// measurement that receives data in a later window must survive with a chunk +// only for that window (never a count=0 chunk for the empty windows). +TEST_F(EmptyChunkRegressionTest, NonAlignedEmptyMeasurementAcrossFlushes) { + std::string device = "root.dev_empty_col_multi"; + std::vector schemas; + for (int i = 0; i < 3; i++) { + schemas.emplace_back("c" + std::to_string(i), TSDataType::INT32, + TSEncoding::PLAIN, CompressionType::UNCOMPRESSED); + ASSERT_EQ( + tsfile_writer_->register_timeseries( + device, MeasurementSchema("c" + std::to_string(i), + TSDataType::INT32, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED)), + E_OK); + } + + // Window 1: only c0 written. + { + storage::Tablet tablet( + device, std::make_shared>(schemas), + 4); + for (int r = 0; r < 4; r++) { + ASSERT_EQ(tablet.add_timestamp(r, 1000 + r), E_OK); + ASSERT_EQ(tablet.add_value(r, 0u, static_cast(1)), E_OK); + } + ASSERT_EQ(tsfile_writer_->write_tablet(tablet), E_OK); + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + } + // Window 2: only c1 written (c0 gets nothing this window). + { + storage::Tablet tablet( + device, std::make_shared>(schemas), + 4); + for (int r = 0; r < 4; r++) { + ASSERT_EQ(tablet.add_timestamp(r, 2000 + r), E_OK); + ASSERT_EQ(tablet.add_value(r, 1u, static_cast(2)), 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); + + storage::TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + auto meta = CollectMeasurementMeta(reader, device); + // c2 never written in any window -> absent. c0/c1 each have one chunk. + ASSERT_EQ(meta.size(), 2u); + EXPECT_EQ(meta.count("c2"), 0u); + EXPECT_EQ(meta["c0"]->get_chunk_meta_list()->size(), 1u); + EXPECT_EQ(meta["c1"]->get_chunk_meta_list()->size(), 1u); + EXPECT_EQ(meta["c0"]->get_statistic()->count_, 4); + EXPECT_EQ(meta["c0"]->get_statistic()->start_time_, 1000); + EXPECT_EQ(meta["c1"]->get_statistic()->count_, 4); + EXPECT_EQ(meta["c1"]->get_statistic()->start_time_, 2000); + reader.close(); +} + +// Data written after an empty flush window keeps flowing into the same +// column: an earlier flush that skipped the column must not corrupt the +// later write (writer reset semantics), and final statistics must span the +// written window only. +TEST_F(EmptyChunkRegressionTest, NonAlignedWriteAfterEmptyWindowSurvives) { + std::string device = "root.dev_empty_then_write"; + std::vector schemas; + schemas.emplace_back("w0", TSDataType::INT32, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED); + ASSERT_EQ(tsfile_writer_->register_timeseries( + device, + MeasurementSchema("w0", TSDataType::INT32, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED)), + E_OK); + + // Window 1: nothing written, flush must succeed and emit no chunks. + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + + // Window 2: real data after the empty flush. + const int rows = 6; + storage::Tablet tablet( + device, std::make_shared>(schemas), + rows); + for (int r = 0; r < rows; r++) { + ASSERT_EQ(tablet.add_timestamp(r, 3000 + r), E_OK); + ASSERT_EQ(tablet.add_value(r, 0u, static_cast(r)), 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); + + storage::TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + auto meta = CollectMeasurementMeta(reader, device); + ASSERT_EQ(meta.size(), 1u); + ASSERT_NE(meta.count("w0"), 0u); + EXPECT_EQ(meta["w0"]->get_statistic()->count_, rows); + EXPECT_EQ(meta["w0"]->get_statistic()->start_time_, 3000); + EXPECT_EQ(meta["w0"]->get_statistic()->end_time_, 3000 + rows - 1); + reader.close(); +} + +// Multiple devices: one device fully written, one device registered with an +// unwritten measurement. The empty column in the second device must not +// poison the first device's chunk group (they flush in the same pass). +TEST_F(EmptyChunkRegressionTest, EmptyColumnDoesNotAffectSiblingDevice) { + std::string dev_full = "root.dev_full"; + std::string dev_partial = "root.dev_partial"; + + std::vector full_schemas; + full_schemas.emplace_back("f0", TSDataType::INT32, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED); + ASSERT_EQ(tsfile_writer_->register_timeseries( + dev_full, + MeasurementSchema("f0", TSDataType::INT32, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED)), + E_OK); + + std::vector partial_schemas; + partial_schemas.emplace_back("q0", TSDataType::INT32, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED); + partial_schemas.emplace_back("q1", TSDataType::INT32, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED); + for (const auto& s : partial_schemas) { + ASSERT_EQ(tsfile_writer_->register_timeseries( + dev_partial, + MeasurementSchema(s.measurement_name_, TSDataType::INT32, + TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED)), + E_OK); + } + + const int rows = 5; + storage::Tablet full_tablet( + dev_full, + std::make_shared>(full_schemas), rows); + storage::Tablet partial_tablet( + dev_partial, + std::make_shared>(partial_schemas), + rows); + for (int r = 0; r < rows; r++) { + ASSERT_EQ(full_tablet.add_timestamp(r, 7000 + r), E_OK); + ASSERT_EQ(full_tablet.add_value(r, 0u, static_cast(r)), E_OK); + ASSERT_EQ(partial_tablet.add_timestamp(r, 7000 + r), E_OK); + ASSERT_EQ(partial_tablet.add_value(r, 0u, static_cast(r)), + E_OK); + } + ASSERT_EQ(tsfile_writer_->write_tablet(full_tablet), E_OK); + ASSERT_EQ(tsfile_writer_->write_tablet(partial_tablet), E_OK); + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + storage::TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + // Query both devices in one get_timeseries_metadata call: the reader + // resets its metadata arena per call, so raw pointers from separate + // calls must not be held simultaneously. + std::vector> devices = { + std::make_shared(dev_full), + std::make_shared(dev_partial)}; + auto meta_map = reader.get_timeseries_metadata(devices); + ASSERT_EQ(meta_map.size(), 2u); + + auto full_list = + meta_map.at(std::make_shared(dev_full)); + ASSERT_EQ(full_list.size(), 1u); + EXPECT_EQ(full_list[0]->get_measurement_name().to_std_string(), "f0"); + EXPECT_EQ(full_list[0]->get_statistic()->count_, rows); + + auto partial_list = + meta_map.at(std::make_shared(dev_partial)); + ASSERT_EQ(partial_list.size(), 1u); + EXPECT_EQ(partial_list[0]->get_measurement_name().to_std_string(), "q0"); + reader.close(); +} + +// Mixed aligned/non-aligned devices in one file: an aligned device's value +// column with no data (all rows null) plus a non-aligned registered-but-empty +// measurement. Both flush paths must skip their empty columns. +TEST_F(EmptyChunkRegressionTest, EmptyColumnMixedAlignedAndNonAligned) { + std::string dev_aligned = "root.dev_mixed_aligned"; + std::string dev_plain = "root.dev_mixed_plain"; + + // Aligned device: register two value columns, only write one (the other + // stays all-null in every tablet). + std::vector aligned_schemas; + aligned_schemas.emplace_back("a0", TSDataType::INT64, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED); + aligned_schemas.emplace_back("a1", TSDataType::INT64, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED); + std::vector aligned_reg; + for (const auto& s : aligned_schemas) { + aligned_reg.push_back(new MeasurementSchema( + s.measurement_name_, TSDataType::INT64, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED)); + } + ASSERT_EQ( + tsfile_writer_->register_aligned_timeseries(dev_aligned, aligned_reg), + E_OK); + + // Non-aligned device: register two measurements, write only one. + std::vector plain_schemas; + plain_schemas.emplace_back("p0", TSDataType::INT32, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED); + plain_schemas.emplace_back("p1", TSDataType::INT32, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED); + for (const auto& s : plain_schemas) { + ASSERT_EQ(tsfile_writer_->register_timeseries( + dev_plain, + MeasurementSchema(s.measurement_name_, TSDataType::INT32, + TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED)), + E_OK); + } + + const int rows = 8; + storage::Tablet aligned_tablet( + dev_aligned, + std::make_shared>(aligned_schemas), + rows); + storage::Tablet plain_tablet( + dev_plain, + std::make_shared>(plain_schemas), rows); + for (int r = 0; r < rows; r++) { + ASSERT_EQ(aligned_tablet.add_timestamp(r, 5000 + r), E_OK); + ASSERT_EQ(aligned_tablet.add_value(r, 0u, static_cast(r)), + E_OK); + // a1 left all-null. + + ASSERT_EQ(plain_tablet.add_timestamp(r, 5000 + r), E_OK); + ASSERT_EQ(plain_tablet.add_value(r, 0u, static_cast(r)), E_OK); + // p1 never written. + } + ASSERT_EQ(tsfile_writer_->write_tablet_aligned(aligned_tablet), E_OK); + ASSERT_EQ(tsfile_writer_->write_tablet(plain_tablet), E_OK); + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + storage::TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + std::vector> devices = { + std::make_shared(dev_aligned), + std::make_shared(dev_plain)}; + auto meta_map = reader.get_timeseries_metadata(devices); + ASSERT_EQ(meta_map.size(), 2u); + + auto aligned_list = + meta_map.at(std::make_shared(dev_aligned)); + // The aligned branch has always had the hasData() guard; an all-null + // value column still counts its (null) rows, so both columns appear. + // The point of this test on the aligned side is that flush succeeds and + // the written column carries correct statistics. + std::map aligned_meta; + for (auto& ts_idx : aligned_list) { + aligned_meta[ts_idx->get_measurement_name().to_std_string()] = + ts_idx.get(); + } + ASSERT_EQ(aligned_meta.size(), 2u); + EXPECT_EQ(aligned_meta["a0"]->get_statistic()->count_, rows); + + auto plain_list = + meta_map.at(std::make_shared(dev_plain)); + // The non-aligned fix: p1 (registered but never written) must be absent. + ASSERT_EQ(plain_list.size(), 1u); + EXPECT_EQ(plain_list[0]->get_measurement_name().to_std_string(), "p0"); + EXPECT_EQ(plain_list[0]->get_statistic()->count_, rows); + reader.close(); +} From cf742dee95a7791af92d63591eca62656a52cb5f Mon Sep 17 00:00:00 2001 From: gx Date: Tue, 25 Aug 2026 23:37:38 +0800 Subject: [PATCH 4/6] test(cpp): adversarial additions to empty-chunk and parallel-write regressions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fix 1 coverage: - memory-threshold auto-flush path: check_memory_size_and_may_flush_chunks() is a second entry into flush_chunk_group that the explicit-flush tests never drive; drive it via a shrunk chunk_group_size_threshold_. - partially-null column counter-check: the hasData() guard must not skip columns with some nulls (null-bitmap fallback path) — only fully empty ones. - TEXT column variant: empty TEXT takes the string write path, distinct from the fixed-width paths used before. Fix 2 coverage: - tiny page size (8 points) so pool-thread tasks seal pages repeatedly and exercise the initial_page_points continuation across batches. - thread-pool boundary configs: 1-thread pool (serialized workers, worst case for slot aliasing) and 8-thread pool, via set_thread_count(). Verified: auto-flush and TEXT tests fail with the hasData() guard reverted; all 17 tests pass with the fixes (and the full TsFile_Test suite, 767 tests). --- ...e_parallel_tablet_write_regression_test.cc | 120 ++++++++++++++ ...file_writer_empty_chunk_regression_test.cc | 153 ++++++++++++++++++ 2 files changed, 273 insertions(+) diff --git a/cpp/test/writer/table_view/tsfile_parallel_tablet_write_regression_test.cc b/cpp/test/writer/table_view/tsfile_parallel_tablet_write_regression_test.cc index bc5417557..6f621df9f 100644 --- a/cpp/test/writer/table_view/tsfile_parallel_tablet_write_regression_test.cc +++ b/cpp/test/writer/table_view/tsfile_parallel_tablet_write_regression_test.cc @@ -354,3 +354,123 @@ TEST_F(ParallelTabletWriteRegressionTest, FlushBetweenBatchesRoundTrip) { reader.close(); delete table_schema; } + +// ===== Adversarial additions ===== + +// Small page size (8 points) makes every task seal pages repeatedly on the +// pool threads and drives the initial_page_points continuation logic hard: +// rows_per_device=50 crosses 6 page boundaries per column, and the +// batch-to-batch continuation keeps partial pages live across write_table +// calls. +TEST_F(ParallelTabletWriteRegressionTest, TinyPageBoundaryRoundTrip) { + const int prev_page_point_num = + common::g_config_value_.page_writer_max_point_num_; + common::g_config_value_.page_writer_max_point_num_ = 8; + + const int device_num = 4; + const int rows_per_device = 50; + const int field_col_num = 3; + auto table_schema = gen_table_schema(field_col_num); + auto writer = + std::make_shared(&write_file_, table_schema); + const int batches = 3; + for (int b = 0; b < batches; b++) { + Tablet tablet(table_schema->get_measurement_names(), + table_schema->get_data_types(), + static_cast(device_num * rows_per_device)); + gen_tablet(tablet, table_schema, 900000 + b * 100000, device_num, + rows_per_device, field_col_num); + ASSERT_EQ(E_OK, writer->write_table(tablet)); + } + ASSERT_EQ(E_OK, writer->flush()); + ASSERT_EQ(E_OK, writer->close()); + + // Verify every batch: rows for batch b carry ts in [base, base+total). + TsFileReader reader; + ASSERT_EQ(E_OK, reader.open(file_name_)); + for (int b = 0; b < batches; b++) { + const int64_t base = 900000 + b * 100000; + // Per-batch duplicate tracking: row_index is batch-relative. + std::vector seen(static_cast(device_num) * + rows_per_device); + ResultSet* tmp_result_set = nullptr; + ASSERT_EQ(E_OK, + reader.query( + "test_table", table_schema->get_measurement_names(), base, + base + device_num * rows_per_device - 1, tmp_result_set)); + auto* result_set = (TableResultSet*)tmp_result_set; + bool has_next = false; + while (IS_SUCC(result_set->next(has_next)) && has_next) { + const int64_t ts = result_set->get_value("time"); + const int64_t row_index = ts - base; + ASSERT_GE(row_index, 0); + ASSERT_LT(row_index, + static_cast(device_num) * rows_per_device); + ASSERT_EQ(seen[row_index], 0) + << "duplicate row at timestamp " << ts; + seen[row_index] = 1; + const int device_idx = + static_cast(row_index / rows_per_device); + char literal[32]; + snprintf(literal, sizeof(literal), "device_%d", device_idx); + common::String* tag = result_set->get_value("id0"); + ASSERT_NE(tag, nullptr); + EXPECT_EQ(0, tag->compare(String(literal, strlen(literal)))); + for (int c = 0; c < field_col_num; c++) { + EXPECT_EQ( + result_set->get_value("s" + std::to_string(c)), + row_index * 100 + c) + << "field s" << c << " corrupted at ts " << ts; + } + } + reader.destroy_query_data_set(result_set); + } + reader.close(); + delete table_schema; + + common::g_config_value_.page_writer_max_point_num_ = prev_page_point_num; +} + +// Thread-pool boundary configs: a 1-thread pool serializes the tasks on one +// worker (worst case for the old same-slot aliasing), and a larger pool +// runs them concurrently. Both must round-trip every row. +TEST_F(ParallelTabletWriteRegressionTest, ThreadCountBoundariesRoundTrip) { + for (int threads : {1, 8}) { + ASSERT_EQ(E_OK, set_thread_count(threads)); + // set_thread_count rebuilds the global pool; TsFileTableWriter is + // constructed per iteration so no writer holds state across the + // rebuild. + WriteFile write_file; + std::string file_name = + std::string("tsfile_parallel_write_regression_thr") + + std::to_string(threads) + "_" + generate_random_string(8) + + ".tsfile"; + remove(file_name.c_str()); + int flags = O_WRONLY | O_CREAT | O_TRUNC; +#ifdef _WIN32 + flags |= O_BINARY; +#endif + write_file.create(file_name, flags, 0666); + + const int device_num = 6; + const int rows_per_device = 400; + const int field_col_num = 4; + auto table_schema = gen_table_schema(field_col_num); + auto writer = + std::make_shared(&write_file, table_schema); + Tablet tablet(table_schema->get_measurement_names(), + table_schema->get_data_types(), + static_cast(device_num * rows_per_device)); + gen_tablet(tablet, table_schema, 0, device_num, rows_per_device, + field_col_num); + ASSERT_EQ(E_OK, writer->write_table(tablet)); + ASSERT_EQ(E_OK, writer->flush()); + ASSERT_EQ(E_OK, writer->close()); + VerifyTableRoundTrip(file_name, table_schema, device_num, + rows_per_device, field_col_num); + delete table_schema; + ASSERT_EQ(0, remove(file_name.c_str())); + } + // Restore the default pool size for the rest of the suite. + ASSERT_EQ(E_OK, set_thread_count(6)); +} diff --git a/cpp/test/writer/tsfile_writer_empty_chunk_regression_test.cc b/cpp/test/writer/tsfile_writer_empty_chunk_regression_test.cc index c729f809a..47ca4a904 100644 --- a/cpp/test/writer/tsfile_writer_empty_chunk_regression_test.cc +++ b/cpp/test/writer/tsfile_writer_empty_chunk_regression_test.cc @@ -527,3 +527,156 @@ TEST_F(EmptyChunkRegressionTest, EmptyColumnMixedAlignedAndNonAligned) { EXPECT_EQ(plain_list[0]->get_statistic()->count_, rows); reader.close(); } + +// ===== Adversarial additions ===== + +// The memory-threshold auto-flush is a second entry into flush_chunk_group +// that the tests above never drive (they all flush explicitly). Shrink +// chunk_group_size_threshold_ so write_tablet() itself triggers the flush, +// and verify the empty column is still skipped on that path. +TEST_F(EmptyChunkRegressionTest, EmptyColumnSkippedOnMemoryAutoFlush) { + const int64_t prev_threshold = + common::g_config_value_.chunk_group_size_threshold_; + const int32_t prev_check_interval = + common::g_config_value_.record_count_for_next_mem_check_; + // Force the next check to fire after the first tablet and flush almost + // immediately (threshold below the smallest realistic meta accounting). + common::g_config_value_.record_count_for_next_mem_check_ = 1; + common::g_config_value_.chunk_group_size_threshold_ = 1; + + std::string device = "root.dev_empty_autoflush"; + std::vector schemas; + for (int i = 0; i < 3; i++) { + schemas.emplace_back("m" + std::to_string(i), TSDataType::INT32, + TSEncoding::PLAIN, CompressionType::UNCOMPRESSED); + ASSERT_EQ( + tsfile_writer_->register_timeseries( + device, MeasurementSchema("m" + std::to_string(i), + TSDataType::INT32, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED)), + E_OK); + } + + { + const int rows = 5; + storage::Tablet tablet( + device, std::make_shared>(schemas), + rows); + for (int r = 0; r < rows; r++) { + ASSERT_EQ(tablet.add_timestamp(r, 100 + r), E_OK); + ASSERT_EQ(tablet.add_value(r, 0u, static_cast(r)), E_OK); + // m1/m2 stay empty for this window. + } + ASSERT_EQ(tsfile_writer_->write_tablet(tablet), E_OK); + } + + common::g_config_value_.chunk_group_size_threshold_ = prev_threshold; + common::g_config_value_.record_count_for_next_mem_check_ = + prev_check_interval; + + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + storage::TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + auto meta = CollectMeasurementMeta(reader, device); + ASSERT_EQ(meta.size(), 1u); + EXPECT_EQ(meta.count("m0"), 1u); + EXPECT_EQ(meta.count("m1"), 0u); + EXPECT_EQ(meta.count("m2"), 0u); + EXPECT_EQ(meta["m0"]->get_statistic()->count_, 5); + reader.close(); +} + +// Adversarial counter-check for the hasData() guard: a column with *some* +// nulls (partial data) must still be sealed — the fix must skip only fully +// empty columns, not partially-null ones. Uses the null-bitmap fallback +// path in write_column (row 2 of 4 left null). +TEST_F(EmptyChunkRegressionTest, PartiallyNullColumnIsStillSealed) { + std::string device = "root.dev_partial_null"; + std::vector schemas; + schemas.emplace_back("p0", TSDataType::INT32, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED); + schemas.emplace_back("p1", TSDataType::INT32, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED); + for (const auto& s : schemas) { + ASSERT_EQ( + tsfile_writer_->register_timeseries( + device, MeasurementSchema(s.measurement_name_, + TSDataType::INT32, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED)), + E_OK); + } + + const int rows = 4; + storage::Tablet tablet( + device, std::make_shared>(schemas), + rows); + for (int r = 0; r < rows; r++) { + ASSERT_EQ(tablet.add_timestamp(r, 200 + r), E_OK); + ASSERT_EQ(tablet.add_value(r, 0u, static_cast(r)), E_OK); + if (r != 2) { // row 2 stays null in p1 -> null-bitmap fallback path + ASSERT_EQ(tablet.add_value(r, 1u, static_cast(r * 3)), + 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); + + storage::TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + auto meta = CollectMeasurementMeta(reader, device); + // Both columns sealed; p1 counts its 3 non-null rows. + ASSERT_EQ(meta.size(), 2u); + ASSERT_NE(meta.count("p1"), 0u) + << "hasData() guard must not skip partially-null columns"; + EXPECT_EQ(meta["p1"]->get_statistic()->count_, rows - 1); + EXPECT_EQ(meta["p1"]->get_statistic()->start_time_, 200); + EXPECT_EQ(meta["p1"]->get_statistic()->end_time_, 203); + reader.close(); +} + +// TEXT columns take a different write path (write_string_batch) than the +// fixed-width columns used above; an empty TEXT column must be skipped too. +TEST_F(EmptyChunkRegressionTest, EmptyTextColumnNotSealed) { + std::string device = "root.dev_empty_text"; + std::vector schemas; + schemas.emplace_back("t0", TSDataType::TEXT, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED); + schemas.emplace_back("t1", TSDataType::TEXT, TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED); + for (const auto& s : schemas) { + ASSERT_EQ( + tsfile_writer_->register_timeseries( + device, MeasurementSchema(s.measurement_name_, TSDataType::TEXT, + TSEncoding::PLAIN, + CompressionType::UNCOMPRESSED)), + E_OK); + } + + const int rows = 3; + storage::Tablet tablet( + device, std::make_shared>(schemas), + rows); + char buf[] = "v"; + String s0(buf, 1); + for (int r = 0; r < rows; r++) { + ASSERT_EQ(tablet.add_timestamp(r, 300 + r), E_OK); + ASSERT_EQ(tablet.add_value(r, 0u, s0), E_OK); + // t1 never written. + } + ASSERT_EQ(tsfile_writer_->write_tablet(tablet), E_OK); + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + storage::TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + auto meta = CollectMeasurementMeta(reader, device); + ASSERT_EQ(meta.size(), 1u); + EXPECT_EQ(meta.count("t0"), 1u); + EXPECT_EQ(meta.count("t1"), 0u) + << "empty TEXT column must not be sealed as a chunk"; + EXPECT_EQ(meta["t0"]->get_statistic()->count_, rows); + reader.close(); +} From a9af789ed05b729f9a57595d5f83a398ee3e2d90 Mon Sep 17 00:00:00 2001 From: gx Date: Wed, 26 Aug 2026 14:19:12 +0800 Subject: [PATCH 5/6] =?UTF-8?q?test(cpp):=20address=20review=20=E2=80=94?= =?UTF-8?q?=20completeness=20check=20and=20comment=20rewording?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per review feedback on PR #909: - TinyPageBoundaryRoundTrip now asserts completeness: every row of every batch must appear in the query results, so a dropped row on the page-boundary path fails the test instead of going unnoticed. - Reword the parallel-write comments (source and tests): the old by-reference captures were lifetime-safe in the current code shape (device_ctxs outlives all future.get() calls), so the change is a defensive cleanup, not a dangling-reference fix. Comments now say exactly that. --- cpp/src/writer/tsfile_writer.cc | 9 +-- ...e_parallel_tablet_write_regression_test.cc | 58 ++++++++++--------- 2 files changed, 37 insertions(+), 30 deletions(-) diff --git a/cpp/src/writer/tsfile_writer.cc b/cpp/src/writer/tsfile_writer.cc index 20512e68a..005c47bff 100644 --- a/cpp/src/writer/tsfile_writer.cc +++ b/cpp/src/writer/tsfile_writer.cc @@ -1351,10 +1351,11 @@ int TsFileWriter::write_table(Tablet& tablet) { common::g_thread_pool_ != nullptr) { std::vector> futures; for (auto& ctx : device_ctxs) { - // Capture pointers by value: the submitted tasks run on pool - // threads asynchronously (after this loop returns), so - // capturing the loop variables by reference would dangle - // (all tasks would read the same/out-of-scope ctx). + // Capture the per-iteration state by pointer value. The + // references stay valid today (device_ctxs outlives all + // future.get() calls below), but the by-value form doesn't + // rely on that lifetime coincidence and stays safe if the + // get() loop is ever moved out of this scope. auto* ctx_ptr = &ctx; futures.push_back(common::g_thread_pool_->submit( [&write_time_segments, ctx_ptr]() { diff --git a/cpp/test/writer/table_view/tsfile_parallel_tablet_write_regression_test.cc b/cpp/test/writer/table_view/tsfile_parallel_tablet_write_regression_test.cc index 6f621df9f..5264e10ff 100644 --- a/cpp/test/writer/table_view/tsfile_parallel_tablet_write_regression_test.cc +++ b/cpp/test/writer/table_view/tsfile_parallel_tablet_write_regression_test.cc @@ -16,15 +16,17 @@ * specific language governing permissions and limitations * under the License. */ -// Regression tests for PR #909 / issue #908 bug 2: the parallel aligned -// tablet write path in TsFileWriter::write_table() submits per-device / -// per-column tasks to the global thread pool. The task lambdas used to -// capture the loop variables (ctx / vt) by reference; since the pool runs -// the tasks asynchronously after the submission loop advanced or exited, -// every task could read the same or out-of-scope state (use-after-scope). -// The fix captures the per-iteration addresses by value. These tests drive -// the parallel path (multiple devices x multiple value columns x enough -// rows to cross page boundaries) and verify every row survives round-trip. +// Coverage for the parallel aligned tablet write path in +// TsFileWriter::write_table(), which submits per-device / per-column tasks +// to the global thread pool. The task lambdas now capture the per-iteration +// state (ctx / vt) by pointer value rather than by reference. The by-value +// form is defensive: in the current code shape the vectors backing those +// references outlive all future.get() calls, so the old by-reference +// captures were lifetime-safe — but they silently relied on the submission +// loop's scope and would dangle under a refactor (e.g. moving the get() +// loop out of that scope). These tests pin the behavior of the parallel +// path (multiple devices x multiple value columns x enough rows to cross +// page boundaries) and verify every row survives the round-trip. #include #ifdef _WIN32 @@ -195,10 +197,9 @@ void VerifyTableRoundTrip(const std::string& file_name, } // namespace -// Core regression for the dangling-reference capture: enough devices and -// columns that the submission loop enqueues many tasks whose loop variables -// would alias (the old [&ctx] / [&vt] captures), plus enough rows to cross -// the 10000-point page boundary so page sealing also runs per task. +// Enough devices and columns that the submission loop enqueues many tasks +// over distinct ctx / vt entries, plus enough rows to cross the +// 10000-point page boundary so page sealing also runs per task. TEST_F(ParallelTabletWriteRegressionTest, MultiDeviceMultiColumnRoundTrip) { const int device_num = 8; const int rows_per_device = 1500; // > page_writer_max_point_num_ (10000/8) @@ -220,8 +221,8 @@ TEST_F(ParallelTabletWriteRegressionTest, MultiDeviceMultiColumnRoundTrip) { } // Single device, many columns: the per-column ValueTask loop is the inner -// one whose [&vt] capture dangled; with enough columns the tasks are queued -// well past the loop's lifetime. +// submission loop; with enough columns the queued tasks span many vt +// entries. TEST_F(ParallelTabletWriteRegressionTest, SingleDeviceManyColumnsRoundTrip) { const int device_num = 1; const int rows_per_device = 12000; // crosses one page boundary @@ -243,9 +244,8 @@ TEST_F(ParallelTabletWriteRegressionTest, SingleDeviceManyColumnsRoundTrip) { } // Many devices but few rows each: maximizes the number of DeviceWriteCtx -// entries whose vector reallocation would move the captured ctx references -// while queued tasks still hold them (the classic dangling scenario the fix -// addresses — device_ctxs grows via push_back as the loop progresses). +// entries (device_ctxs grows via push_back as the loop progresses), the +// shape most sensitive to how task state is captured. TEST_F(ParallelTabletWriteRegressionTest, ManyDevicesVectorReallocation) { const int device_num = 64; const int rows_per_device = 50; @@ -267,9 +267,8 @@ TEST_F(ParallelTabletWriteRegressionTest, ManyDevicesVectorReallocation) { } // Repeated write_table() calls on the same writer: each call builds a fresh -// device_ctxs vector on the stack, so previously-submitted tasks referencing -// the destroyed vector are exactly the use-after-scope hazard. Several -// sequential batches must all survive. +// device_ctxs vector on the stack. Several sequential batches must all +// survive the round-trip. TEST_F(ParallelTabletWriteRegressionTest, SequentialBatchesRoundTrip) { const int device_num = 4; const int rows_per_device = 300; @@ -313,9 +312,8 @@ TEST_F(ParallelTabletWriteRegressionTest, SequentialBatchesRoundTrip) { // Interleaved flushes between batches: the parallel path runs while earlier // chunk groups are already sealed, and the value writers reused across -// batches carry non-zero initial_page_points (partial pages). The dangling -// capture could make tasks read the wrong ctx.initial_page_points and -// mis-align page boundaries; the round-trip check catches that. +// batches carry non-zero initial_page_points (partial pages). The +// round-trip check catches any mis-aligned page boundaries. TEST_F(ParallelTabletWriteRegressionTest, FlushBetweenBatchesRoundTrip) { const int device_num = 3; const int rows_per_device = 800; @@ -423,6 +421,14 @@ TEST_F(ParallelTabletWriteRegressionTest, TinyPageBoundaryRoundTrip) { << "field s" << c << " corrupted at ts " << ts; } } + // Completeness: every row of this batch must have been returned — + // a dropped row on the page-boundary path must fail the test, not + // just go unnoticed (see review on PR #909). + for (size_t i = 0; i < seen.size(); i++) { + EXPECT_EQ(seen[i], 1) + << "row " << i << " of batch " << b + << " missing from query results (base ts " << base << ")"; + } reader.destroy_query_data_set(result_set); } reader.close(); @@ -432,8 +438,8 @@ TEST_F(ParallelTabletWriteRegressionTest, TinyPageBoundaryRoundTrip) { } // Thread-pool boundary configs: a 1-thread pool serializes the tasks on one -// worker (worst case for the old same-slot aliasing), and a larger pool -// runs them concurrently. Both must round-trip every row. +// worker, and a larger pool runs them concurrently. Both must round-trip +// every row. TEST_F(ParallelTabletWriteRegressionTest, ThreadCountBoundariesRoundTrip) { for (int threads : {1, 8}) { ASSERT_EQ(E_OK, set_thread_count(threads)); From f5a3847df259fbb8ec6b25e652e21ee3e3a2c5c2 Mon Sep 17 00:00:00 2001 From: gx Date: Wed, 26 Aug 2026 14:28:39 +0800 Subject: [PATCH 6/6] docs(cpp): correct lifetime claim in parallel-write comments Per Copilot's review on PR #909: the by-value pointer capture is equivalent in lifetime to the old by-reference captures (each referred to its own vector element, and the vectors outlive all future.get() calls), so the previous wording overclaimed that the by-value form would stay safe if the get() loop were moved out of scope. Task-owned lifetime would require copying the task inputs. Comments (source and tests) now state the equivalence instead. --- cpp/src/writer/tsfile_writer.cc | 11 ++++++----- .../tsfile_parallel_tablet_write_regression_test.cc | 13 +++++++------ 2 files changed, 13 insertions(+), 11 deletions(-) diff --git a/cpp/src/writer/tsfile_writer.cc b/cpp/src/writer/tsfile_writer.cc index 005c47bff..564d1f203 100644 --- a/cpp/src/writer/tsfile_writer.cc +++ b/cpp/src/writer/tsfile_writer.cc @@ -1351,11 +1351,12 @@ int TsFileWriter::write_table(Tablet& tablet) { common::g_thread_pool_ != nullptr) { std::vector> futures; for (auto& ctx : device_ctxs) { - // Capture the per-iteration state by pointer value. The - // references stay valid today (device_ctxs outlives all - // future.get() calls below), but the by-value form doesn't - // rely on that lifetime coincidence and stays safe if the - // get() loop is ever moved out of this scope. + // Capture the per-iteration state by pointer value. This + // is equivalent in lifetime to the old by-reference + // captures (each referred to its own vector element, and + // device_ctxs outlives all future.get() calls below) — it + // only makes the per-task address explicit. Truly task- + // owned lifetime would require copying the task inputs. auto* ctx_ptr = &ctx; futures.push_back(common::g_thread_pool_->submit( [&write_time_segments, ctx_ptr]() { diff --git a/cpp/test/writer/table_view/tsfile_parallel_tablet_write_regression_test.cc b/cpp/test/writer/table_view/tsfile_parallel_tablet_write_regression_test.cc index 5264e10ff..034b4baa2 100644 --- a/cpp/test/writer/table_view/tsfile_parallel_tablet_write_regression_test.cc +++ b/cpp/test/writer/table_view/tsfile_parallel_tablet_write_regression_test.cc @@ -19,12 +19,13 @@ // Coverage for the parallel aligned tablet write path in // TsFileWriter::write_table(), which submits per-device / per-column tasks // to the global thread pool. The task lambdas now capture the per-iteration -// state (ctx / vt) by pointer value rather than by reference. The by-value -// form is defensive: in the current code shape the vectors backing those -// references outlive all future.get() calls, so the old by-reference -// captures were lifetime-safe — but they silently relied on the submission -// loop's scope and would dangle under a refactor (e.g. moving the get() -// loop out of that scope). These tests pin the behavior of the parallel +// state (ctx / vt) by pointer value rather than by reference. The two forms +// are equivalent in lifetime — each by-reference capture referred to its own +// vector element (not a shared loop slot), and the vectors outlive all +// future.get() calls, so the old captures were lifetime-safe — and the +// by-value form merely makes the per-task address explicit. Task-owned +// lifetime would require copying the task inputs. These tests pin the +// behavior of the parallel // path (multiple devices x multiple value columns x enough rows to cross // page boundaries) and verify every row survives the round-trip. #include