Skip to content
77 changes: 60 additions & 17 deletions cpp/src/reader/aligned_chunk_reader.cc
Original file line number Diff line number Diff line change
Expand Up @@ -1537,6 +1537,10 @@ int AlignedChunkReader::build_page_plan(Filter* filter, int& row_offset) {
}

const uint32_t num_cols = value_columns_.size();
// For single-page chunks the times decoded while classifying the page are
// exactly the ones the lazy decode path would decode again; keep them so
// each page is read and decoded once.
std::vector<int64_t> single_page_times;
while (IS_SUCC(ret)) {
if (time_chunk_visit_offset_ - time_chunk_header_.serialized_size_ >=
time_chunk_header_.data_size_) {
Expand All @@ -1554,6 +1558,9 @@ int AlignedChunkReader::build_page_plan(Filter* filter, int& row_offset) {
}

ChunkPageInfo page_info;
// Times decoded below for classification are handed to the lazy
// decode path for single-page chunks, so the page is decoded once.
std::vector<int64_t> decoded_page_times;
page_info.time_file_offset = time_chunk_meta_->offset_of_chunk_header_ +
time_chunk_visit_offset_;
page_info.time_compressed_size = cur_time_page_header_.compressed_size_;
Expand Down Expand Up @@ -1625,24 +1632,43 @@ int AlignedChunkReader::build_page_plan(Filter* filter, int& row_offset) {
page_info.row_begin = 0;
page_info.row_end = stat != nullptr ? stat->count_ : 0;
} else {
std::vector<int64_t> times;
if (RET_FAIL(decode_time_page_direct(page_info, times))) {
if (RET_FAIL(decode_time_page_direct(page_info,
decoded_page_times))) {
break;
}
int32_t first = -1;
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 (field_filter_ == nullptr && row_offset > 0) {
row_offset--;
continue;
const int32_t n =
static_cast<int32_t>(decoded_page_times.size());
if (filter->is_unbounded_time_filter()) {
// Every timestamp matches, so the per-row predicate can
// be skipped. Without a FIELD filter the whole page is
// passed and takes the bulk-copy path; the shared
// FULL_PASS handling below consumes OFFSET. With a FIELD
// filter the exact rows depend on statistics that are not
// available here, so the page stays a boundary page and
// the mask decides (OFFSET is applied after the mask).
if (n > 0) {
first = 0;
last = n - 1;
if (field_filter_ == nullptr) {
page_info.pass_type = PagePassType::FULL_PASS;
}
}
} else {
for (int32_t i = 0; i < n; i++) {
if (filter->satisfy_start_end_time(
decoded_page_times[i], decoded_page_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 (field_filter_ == nullptr && row_offset > 0) {
row_offset--;
continue;
}
if (first < 0) first = i;
last = i;
}
if (first < 0) first = i;
last = i;
}
}
if (first >= 0) {
Expand All @@ -1656,11 +1682,12 @@ int AlignedChunkReader::build_page_plan(Filter* filter, int& row_offset) {

if (page_info.pass_type != PagePassType::SKIP) {
if (page_info.row_end == 0) {
std::vector<int64_t> times;
if (RET_FAIL(decode_time_page_direct(page_info, times))) {
if (RET_FAIL(decode_time_page_direct(page_info,
decoded_page_times))) {
break;
}
page_info.row_end = static_cast<int32_t>(times.size());
page_info.row_end =
static_cast<int32_t>(decoded_page_times.size());
}
if (page_info.pass_type == PagePassType::FULL_PASS &&
row_offset > 0) {
Expand All @@ -1676,6 +1703,9 @@ int AlignedChunkReader::build_page_plan(Filter* filter, int& row_offset) {
}
if (page_info.row_begin < page_info.row_end) {
chunk_pages_.push_back(std::move(page_info));
if (chunk_pages_.size() == 1 && !decoded_page_times.empty()) {
single_page_times = std::move(decoded_page_times);
}
}
}

Expand All @@ -1699,6 +1729,9 @@ int AlignedChunkReader::build_page_plan(Filter* filter, int& row_offset) {
out_row += page.row_end - page.row_begin;
}
per_page_times_.assign(chunk_pages_.size(), std::vector<int64_t>{});
if (chunk_pages_.size() == 1 && !single_page_times.empty()) {
per_page_times_[0] = std::move(single_page_times);
}
for (auto* col : value_columns_) {
col->per_page_state.clear();
col->per_page_state.resize(chunk_pages_.size());
Expand Down Expand Up @@ -2123,6 +2156,12 @@ int AlignedChunkReader::decode_all_planned_pages() {
Compressor* comp = time_compressor_pool_[k];
for (size_t p = k; p < chunk_pages_.size();
p += time_task_count) {
// The planner may already have decoded this page to
// classify it (single-page plans prefill slot 0);
// decoding it again would undo that reuse.
if (!per_page_times_[p].empty()) {
continue;
}
int r = decode_time_page_with(
chunk_pages_[p], per_page_times_[p], dec, comp);
if (IS_FAIL(r)) {
Expand Down Expand Up @@ -2154,7 +2193,11 @@ int AlignedChunkReader::decode_all_planned_pages() {
// matching row.
int AlignedChunkReader::decode_page_lazy(size_t page_idx) {
int ret = E_OK;
if (RET_FAIL(decode_time_page_direct(chunk_pages_[page_idx],
// build_page_plan() already decoded (and kept) this page's times for
// single-page chunks; decoding again would re-read and re-decode the same
// page.
if (per_page_times_[page_idx].empty() &&
RET_FAIL(decode_time_page_direct(chunk_pages_[page_idx],
per_page_times_[page_idx]))) {
Comment thread
ColinLeeo marked this conversation as resolved.
Resolved
return ret;
}
Expand Down
3 changes: 3 additions & 0 deletions cpp/src/reader/filter/filter.h
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,9 @@ class Filter {
return nullptr;
}

// True when the predicate accepts every timestamp (e.g. TimeBetween over
// the full int64 range). Callers may then skip per-row evaluation.
virtual bool is_unbounded_time_filter() const { return false; }
// True when this time filter selects one contiguous interval: inside a
// BOUNDARY page's [row_begin, row_end) range every timestamp matches (e.g.
// TimeBetween, one-sided Gt/Lt), so the range can be bulk-copied. Discrete
Expand Down
5 changes: 5 additions & 0 deletions cpp/src/reader/filter/time_operator.h
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,11 @@ class TimeBetween : public Filter {

std::vector<TimeRange*>* get_time_ranges();

bool is_unbounded_time_filter() const override {
return !not_ && value1_ == std::numeric_limits<int64_t>::min() &&
value2_ == std::numeric_limits<int64_t>::max();
}
Comment thread
ColinLeeo marked this conversation as resolved.
Resolved
Comment thread
ColinLeeo marked this conversation as resolved.
Resolved

int satisfy_batch_time(const int64_t* times, int count, bool* mask);

FilterType get_filter_type() { return type_; }
Expand Down
31 changes: 31 additions & 0 deletions cpp/test/reader/filter/time_in_filter_test.cc
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@
*/
#include <gtest/gtest.h>

#include <limits>

#include "reader/filter/time_operator.h"

using namespace storage;
Expand Down Expand Up @@ -82,3 +84,32 @@ TEST(TimeInFilterTest, BatchTimeFallbackUsesScalarSemantics) {
EXPECT_TRUE(mask[4]);
EXPECT_FALSE(mask[5]);
}

// The aligned chunk reader uses this predicate to skip the per-row time check
// and to take the bulk path; it must only be true for a plain TimeBetween that
// accepts every int64 value.
TEST(TimeBetweenFilterTest, IsUnboundedTimeFilterOnlyForFullRangeBetween) {
const int64_t min_time = std::numeric_limits<int64_t>::min();
const int64_t max_time = std::numeric_limits<int64_t>::max();
Comment thread
ColinLeeo marked this conversation as resolved.
Resolved
Comment thread
ColinLeeo marked this conversation as resolved.
Resolved

TimeBetween full(min_time, max_time, /*not_between=*/false);
EXPECT_TRUE(full.is_unbounded_time_filter());

// Negated full range accepts nothing, so it must not take the shortcut.
TimeBetween negated(min_time, max_time, /*not_between=*/true);
EXPECT_FALSE(negated.is_unbounded_time_filter());

// Inverted bounds are not "everything".
TimeBetween inverted(max_time, min_time, /*not_between=*/false);
EXPECT_FALSE(inverted.is_unbounded_time_filter());

// One missing bound is a real restriction.
TimeBetween half_open(min_time + 1, max_time, /*not_between=*/false);
EXPECT_FALSE(half_open.is_unbounded_time_filter());
TimeBetween narrow(0, 100, /*not_between=*/false);
EXPECT_FALSE(narrow.is_unbounded_time_filter());

// Other filter types never claim to be unbounded.
TimeIn time_in({1, 2}, /*not_in=*/false);
EXPECT_FALSE(time_in.is_unbounded_time_filter());
}
Loading
Loading