perf(cpp): enlarge the TsFile write stream page size to 64 KiB

cpp/src/file/tsfile_io_writer.cc

@@ -36,6 +36,12 @@ using namespace common;
 
 namespace storage {
 
+// Definition for the in-class `static constexpr` used as the write-stream
+// page size.  Before C++17 a static constexpr data member needs an
+// out-of-class definition when it is odr-used (e.g. bound to a reference
+// by a test assertion); without one the link fails.
+constexpr uint32_t TsFileIOWriter::WRITE_STREAM_PAGE_SIZE;
+
 #if 0
 #define OFFSET_DEBUG(msg)                \
     std::cout << "OFFSET_DEBUG: " << msg \

cpp/src/file/tsfile_io_writer.h

@@ -56,7 +56,13 @@ class TsFileIOWriter {
     typedef DeviceNodeMap::iterator DeviceNodeMapIterator;
 
    public:
-    static const uint32_t WRITE_STREAM_PAGE_SIZE = 512;  // FIXME
+    // Pending bytes drain to the file one page at a time, so this is both the
+    // write() granularity and the per-flush allocation unit. 512 bytes turned
+    // a chunk flush into dozens of tiny write() syscalls and page allocations;
+    // 64 KiB keeps a chunk's data in one or two pages without wasting much
+    // memory on the trailing page.
+    static constexpr uint32_t WRITE_STREAM_PAGE_SIZE = 64 * 1024;
+
    public:
     TsFileIOWriter()
         : meta_allocator_(),

cpp/test/common/allocator/byte_stream_test.cc

@@ -434,4 +434,53 @@ TEST(OptionalAtomicStorageTest, BackingStorageIsRealAtomic) {
     EXPECT_EQ(oa.load(), 51u);
 }
 
+// The write path drains a ByteStream page by page via Consumer::get_next_buf().
+// Its exact-multiple branch (a stream whose total size is a multiple of the
+// page size) decides whether the last page is full, and had no coverage; a
+// regression there would silently truncate or over-read the final page.
+TEST(ByteStreamConsumerTest, DrainsExactPageMultiplesAndPartialTails) {
+    constexpr uint32_t kPageSize = 1024;
+    struct Case {
+        uint32_t total;
+        std::vector<uint32_t> expected_lens;
+    };
+    const std::vector<Case> cases = {
+        {0, {}},
+        {1, {1}},
+        {kPageSize - 1, {kPageSize - 1}},
+        {kPageSize, {kPageSize}},
+        {kPageSize + 1, {kPageSize, 1}},
+        {2 * kPageSize, {kPageSize, kPageSize}},
+        {3 * kPageSize, {kPageSize, kPageSize, kPageSize}},
+        {3 * kPageSize + 17, {kPageSize, kPageSize, kPageSize, 17}},
+    };
+
+    for (const auto& test_case : cases) {
+        SCOPED_TRACE(::testing::Message() << "total=" << test_case.total);
+        ByteStream stream(kPageSize, MOD_DEFAULT,
+                          /*enable_atomic=*/true);
+        std::vector<uint8_t> payload(test_case.total);
+        for (uint32_t i = 0; i < test_case.total; i++) {
+            payload[i] = static_cast<uint8_t>((i * 13 + 5) & 0xff);
+        }
+        if (test_case.total > 0) {
+            ASSERT_EQ(stream.write_buf(payload.data(), test_case.total), E_OK);
+        }
+
+        ByteStream::Consumer consumer(stream);
+        std::vector<uint32_t> lens;
+        std::vector<uint8_t> drained;
+        while (true) {
+            ByteStream::Buffer buf = consumer.get_next_buf(stream);
+            if (buf.buf_ == nullptr) {
+                break;
+            }
+            lens.push_back(buf.len_);
+            drained.insert(drained.end(), buf.buf_, buf.buf_ + buf.len_);
+        }
+        EXPECT_EQ(lens, test_case.expected_lens);
+        EXPECT_EQ(drained, payload);
+    }
+}
+
 }  // namespace common

cpp/test/reader/tsfile_reader_test.cc

@@ -21,6 +21,7 @@
 #include <gtest/gtest.h>
 #include <sys/stat.h>
 
+#include <atomic>
 #include <cmath>
 #include <cstring>
 #include <fstream>
@@ -43,6 +44,12 @@
 #include "reader/block/single_device_tsblock_reader.h"
 #include "reader/filter/time_operator.h"
 #include "reader/qds_without_timegenerator.h"
+
+#ifdef _WIN32
+#include <process.h>
+#else
+#include <unistd.h>
+#endif
 #include "reader/tsfile_series_scan_iterator.h"
 #include "writer/chunk_writer.h"
 #include "writer/tsfile_writer.h"
@@ -118,6 +125,18 @@ class TsFileReaderTest : public ::testing::Test {
             random_string += chars[dis(gen)];
         }
 
+        // CTest runs test cases in parallel processes, and two of them can
+        // seed the generator in the same clock tick.  Without a per-process
+        // unique suffix one process can remove the file another is using,
+        // which surfaced as E_ALREADY_EXIST from TsFileWriter::open.
+#ifdef _WIN32
+        const auto process_id = static_cast<uint64_t>(_getpid());
+#else
+        const auto process_id = static_cast<uint64_t>(getpid());
+#endif
+        static std::atomic<uint64_t> counter{0};
+        random_string += "_" + std::to_string(process_id) + "_" +
+                         std::to_string(counter.fetch_add(1));
         return random_string;
     }
 

cpp/test/writer/tsfile_writer_test.cc

@@ -333,6 +333,67 @@ TEST_F(TsFileWriterTest, MultiFlushWriteAndRead) {
     ASSERT_EQ(reader.close(), E_OK);
 }
 
+// The I/O writer drains its pending bytes one stream page at a time, so a
+// chunk larger than one page is written as a chain of write() calls.  These
+// sizes produce 1-, 2- and 3-page chains with partial tails, which checks that
+// the chain is flushed in order.  (A chunk ending exactly on a page boundary is
+// covered by ByteStreamConsumerTest.DrainsExactPageMultiplesAndPartialTails,
+// since the per-chunk metadata overhead makes row counts alone unable to land
+// on the boundary.)
+TEST_F(TsFileWriterTest, WriteChunkSpanningMultipleWriteStreamPages) {
+    const std::vector<int64_t> row_counts = {1000, 8191,  8192,
+                                             8193, 16384, 20000};
+    // Keep at least one case spanning several pages even if the page size or
+    // the value width changes: the largest payload alone must exceed one page
+    // (per-chunk metadata only adds bytes on top of this).
+    EXPECT_LT(storage::TsFileIOWriter::WRITE_STREAM_PAGE_SIZE,
+              row_counts.back() * static_cast<int64_t>(sizeof(int64_t)));
+    std::vector<std::string> select_list;
+    for (size_t i = 0; i < row_counts.size(); ++i) {
+        const std::string device = "device_" + std::to_string(i);
+        const std::string measurement = "value";
+        ASSERT_EQ(tsfile_writer_->register_timeseries(
+                      device, storage::MeasurementSchema(
+                                  measurement, common::TSDataType::INT64,
+                                  common::TSEncoding::PLAIN,
+                                  common::CompressionType::UNCOMPRESSED)),
+                  E_OK);
+        select_list.push_back(device + "." + measurement);
+        for (int64_t row = 0; row < row_counts[i]; ++row) {
+            TsRecord record(row, device);
+            record.add_point(measurement, row);
+            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);
+    for (size_t i = 0; i < row_counts.size(); ++i) {
+        SCOPED_TRACE(::testing::Message() << "rows=" << row_counts[i]
+                                          << ", select=" << select_list[i]);
+        storage::ResultSet* tmp_qds = nullptr;
+        std::vector<std::string> columns{select_list[i]};
+        ASSERT_EQ(reader.query(columns, 0, row_counts[i] - 1, tmp_qds), E_OK);
+        auto* qds = static_cast<QDSWithoutTimeGenerator*>(tmp_qds);
+        int64_t read_rows = 0;
+        bool has_next = false;
+        while (true) {
+            ASSERT_EQ(qds->next(has_next), E_OK);
+            if (!has_next) {
+                break;
+            }
+            ASSERT_EQ(qds->get_value<int64_t>(1), read_rows);
+            ASSERT_EQ(qds->get_value<int64_t>(2), read_rows);
+            ++read_rows;
+        }
+        ASSERT_EQ(read_rows, row_counts[i]);
+        reader.destroy_query_data_set(qds);
+    }
+    ASSERT_EQ(reader.close(), E_OK);
+}
+
 #if defined(ENABLE_ZLIB) && defined(ENABLE_SNAPPY) && defined(ENABLE_LZ4) && \
     defined(ENABLE_LZOKAY)
 TEST_F(TsFileWriterTest, WriteDiffrentTypeCombination) {