perf(cpp): reuse one output TsBlock across the devices of a scan

cpp/src/reader/block/device_ordered_tsblock_reader.cc

@@ -69,6 +69,7 @@ int DeviceOrderedTsBlockReader::has_next(bool& has_next) {
         current_reader_ = new SingleDeviceTsBlockReader(
             task, block_size_, metadata_querier_, tsfile_io_reader_,
             time_filter_, field_filter_);
+        current_reader_->set_reusable_block(&reusable_block_);
         if (current_reader_ == nullptr) {
             return common::E_OOM;
         }

cpp/src/reader/block/device_ordered_tsblock_reader.h

@@ -58,6 +58,10 @@ class DeviceOrderedTsBlockReader : public TsBlockReader {
     Filter* field_filter_ = nullptr;
     int remaining_offset_ = 0;
     int remaining_limit_ = -1;
+    // Output block shared by the per-device readers this iterator creates.
+    // Each device otherwise allocates (and faults in) its own block and column
+    // buffers; the layout is identical across devices, so one block is enough.
+    ReusableTsBlock reusable_block_;
 };
 }  // namespace storage
 

cpp/src/reader/block/single_device_tsblock_reader.cc

@@ -30,7 +30,12 @@ namespace storage {
 
 namespace {
 const char* kTimeOnlyContextName = "__time_only_aligned_context__";
+
+common::TsBlock* reusable_block_factory(void* ctx, common::TupleDesc& desc,
+                                        uint32_t max_rows) {
+    return static_cast<ReusableTsBlock*>(ctx)->acquire(desc, max_rows);
 }
+}  // namespace
 
 SingleDeviceTsBlockReader::SingleDeviceTsBlockReader(
     DeviceQueryTask* device_query_task, uint32_t block_size,
@@ -283,9 +288,9 @@ int SingleDeviceTsBlockReader::init_internal(DeviceQueryTask* device_query_task,
                  dense_row_count_ >= 0)
                     ? remaining_offset_
                     : 0;
-            int ctx_ret =
-                ctx->init(device_query_task_, meas_names, time_filter, pos_list,
-                          pa_, aligned_field_filter_, ssi_offset, -1);
+            int ctx_ret = ctx->init(device_query_task_, meas_names, time_filter,
+                                    pos_list, pa_, aligned_field_filter_,
+                                    ssi_offset, -1, reusable_block_);
             if (ctx_ret == common::E_OK) {
                 if (ssi_offset > 0) {
                     row_offset_pushed_to_ssi_ = true;
@@ -1209,7 +1214,8 @@ int VectorMeasurementColumnContext::init(
     DeviceQueryTask* device_query_task,
     const std::vector<std::string>& measurement_names, Filter* time_filter,
     std::vector<std::vector<int32_t>>& pos_in_result, common::PageArena& pa,
-    AlignedFieldFilter* field_filter, int ssi_offset, int ssi_limit) {
+    AlignedFieldFilter* field_filter, int ssi_offset, int ssi_limit,
+    ReusableTsBlock* reusable_block) {
     int ret = common::E_OK;
     pos_in_result_ = pos_in_result;
     column_names_ = measurement_names;
@@ -1218,6 +1224,9 @@ int VectorMeasurementColumnContext::init(
             time_filter, field_filter))) {
         return ret;
     }
+    if (reusable_block != nullptr) {
+        ssi_->set_tsblock_factory(reusable_block_factory, reusable_block);
+    }
     ssi_->set_row_range(ssi_offset, ssi_limit);
     if (RET_FAIL(get_next_tsblock(true))) {
         return ret;

cpp/src/reader/block/single_device_tsblock_reader.h

@@ -33,6 +33,77 @@ class DeviceQueryTask;
 class MeasurementColumnContext;
 class IdColumnContext;
 
+// One output TsBlock shared by every device a reader visits. A device-ordered
+// scan builds the same time+value layout for each series; allocating a fresh
+// block (and faulting in its per-column buffers) per device is a large share of
+// the query cost once the table is wide, so the layout is kept and only rebuilt
+// when the column count or value types change.
+class ReusableTsBlock {
+   public:
+    ReusableTsBlock() {}
+    ~ReusableTsBlock() { release(); }
+    ReusableTsBlock(const ReusableTsBlock&) = delete;
+    ReusableTsBlock& operator=(const ReusableTsBlock&) = delete;
+
+    // Returns a block matching `wanted` (time + value columns), reusing the
+    // previous buffers when the shape matches. The caller resets it before use.
+    common::TsBlock* acquire(common::TupleDesc& wanted, uint32_t max_rows) {
+        bool rebuild = block_ == nullptr || max_rows_ != max_rows ||
+                       desc_.get_column_count() != wanted.get_column_count();
+        if (!rebuild) {
+            for (uint32_t i = 0; i < wanted.get_column_count(); i++) {
+                if (desc_[i].data_type_ != wanted[i].data_type_) {
+                    rebuild = true;
+                    break;
+                }
+            }
+        }
+        if (rebuild) {
+            release();
+            for (uint32_t i = 0; i < wanted.get_column_count(); i++) {
+                desc_.push_back(wanted[i]);
+            }
+            max_rows_ = max_rows;
+            block_ = new common::TsBlock(&desc_, max_rows_);
+            if (block_->init() != common::E_OK) {
+                delete block_;
+                block_ = nullptr;
+            }
+#ifdef ENABLE_TEST
+            ++build_count_;
+#endif
+            return block_;
+        }
+        // Same layout: refresh names in place, the block points at desc_.
+        for (uint32_t i = 0; i < wanted.get_column_count(); i++) {
+            desc_[i] = wanted[i];
+        }
+        return block_;
+    }
+
+    void release() {
+        if (block_ != nullptr) {
+            delete block_;
+            block_ = nullptr;
+        }
+        desc_.release();
+        max_rows_ = 0;
+    }
+
+#ifdef ENABLE_TEST
+    // Blocks allocated so far; a reuse of a matching layout does not count.
+    size_t TEST_build_count() const { return build_count_; }
+#endif
+
+   private:
+    common::TupleDesc desc_;
+    common::TsBlock* block_ = nullptr;
+    uint32_t max_rows_ = 0;
+#ifdef ENABLE_TEST
+    size_t build_count_ = 0;
+#endif
+};
+
 class SingleDeviceTsBlockReader : public TsBlockReader {
    public:
     explicit SingleDeviceTsBlockReader(DeviceQueryTask* device_query_task,
@@ -51,6 +122,9 @@ class SingleDeviceTsBlockReader : public TsBlockReader {
              int row_limit);
     void close() override;
 
+    // Borrow a block that can be reused across devices (see ReusableTsBlock).
+    void set_reusable_block(ReusableTsBlock* block) { reusable_block_ = block; }
+
     int get_remaining_offset() const { return remaining_offset_; }
     int get_remaining_limit() const { return remaining_limit_; }
     int32_t get_dense_row_count() const { return dense_row_count_; }
@@ -103,6 +177,10 @@ class SingleDeviceTsBlockReader : public TsBlockReader {
     SelectionMask aligned_filter_mask_;
     uint32_t aligned_filter_next_row_ = 0;
     bool aligned_filter_mask_valid_ = false;
+    // Output block owned by the enclosing device-ordered reader and handed to
+    // this device's aligned multi-value reader, so the per-column buffers
+    // survive across devices. Null when this reader owns nothing (standalone).
+    ReusableTsBlock* reusable_block_ = nullptr;
 };
 
 class MeasurementColumnContext {
@@ -205,7 +283,8 @@ class VectorMeasurementColumnContext final : public MeasurementColumnContext {
              Filter* time_filter,
              std::vector<std::vector<int32_t>>& pos_in_result,
              common::PageArena& pa, AlignedFieldFilter* field_filter = nullptr,
-             int ssi_offset = 0, int ssi_limit = -1);
+             int ssi_offset = 0, int ssi_limit = -1,
+             ReusableTsBlock* reusable_block = nullptr);
     int get_next_tsblock(bool alloc_mem) override;
     int get_current_time(int64_t& time) override;
     int get_current_value(char*& value, uint32_t& len) override;

cpp/src/reader/tsfile_series_scan_iterator.cc

@@ -363,8 +363,7 @@ int TsFileSeriesScanIterator::get_next(TsBlock*& ret_tsblock, bool alloc,
         }
         if (IS_SUCC(ret)) {
             if (alloc && ret_tsblock == nullptr) {
-                ret_tsblock =
-                    is_multi_value_ ? alloc_tsblock_multi() : alloc_tsblock();
+                ret_tsblock = alloc_or_reuse_tsblock();
             }
             ret = chunk_reader_->get_next_page(ret_tsblock, filter, *data_pa_,
                                                min_time_hint, row_offset_,
@@ -540,6 +539,35 @@ int TsFileSeriesScanIterator::init_chunk_reader_multi() {
     return ret;
 }
 
+TsBlock* TsFileSeriesScanIterator::alloc_or_reuse_tsblock() {
+    if (!is_multi_value_ || tsblock_factory_ == nullptr) {
+        return is_multi_value_ ? alloc_tsblock_multi() : alloc_tsblock();
+    }
+    auto* acr = static_cast<AlignedChunkReader*>(chunk_reader_);
+    // Same layout alloc_tsblock_multi() would build: time + one column per
+    // value measurement of the series now loaded.
+    tuple_desc_.reset();
+    tuple_desc_.push_back(ColumnSchema("time", common::INT64,
+                                       common::INVALID_COMPRESSION,
+                                       common::INVALID_ENCODING));
+    const uint32_t num_cols = acr->get_value_column_count();
+    for (uint32_t c = 0; c < num_cols; c++) {
+        ChunkHeader& ch = acr->get_value_chunk_header(c);
+        tuple_desc_.push_back(ColumnSchema(ch.measurement_name_, ch.data_type_,
+                                           ch.compression_type_,
+                                           ch.encoding_type_));
+    }
+    common::TsBlock* block =
+        tsblock_factory_(tsblock_factory_ctx_, tuple_desc_, max_block_rows_);
+    if (block != nullptr) {
+        block->reset();
+        return block;
+    }
+    // The factory could not match this layout; a fresh block is allocated and
+    // owned by this iterator (destroyed by revert_tsblock()/destroy()).
+    return alloc_tsblock_multi();
+}
+
 TsBlock* TsFileSeriesScanIterator::alloc_tsblock() {
     ChunkHeader& ch = chunk_reader_->get_chunk_header();
 

cpp/src/reader/tsfile_series_scan_iterator.h

@@ -96,6 +96,20 @@ class TsFileSeriesScanIterator {
         max_block_rows_ = max_block_rows;
     }
 
+    // Optional factory for the output TsBlock. A device-ordered scan reads one
+    // series at a time and would otherwise allocate (and fault in) a fresh
+    // block plus all its per-column buffers for every device. The factory is
+    // handed the exact layout the block must have and may return a matching
+    // block that is reused across series; returning nullptr makes the iterator
+    // allocate a new one. The returned block must already be reset().
+    using TsBlockFactory = common::TsBlock* (*)(void* ctx,
+                                                common::TupleDesc& desc,
+                                                uint32_t max_block_rows);
+    void set_tsblock_factory(TsBlockFactory factory, void* ctx) {
+        tsblock_factory_ = factory;
+        tsblock_factory_ctx_ = ctx;
+    }
+
     /** Current row offset/limit after chunk/page skip; used to sync with QDS
      * for single-path. */
     int get_row_offset() const { return row_offset_; }
@@ -215,6 +229,8 @@ class TsFileSeriesScanIterator {
         ChunkMeta* time_cm, const std::vector<ChunkMeta*>& value_cms);
     common::TsBlock* alloc_tsblock();
     common::TsBlock* alloc_tsblock_multi();
+    // Reuse a block supplied by tsblock_factory_, or allocate a new one.
+    common::TsBlock* alloc_or_reuse_tsblock();
 
    private:
     RandomAccessReadFile* read_file_;
@@ -247,6 +263,8 @@ class TsFileSeriesScanIterator {
     uint32_t max_block_rows_;
     int row_offset_;
     int row_limit_;
+    TsBlockFactory tsblock_factory_ = nullptr;
+    void* tsblock_factory_ctx_ = nullptr;
 };
 
 }  // end namespace storage