Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions cpp/src/common/container/byte_buffer.h
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,14 @@ class ByteBuffer {

FORCE_INLINE uint32_t get_data_size() const { return real_data_size_; }

// Declare bytes that were written directly through get_data() without going
// through append_*(). Used when a caller decodes straight into this buffer
// at a known offset.
FORCE_INLINE void set_data_size(uint32_t size) {
ASSERT(size <= reserved_size_);
real_data_size_ = size;
}

private:
char* data_;
uint8_t variable_type_len_;
Expand Down
245 changes: 219 additions & 26 deletions cpp/src/reader/aligned_chunk_reader.cc

Large diffs are not rendered by default.

40 changes: 40 additions & 0 deletions cpp/src/reader/aligned_chunk_reader.h
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,9 @@ struct ChunkPageInfo {
uint32_t time_uncompressed_size = 0;
int32_t row_begin = 0; // inclusive
int32_t row_end = 0; // exclusive
// Row index of this page's first retained row in the chunk reader's output
// stream. Filled by build_page_plan() once the whole plan is known.
int32_t out_row_begin = 0;
std::vector<int64_t> value_file_offsets;
std::vector<uint32_t> value_compressed_sizes;
std::vector<uint32_t> value_uncompressed_sizes;
Expand All @@ -65,6 +68,17 @@ struct PageDecodedState {
common::PageArena predecode_pa;
int32_t predecoded_count = 0;
int32_t predecoded_read_pos = 0;
// Non-null when the page was decoded straight into the destination TsBlock
// column buffer (see AlignedChunkReader::prepare_direct_decode()). Aliases
// memory owned by the Vector; predecoded_values stays empty in that case.
char* direct_values = nullptr;

// Where this slot's fixed-length values live: the direct view when present,
// otherwise the slot's own buffer.
char* values() {
return direct_values != nullptr ? direct_values
: predecoded_values.data();
}
};

// Per-value-column state for multi-value AlignedChunkReader.
Expand Down Expand Up @@ -177,6 +191,15 @@ class AlignedChunkReader : public IChunkReader {

bool is_multi_value_mode() const { return multi_value_mode_; }

#ifdef ENABLE_TEST
// Pages scattered by the bulk-copy path so far (FULL_PASS pages and
// BOUNDARY pages a contiguous time filter proves fully matched).
size_t TEST_bulk_page_count() const { return bulk_page_count_; }
// Leading pages of the current plan that decode straight into the
// destination block's column buffers (see prepare_direct_decode()).
size_t TEST_direct_page_end() const { return direct_page_end_; }
#endif

#ifdef ENABLE_THREADS
// Set external thread pool for parallel decode (not owned).
void set_decode_pool(common::ThreadPool* pool) { decode_pool_ = pool; }
Expand Down Expand Up @@ -274,6 +297,7 @@ class AlignedChunkReader : public IChunkReader {
common::RowAppender& row_appender, Filter* filter,
common::PageArena* pa);
int build_page_plan(Filter* filter, int& row_offset);
bool prepare_direct_decode(common::TsBlock* tsblock, Filter* filter);
void build_filter_column_mask();
int evaluate_page_field_mask(size_t page_idx);
int decode_output_columns_for_page(size_t page_idx);
Expand Down Expand Up @@ -346,6 +370,22 @@ class AlignedChunkReader : public IChunkReader {
SelectionMask current_page_field_mask_;
bool current_page_field_mask_valid_ = false;

// Direct decode: pages [0, direct_page_end_) are decoded straight into the
// destination Vector memory instead of a per-page staging buffer. See
// prepare_direct_decode().
size_t direct_page_end_ = 0;
int32_t direct_base_row_ = 0;
common::TsBlock* direct_tsblock_ = nullptr;
// Per value column: byte offset in the Vector's buffer where the next
// decoded value goes. TsBlock columns store non-null values compactly
// (RowIterator advances offsets only for non-null rows), so this advances
// by the non-null count of each page, not by its row index.
std::vector<size_t> direct_col_off_;

#ifdef ENABLE_TEST
size_t bulk_page_count_ = 0;
#endif

#ifdef ENABLE_THREADS
common::ThreadPool* decode_pool_ = nullptr; // borrowed, not owned
// Per-task time decoder + compressor pool for parallel time-page decode.
Expand Down
8 changes: 8 additions & 0 deletions cpp/src/reader/filter/filter.h
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,14 @@ class Filter {
return nullptr;
}

// 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
// sets (TimeIn) or negated forms (NOT BETWEEN / NotEq) may contain holes
// and must still be checked per value. Defaults to false; only explicitly
// contiguous time filters override it to true.
virtual bool is_contiguous_time_filter() const { return false; }
Comment thread
ColinLeeo marked this conversation as resolved.
Resolved

// Batch time filter: evaluate time filter on an array of timestamps.
// Writes true/false into @mask for each element.
// Returns the number of elements that passed (mask[i] == true).
Expand Down
2 changes: 2 additions & 0 deletions cpp/src/reader/filter/time_operator.h
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,8 @@ class TimeBetween : public Filter {

FilterType get_filter_type() { return type_; }

bool is_contiguous_time_filter() const override { return !not_; }

private:
int64_t value1_;
int64_t value2_;
Expand Down
Loading
Loading