From f8fed9389eeb958687bd249a25f6adfea0e2f1d5 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Wed, 23 Sep 2026 16:42:14 +0800 Subject: [PATCH] fix(cpp): flush pending data on writer close --- cpp/src/cwrapper/tsfile_cwrapper.cc | 12 +----- cpp/src/cwrapper/tsfile_cwrapper.h | 3 +- cpp/src/writer/tsfile_table_writer.h | 2 +- cpp/src/writer/tsfile_writer.cc | 4 ++ cpp/src/writer/tsfile_writer.h | 5 ++- .../table_view/tsfile_writer_table_test.cc | 22 +++++++++- cpp/test/writer/tsfile_writer_test.cc | 43 +++++++++++++++++++ 7 files changed, 76 insertions(+), 15 deletions(-) diff --git a/cpp/src/cwrapper/tsfile_cwrapper.cc b/cpp/src/cwrapper/tsfile_cwrapper.cc index a31e66d1b..5cae8f01d 100644 --- a/cpp/src/cwrapper/tsfile_cwrapper.cc +++ b/cpp/src/cwrapper/tsfile_cwrapper.cc @@ -318,11 +318,7 @@ ERRNO tsfile_writer_close(TsFileWriter writer) { return common::E_OK; } auto* w = static_cast(writer); - int ret = w->flush(); - if (ret != common::E_OK) { - return ret; - } - ret = w->close(); + int ret = w->close(); if (ret != common::E_OK) { return ret; } @@ -2300,11 +2296,7 @@ ERRNO _tsfile_writer_write_ts_record(TsFileWriter writer, TsRecord data) { ERRNO _tsfile_writer_close(TsFileWriter writer) { auto* w = static_cast(writer); - int ret = w->flush(); - if (ret != common::E_OK) { - return ret; - } - ret = w->close(); + int ret = w->close(); if (ret != common::E_OK) { return ret; } diff --git a/cpp/src/cwrapper/tsfile_cwrapper.h b/cpp/src/cwrapper/tsfile_cwrapper.h index 4656c5545..003e59198 100644 --- a/cpp/src/cwrapper/tsfile_cwrapper.h +++ b/cpp/src/cwrapper/tsfile_cwrapper.h @@ -496,7 +496,8 @@ TsFileWriter tsfile_writer_new_with_memory_threshold(WriteFile file, TsFileReader tsfile_reader_new(const char* pathname, ERRNO* err_code); /** - * @brief Releases resources associated with a TsFileWriter. + * @brief Flushes pending data, finalizes the TsFile, and releases resources + * associated with a TsFileWriter. * * @param writer [in] Writer handle obtained from tsfile_writer_new(). * After call: handle becomes invalid and must not be reused. diff --git a/cpp/src/writer/tsfile_table_writer.h b/cpp/src/writer/tsfile_table_writer.h index cc7b59bde..d0b0bca1a 100644 --- a/cpp/src/writer/tsfile_table_writer.h +++ b/cpp/src/writer/tsfile_table_writer.h @@ -125,7 +125,7 @@ class TsFileTableWriter { int add_tsfile_property(const std::string& key, const std::vector& value); /** - * Closes the writer and releases any resources held by it. + * Flushes pending data, finalizes the file, and releases writer resources. * After calling this method, no further operations should be performed on * this instance. * diff --git a/cpp/src/writer/tsfile_writer.cc b/cpp/src/writer/tsfile_writer.cc index 41a485c00..110fcf357 100644 --- a/cpp/src/writer/tsfile_writer.cc +++ b/cpp/src/writer/tsfile_writer.cc @@ -1986,6 +1986,10 @@ int TsFileWriter::flush_chunk_group(MeasurementSchemaGroup* chunk_group, int TsFileWriter::close() { if (UNLIKELY(unrecoverable_)) return E_DATA_INCONSISTENCY; + int ret = flush(); + if (ret != E_OK) { + return ret; + } return io_writer_->end_file(); } diff --git a/cpp/src/writer/tsfile_writer.h b/cpp/src/writer/tsfile_writer.h index 55e9e7f3a..842c75446 100644 --- a/cpp/src/writer/tsfile_writer.h +++ b/cpp/src/writer/tsfile_writer.h @@ -115,8 +115,9 @@ class TsFileWriter { int flush(); /* - * Flush file index part of the whole file (it may be flushed many times - * before close, the index part should cover all data in disk file). + * Flushes remaining buffered data, writes the file index and footer, and + * closes the file. Any flush failure is returned without finalizing the + * file, so the caller can handle the error and retry if possible. */ int close(); diff --git a/cpp/test/writer/table_view/tsfile_writer_table_test.cc b/cpp/test/writer/table_view/tsfile_writer_table_test.cc index eeb82dbf4..9e716558f 100644 --- a/cpp/test/writer/table_view/tsfile_writer_table_test.cc +++ b/cpp/test/writer/table_view/tsfile_writer_table_test.cc @@ -139,8 +139,28 @@ TEST_F(TsFileWriterTableTest, WriteTableTest) { std::make_shared(&write_file_, table_schema); auto tablet = gen_tablet(table_schema, 0, 1); ASSERT_EQ(tsfile_table_writer_->write_table(tablet), common::E_OK); - ASSERT_EQ(tsfile_table_writer_->flush(), common::E_OK); ASSERT_EQ(tsfile_table_writer_->close(), common::E_OK); + + TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), common::E_OK); + ResultSet* result_set = nullptr; + ASSERT_EQ(reader.query(table_schema->get_table_name(), {"s0"}, 0, INT32_MAX, + result_set), + common::E_OK); + auto* table_result_set = static_cast(result_set); + bool has_next = false; + int64_t row_count = 0; + while (true) { + ASSERT_EQ(table_result_set->next(has_next), common::E_OK); + if (!has_next) { + break; + } + ++row_count; + } + EXPECT_EQ(row_count, 10); + table_result_set->close(); + reader.destroy_query_data_set(table_result_set); + ASSERT_EQ(reader.close(), common::E_OK); delete table_schema; } diff --git a/cpp/test/writer/tsfile_writer_test.cc b/cpp/test/writer/tsfile_writer_test.cc index 3b9dae92a..6eee36130 100644 --- a/cpp/test/writer/tsfile_writer_test.cc +++ b/cpp/test/writer/tsfile_writer_test.cc @@ -871,6 +871,49 @@ TEST_F(TsFileWriterTest, FlushWithoutWriteAfterRegisterTS) { ASSERT_EQ(tsfile_writer_->close(), E_OK); } +TEST_F(TsFileWriterTest, CloseFlushesDataWrittenAfterLastFlush) { + const std::string device_path = "device_close_flush"; + const std::string measurement_name = "value"; + ASSERT_EQ(tsfile_writer_->register_timeseries( + device_path, storage::MeasurementSchema( + measurement_name, common::TSDataType::INT64, + common::TSEncoding::PLAIN, + common::CompressionType::UNCOMPRESSED)), + E_OK); + + TsRecord first_record(100, device_path); + first_record.add_point(measurement_name, static_cast(1)); + ASSERT_EQ(tsfile_writer_->write_record(first_record), E_OK); + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + + TsRecord final_record(101, device_path); + final_record.add_point(measurement_name, static_cast(2)); + ASSERT_EQ(tsfile_writer_->write_record(final_record), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + ResultSet* result_set = nullptr; + std::vector select_list = {device_path + "." + + measurement_name}; + ASSERT_EQ(reader.query(select_list, 100, 102, result_set), E_OK); + auto* query_result = static_cast(result_set); + bool has_next = false; + ASSERT_EQ(query_result->next(has_next), E_OK); + ASSERT_TRUE(has_next); + EXPECT_EQ(query_result->get_value(1), 100); + EXPECT_EQ(query_result->get_value(2), 1); + ASSERT_EQ(query_result->next(has_next), E_OK); + ASSERT_TRUE(has_next); + EXPECT_EQ(query_result->get_value(1), 101); + EXPECT_EQ(query_result->get_value(2), 2); + ASSERT_EQ(query_result->next(has_next), E_OK); + EXPECT_FALSE(has_next); + + reader.destroy_query_data_set(result_set); + ASSERT_EQ(reader.close(), E_OK); +} + TEST_F(TsFileWriterTest, WriteAlignedTimeseries) { int measurement_num = 100, row_num = 150; std::string device_name = "device";