This is an automated email from the ASF dual-hosted git repository.
ColinLeeo pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/tsfile.git
The following commit(s) were added to refs/heads/develop by this push:
new 7a7d7f96e perf(cpp): push down multi-value aligned offsets (#863)
7a7d7f96e is described below
commit 7a7d7f96e1015de2774ec936f498df4e05bce293
Author: Colin Lee <[email protected]>
AuthorDate: Tue Jul 21 09:46:26 2026 +0800
perf(cpp): push down multi-value aligned offsets (#863)
---
cpp/src/reader/aligned_chunk_reader.cc | 140 +++++--
cpp/src/reader/aligned_chunk_reader.h | 21 +-
.../reader/block/single_device_tsblock_reader.cc | 64 ++-
.../reader/block/single_device_tsblock_reader.h | 12 +-
cpp/src/reader/chunk_reader.cc | 21 +-
cpp/src/reader/chunk_reader.h | 3 +-
cpp/src/reader/tsfile_series_scan_iterator.cc | 60 ++-
cpp/src/reader/tsfile_series_scan_iterator.h | 8 +
cpp/test/reader/tsfile_reader_test.cc | 461 ++++++++++++++++++++-
9 files changed, 697 insertions(+), 93 deletions(-)
diff --git a/cpp/src/reader/aligned_chunk_reader.cc
b/cpp/src/reader/aligned_chunk_reader.cc
index f02497363..4b89e659b 100644
--- a/cpp/src/reader/aligned_chunk_reader.cc
+++ b/cpp/src/reader/aligned_chunk_reader.cc
@@ -333,7 +333,9 @@ int AlignedChunkReader::alloc_compressor_and_decoder(
int AlignedChunkReader::get_next_page(TsBlock* ret_tsblock,
Filter* oneshoot_filter, PageArena& pa) {
if (multi_value_mode_) {
- return get_next_page_multi(ret_tsblock, oneshoot_filter, pa);
+ int row_offset = 0;
+ return get_next_page_multi(ret_tsblock, oneshoot_filter, pa,
+ row_offset);
}
int ret = E_OK;
Filter* filter =
@@ -353,7 +355,7 @@ int AlignedChunkReader::get_next_page(TsBlock* ret_tsblock,
value_chunk_meta_, value_in_stream_,
cur_value_page_header_, value_chunk_visit_offset_,
value_chunk_header_))) {
- } else if (cur_page_statisify_filter(filter)) {
+ } else if (cur_page_may_satisfy_filter(filter)) {
break;
} else if (RET_FAIL(skip_cur_page())) {
}
@@ -467,7 +469,10 @@ int AlignedChunkReader::read_from_file_and_rewrap(
return ret;
}
-bool AlignedChunkReader::cur_page_statisify_filter(Filter* filter) {
+// Page statistics provide two levels of certainty: "may satisfy" means the
+// page cannot be ruled out, while "fully satisfies" means every row is known
+// to match and the page count can safely be used for offset pushdown.
+bool AlignedChunkReader::cur_page_may_satisfy_filter(Filter* filter) {
bool value_satisfy = filter == nullptr ||
cur_value_page_header_.statistic_ == nullptr ||
filter->satisfy(cur_value_page_header_.statistic_);
@@ -477,6 +482,16 @@ bool AlignedChunkReader::cur_page_statisify_filter(Filter*
filter) {
return time_satisfy && value_satisfy;
}
+bool AlignedChunkReader::cur_page_fully_satisfies_filter(Filter* filter) {
+ Statistic* stat = cur_time_page_header_.statistic_;
+ if (stat == nullptr) {
+ stat = cur_value_page_header_.statistic_;
+ }
+ return filter == nullptr ||
+ (stat != nullptr &&
+ filter->contain_start_end_time(stat->start_time_,
stat->end_time_));
+}
+
int AlignedChunkReader::skip_cur_page() {
int ret = E_OK;
// visit a page tv data
@@ -1082,17 +1097,19 @@ int AlignedChunkReader::get_next_page(TsBlock*
ret_tsblock,
int64_t min_time_hint, int& row_offset,
int& row_limit) {
if (multi_value_mode_) {
- // Multi-value aligned path doesn't yet honour row_offset / row_limit
- // / min_time_hint — they get dropped on the floor, which silently
- // returns full chunk data when the caller asked for a sub-range.
- // Refuse the combination so the caller sees an actual error instead
- // of garbage results. set_row_range(0, -1) keeps the all-rows
- // contract intact for normal queries.
- if (row_offset > 0 || row_limit >= 0 ||
+ if (row_limit == 0) {
+ return E_NO_MORE_DATA;
+ }
+ // The multi-value path can consume offset through chunk/page count
+ // statistics. Limit and min-time pushdown still need separate state
+ // handling, so keep those combinations explicit instead of silently
+ // returning a wider range.
+ if (row_limit >= 0 ||
min_time_hint != std::numeric_limits<int64_t>::min()) {
return common::E_NOT_SUPPORT;
}
- return get_next_page_multi(ret_tsblock, oneshoot_filter, pa);
+ return get_next_page_multi(ret_tsblock, oneshoot_filter, pa,
+ row_offset);
}
int ret = E_OK;
Filter* filter =
@@ -1118,13 +1135,14 @@ int AlignedChunkReader::get_next_page(TsBlock*
ret_tsblock,
value_chunk_meta_, value_in_stream_,
cur_value_page_header_, value_chunk_visit_offset_,
value_chunk_header_))) {
- } else if (!cur_page_statisify_filter(filter)) {
+ } else if (!cur_page_may_satisfy_filter(filter)) {
if (RET_FAIL(skip_cur_page())) {
}
} else if (should_skip_page_by_time(min_time_hint)) {
if (RET_FAIL(skip_cur_page())) {
}
- } else if (should_skip_page_by_offset(row_offset)) {
+ } else if (cur_page_fully_satisfies_filter(filter) &&
+ should_skip_page_by_offset(row_offset)) {
if (RET_FAIL(skip_cur_page())) {
}
} else {
@@ -1383,7 +1401,7 @@ int AlignedChunkReader::decode_time_page_with(const
ChunkPageInfo& page_info,
return ret;
}
-int AlignedChunkReader::build_page_plan(Filter* filter) {
+int AlignedChunkReader::build_page_plan(Filter* filter, int& row_offset) {
int ret = E_OK;
chunk_pages_.clear();
current_page_plan_index_ = 0;
@@ -1459,6 +1477,13 @@ int AlignedChunkReader::build_page_plan(Filter* filter) {
int32_t last = -1;
for (int32_t i = 0; i < static_cast<int32_t>(times.size()); i++) {
if (filter->satisfy_start_end_time(times[i], times[i])) {
+ // Offset is defined on the filtered row stream. Consume
+ // matching rows only; holes inside a boundary page do not
+ // count toward it.
+ if (row_offset > 0) {
+ row_offset--;
+ continue;
+ }
if (first < 0) first = i;
last = i;
}
@@ -1479,6 +1504,18 @@ int AlignedChunkReader::build_page_plan(Filter* filter) {
}
page_info.row_end = static_cast<int32_t>(times.size());
}
+ if (page_info.pass_type == PagePassType::FULL_PASS &&
+ row_offset > 0) {
+ const int32_t page_row_count =
+ page_info.row_end - page_info.row_begin;
+ const int32_t rows_to_skip =
+ std::min(page_row_count, row_offset);
+ page_info.row_begin += rows_to_skip;
+ row_offset -= rows_to_skip;
+ if (page_info.row_begin == page_info.row_end) {
+ page_info.pass_type = PagePassType::SKIP;
+ }
+ }
if (page_info.row_begin < page_info.row_end) {
chunk_pages_.push_back(std::move(page_info));
}
@@ -1833,34 +1870,36 @@ void AlignedChunkReader::release_page_slot(size_t
page_idx) {
int AlignedChunkReader::get_next_page_multi(TsBlock* ret_tsblock,
Filter* oneshoot_filter,
- PageArena& pa) {
+ PageArena& pa, int& row_offset) {
int ret = E_OK;
Filter* filter =
(oneshoot_filter != nullptr ? oneshoot_filter : time_filter_);
// Dispatch:
- // - Multi-column with a thread pool → chunk-level pre-decode: one task
- // per value column decodes that column's whole chunk up front, then
the
- // scatter loop bulk-memcpys. decode_all_planned_pages() works for any
- // column count. (An earlier cutoff sent >6 columns down the serial
- // path because per_page_state — the upfront predecode buffer — grows
- // with column count and was feared to thrash cache; it still grows, so
- // very wide aligned chunks are the case to watch if reads regress.)
- // - Single column, or no thread pool → serial path: decode the current
- // page's columns inline (multi_DECODE_TV_BATCH), no thread-pool
- // fan-out.
+ // - Offset pushdown always uses the page plan so fully covered pages and
+ // the prefix of the first retained page can be removed before value
+ // decoding, even when no worker pool is available.
+ // - Multi-column with a thread pool also uses the page plan for
+ // chunk-level parallel pre-decode.
+ // - Otherwise decode the current page's columns inline.
#ifdef ENABLE_THREADS
- const bool use_chunk_level =
+ const bool use_parallel_page_plan =
decode_pool_ != nullptr && value_columns_.size() > 1;
#else
- const bool use_chunk_level = false;
+ const bool use_parallel_page_plan = false;
#endif
- if (!use_chunk_level) {
- return get_next_page_multi_serial(ret_tsblock, filter, pa);
+ const bool serial_page_in_progress =
+ !page_plan_built_ &&
+ (prev_time_page_not_finish() ||
prev_any_value_page_not_finish_multi());
+ const bool use_page_plan =
+ page_plan_built_ || (!serial_page_in_progress &&
+ (row_offset > 0 || use_parallel_page_plan));
+ if (!use_page_plan) {
+ return get_next_page_multi_serial(ret_tsblock, filter, pa, row_offset);
}
if (!page_plan_built_) {
- if (RET_FAIL(build_page_plan(filter))) {
+ if (RET_FAIL(build_page_plan(filter, row_offset))) {
return ret;
}
if (RET_FAIL(decode_all_planned_pages())) {
@@ -2044,7 +2083,8 @@ int AlignedChunkReader::get_next_page_multi(TsBlock*
ret_tsblock,
int AlignedChunkReader::get_next_page_multi_serial(TsBlock* ret_tsblock,
Filter* filter,
- PageArena& pa) {
+ PageArena& pa,
+ int& row_offset) {
int ret = E_OK;
bool pt = prev_time_page_not_finish();
bool pv = prev_any_value_page_not_finish_multi();
@@ -2069,8 +2109,14 @@ int
AlignedChunkReader::get_next_page_multi_serial(TsBlock* ret_tsblock,
}
}
if (IS_FAIL(ret)) break;
- if (cur_page_statisify_filter_multi(filter)) break;
- if (RET_FAIL(skip_cur_page_multi())) break;
+ if (!cur_page_may_satisfy_filter_multi(filter)) {
+ if (RET_FAIL(skip_cur_page_multi())) break;
+ } else if (cur_page_fully_satisfies_filter_multi(filter) &&
+ should_skip_page_by_offset_multi(row_offset)) {
+ if (RET_FAIL(skip_cur_page_multi())) break;
+ } else {
+ break;
+ }
if (!has_more_data()) {
ret = E_NO_MORE_DATA;
break;
@@ -2088,13 +2134,39 @@ int
AlignedChunkReader::get_next_page_multi_serial(TsBlock* ret_tsblock,
return ret;
}
-bool AlignedChunkReader::cur_page_statisify_filter_multi(Filter* filter) {
+// Keep the same conservative distinction as the single-value path: a page
+// that may satisfy the filter still needs row-level filtering unless its full
+// time range is covered by the filter.
+bool AlignedChunkReader::cur_page_may_satisfy_filter_multi(Filter* filter) {
bool time_satisfy = filter == nullptr ||
cur_time_page_header_.statistic_ == nullptr ||
filter->satisfy(cur_time_page_header_.statistic_);
return time_satisfy;
}
+bool AlignedChunkReader::cur_page_fully_satisfies_filter_multi(Filter* filter)
{
+ Statistic* stat = cur_time_page_header_.statistic_;
+ return filter == nullptr ||
+ (stat != nullptr &&
+ filter->contain_start_end_time(stat->start_time_,
stat->end_time_));
+}
+
+bool AlignedChunkReader::should_skip_page_by_offset_multi(int& row_offset) {
+ if (row_offset <= 0) {
+ return false;
+ }
+ Statistic* stat = cur_time_page_header_.statistic_;
+ if (stat == nullptr || stat->count_ == 0) {
+ return false;
+ }
+ int32_t count = stat->count_;
+ if (row_offset >= count) {
+ row_offset -= count;
+ return true;
+ }
+ return false;
+}
+
int AlignedChunkReader::skip_cur_page_multi() {
time_chunk_visit_offset_ += cur_time_page_header_.compressed_size_;
time_in_stream_.wrapped_buf_advance_read_pos(
diff --git a/cpp/src/reader/aligned_chunk_reader.h
b/cpp/src/reader/aligned_chunk_reader.h
index b92c4d7b5..45dd2ef99 100644
--- a/cpp/src/reader/aligned_chunk_reader.h
+++ b/cpp/src/reader/aligned_chunk_reader.h
@@ -36,7 +36,7 @@ class ThreadPool;
namespace storage {
-// Page classification for chunk-level parallel decode.
+// Page classification for chunk-level planned decode.
enum class PagePassType { SKIP, FULL_PASS, BOUNDARY };
// Metadata collected per page during the chunk scan phase.
@@ -53,8 +53,9 @@ struct ChunkPageInfo {
std::vector<uint32_t> value_uncompressed_sizes;
};
-// Decoded state for one (column, page) slot. Populated by chunk-level
-// parallel decode; consumed by the scatter loop.
+// Decoded state for one (column, page) slot. Populated eagerly by parallel
+// decode or lazily by the single-thread page-plan path, then consumed by the
+// scatter loop.
struct PageDecodedState {
std::vector<uint8_t> notnull_bitmap;
std::vector<char> predecoded_values;
@@ -194,7 +195,8 @@ class AlignedChunkReader : public IChunkReader {
uint32_t& chunk_visit_offset,
int32_t& file_data_buf_size,
int want_size = 0, bool may_shrink = true);
- bool cur_page_statisify_filter(Filter* filter);
+ bool cur_page_may_satisfy_filter(Filter* filter);
+ bool cur_page_fully_satisfies_filter(Filter* filter);
int skip_cur_page();
int decode_cur_time_page_data();
int decode_cur_value_page_data();
@@ -238,11 +240,14 @@ class AlignedChunkReader : public IChunkReader {
bool has_more_data_multi() const;
bool prev_any_value_page_not_finish_multi() const;
int get_next_page_multi(common::TsBlock* ret_tsblock,
- Filter* oneshoot_filter, common::PageArena& pa);
+ Filter* oneshoot_filter, common::PageArena& pa,
+ int& row_offset);
int get_next_page_multi_serial(common::TsBlock* ret_tsblock, Filter*
filter,
- common::PageArena& pa);
+ common::PageArena& pa, int& row_offset);
int skip_cur_page_multi();
- bool cur_page_statisify_filter_multi(Filter* filter);
+ bool cur_page_may_satisfy_filter_multi(Filter* filter);
+ bool cur_page_fully_satisfies_filter_multi(Filter* filter);
+ bool should_skip_page_by_offset_multi(int& row_offset);
int decode_cur_value_pages_multi();
int decode_cur_value_page_data_for(ValueColumnState& col);
int ensure_value_page_loaded(ValueColumnState& col);
@@ -255,7 +260,7 @@ class AlignedChunkReader : public IChunkReader {
int multi_DECODE_TV_BATCH(common::TsBlock* ret_tsblock,
common::RowAppender& row_appender, Filter*
filter,
common::PageArena* pa);
- int build_page_plan(Filter* filter);
+ int build_page_plan(Filter* filter, int& row_offset);
int decode_time_page_direct(const ChunkPageInfo& page_info,
std::vector<int64_t>& out_times);
int decode_time_page_with(const ChunkPageInfo& page_info,
diff --git a/cpp/src/reader/block/single_device_tsblock_reader.cc
b/cpp/src/reader/block/single_device_tsblock_reader.cc
index d842a35d9..5d66bb7e9 100644
--- a/cpp/src/reader/block/single_device_tsblock_reader.cc
+++ b/cpp/src/reader/block/single_device_tsblock_reader.cc
@@ -47,6 +47,7 @@ int SingleDeviceTsBlockReader::init(DeviceQueryTask*
device_query_task,
remaining_offset_ = 0;
remaining_limit_ = -1;
dense_row_count_ = -1;
+ row_offset_pushed_to_ssi_ = false;
return init_internal(device_query_task, block_size, time_filter,
field_filter);
}
@@ -58,6 +59,7 @@ int SingleDeviceTsBlockReader::init(DeviceQueryTask*
device_query_task,
remaining_offset_ = row_offset;
remaining_limit_ = row_limit;
dense_row_count_ = -1;
+ row_offset_pushed_to_ssi_ = false;
return init_internal(device_query_task, block_size, time_filter,
field_filter);
}
@@ -162,15 +164,6 @@ int
SingleDeviceTsBlockReader::init_internal(DeviceQueryTask* device_query_task,
}
}
time_column_index_ = 0;
- if (RET_FAIL(common::TsBlock::create_tsblock(&tuple_desc_, current_block_,
- block_size))) {
- return ret;
- }
- col_appenders_.resize(tuple_desc_.get_column_count());
- for (uint32_t i = 0; i < tuple_desc_.get_column_count(); i++) {
- col_appenders_[i] = new common::ColAppender(i, current_block_);
- }
- row_appender_ = new common::RowAppender(current_block_);
std::vector<ITimeseriesIndex*> time_series_indexs(
device_query_task_->get_column_mapping()
->get_measurement_columns()
@@ -182,6 +175,11 @@ int
SingleDeviceTsBlockReader::init_internal(DeviceQueryTask* device_query_task,
return ret;
}
dense_row_count_ = compute_dense_row_count(time_series_indexs);
+ if (time_filter == nullptr && field_filter == nullptr &&
+ dense_row_count_ >= 0 && remaining_offset_ >= dense_row_count_) {
+ remaining_offset_ -= dense_row_count_;
+ return common::E_OK;
+ }
// Early device-level time skip: if time_filter is set and ALL chunks of
// this device have statistics that fall outside the filter range, skip the
// entire device. Chunks without statistics are assumed to satisfy.
@@ -217,8 +215,6 @@ int
SingleDeviceTsBlockReader::init_internal(DeviceQueryTask* device_query_task,
}
if (examined_any && all_outside) {
// No data in this device matches the time filter.
- delete current_block_;
- current_block_ = nullptr;
return common::E_OK;
}
}
@@ -278,8 +274,20 @@ int
SingleDeviceTsBlockReader::init_internal(DeviceQueryTask* device_query_task,
}
auto* ctx = new VectorMeasurementColumnContext(tsfile_io_reader_);
+ const int ssi_offset =
+ (time_filter == nullptr && field_filter == nullptr &&
+ dense_row_count_ >= 0)
+ ? remaining_offset_
+ : 0;
if (common::E_OK == ctx->init(device_query_task_, meas_names,
- time_filter, pos_list, pa_)) {
+ time_filter, pos_list, pa_,
+ ssi_offset, -1)) {
+ if (ssi_offset > 0) {
+ row_offset_pushed_to_ssi_ = true;
+ // init() prefetches the first TsBlock and may consume part
+ // of the offset by skipping whole chunks/pages.
+ remaining_offset_ = ctx->get_ssi_row_offset();
+ }
// The shared ctx is referenced from N map entries; close()
// and the merge loop dedupe by pointer (already in place).
for (const auto& name : meas_names) {
@@ -326,8 +334,6 @@ int
SingleDeviceTsBlockReader::init_internal(DeviceQueryTask* device_query_task,
}
}
if (any_value_column_requested) {
- delete current_block_;
- current_block_ = nullptr;
return common::E_OK;
}
@@ -348,8 +354,6 @@ int
SingleDeviceTsBlockReader::init_internal(DeviceQueryTask* device_query_task,
// propagate so the caller sees the actual failure instead of
// an empty resultset wearing E_OK.
if (time_only_ret != common::E_NO_MORE_DATA) {
- delete current_block_;
- current_block_ = nullptr;
return time_only_ret;
}
}
@@ -373,11 +377,23 @@ int
SingleDeviceTsBlockReader::init_internal(DeviceQueryTask* device_query_task,
}
if (field_column_contexts_.empty()) {
- delete current_block_;
- current_block_ = nullptr;
return common::E_OK;
}
+ // Metadata and context initialization above can prove that a device has no
+ // rows to return. Delay result-buffer allocation until those early exits
+ // have been ruled out, so skipped devices never create appenders that
+ // briefly point at a deleted TsBlock.
+ if (RET_FAIL(common::TsBlock::create_tsblock(&tuple_desc_, current_block_,
+ block_size))) {
+ return ret;
+ }
+ col_appenders_.resize(tuple_desc_.get_column_count());
+ for (uint32_t i = 0; i < tuple_desc_.get_column_count(); i++) {
+ col_appenders_[i] = new common::ColAppender(i, current_block_);
+ }
+ row_appender_ = new common::RowAppender(current_block_);
+
for (const auto& id_column :
device_query_task->get_column_mapping()->get_id_columns()) {
const auto& column_pos_in_result =
@@ -540,7 +556,11 @@ int SingleDeviceTsBlockReader::has_next_aligned(bool&
result_has_next) {
int sr = ctx->skip_rows(skip);
if (sr != common::E_OK) return sr;
}
- remaining_offset_ -= skip;
+ if (row_offset_pushed_to_ssi_ && !aligned_vec_.empty()) {
+ remaining_offset_ = aligned_vec_[0]->get_ssi_row_offset();
+ } else {
+ remaining_offset_ -= skip;
+ }
continue;
}
@@ -973,6 +993,7 @@ int SingleMeasurementColumnContext::skip_rows(uint32_t
count) {
const uint32_t val_elem_size = common::get_data_type_size(dt);
value_iter_->advance(to_skip, val_elem_size);
}
+ consume_ssi_row_offset(static_cast<int>(to_skip));
if (time_iter_->end()) {
// Propagate hard errors from the next-tsblock load; E_NO_MORE_DATA
// is the legitimate end-of-stream signal and gets squashed back to
@@ -1004,7 +1025,8 @@
VectorMeasurementColumnContext::~VectorMeasurementColumnContext() {
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) {
+ std::vector<std::vector<int32_t>>& pos_in_result, common::PageArena& pa,
+ int ssi_offset, int ssi_limit) {
int ret = common::E_OK;
pos_in_result_ = pos_in_result;
column_names_ = measurement_names;
@@ -1013,6 +1035,7 @@ int VectorMeasurementColumnContext::init(
time_filter))) {
return ret;
}
+ ssi_->set_row_range(ssi_offset, ssi_limit);
if (RET_FAIL(get_next_tsblock(true))) {
return ret;
}
@@ -1224,6 +1247,7 @@ int VectorMeasurementColumnContext::skip_rows(uint32_t
count) {
}
}
}
+ consume_ssi_row_offset(static_cast<int>(to_skip));
if (time_iter_->end()) {
int r = get_next_tsblock(false);
if (r != common::E_OK && r != common::E_NO_MORE_DATA) return r;
diff --git a/cpp/src/reader/block/single_device_tsblock_reader.h
b/cpp/src/reader/block/single_device_tsblock_reader.h
index e74304baf..4bf0cca0f 100644
--- a/cpp/src/reader/block/single_device_tsblock_reader.h
+++ b/cpp/src/reader/block/single_device_tsblock_reader.h
@@ -74,7 +74,7 @@ class SingleDeviceTsBlockReader : public TsBlockReader {
uint32_t block_size_;
common::TsBlock* current_block_ = nullptr;
std::vector<common::ColAppender*> col_appenders_;
- common::RowAppender* row_appender_;
+ common::RowAppender* row_appender_ = nullptr;
common::TupleDesc tuple_desc_;
bool last_block_returned_ = true;
std::map<std::string, MeasurementColumnContext*> field_column_contexts_;
@@ -86,6 +86,7 @@ class SingleDeviceTsBlockReader : public TsBlockReader {
int remaining_offset_ = 0;
int remaining_limit_ = -1;
int32_t dense_row_count_ = -1;
+ bool row_offset_pushed_to_ssi_ = false;
// Populated in init() when every field column comes from an aligned chunk.
// Provides cache-friendly vector iteration for has_next_aligned().
bool all_aligned_ = false;
@@ -114,14 +115,11 @@ class MeasurementColumnContext {
virtual int move_iter() = 0;
- virtual void set_ssi_row_range(int offset, int limit) {
- if (ssi_) ssi_->set_row_range(offset, limit);
- }
virtual int get_ssi_row_offset() const {
return ssi_ ? ssi_->get_row_offset() : 0;
}
- virtual int get_ssi_row_limit() const {
- return ssi_ ? ssi_->get_row_limit() : -1;
+ virtual void consume_ssi_row_offset(int count) {
+ if (ssi_) ssi_->consume_row_offset(count);
}
virtual uint32_t available_rows() const = 0;
@@ -195,7 +193,7 @@ class VectorMeasurementColumnContext final : public
MeasurementColumnContext {
const std::vector<std::string>& measurement_names,
Filter* time_filter,
std::vector<std::vector<int32_t>>& pos_in_result,
- common::PageArena& pa);
+ common::PageArena& pa, int ssi_offset = 0, int ssi_limit = -1);
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/chunk_reader.cc b/cpp/src/reader/chunk_reader.cc
index 7c36ea07f..6b3d853d9 100644
--- a/cpp/src/reader/chunk_reader.cc
+++ b/cpp/src/reader/chunk_reader.cc
@@ -192,7 +192,7 @@ int ChunkReader::get_next_page(TsBlock* ret_tsblock,
Filter* oneshoot_filter,
return E_NO_MORE_DATA;
}
if (RET_FAIL(get_cur_page_header())) {
- } else if (cur_page_statisify_filter(filter)) {
+ } else if (cur_page_may_satisfy_filter(filter)) {
break;
} else if (RET_FAIL(skip_cur_page())) {
}
@@ -265,11 +265,21 @@ int ChunkReader::read_from_file_and_rewrap(int want_size)
{
return ret;
}
-bool ChunkReader::cur_page_statisify_filter(Filter* filter) {
+// Page statistics provide two levels of certainty: "may satisfy" means the
+// page cannot be ruled out, while "fully satisfies" means every row is known
+// to match and the page count can safely be used for offset pushdown.
+bool ChunkReader::cur_page_may_satisfy_filter(Filter* filter) {
return filter == nullptr || cur_page_header_.statistic_ == nullptr ||
filter->satisfy(cur_page_header_.statistic_);
}
+bool ChunkReader::cur_page_fully_satisfies_filter(Filter* filter) {
+ return filter == nullptr || (cur_page_header_.statistic_ != nullptr &&
+ filter->contain_start_end_time(
+ cur_page_header_.statistic_->start_time_,
+ cur_page_header_.statistic_->end_time_));
+}
+
int ChunkReader::skip_cur_page() {
int ret = E_OK;
// visit a page tv data
@@ -904,13 +914,14 @@ int ChunkReader::get_next_page(TsBlock* ret_tsblock,
Filter* oneshoot_filter,
return E_NO_MORE_DATA;
}
if (RET_FAIL(get_cur_page_header())) {
- } else if (!cur_page_statisify_filter(filter)) {
+ } else if (!cur_page_may_satisfy_filter(filter)) {
if (RET_FAIL(skip_cur_page())) {
}
} else if (should_skip_page_by_time(min_time_hint)) {
if (RET_FAIL(skip_cur_page())) {
}
- } else if (should_skip_page_by_offset(row_offset)) {
+ } else if (cur_page_fully_satisfies_filter(filter) &&
+ should_skip_page_by_offset(row_offset)) {
if (RET_FAIL(skip_cur_page())) {
}
} else {
@@ -924,4 +935,4 @@ int ChunkReader::get_next_page(TsBlock* ret_tsblock,
Filter* oneshoot_filter,
return ret;
}
-} // end namespace storage
\ No newline at end of file
+} // end namespace storage
diff --git a/cpp/src/reader/chunk_reader.h b/cpp/src/reader/chunk_reader.h
index a1196c330..b92be8169 100644
--- a/cpp/src/reader/chunk_reader.h
+++ b/cpp/src/reader/chunk_reader.h
@@ -87,7 +87,8 @@ class ChunkReader : public IChunkReader {
common::CompressionType compression_type);
int get_cur_page_header();
int read_from_file_and_rewrap(int want_size = 0);
- bool cur_page_statisify_filter(Filter* filter);
+ bool cur_page_may_satisfy_filter(Filter* filter);
+ bool cur_page_fully_satisfies_filter(Filter* filter);
int skip_cur_page();
int decode_cur_page_data(common::TsBlock*& ret_tsblock, Filter* filter,
common::PageArena& pa);
diff --git a/cpp/src/reader/tsfile_series_scan_iterator.cc
b/cpp/src/reader/tsfile_series_scan_iterator.cc
index 6e318b38e..79ac8d308 100644
--- a/cpp/src/reader/tsfile_series_scan_iterator.cc
+++ b/cpp/src/reader/tsfile_series_scan_iterator.cc
@@ -30,6 +30,21 @@ using namespace common;
namespace storage {
+namespace {
+bool chunk_may_satisfy_filter(ChunkMeta* chunk_meta, Filter* filter) {
+ return filter == nullptr || chunk_meta == nullptr ||
+ chunk_meta->statistic_ == nullptr ||
+ filter->satisfy(chunk_meta->statistic_);
+}
+
+bool chunk_fully_satisfies_filter(ChunkMeta* chunk_meta, Filter* filter) {
+ return filter == nullptr ||
+ (chunk_meta != nullptr && chunk_meta->statistic_ != nullptr &&
+ filter->contain_start_end_time(chunk_meta->statistic_->start_time_,
+ chunk_meta->statistic_->end_time_));
+}
+} // namespace
+
void TsFileSeriesScanIterator::destroy() {
// MultiAlignedTimeseriesIndex is placement-new'd inside
// timeseries_index_pa_ (see TsFileIOReader::alloc_multi_ssi). The arena's
@@ -122,6 +137,31 @@ bool
TsFileSeriesScanIterator::should_skip_aligned_chunk_by_offset(
return false;
}
+bool TsFileSeriesScanIterator::should_skip_multi_aligned_chunk_by_offset(
+ ChunkMeta* time_cm, const std::vector<ChunkMeta*>& value_cms) {
+ if (row_offset_ <= 0) {
+ return false;
+ }
+ if (time_cm == nullptr || time_cm->statistic_ == nullptr) {
+ return false;
+ }
+ int32_t time_count = time_cm->statistic_->count_;
+ if (time_count <= 0) {
+ return false;
+ }
+ for (const auto* value_cm : value_cms) {
+ if (value_cm == nullptr || value_cm->statistic_ == nullptr ||
+ value_cm->statistic_->count_ != time_count) {
+ return false;
+ }
+ }
+ if (row_offset_ >= time_count) {
+ row_offset_ -= time_count;
+ return true;
+ }
+ return false;
+}
+
int TsFileSeriesScanIterator::get_next(TsBlock*& ret_tsblock, bool alloc,
Filter* oneshoot_filter,
int64_t min_time_hint) {
@@ -151,13 +191,17 @@ int TsFileSeriesScanIterator::get_next(TsBlock*&
ret_tsblock, bool alloc,
}
advance_to_next_chunk();
// Skip chunk by time filter using time chunk statistics.
- if (filter != nullptr && time_cm->statistic_ != nullptr &&
- !filter->satisfy(time_cm->statistic_)) {
+ if (!chunk_may_satisfy_filter(time_cm, filter)) {
continue;
}
if (should_skip_chunk_by_time(time_cm, min_time_hint)) {
continue;
}
+ if (chunk_fully_satisfies_filter(time_cm, filter) &&
+ should_skip_multi_aligned_chunk_by_offset(time_cm,
+ value_cms)) {
+ continue;
+ }
chunk_reader_->reset();
auto* acr =
static_cast<AlignedChunkReader*>(chunk_reader_);
if (RET_FAIL(acr->load_by_aligned_meta_multi(time_cm,
@@ -167,8 +211,7 @@ int TsFileSeriesScanIterator::get_next(TsBlock*&
ret_tsblock, bool alloc,
} else if (!is_aligned_) {
ChunkMeta* cm = get_current_chunk_meta();
advance_to_next_chunk();
- if (filter != nullptr && cm->statistic_ != nullptr &&
- !filter->satisfy(cm->statistic_)) {
+ if (!chunk_may_satisfy_filter(cm, filter)) {
continue;
}
// Skip by min_time_hint (merge cursor).
@@ -176,7 +219,8 @@ int TsFileSeriesScanIterator::get_next(TsBlock*&
ret_tsblock, bool alloc,
continue;
}
// Single-path: skip entire chunk by offset using count.
- if (should_skip_chunk_by_offset(cm)) {
+ if (chunk_fully_satisfies_filter(cm, filter) &&
+ should_skip_chunk_by_offset(cm)) {
continue;
}
chunk_reader_->reset();
@@ -190,14 +234,14 @@ int TsFileSeriesScanIterator::get_next(TsBlock*&
ret_tsblock, bool alloc,
// Use time chunk statistics for time-based filtering.
ChunkMeta* filter_cm =
(time_cm->statistic_ != nullptr) ? time_cm : value_cm;
- if (filter != nullptr && filter_cm->statistic_ != nullptr
&&
- !filter->satisfy(filter_cm->statistic_)) {
+ if (!chunk_may_satisfy_filter(filter_cm, filter)) {
continue;
}
if (should_skip_chunk_by_time(filter_cm, min_time_hint)) {
continue;
}
- if (should_skip_aligned_chunk_by_offset(time_cm,
+ if (chunk_fully_satisfies_filter(time_cm, filter) &&
+ should_skip_aligned_chunk_by_offset(time_cm,
value_cm)) {
continue;
}
diff --git a/cpp/src/reader/tsfile_series_scan_iterator.h
b/cpp/src/reader/tsfile_series_scan_iterator.h
index 68f1a1f32..cb3832787 100644
--- a/cpp/src/reader/tsfile_series_scan_iterator.h
+++ b/cpp/src/reader/tsfile_series_scan_iterator.h
@@ -81,6 +81,12 @@ class TsFileSeriesScanIterator {
* for single-path. */
int get_row_offset() const { return row_offset_; }
int get_row_limit() const { return row_limit_; }
+ void consume_row_offset(int count) {
+ if (count <= 0 || row_offset_ <= 0) {
+ return;
+ }
+ row_offset_ = count >= row_offset_ ? 0 : row_offset_ - count;
+ }
/*
* If oneshoot filter specified, use it instead of this->time_filter_.
@@ -180,6 +186,8 @@ class TsFileSeriesScanIterator {
bool should_skip_chunk_by_offset(ChunkMeta* cm);
bool should_skip_aligned_chunk_by_offset(ChunkMeta* time_cm,
ChunkMeta* value_cm);
+ bool should_skip_multi_aligned_chunk_by_offset(
+ ChunkMeta* time_cm, const std::vector<ChunkMeta*>& value_cms);
common::TsBlock* alloc_tsblock();
common::TsBlock* alloc_tsblock_multi();
diff --git a/cpp/test/reader/tsfile_reader_test.cc
b/cpp/test/reader/tsfile_reader_test.cc
index 2e74e870f..f7df9c8c9 100644
--- a/cpp/test/reader/tsfile_reader_test.cc
+++ b/cpp/test/reader/tsfile_reader_test.cc
@@ -42,6 +42,23 @@
using namespace storage;
using namespace common;
+TEST(TsFileSeriesScanIteratorTest, ConsumeRowOffsetSaturates) {
+ storage::TsFileSeriesScanIterator ssi;
+ ssi.set_row_range(/*offset=*/10, /*limit=*/-1);
+
+ ssi.consume_row_offset(-1);
+ EXPECT_EQ(ssi.get_row_offset(), 10);
+ ssi.consume_row_offset(std::numeric_limits<int>::min());
+ EXPECT_EQ(ssi.get_row_offset(), 10);
+
+ ssi.consume_row_offset(4);
+ EXPECT_EQ(ssi.get_row_offset(), 6);
+ ssi.consume_row_offset(6);
+ EXPECT_EQ(ssi.get_row_offset(), 0);
+ ssi.consume_row_offset(1);
+ EXPECT_EQ(ssi.get_row_offset(), 0);
+}
+
class TsFileReaderTest : public ::testing::Test {
protected:
void SetUp() override {
@@ -463,12 +480,10 @@ TEST_F(TsFileReaderTest,
reader.close();
}
-// Multi-value aligned chunk reader doesn't honour row_offset / row_limit /
-// min_time_hint pushdown — silently dropping those args would hand the caller
-// full-chunk data when it asked for a sub-range. The guard at the top of
-// AlignedChunkReader::get_next_page must turn the unsupported combination
-// into an explicit E_NOT_SUPPORT.
-TEST_F(TsFileReaderTest, MultiValueAlignedRowOffsetReturnsNotSupport) {
+// The multi-value page plan consumes row_offset before value decoding. When
+// the offset lands inside a page, the returned TsBlock starts at the first row
+// after the offset instead of exposing a residual to the row reader.
+TEST_F(TsFileReaderTest, MultiValueAlignedRowOffsetTrimsPartialPage) {
const std::string device = "root.dev_multi_offset";
std::vector<MeasurementSchema> schema_vec;
schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED);
@@ -505,12 +520,21 @@ TEST_F(TsFileReaderTest,
MultiValueAlignedRowOffsetReturnsNotSupport) {
E_OK);
ASSERT_NE(ssi, nullptr);
- // row_offset > 0 hits the multi-value guard at the top of
- // AlignedChunkReader::get_next_page; the SSI propagates the error code.
ssi->set_row_range(/*offset=*/5, /*limit=*/-1);
common::TsBlock* block = nullptr;
- EXPECT_EQ(ssi->get_next(block, /*alloc_tsblock=*/true),
- common::E_NOT_SUPPORT);
+ EXPECT_EQ(ssi->get_next(block, /*alloc_tsblock=*/true), common::E_OK);
+ ASSERT_NE(block, nullptr);
+ EXPECT_EQ(block->get_row_count(), static_cast<uint32_t>(N - 5));
+ EXPECT_EQ(ssi->get_row_offset(), 0);
+ {
+ common::ColIterator time_iter(0, block);
+ common::ColIterator v0_iter(1, block);
+ common::ColIterator v1_iter(2, block);
+ uint32_t len = 0;
+ EXPECT_EQ(*reinterpret_cast<int64_t*>(time_iter.read(&len)), 1005);
+ EXPECT_EQ(*reinterpret_cast<int64_t*>(v0_iter.read(&len)), 5);
+ EXPECT_EQ(*reinterpret_cast<int64_t*>(v1_iter.read(&len)), 10);
+ }
if (block != nullptr) {
ssi->revert_tsblock();
@@ -521,6 +545,423 @@ TEST_F(TsFileReaderTest,
MultiValueAlignedRowOffsetReturnsNotSupport) {
// it, then ~TsFileMeta would call self_deleter on freed memory.
}
+TEST_F(TsFileReaderTest, MultiValueAlignedRowOffsetSkipsWholePage) {
+ struct PagePointGuard {
+ explicit PagePointGuard(uint32_t page_points)
+ : saved_(common::g_config_value_.page_writer_max_point_num_) {
+ common::g_config_value_.page_writer_max_point_num_ = page_points;
+ }
+ ~PagePointGuard() {
+ common::g_config_value_.page_writer_max_point_num_ = saved_;
+ }
+ uint32_t saved_;
+ } page_point_guard(16);
+
+ const std::string device = "root.dev_multi_page_offset";
+ std::vector<MeasurementSchema> schema_vec;
+ schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED);
+ schema_vec.emplace_back("v1", INT64, PLAIN, UNCOMPRESSED);
+ {
+ std::vector<MeasurementSchema*> reg;
+ for (auto& s : schema_vec) reg.push_back(new MeasurementSchema(s));
+ ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device, reg),
+ E_OK);
+ }
+
+ const int rows_per_page = 16;
+ const int N = rows_per_page * 4;
+ Tablet tablet(device,
+ std::make_shared<std::vector<MeasurementSchema>>(schema_vec),
+ N);
+ for (int i = 0; i < N; ++i) {
+ ASSERT_EQ(tablet.add_timestamp(i, static_cast<int64_t>(i)), E_OK);
+ ASSERT_EQ(tablet.add_value(i, 0u, static_cast<int64_t>(i)), E_OK);
+ ASSERT_EQ(tablet.add_value(i, 1u, static_cast<int64_t>(i * 2)), E_OK);
+ }
+ ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK);
+ ASSERT_EQ(tsfile_writer_->flush(), E_OK);
+ ASSERT_EQ(tsfile_writer_->close(), E_OK);
+
+ storage::TsFileIOReader io_reader;
+ ASSERT_EQ(io_reader.init(file_name_), E_OK);
+
+ auto device_id = std::make_shared<StringArrayDeviceID>(device);
+ std::vector<std::string> measurements = {"v0", "v1"};
+ storage::TsFileSeriesScanIterator* ssi = nullptr;
+ common::PageArena pa;
+ pa.init(512, common::MOD_TSFILE_READER);
+ ASSERT_EQ(io_reader.alloc_multi_ssi(device_id, measurements, ssi, pa,
+ /*time_filter=*/nullptr),
+ E_OK);
+ ASSERT_NE(ssi, nullptr);
+
+ // The first chunk is preloaded before set_row_range(). The page plan skips
+ // page 0 and trims the first four rows from page 1 before value decoding.
+ ssi->set_row_range(/*offset=*/20, /*limit=*/-1);
+ common::TsBlock* block = nullptr;
+ ASSERT_EQ(ssi->get_next(block, /*alloc_tsblock=*/true), common::E_OK);
+ ASSERT_NE(block, nullptr);
+ ASSERT_EQ(block->get_row_count(), static_cast<uint32_t>(N - 20));
+ EXPECT_EQ(ssi->get_row_offset(), 0);
+
+ {
+ common::ColIterator time_iter(0, block);
+ common::ColIterator v0_iter(1, block);
+ common::ColIterator v1_iter(2, block);
+ uint32_t len = 0;
+ EXPECT_EQ(*reinterpret_cast<int64_t*>(time_iter.read(&len)), 20);
+ EXPECT_EQ(*reinterpret_cast<int64_t*>(v0_iter.read(&len)), 20);
+ EXPECT_EQ(*reinterpret_cast<int64_t*>(v1_iter.read(&len)), 40);
+ }
+
+ ssi->revert_tsblock();
+ io_reader.revert_ssi(ssi);
+}
+
+TEST_F(TsFileReaderTest, MultiValueAlignedRowOffsetSkipsWholeChunk) {
+ const std::string device = "root.dev_multi_chunk_offset";
+ std::vector<MeasurementSchema> schema_vec;
+ schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED);
+ schema_vec.emplace_back("v1", INT64, PLAIN, UNCOMPRESSED);
+ {
+ std::vector<MeasurementSchema*> reg;
+ for (auto& s : schema_vec) reg.push_back(new MeasurementSchema(s));
+ ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device, reg),
+ E_OK);
+ }
+
+ const int rows_per_chunk = 64;
+ const int chunk_count = 4;
+ for (int chunk = 0; chunk < chunk_count; ++chunk) {
+ Tablet tablet(
+ device,
+ std::make_shared<std::vector<MeasurementSchema>>(schema_vec),
+ rows_per_chunk);
+ const int base = chunk * rows_per_chunk;
+ for (int i = 0; i < rows_per_chunk; ++i) {
+ const int row = base + i;
+ ASSERT_EQ(tablet.add_timestamp(i, static_cast<int64_t>(row)),
E_OK);
+ ASSERT_EQ(tablet.add_value(i, 0u, static_cast<int64_t>(row)),
E_OK);
+ ASSERT_EQ(tablet.add_value(i, 1u, static_cast<int64_t>(row * 2)),
+ E_OK);
+ }
+ ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK);
+ ASSERT_EQ(tsfile_writer_->flush(), E_OK);
+ }
+ ASSERT_EQ(tsfile_writer_->close(), E_OK);
+
+ storage::TsFileIOReader io_reader;
+ ASSERT_EQ(io_reader.init(file_name_), E_OK);
+
+ auto device_id = std::make_shared<StringArrayDeviceID>(device);
+ std::vector<std::string> measurements = {"v0", "v1"};
+ storage::TsFileSeriesScanIterator* ssi = nullptr;
+ common::PageArena pa;
+ pa.init(512, common::MOD_TSFILE_READER);
+ ASSERT_EQ(io_reader.alloc_multi_ssi(device_id, measurements, ssi, pa,
+ /*time_filter=*/nullptr),
+ E_OK);
+ ASSERT_NE(ssi, nullptr);
+
+ // alloc_multi_ssi() preloads chunk 0 before row range is set. Read it
+ // normally first, then set an offset that skips the next whole chunk by
+ // chunk metadata count and trims the first 32 rows from chunk 2's plan.
+ common::TsBlock* block = nullptr;
+ ASSERT_EQ(ssi->get_next(block, /*alloc_tsblock=*/true), common::E_OK);
+ ASSERT_NE(block, nullptr);
+ ASSERT_EQ(block->get_row_count(), static_cast<uint32_t>(rows_per_chunk));
+ {
+ common::ColIterator time_iter(0, block);
+ common::ColIterator v0_iter(1, block);
+ common::ColIterator v1_iter(2, block);
+ uint32_t len = 0;
+ EXPECT_EQ(*reinterpret_cast<int64_t*>(time_iter.read(&len)), 0);
+ EXPECT_EQ(*reinterpret_cast<int64_t*>(v0_iter.read(&len)), 0);
+ EXPECT_EQ(*reinterpret_cast<int64_t*>(v1_iter.read(&len)), 0);
+ }
+ ssi->revert_tsblock();
+
+ ssi->set_row_range(/*offset=*/96, /*limit=*/-1);
+ block = nullptr;
+ ASSERT_EQ(ssi->get_next(block, /*alloc_tsblock=*/true), common::E_OK);
+ ASSERT_NE(block, nullptr);
+ ASSERT_EQ(block->get_row_count(),
+ static_cast<uint32_t>(rows_per_chunk / 2));
+ EXPECT_EQ(ssi->get_row_offset(), 0);
+
+ {
+ common::ColIterator time_iter(0, block);
+ common::ColIterator v0_iter(1, block);
+ common::ColIterator v1_iter(2, block);
+ uint32_t len = 0;
+ EXPECT_EQ(*reinterpret_cast<int64_t*>(time_iter.read(&len)), 160);
+ EXPECT_EQ(*reinterpret_cast<int64_t*>(v0_iter.read(&len)), 160);
+ EXPECT_EQ(*reinterpret_cast<int64_t*>(v1_iter.read(&len)), 320);
+ }
+
+ ssi->revert_tsblock();
+ io_reader.revert_ssi(ssi);
+}
+
+// Offset is defined over the rows left after filtering. A partially matching
+// chunk therefore cannot consume its full time-row count from the offset.
+TEST_F(TsFileReaderTest, MultiValueAlignedRowOffsetCountsOnlyFilteredRows) {
+ const std::string device = "root.dev_multi_filtered_offset";
+ std::vector<MeasurementSchema> schema_vec;
+ schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED);
+ schema_vec.emplace_back("v1", INT64, PLAIN, UNCOMPRESSED);
+ {
+ std::vector<MeasurementSchema*> reg;
+ for (auto& s : schema_vec) reg.push_back(new MeasurementSchema(s));
+ ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device, reg),
+ E_OK);
+ }
+
+ const int rows_per_chunk = 64;
+ const int chunk_count = 4;
+ for (int chunk = 0; chunk < chunk_count; ++chunk) {
+ Tablet tablet(
+ device,
+ std::make_shared<std::vector<MeasurementSchema>>(schema_vec),
+ rows_per_chunk);
+ const int base = chunk * rows_per_chunk;
+ for (int i = 0; i < rows_per_chunk; ++i) {
+ const int row = base + i;
+ ASSERT_EQ(tablet.add_timestamp(i, static_cast<int64_t>(row)),
E_OK);
+ ASSERT_EQ(tablet.add_value(i, 0u, static_cast<int64_t>(row)),
E_OK);
+ ASSERT_EQ(tablet.add_value(i, 1u, static_cast<int64_t>(row * 2)),
+ E_OK);
+ }
+ ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK);
+ ASSERT_EQ(tsfile_writer_->flush(), E_OK);
+ }
+ ASSERT_EQ(tsfile_writer_->close(), E_OK);
+
+ storage::TsFileIOReader io_reader;
+ ASSERT_EQ(io_reader.init(file_name_), E_OK);
+
+ auto device_id = std::make_shared<StringArrayDeviceID>(device);
+ std::vector<std::string> measurements = {"v0", "v1"};
+ storage::TsFileSeriesScanIterator* ssi = nullptr;
+ common::PageArena pa;
+ pa.init(512, common::MOD_TSFILE_READER);
+
+ // Keep all 64 rows from chunk 0, only 10 rows from chunk 1, and all 128
+ // rows from chunks 2 and 3. The filtered stream has 202 rows, so offset
+ // 200 must leave only timestamps 254 and 255.
+ std::vector<int64_t> selected_times;
+ for (int64_t t = 0; t < 64; ++t) selected_times.push_back(t);
+ for (int64_t t = 64; t < 74; ++t) selected_times.push_back(t);
+ for (int64_t t = 128; t < 256; ++t) selected_times.push_back(t);
+ storage::TimeIn time_filter(selected_times, /*not_in=*/false);
+
+ ASSERT_EQ(io_reader.alloc_multi_ssi(device_id, measurements, ssi, pa,
+ &time_filter),
+ E_OK);
+ ASSERT_NE(ssi, nullptr);
+ ssi->set_row_range(/*offset=*/200, /*limit=*/-1);
+
+ std::vector<int64_t> actual_times;
+ while (true) {
+ common::TsBlock* block = nullptr;
+ int ret = ssi->get_next(block, /*alloc_tsblock=*/true, &time_filter);
+ if (ret == common::E_NO_MORE_DATA) break;
+ ASSERT_EQ(ret, common::E_OK);
+ ASSERT_NE(block, nullptr);
+
+ const uint32_t rows = block->get_row_count();
+ const int remaining_offset = ssi->get_row_offset();
+ const uint32_t rows_to_skip = std::min(
+ rows, static_cast<uint32_t>(std::max(remaining_offset, 0)));
+ {
+ common::ColIterator time_iter(0, block);
+ for (uint32_t row = 0; row < rows; ++row) {
+ uint32_t len = 0;
+ int64_t time =
+ *reinterpret_cast<int64_t*>(time_iter.read(&len));
+ if (row >= rows_to_skip) actual_times.push_back(time);
+ time_iter.next();
+ }
+ }
+ ssi->set_row_range(remaining_offset - static_cast<int>(rows_to_skip),
+ /*limit=*/-1);
+ ssi->revert_tsblock();
+ }
+
+ ASSERT_EQ(actual_times.size(), 2u);
+ EXPECT_EQ(actual_times[0], 254);
+ EXPECT_EQ(actual_times[1], 255);
+
+ io_reader.revert_ssi(ssi);
+}
+
+// A boundary page may contain holes between matching timestamps. Offset must
+// consume only matching rows and keep every value column aligned with the
+// first retained match.
+TEST_F(TsFileReaderTest, MultiValueAlignedPagePlanOffsetSkipsBoundaryMatches) {
+ const std::string device = "root.dev_multi_boundary_offset";
+ std::vector<MeasurementSchema> schema_vec;
+ schema_vec.emplace_back("v0", INT64, PLAIN, UNCOMPRESSED);
+ schema_vec.emplace_back("v1", INT64, PLAIN, UNCOMPRESSED);
+ {
+ std::vector<MeasurementSchema*> reg;
+ for (auto& s : schema_vec) reg.push_back(new MeasurementSchema(s));
+ ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device, reg),
+ E_OK);
+ }
+
+ constexpr int kRowCount = 16;
+ Tablet tablet(device,
+ std::make_shared<std::vector<MeasurementSchema>>(schema_vec),
+ kRowCount);
+ for (int i = 0; i < kRowCount; ++i) {
+ ASSERT_EQ(tablet.add_timestamp(i, i), E_OK);
+ ASSERT_EQ(tablet.add_value(i, 0u, static_cast<int64_t>(i)), E_OK);
+ ASSERT_EQ(tablet.add_value(i, 1u, static_cast<int64_t>(i * 10)), E_OK);
+ }
+ ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK);
+ ASSERT_EQ(tsfile_writer_->flush(), E_OK);
+ ASSERT_EQ(tsfile_writer_->close(), E_OK);
+
+ storage::TsFileIOReader io_reader;
+ ASSERT_EQ(io_reader.init(file_name_), E_OK);
+
+ auto device_id = std::make_shared<StringArrayDeviceID>(device);
+ std::vector<std::string> measurements = {"v0", "v1"};
+ storage::TsFileSeriesScanIterator* ssi = nullptr;
+ common::PageArena pa;
+ pa.init(512, common::MOD_TSFILE_READER);
+ const std::vector<int64_t> selected_times = {1, 3, 5, 7, 9};
+ storage::TimeIn time_filter(selected_times, /*not_in=*/false);
+ ASSERT_EQ(io_reader.alloc_multi_ssi(device_id, measurements, ssi, pa,
+ &time_filter),
+ E_OK);
+ ASSERT_NE(ssi, nullptr);
+
+ ssi->set_row_range(/*offset=*/2, /*limit=*/-1);
+ common::TsBlock* block = nullptr;
+ ASSERT_EQ(ssi->get_next(block, /*alloc_tsblock=*/true, &time_filter),
E_OK);
+ ASSERT_NE(block, nullptr);
+ EXPECT_EQ(ssi->get_row_offset(), 0);
+ ASSERT_EQ(block->get_row_count(), 3u);
+
+ const std::vector<int64_t> expected_times = {5, 7, 9};
+ {
+ common::ColIterator time_iter(0, block);
+ common::ColIterator v0_iter(1, block);
+ common::ColIterator v1_iter(2, block);
+ for (int64_t expected : expected_times) {
+ uint32_t len = 0;
+ EXPECT_EQ(*reinterpret_cast<int64_t*>(time_iter.read(&len)),
+ expected);
+ EXPECT_EQ(*reinterpret_cast<int64_t*>(v0_iter.read(&len)),
+ expected);
+ EXPECT_EQ(*reinterpret_cast<int64_t*>(v1_iter.read(&len)),
+ expected * 10);
+ time_iter.next();
+ v0_iter.next();
+ v1_iter.next();
+ }
+ }
+
+ ssi->revert_tsblock();
+ io_reader.revert_ssi(ssi);
+}
+
+// The single-value SSI paths must apply offset to the filtered stream too.
+// Cover both a non-aligned series (ChunkReader) and a single-value aligned
+// series (AlignedChunkReader), including their chunk- and page-level skips.
+TEST_F(TsFileReaderTest, SingleValueRowOffsetCountsOnlyFilteredRows) {
+ const std::string non_aligned_device = "root.dev_filtered_offset";
+ const std::string aligned_device = "root.dev_aligned_filtered_offset";
+ MeasurementSchema schema("v0", INT64, PLAIN, UNCOMPRESSED);
+ ASSERT_EQ(tsfile_writer_->register_timeseries(non_aligned_device, schema),
+ E_OK);
+ ASSERT_EQ(
+ tsfile_writer_->register_aligned_timeseries(aligned_device, schema),
+ E_OK);
+
+ const int rows_per_chunk = 64;
+ const int chunk_count = 4;
+ auto schemas = std::make_shared<std::vector<MeasurementSchema>>(1, schema);
+ for (int chunk = 0; chunk < chunk_count; ++chunk) {
+ const int base = chunk * rows_per_chunk;
+ Tablet non_aligned_tablet(non_aligned_device, schemas, rows_per_chunk);
+ Tablet aligned_tablet(aligned_device, schemas, rows_per_chunk);
+ for (int i = 0; i < rows_per_chunk; ++i) {
+ const int row = base + i;
+ ASSERT_EQ(non_aligned_tablet.add_timestamp(i, row), E_OK);
+ ASSERT_EQ(
+ non_aligned_tablet.add_value(i, 0u, static_cast<int64_t>(row)),
+ E_OK);
+ ASSERT_EQ(aligned_tablet.add_timestamp(i, row), E_OK);
+ ASSERT_EQ(
+ aligned_tablet.add_value(i, 0u, static_cast<int64_t>(row)),
+ E_OK);
+ }
+ ASSERT_EQ(tsfile_writer_->write_tablet(non_aligned_tablet), E_OK);
+ ASSERT_EQ(tsfile_writer_->write_tablet_aligned(aligned_tablet), E_OK);
+ ASSERT_EQ(tsfile_writer_->flush(), E_OK);
+ }
+ ASSERT_EQ(tsfile_writer_->close(), E_OK);
+
+ std::vector<int64_t> selected_times;
+ for (int64_t t = 0; t < 64; ++t) selected_times.push_back(t);
+ for (int64_t t = 64; t < 74; ++t) selected_times.push_back(t);
+ for (int64_t t = 128; t < 256; ++t) selected_times.push_back(t);
+ storage::TimeIn time_filter(selected_times, /*not_in=*/false);
+
+ storage::TsFileIOReader io_reader;
+ ASSERT_EQ(io_reader.init(file_name_), E_OK);
+ for (const std::string& device : {non_aligned_device, aligned_device}) {
+ SCOPED_TRACE(device);
+ auto device_id = std::make_shared<StringArrayDeviceID>(device);
+ storage::TsFileSeriesScanIterator* ssi = nullptr;
+ common::PageArena pa;
+ pa.init(512, common::MOD_TSFILE_READER);
+ ASSERT_EQ(io_reader.alloc_ssi(device_id, "v0", ssi, pa, &time_filter),
+ E_OK);
+ ASSERT_NE(ssi, nullptr);
+ ssi->set_row_range(/*offset=*/200, /*limit=*/-1);
+
+ std::vector<int64_t> actual_times;
+ while (true) {
+ common::TsBlock* block = nullptr;
+ int ret =
+ ssi->get_next(block, /*alloc_tsblock=*/true, &time_filter);
+ if (ret == common::E_NO_MORE_DATA) break;
+ ASSERT_EQ(ret, common::E_OK);
+ ASSERT_NE(block, nullptr);
+
+ const uint32_t rows = block->get_row_count();
+ const int remaining_offset = ssi->get_row_offset();
+ const uint32_t rows_to_skip = std::min(
+ rows, static_cast<uint32_t>(std::max(remaining_offset, 0)));
+ {
+ common::ColIterator time_iter(0, block);
+ for (uint32_t row = 0; row < rows; ++row) {
+ uint32_t len = 0;
+ int64_t time =
+ *reinterpret_cast<int64_t*>(time_iter.read(&len));
+ if (row >= rows_to_skip) actual_times.push_back(time);
+ time_iter.next();
+ }
+ }
+ ssi->set_row_range(
+ remaining_offset - static_cast<int>(rows_to_skip),
+ /*limit=*/-1);
+ ssi->revert_tsblock();
+ }
+
+ EXPECT_EQ(actual_times.size(), 2u);
+ if (actual_times.size() == 2u) {
+ EXPECT_EQ(actual_times[0], 254);
+ EXPECT_EQ(actual_times[1], 255);
+ }
+ io_reader.revert_ssi(ssi);
+ }
+}
+
namespace storage {
// Subclass that lets the test (a) inject an error from the next-tsblock load
// and (b) wire a manually constructed TsBlock into the inherited iterator
