This is an automated email from the ASF dual-hosted git repository.
ColinLeeo pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/tsfile.git
The following commit(s) were added to refs/heads/develop by this push:
new bf36d8046 fix(cpp): never seal empty chunks and fix dangling ref in
parallel tablet write (#909)
bf36d8046 is described below
commit bf36d80460c6ef73649b2ea2e57e30ad2b5c42fc
Author: kkzi <[email protected]>
AuthorDate: Wed Aug 26 14:59:39 2026 +0800
fix(cpp): never seal empty chunks and fix dangling ref in parallel tablet
write (#909)
* 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.
* style(cpp): fix spotless clang-format violations in parallel tablet write
* 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.
* test(cpp): adversarial additions to empty-chunk and parallel-write
regressions
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).
* test(cpp): address review — completeness check and comment rewording
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.
* 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.
---------
Co-authored-by: gx <[email protected]>
---
cpp/src/writer/tsfile_writer.cc | 34 +-
...tsfile_parallel_tablet_write_regression_test.cc | 483 +++++++++++++++
.../tsfile_writer_empty_chunk_regression_test.cc | 682 +++++++++++++++++++++
3 files changed, 1191 insertions(+), 8 deletions(-)
diff --git a/cpp/src/writer/tsfile_writer.cc b/cpp/src/writer/tsfile_writer.cc
index aa0e555f8..564d1f203 100644
--- a/cpp/src/writer/tsfile_writer.cc
+++ b/cpp/src/writer/tsfile_writer.cc
@@ -1351,17 +1351,26 @@ int TsFileWriter::write_table(Tablet& tablet) {
common::g_thread_pool_ != nullptr) {
std::vector<std::future<int>> futures;
for (auto& ctx : device_ctxs) {
+ // 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]() {
- 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 +1907,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 +1950,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_,
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..034b4baa2
--- /dev/null
+++ b/cpp/test/writer/table_view/tsfile_parallel_tablet_write_regression_test.cc
@@ -0,0 +1,483 @@
+/*
+ * 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.
+ */
+// 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 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 <gtest/gtest.h>
+
+#ifdef _WIN32
+#include <process.h>
+#else
+#include <unistd.h>
+#endif
+
+#include <atomic>
+#include <chrono>
+#include <random>
+#include <string>
+#include <vector>
+
+#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<uint64_t> counter{0};
+ std::mt19937 gen(static_cast<unsigned int>(
+ std::chrono::system_clock::now().time_since_epoch().count()));
+ std::uniform_int_distribution<> dis(0, 61);
+ const std::string chars =
+ "0123456789"
+ "abcdefghijklmnopqrstuvwxyz"
+ "ABCDEFGHIJKLMNOPQRSTUVWXYZ";
+ std::string random_string;
+ for (int i = 0; i < length; ++i) {
+ random_string += chars[dis(gen)];
+ }
+#ifdef _WIN32
+ const auto process_id = static_cast<uint64_t>(_getpid());
+#else
+ const auto process_id = static_cast<uint64_t>(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<MeasurementSchema*> measurement_schemas;
+ std::vector<ColumnCategory> 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<int64_t>(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<c> = row_index * 100 + c, and the tag
+// column = "device_<row_index / rows_per_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<char> seen(static_cast<size_t>(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<int64_t>("time");
+ ASSERT_GE(ts, time_base);
+ const int64_t row_index = ts - time_base;
+ ASSERT_LT(row_index, static_cast<int64_t>(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<int>(row_index / rows_per_device);
+ char literal[32];
+ snprintf(literal, sizeof(literal), "device_%d", device_idx);
+ common::String* tag = result_set->get_value<common::String*>("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<int64_t>("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<int64_t>(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
+
+// 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)
+ const int field_col_num = 6;
+ auto table_schema = gen_table_schema(field_col_num);
+ auto writer =
+ std::make_shared<TsFileTableWriter>(&write_file_, table_schema);
+ Tablet tablet(table_schema->get_measurement_names(),
+ table_schema->get_data_types(),
+ static_cast<uint32_t>(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
+// 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
+ const int field_col_num = 16;
+ auto table_schema = gen_table_schema(field_col_num);
+ auto writer =
+ std::make_shared<TsFileTableWriter>(&write_file_, table_schema);
+ Tablet tablet(table_schema->get_measurement_names(),
+ table_schema->get_data_types(),
+ static_cast<uint32_t>(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 (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;
+ const int field_col_num = 4;
+ auto table_schema = gen_table_schema(field_col_num);
+ auto writer =
+ std::make_shared<TsFileTableWriter>(&write_file_, table_schema);
+ Tablet tablet(table_schema->get_measurement_names(),
+ table_schema->get_data_types(),
+ static_cast<uint32_t>(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. Several sequential batches must all
+// survive the round-trip.
+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<TsFileTableWriter>(&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<uint32_t>(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<int64_t>(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
+// round-trip check catches any mis-aligned page boundaries.
+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<TsFileTableWriter>(&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<uint32_t>(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<int64_t>(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;
+}
+
+// ===== 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<TsFileTableWriter>(&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<uint32_t>(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<char> seen(static_cast<size_t>(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<int64_t>("time");
+ const int64_t row_index = ts - base;
+ ASSERT_GE(row_index, 0);
+ ASSERT_LT(row_index,
+ static_cast<int64_t>(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<int>(row_index / rows_per_device);
+ char literal[32];
+ snprintf(literal, sizeof(literal), "device_%d", device_idx);
+ common::String* tag =
result_set->get_value<common::String*>("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<int64_t>("s" + std::to_string(c)),
+ row_index * 100 + c)
+ << "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();
+ 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, 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<TsFileTableWriter>(&write_file, table_schema);
+ Tablet tablet(table_schema->get_measurement_names(),
+ table_schema->get_data_types(),
+ static_cast<uint32_t>(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
new file mode 100644
index 000000000..47ca4a904
--- /dev/null
+++ b/cpp/test/writer/tsfile_writer_empty_chunk_regression_test.cc
@@ -0,0 +1,682 @@
+/*
+ * 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 <gtest/gtest.h>
+
+#include "writer/tsfile_writer.h"
+
+#ifdef _WIN32
+#include <process.h>
+#else
+#include <unistd.h>
+#endif
+
+#include <atomic>
+#include <map>
+#include <memory>
+#include <random>
+#include <string>
+#include <vector>
+
+#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<uint64_t> counter{0};
+ std::mt19937 gen(static_cast<unsigned int>(
+ std::chrono::system_clock::now().time_since_epoch().count()));
+ std::uniform_int_distribution<> dis(0, 61);
+ const std::string chars =
+ "0123456789"
+ "abcdefghijklmnopqrstuvwxyz"
+ "ABCDEFGHIJKLMNOPQRSTUVWXYZ";
+ std::string random_string;
+ for (int i = 0; i < length; ++i) {
+ random_string += chars[dis(gen)];
+ }
+#ifdef _WIN32
+ const auto process_id = static_cast<uint64_t>(_getpid());
+#else
+ const auto process_id = static_cast<uint64_t>(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<std::string, storage::ITimeseriesIndex*> CollectMeasurementMeta(
+ storage::TsFileReader& reader, const std::string& device) {
+ std::map<std::string, storage::ITimeseriesIndex*> out;
+ std::vector<std::shared_ptr<IDeviceID>> devices = {
+ std::make_shared<StringArrayDeviceID>(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<MeasurementSchema> 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<std::vector<MeasurementSchema>>(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<int32_t>(r)), E_OK);
+ ASSERT_EQ(tablet.add_value(r, 1u, static_cast<int32_t>(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<MeasurementSchema> 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<std::vector<MeasurementSchema>>(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<int32_t>(r)), E_OK);
+ ASSERT_EQ(tablet.add_value(r, 2u, static_cast<int32_t>(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<std::string> 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<std::string> 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<int64_t>(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<MeasurementSchema> 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<std::vector<MeasurementSchema>>(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<int32_t>(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<std::vector<MeasurementSchema>>(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<int32_t>(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<MeasurementSchema> 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<std::vector<MeasurementSchema>>(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<int32_t>(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<MeasurementSchema> 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<MeasurementSchema> 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<std::vector<MeasurementSchema>>(full_schemas), rows);
+ storage::Tablet partial_tablet(
+ dev_partial,
+ std::make_shared<std::vector<MeasurementSchema>>(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<int32_t>(r)), E_OK);
+ ASSERT_EQ(partial_tablet.add_timestamp(r, 7000 + r), E_OK);
+ ASSERT_EQ(partial_tablet.add_value(r, 0u, static_cast<int32_t>(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<std::shared_ptr<IDeviceID>> devices = {
+ std::make_shared<StringArrayDeviceID>(dev_full),
+ std::make_shared<StringArrayDeviceID>(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<StringArrayDeviceID>(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<StringArrayDeviceID>(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<MeasurementSchema> 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<MeasurementSchema*> 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<MeasurementSchema> 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<std::vector<MeasurementSchema>>(aligned_schemas),
+ rows);
+ storage::Tablet plain_tablet(
+ dev_plain,
+ std::make_shared<std::vector<MeasurementSchema>>(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<int64_t>(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<int32_t>(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<std::shared_ptr<IDeviceID>> devices = {
+ std::make_shared<StringArrayDeviceID>(dev_aligned),
+ std::make_shared<StringArrayDeviceID>(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<StringArrayDeviceID>(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<std::string, storage::ITimeseriesIndex*> 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<StringArrayDeviceID>(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();
+}
+
+// ===== 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<MeasurementSchema> 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<std::vector<MeasurementSchema>>(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<int32_t>(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<MeasurementSchema> 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<std::vector<MeasurementSchema>>(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<int32_t>(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<int32_t>(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<MeasurementSchema> 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<std::vector<MeasurementSchema>>(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();
+}
