diff --git a/cpp/src/reader/block/device_ordered_tsblock_reader.cc b/cpp/src/reader/block/device_ordered_tsblock_reader.cc index 5c30d9bc6..0ccde9144 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 8018725c9..0190d11a8 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 619201e63..19e6dd16d 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 fb592a6c4..dede6b627 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 948d1a553..adc1bf130 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 19c261694..53c63058c 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 000000000..7031e0e60 --- /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