perf(cpp): bulk-copy contiguous boundary pages and decode aligned pages in place (#1096)

* perf(cpp): bulk-copy contiguous boundary pages and decode aligned pages in place

The aligned chunk reader bulk-copied FULL_PASS pages only. A page can also be
fully matched while classified BOUNDARY: under a contiguous time predicate
(TimeBetween without negation) every timestamp inside the page's retained
[row_begin, row_end] range passes, so the range can be copied as a block
instead of re-checking and appending row by row. Discrete predicates (TimeIn)
and negated forms keep the row-by-row path, and a FIELD predicate keeps it
too: a page mask can still reject rows inside the retained range.

Second, on the page-plan path the leading pages whose output rows fit in the
current TsBlock are now decoded straight into the destination column buffers
instead of a per-page staging buffer that is copied afterwards. That needs a
fixed destination address, so it applies only to fixed-length columns, only
when a page is retained from row 0 through its end, and only to the prefix
that the scatter consumes from the same block; eager (parallel) decode can
use it for every page that fits. Pages with nulls, partial ranges and
variable-length columns keep their staging buffers.

ByteBuffer gains set_data_size() to declare bytes written directly through
get_data().

On a scan of 200k rows x 8 INT64 columns with a window covering the middle
half, 64 KiB pages and the default block size, a full read takes 2.08 ms
instead of 2.39 ms per scan. With 4 KiB pages the read is I/O bound and the
change is neutral (5.1 ms both ways).

* test(cpp): count bulk-copied pages once each

TEST_bulk_page_count() counted bulk-copy operations, so a page split across
several calls (small destination block) was counted more than once. Count a
page when its first bulk operation starts, and have the test read the
cumulative counter as a delta instead of adding it up per call.

Also scope the skip macro comment to the tests that actually need a decode
pool; the FIELD filter test reaches the page plan either way.

* fix(cpp): derive the direct-decode offset from the rows the scatter emits

A page inside the direct-decode prefix that cannot be written in place (it
carries nulls, or its retained range is shorter than the page) still has to
reserve space in the destination column, because the pages after it were
already decoded at offsets computed from it. The reservation used the whole
page's non-null count; derive it from the non-null values in
[row_begin, row_end) instead, which is exactly what the scatter appends for
that page, so the offsets stay right whatever mix of pages the prefix holds.

Also fix the comment to use the half-open [row_begin, row_end) range, and
let read_all() zero its out parameters so callers need not pre-zero them.

* test(cpp): harden the aligned bulk-copy test setup

Give each case's file a wall-clock suffix so parallel or repeated runs
cannot collide on the name, take the shared aligned time chunk from the
first index and assert the rest agree instead of silently overwriting it,
and stop the read helper on a failed load or block init instead of
asserting further against an unusable block.

* test(cpp): avoid clashing fixture file names across concurrent cases

The CI run for this PR failed in HotDeviceMetadataTest.RepeatedLookups...
with the fixture's open() returning E_ALREADY_EXIST, and the test body then
segfaulted on the unopened writer. The name is seeded from the wall clock,
and cases of one suite run as separate processes in a shared directory
(ctest --parallel), so two of them can land on the same seed and race on
open() with O_EXCL. Seed from random_device instead, and assert the open so
a failure is reported rather than crashing the body.

Author: ColinLeeo

cpp/src/common/container/byte_buffer.h

@@ -133,6 +133,14 @@ class ByteBuffer {
 
     FORCE_INLINE uint32_t get_data_size() const { return real_data_size_; }
 
+    // Declare bytes that were written directly through get_data() without going
+    // through append_*(). Used when a caller decodes straight into this buffer
+    // at a known offset.
+    FORCE_INLINE void set_data_size(uint32_t size) {
+        ASSERT(size <= reserved_size_);
+        real_data_size_ = size;
+    }
+
    private:
     char* data_;
     uint8_t variable_type_len_;

cpp/src/reader/aligned_chunk_reader.cc

@@ -49,6 +49,9 @@ struct ChunkPageInfo {
     uint32_t time_uncompressed_size = 0;
     int32_t row_begin = 0;  // inclusive
     int32_t row_end = 0;    // exclusive
+    // Row index of this page's first retained row in the chunk reader's output
+    // stream. Filled by build_page_plan() once the whole plan is known.
+    int32_t out_row_begin = 0;
     std::vector<int64_t> value_file_offsets;
     std::vector<uint32_t> value_compressed_sizes;
     std::vector<uint32_t> value_uncompressed_sizes;
@@ -65,6 +68,17 @@ struct PageDecodedState {
     common::PageArena predecode_pa;
     int32_t predecoded_count = 0;
     int32_t predecoded_read_pos = 0;
+    // Non-null when the page was decoded straight into the destination TsBlock
+    // column buffer (see AlignedChunkReader::prepare_direct_decode()). Aliases
+    // memory owned by the Vector; predecoded_values stays empty in that case.
+    char* direct_values = nullptr;
+
+    // Where this slot's fixed-length values live: the direct view when present,
+    // otherwise the slot's own buffer.
+    char* values() {
+        return direct_values != nullptr ? direct_values
+                                        : predecoded_values.data();
+    }
 };
 
 // Per-value-column state for multi-value AlignedChunkReader.
@@ -177,6 +191,15 @@ class AlignedChunkReader : public IChunkReader {
 
     bool is_multi_value_mode() const { return multi_value_mode_; }
 
+#ifdef ENABLE_TEST
+    // Pages scattered by the bulk-copy path so far (FULL_PASS pages and
+    // BOUNDARY pages a contiguous time filter proves fully matched).
+    size_t TEST_bulk_page_count() const { return bulk_page_count_; }
+    // Leading pages of the current plan that decode straight into the
+    // destination block's column buffers (see prepare_direct_decode()).
+    size_t TEST_direct_page_end() const { return direct_page_end_; }
+#endif
+
 #ifdef ENABLE_THREADS
     // Set external thread pool for parallel decode (not owned).
     void set_decode_pool(common::ThreadPool* pool) { decode_pool_ = pool; }
@@ -274,6 +297,7 @@ class AlignedChunkReader : public IChunkReader {
                               common::RowAppender& row_appender, Filter* filter,
                               common::PageArena* pa);
     int build_page_plan(Filter* filter, int& row_offset);
+    bool prepare_direct_decode(common::TsBlock* tsblock, Filter* filter);
     void build_filter_column_mask();
     int evaluate_page_field_mask(size_t page_idx);
     int decode_output_columns_for_page(size_t page_idx);
@@ -346,6 +370,22 @@ class AlignedChunkReader : public IChunkReader {
     SelectionMask current_page_field_mask_;
     bool current_page_field_mask_valid_ = false;
 
+    // Direct decode: pages [0, direct_page_end_) are decoded straight into the
+    // destination Vector memory instead of a per-page staging buffer. See
+    // prepare_direct_decode().
+    size_t direct_page_end_ = 0;
+    int32_t direct_base_row_ = 0;
+    common::TsBlock* direct_tsblock_ = nullptr;
+    // Per value column: byte offset in the Vector's buffer where the next
+    // decoded value goes. TsBlock columns store non-null values compactly
+    // (RowIterator advances offsets only for non-null rows), so this advances
+    // by the non-null count of each page, not by its row index.
+    std::vector<size_t> direct_col_off_;
+
+#ifdef ENABLE_TEST
+    size_t bulk_page_count_ = 0;
+#endif
+
 #ifdef ENABLE_THREADS
     common::ThreadPool* decode_pool_ = nullptr;  // borrowed, not owned
     // Per-task time decoder + compressor pool for parallel time-page decode.

cpp/src/reader/aligned_chunk_reader.h

@@ -64,6 +64,14 @@ class Filter {
         return nullptr;
     }
 
+    // True when this time filter selects one contiguous interval: inside a
+    // BOUNDARY page's [row_begin, row_end) range every timestamp matches (e.g.
+    // TimeBetween, one-sided Gt/Lt), so the range can be bulk-copied. Discrete
+    // sets (TimeIn) or negated forms (NOT BETWEEN / NotEq) may contain holes
+    // and must still be checked per value. Defaults to false; only explicitly
+    // contiguous time filters override it to true.
+    virtual bool is_contiguous_time_filter() const { return false; }
+
     // Batch time filter: evaluate time filter on an array of timestamps.
     // Writes true/false into @mask for each element.
     // Returns the number of elements that passed (mask[i] == true).

cpp/src/reader/filter/filter.h

@@ -52,6 +52,8 @@ class TimeBetween : public Filter {
 
     FilterType get_filter_type() { return type_; }
 
+    bool is_contiguous_time_filter() const override { return !not_; }
+
    private:
     int64_t value1_;
     int64_t value2_;