From 93cf2ca6b68936331577864066691f135d9f7e4e Mon Sep 17 00:00:00 2001 From: ColinLee Date: Thu, 12 Nov 2026 18:38:48 +0800 Subject: [PATCH] perf(cpp): reuse one output TsBlock across the devices of a scan A device-ordered table scan creates a SingleDeviceTsBlockReader per device, and each of them allocated its own output TsBlock: one Vector (and one column buffer) per projected column, faulted in again for every device. The layout is the same for every device of a table, so the allocation only has to happen once. DeviceOrderedTsBlockReader now owns a ReusableTsBlock and hands it to each per-device reader, which passes it down to the multi-value aligned column context; the series scan iterator asks it for the block through a factory instead of allocating one. The block is rebuilt only when the row budget, the column count or a column type changes, and a renamed column updates the tuple description in place. Every acquire is followed by reset(), so each device starts from an empty block. The returned block stays owned by the reader and is reused across devices, as TableResultSet already consumes one block before requesting the next. Reading a 200-column table over 30 devices of 20 rows takes 68.0 ms instead of 71.9 ms per query (mean of 3 runs; the gain grows with the column count, but per-device index loading still dominates the fixed cost). --- .../block/device_ordered_tsblock_reader.cc | 1 + .../block/device_ordered_tsblock_reader.h | 4 + .../block/single_device_tsblock_reader.cc | 17 +- .../block/single_device_tsblock_reader.h | 81 ++++++- cpp/src/reader/tsfile_series_scan_iterator.cc | 32 ++- cpp/src/reader/tsfile_series_scan_iterator.h | 18 ++ .../reader/block/reusable_tsblock_test.cc | 226 ++++++++++++++++++ 7 files changed, 372 insertions(+), 7 deletions(-) create mode 100644 cpp/test/reader/block/reusable_tsblock_test.cc diff --git a/cpp/src/reader/block/device_ordered_tsblock_reader.cc b/cpp/src/reader/block/device_ordered_tsblock_reader.cc index 5c30d9bc6a..0ccde9144a 100644 --- a/cpp/src/reader/block/device_ordered_tsblock_reader.cc +++ b/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; } diff --git a/cpp/src/reader/block/device_ordered_tsblock_reader.h b/cpp/src/reader/block/device_ordered_tsblock_reader.h index 8018725c9b..0190d11a8b 100644 --- a/cpp/src/reader/block/device_ordered_tsblock_reader.h +++ b/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 diff --git a/cpp/src/reader/block/single_device_tsblock_reader.cc b/cpp/src/reader/block/single_device_tsblock_reader.cc index 619201e636..19e6dd16d5 100644 --- a/cpp/src/reader/block/single_device_tsblock_reader.cc +++ b/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(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& measurement_names, Filter* time_filter, std::vector>& 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; diff --git a/cpp/src/reader/block/single_device_tsblock_reader.h b/cpp/src/reader/block/single_device_tsblock_reader.h index fb592a6c4a..dede6b627e 100644 --- a/cpp/src/reader/block/single_device_tsblock_reader.h +++ b/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>& 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; diff --git a/cpp/src/reader/tsfile_series_scan_iterator.cc b/cpp/src/reader/tsfile_series_scan_iterator.cc index 948d1a5538..adc1bf1305 100644 --- a/cpp/src/reader/tsfile_series_scan_iterator.cc +++ b/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(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(); diff --git a/cpp/src/reader/tsfile_series_scan_iterator.h b/cpp/src/reader/tsfile_series_scan_iterator.h index 19c2616945..53c63058c9 100644 --- a/cpp/src/reader/tsfile_series_scan_iterator.h +++ b/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& 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 diff --git a/cpp/test/reader/block/reusable_tsblock_test.cc b/cpp/test/reader/block/reusable_tsblock_test.cc new file mode 100644 index 0000000000..7031e0e600 --- /dev/null +++ b/cpp/test/reader/block/reusable_tsblock_test.cc @@ -0,0 +1,226 @@ +/* + * 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. + */ +#include + +#include +#include +#include +#include +#include + +#include "common/config/config.h" +#include "common/tablet.h" +#include "common/tsblock/tsblock.h" +#include "file/local_random_access_read_file.h" +#include "file/tsfile_io_reader.h" +#include "file/write_file.h" +#include "reader/block/single_device_tsblock_reader.h" +#include "reader/task/device_query_task.h" +#include "writer/tsfile_table_writer.h" + +namespace { +using namespace common; +using namespace storage; + +constexpr int kRows = 40; +constexpr int64_t kBaseTime = 1000; + +TupleDesc MakeDesc(const char* second_name, TSDataType second_type) { + TupleDesc desc; + desc.push_back(ColumnSchema("time", INT64, UNCOMPRESSED, PLAIN)); + desc.push_back(ColumnSchema(second_name, second_type, UNCOMPRESSED, PLAIN)); + return desc; +} + +TEST(ReusableTsBlockTest, ReusesTheBlockForTheSameLayout) { + ReusableTsBlock reusable; + TupleDesc desc = MakeDesc("v", INT64); + TsBlock* first = reusable.acquire(desc, 32); + ASSERT_NE(first, nullptr); + EXPECT_EQ(first->get_column_count(), 2u); + EXPECT_EQ(first->get_max_row_count(), 32u); + + RowAppender rows(first); + for (int i = 0; i < 5; i++) { + ASSERT_TRUE(rows.add_row()); + int64_t t = i; + rows.append(0, reinterpret_cast(&t), sizeof(t)); + } + first->reset(); + EXPECT_EQ(first->get_row_count(), 0u); + + EXPECT_EQ(reusable.acquire(desc, 32), first); + EXPECT_EQ(reusable.TEST_build_count(), 1u); +} + +// Devices of one table share the column types but not their names; a rename +// must not throw the buffers away. +TEST(ReusableTsBlockTest, RenamedColumnKeepsTheBlock) { + ReusableTsBlock reusable; + TupleDesc first_desc = MakeDesc("v", INT64); + TsBlock* first = reusable.acquire(first_desc, 16); + ASSERT_NE(first, nullptr); + + TupleDesc renamed = MakeDesc("renamed", INT64); + TsBlock* second = reusable.acquire(renamed, 16); + EXPECT_EQ(second, first); + EXPECT_EQ(reusable.TEST_build_count(), 1u); + ASSERT_EQ(second->get_column_count(), 2u); + EXPECT_EQ(second->get_tuple_desc()->get_column_name(1), "renamed"); +} + +TEST(ReusableTsBlockTest, RebuildsOnADifferentLayout) { + ReusableTsBlock reusable; + TupleDesc two_columns = MakeDesc("v", INT64); + ASSERT_NE(reusable.acquire(two_columns, 16), nullptr); + EXPECT_EQ(reusable.TEST_build_count(), 1u); + + // Row budget changed: the buffers were sized for the old one. + TsBlock* wider = reusable.acquire(two_columns, 64); + ASSERT_NE(wider, nullptr); + EXPECT_EQ(wider->get_max_row_count(), 64u); + EXPECT_EQ(reusable.TEST_build_count(), 2u); + + // Column type changed. + TupleDesc other_type = MakeDesc("v", DOUBLE); + ASSERT_NE(reusable.acquire(other_type, 64), nullptr); + EXPECT_EQ(reusable.TEST_build_count(), 3u); + + // Column count changed. + TupleDesc three_columns = MakeDesc("v", DOUBLE); + three_columns.push_back(ColumnSchema("w", DOUBLE, UNCOMPRESSED, PLAIN)); + ASSERT_NE(reusable.acquire(three_columns, 64), nullptr); + EXPECT_EQ(reusable.TEST_build_count(), 4u); + + // Releasing frees the block; the next acquire builds a new one. + reusable.release(); + ASSERT_NE(reusable.acquire(two_columns, 16), nullptr); + EXPECT_EQ(reusable.TEST_build_count(), 5u); +} + +class ReusableTsBlockScanTest : public ::testing::Test { + protected: + void SetUp() override { + libtsfile_init(); + saved_ = g_config_value_; + // 8 rows per TsBlock (time + v + w = 24 bytes per row) so one scan + // asks the factory for several blocks, and 8 rows per page. + g_config_value_.tsblock_max_memory_ = 24 * 8; + g_config_value_.page_writer_max_point_num_ = 8; + filename_ = + std::string("reusable_tsblock_") + + ::testing::UnitTest::GetInstance()->current_test_info()->name() + + ".tsfile"; + std::remove(filename_.c_str()); + const std::vector columns{ + {"v", INT64, UNCOMPRESSED, PLAIN, ColumnCategory::FIELD}, + {"w", INT64, UNCOMPRESSED, PLAIN, ColumnCategory::FIELD}}; + schema_ = std::make_shared("t", columns); + { + WriteFile file; + int flags = O_WRONLY | O_CREAT | O_TRUNC; +#ifdef _WIN32 + flags |= O_BINARY; +#endif + ASSERT_EQ(file.create(filename_, flags, 0666), E_OK); + TsFileTableWriter writer(&file, schema_.get()); + Tablet tablet("t", {"v", "w"}, {INT64, INT64}, + {ColumnCategory::FIELD, ColumnCategory::FIELD}, + kRows); + for (int r = 0; r < kRows; r++) { + ASSERT_EQ(tablet.add_timestamp(r, kBaseTime + r), E_OK); + ASSERT_EQ(tablet.add_value(r, "v", static_cast(r)), + E_OK); + ASSERT_EQ(tablet.add_value(r, "w", static_cast(r) * 2), + E_OK); + } + ASSERT_EQ(writer.write_table(tablet), E_OK); + ASSERT_EQ(writer.close(), E_OK); + } + ASSERT_EQ(read_file_.open(filename_), E_OK); + ASSERT_EQ(reader_.init(&read_file_), E_OK); + TsFileMeta* meta = nullptr; + ASSERT_EQ(reader_.get_tsfile_meta(meta), E_OK); + } + + void TearDown() override { + std::remove(filename_.c_str()); + g_config_value_ = saved_; + libtsfile_destroy(); + } + + // Reads every row of the table once, exactly as a device-ordered scan does + // for one device: the series scan iterator asks @reusable for its output + // block. + std::vector scan(ReusableTsBlock& reusable) { + PageArena arena; + arena.init(512, MOD_TSFILE_READER); + auto device = std::make_shared( + std::vector{"t"}); + std::vector names{"v", "w"}; + DeviceQueryTask task(device, names, std::make_shared(), + nullptr, schema_); + std::vector> positions{{0}, {1}}; + VectorMeasurementColumnContext context(&reader_); + EXPECT_EQ(context.init(&task, names, nullptr, positions, arena, nullptr, + 0, -1, &reusable), + common::E_OK); + std::vector times; + while (context.available_rows() > 0) { + int64_t time = 0; + EXPECT_EQ(context.get_current_time(time), common::E_OK); + char* value = nullptr; + uint32_t len = 0; + EXPECT_EQ(context.get_current_value(value, len), common::E_OK); + int64_t v = 0; + std::memcpy(&v, value, sizeof(v)); + EXPECT_EQ(v, time - kBaseTime); + times.push_back(time); + const int moved = context.move_iter(); + if (moved == common::E_NO_MORE_DATA) { + break; + } + EXPECT_EQ(moved, common::E_OK); + } + return times; + } + + ConfigValue saved_; + std::string filename_; + std::shared_ptr schema_; + LocalRandomAccessReadFile read_file_; + TsFileIOReader reader_; +}; + +// Each device of a device-ordered scan gets its own series scan iterator but +// shares the output block: the second scan must see the same buffer, already +// reset, and produce the same rows. +TEST_F(ReusableTsBlockScanTest, ConsecutiveDeviceScansShareOneBlock) { + ReusableTsBlock reusable; + for (int device = 0; device < 3; device++) { + const std::vector times = scan(reusable); + ASSERT_EQ(times.size(), static_cast(kRows)) << device; + for (int r = 0; r < kRows; r++) { + EXPECT_EQ(times[r], kBaseTime + r) << device; + } + } + EXPECT_EQ(reusable.TEST_build_count(), 1u); +} + +} // namespace