You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tsfile.apache.org by co...@apache.org on 2026/10/22 01:56:34 UTC
(tsfile) 01/01: [C++] Batch-evaluate FIELD predicates on aligned table rows
This is an automated email from the ASF dual-hosted git repository.
ColinLeeo pushed a commit to branch colin/aligned-batch-field-filter-design
in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit f5d1243a44fd8a6adddb24d23e8718ab8ccd0fed
Author: ColinLee <sh...@163.com>
AuthorDate: Thu Oct 22 09:54:08 2026 +0800
[C++] Batch-evaluate FIELD predicates on aligned table rows
Add AlignedFieldFilter (a columnar predicate tree) and SelectionMask
(a row bitmask), and thread them through the aligned chunk/block readers.
Single-threaded flow: page statistics (may_match/all_match) prune pages
that cannot match or mark pages that fully match; only filter-referenced
columns are decoded first; output columns are decoded lazily and only for
pages with at least one matching row; the predicate is evaluated once per
page into a bitmask, and the scatter loop only tests bits.
NOT is pushed down to leaf comparisons so NULL rows keep SQL three-valued
semantics, get_data_type_size is no longer called on string columns, and
query() error paths free the field filter.
---
cpp/src/file/tsfile_io_reader.cc | 6 +-
cpp/src/file/tsfile_io_reader.h | 3 +-
cpp/src/reader/aligned_chunk_reader.cc | 230 +++++-
cpp/src/reader/aligned_chunk_reader.h | 14 +
.../reader/block/single_device_tsblock_reader.cc | 205 +++++-
.../reader/block/single_device_tsblock_reader.h | 19 +-
cpp/src/reader/column_mapping.h | 17 +
cpp/src/reader/filter/aligned_field_filter.cc | 775 +++++++++++++++++++++
cpp/src/reader/filter/aligned_field_filter.h | 129 ++++
cpp/src/reader/filter/selection_mask.cc | 157 +++++
cpp/src/reader/filter/selection_mask.h | 65 ++
cpp/src/reader/table_query_executor.cc | 29 +-
cpp/src/reader/tsfile_series_scan_iterator.cc | 4 +-
cpp/src/reader/tsfile_series_scan_iterator.h | 13 +-
.../reader/filter/aligned_field_filter_test.cc | 147 ++++
cpp/test/reader/table_field_filter_test.cc | 94 +++
16 files changed, 1848 insertions(+), 59 deletions(-)
diff --git a/cpp/src/file/tsfile_io_reader.cc b/cpp/src/file/tsfile_io_reader.cc
index 37225677c..bc0fa62f6 100644
--- a/cpp/src/file/tsfile_io_reader.cc
+++ b/cpp/src/file/tsfile_io_reader.cc
@@ -297,8 +297,8 @@ int TsFileIOReader::alloc_prepared_multi_ssi(
int TsFileIOReader::alloc_multi_ssi(
std::shared_ptr<IDeviceID> device_id,
const std::vector<std::string>& measurement_names,
- TsFileSeriesScanIterator*& ssi, common::PageArena& pa,
- Filter* time_filter) {
+ TsFileSeriesScanIterator*& ssi, common::PageArena& pa, Filter* time_filter,
+ AlignedFieldFilter* field_filter) {
int ret = E_OK;
if (RET_FAIL(load_tsfile_meta_if_necessary())) return ret;
@@ -307,7 +307,7 @@ int TsFileIOReader::alloc_multi_ssi(
if (IS_NULL(ssi_buf)) return E_OOM;
ssi = new (ssi_buf) TsFileSeriesScanIterator;
ssi->init(device_id, measurement_names.empty() ? "" : measurement_names[0],
- read_file_, time_filter, pa);
+ read_file_, time_filter, pa, field_filter);
auto& ssi_pa = ssi->timeseries_index_pa_;
diff --git a/cpp/src/file/tsfile_io_reader.h b/cpp/src/file/tsfile_io_reader.h
index f0c18f60f..a7105739d 100644
--- a/cpp/src/file/tsfile_io_reader.h
+++ b/cpp/src/file/tsfile_io_reader.h
@@ -89,7 +89,8 @@ class TsFileIOReader {
int alloc_multi_ssi(std::shared_ptr<IDeviceID> device_id,
const std::vector<std::string>& measurement_names,
TsFileSeriesScanIterator*& ssi, common::PageArena& pa,
- Filter* time_filter = nullptr);
+ Filter* time_filter = nullptr,
+ AlignedFieldFilter* field_filter = nullptr);
int prepare_series(const FileGeneration& generation,
const PreparedLocator& locator,
diff --git a/cpp/src/reader/aligned_chunk_reader.cc b/cpp/src/reader/aligned_chunk_reader.cc
index a2005cdac..2fa694079 100644
--- a/cpp/src/reader/aligned_chunk_reader.cc
+++ b/cpp/src/reader/aligned_chunk_reader.cc
@@ -1421,6 +1421,68 @@ int AlignedChunkReader::decode_time_page_with(const ChunkPageInfo& page_info,
return ret;
}
+void AlignedChunkReader::build_filter_column_mask() {
+ filter_column_mask_.assign(value_columns_.size(), false);
+ if (field_filter_ == nullptr) return;
+ std::set<std::string> filter_columns;
+ for (const auto& name : field_filter_->referenced_columns()) {
+ filter_columns.insert(name);
+ }
+ for (uint32_t c = 0; c < value_columns_.size(); ++c) {
+ filter_column_mask_[c] =
+ filter_columns.count(
+ value_columns_[c]
+ ->chunk_meta->measurement_name_.to_std_string()) > 0;
+ }
+}
+
+int AlignedChunkReader::evaluate_page_field_mask(size_t page_idx) {
+ if (field_filter_ == nullptr) return E_OK;
+ const ChunkPageInfo& page_info = chunk_pages_[page_idx];
+ std::vector<AlignedFieldFilter::ColumnView> views;
+ for (uint32_t c = 0; c < value_columns_.size(); ++c) {
+ if (!filter_column_mask_[c]) continue;
+ auto* col = value_columns_[c];
+ auto& pps = col->per_page_state[page_idx];
+ AlignedFieldFilter::ColumnView view;
+ view.name = col->chunk_meta->measurement_name_.to_std_string();
+ view.type = col->chunk_header.data_type_;
+ view.notnull_bitmap =
+ pps.notnull_bitmap.empty() ? nullptr : pps.notnull_bitmap.data();
+ view.notnull_bitmap_size =
+ static_cast<uint32_t>(pps.notnull_bitmap.size());
+ view.row_offset = static_cast<uint32_t>(page_info.row_begin);
+ view.nonnull_start = static_cast<uint32_t>(
+ count_non_null_prefix(pps.notnull_bitmap, page_info.row_begin));
+ if (view.type == common::STRING || view.type == common::TEXT ||
+ view.type == common::BLOB) {
+ view.string_values = pps.predecoded_strings.data();
+ } else {
+ view.fixed_values = pps.predecoded_values.data();
+ }
+ views.push_back(std::move(view));
+ }
+
+ const uint32_t row_count =
+ static_cast<uint32_t>(page_info.row_end - page_info.row_begin);
+ current_page_field_mask_.reset(row_count);
+ int ret = field_filter_->evaluate_columns(views, row_count,
+ current_page_field_mask_);
+ if (ret != E_OK) return ret;
+ current_page_field_mask_valid_ = true;
+ return E_OK;
+}
+
+int AlignedChunkReader::decode_output_columns_for_page(size_t page_idx) {
+ if (field_filter_ == nullptr) return E_OK;
+ for (uint32_t c = 0; c < value_columns_.size(); ++c) {
+ if (filter_column_mask_[c]) continue;
+ int ret = decode_value_page_for_slot(c, page_idx);
+ if (IS_FAIL(ret)) return ret;
+ }
+ return E_OK;
+}
+
int AlignedChunkReader::build_page_plan(Filter* filter, int& row_offset) {
int ret = E_OK;
chunk_pages_.clear();
@@ -1475,44 +1537,74 @@ int AlignedChunkReader::build_page_plan(Filter* filter, int& row_offset) {
break;
}
+ bool field_page_skip = false;
+ bool field_all_match = false;
+ if (field_filter_ != nullptr) {
+ std::map<std::string, Statistic*> field_statistics;
+ for (uint32_t c = 0; c < num_cols; ++c) {
+ auto* col = value_columns_[c];
+ field_statistics[col->chunk_meta->measurement_name_
+ .to_std_string()] =
+ col->cur_page_header.statistic_;
+ }
+ const int64_t row_count =
+ cur_time_page_header_.statistic_ != nullptr
+ ? cur_time_page_header_.statistic_->count_
+ : 0;
+ field_page_skip =
+ !field_filter_->may_match(field_statistics, row_count);
+ field_all_match =
+ !field_page_skip &&
+ field_filter_->all_match(field_statistics, row_count);
+ }
+
Statistic* stat = cur_time_page_header_.statistic_;
- if (filter == nullptr) {
- page_info.pass_type = PagePassType::FULL_PASS;
- page_info.row_begin = 0;
- page_info.row_end = stat != nullptr ? stat->count_ : 0;
- } else if (stat != nullptr && !filter->satisfy(stat)) {
+ const bool time_full_match =
+ filter == nullptr ||
+ (stat != nullptr && filter->contain_start_end_time(
+ stat->start_time_, stat->end_time_));
+ if (field_page_skip ||
+ (stat != nullptr && filter != nullptr && !filter->satisfy(stat))) {
page_info.pass_type = PagePassType::SKIP;
- } else if (stat != nullptr && filter->contain_start_end_time(
- stat->start_time_, stat->end_time_)) {
+ } else if (time_full_match &&
+ (field_filter_ == nullptr || field_all_match)) {
page_info.pass_type = PagePassType::FULL_PASS;
page_info.row_begin = 0;
- page_info.row_end = stat->count_;
+ page_info.row_end = stat != nullptr ? stat->count_ : 0;
} else {
page_info.pass_type = PagePassType::BOUNDARY;
- std::vector<int64_t> times;
- if (RET_FAIL(decode_time_page_direct(page_info, 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 (row_offset > 0) {
- row_offset--;
- continue;
+ if (filter == nullptr) {
+ // There is no time predicate to narrow the rows; keep the
+ // whole page and let the FIELD mask decide exact rows.
+ 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))) {
+ 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;
+ }
+ if (first < 0) first = i;
+ last = i;
}
- if (first < 0) first = i;
- last = i;
}
- }
- if (first >= 0) {
- page_info.row_begin = first;
- page_info.row_end = last + 1;
- } else {
- page_info.pass_type = PagePassType::SKIP;
+ if (first >= 0) {
+ page_info.row_begin = first;
+ page_info.row_end = last + 1;
+ } else {
+ page_info.pass_type = PagePassType::SKIP;
+ }
}
}
@@ -1774,6 +1866,7 @@ int AlignedChunkReader::decode_value_page_for_slot(uint32_t col_idx,
// (see decode_page_lazy() / release_page_slot()). Bounds memory to one page.
int AlignedChunkReader::decode_all_planned_pages() {
if (chunk_pages_.empty()) return E_OK;
+ build_filter_column_mask();
#ifdef ENABLE_THREADS
if (decode_pool_ != nullptr && value_columns_.size() > 1) {
@@ -1803,6 +1896,7 @@ int AlignedChunkReader::decode_all_planned_pages() {
std::vector<std::future<void>> futures;
std::vector<int> col_rets(value_columns_.size(), E_OK);
for (uint32_t c = 0; c < value_columns_.size(); c++) {
+ if (field_filter_ != nullptr && !filter_column_mask_[c]) continue;
int* col_ret = &col_rets[c];
futures.push_back(decode_pool_->submit([this, c, col_ret]() {
for (size_t p = 0; p < chunk_pages_.size(); p++) {
@@ -1854,20 +1948,47 @@ int AlignedChunkReader::decode_all_planned_pages() {
return E_OK;
}
-// Decode time + all value columns for a single page slot on demand.
-// Used by the single-thread path to keep memory bounded to one page.
+// Decode one page slot on demand. With a FIELD filter, decode only its filter
+// columns first and leave output-only columns untouched when the page has no
+// 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],
per_page_times_[page_idx]))) {
return ret;
}
- for (uint32_t c = 0; c < value_columns_.size(); c++) {
- if (RET_FAIL(decode_value_page_for_slot(c, page_idx))) {
- return ret;
+ if (field_filter_ != nullptr &&
+ filter_column_mask_.size() != value_columns_.size()) {
+ build_filter_column_mask();
+ }
+ if (field_filter_ == nullptr) {
+ for (uint32_t c = 0; c < value_columns_.size(); ++c) {
+ if (RET_FAIL(decode_value_page_for_slot(c, page_idx))) return ret;
}
+ return E_OK;
}
- return E_OK;
+ if (chunk_pages_[page_idx].pass_type == PagePassType::FULL_PASS) {
+ // Every row matches; no mask evaluation is needed, so decode all
+ // columns once instead of first decoding the filter columns.
+ for (uint32_t c = 0; c < value_columns_.size(); ++c) {
+ if (RET_FAIL(decode_value_page_for_slot(c, page_idx))) return ret;
+ }
+ current_page_field_mask_.reset(static_cast<uint32_t>(
+ chunk_pages_[page_idx].row_end - chunk_pages_[page_idx].row_begin));
+ current_page_field_mask_.set_all();
+ current_page_field_mask_valid_ = true;
+ return E_OK;
+ }
+
+ // BOUNDARY page: decode only the filter columns, evaluate the mask, then
+ // decode output columns only when at least one row survives.
+ for (uint32_t c = 0; c < value_columns_.size(); ++c) {
+ if (!filter_column_mask_[c]) continue;
+ if (RET_FAIL(decode_value_page_for_slot(c, page_idx))) return ret;
+ }
+ if (RET_FAIL(evaluate_page_field_mask(page_idx))) return ret;
+ if (current_page_field_mask_.empty()) return E_OK;
+ return decode_output_columns_for_page(page_idx);
}
// Release the decoded buffers of one page slot so they can be reused by the
@@ -1911,7 +2032,8 @@ int AlignedChunkReader::get_next_page_multi(TsBlock* ret_tsblock,
(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));
+ (row_offset > 0 || use_parallel_page_plan ||
+ field_filter_ != nullptr));
if (!use_page_plan) {
return get_next_page_multi_serial(ret_tsblock, filter, pa, row_offset);
}
@@ -1947,7 +2069,33 @@ int AlignedChunkReader::get_next_page_multi(TsBlock* ret_tsblock,
page_time_cursor_ = page_info.row_begin;
page_time_count_ = page_info.row_end;
current_page_loaded_ = true;
+ current_page_field_mask_valid_ = false;
+ }
+
+ if (field_filter_ != nullptr && !current_page_field_mask_valid_) {
+ if (page_info.pass_type == PagePassType::FULL_PASS) {
+ current_page_field_mask_.reset(static_cast<uint32_t>(
+ page_info.row_end - page_info.row_begin));
+ current_page_field_mask_.set_all();
+ current_page_field_mask_valid_ = true;
+ } else if (RET_FAIL(evaluate_page_field_mask(
+ current_page_plan_index_))) {
+ return ret;
+ }
+ if (current_page_field_mask_.empty()) {
+ if (single_thread_lazy)
+ release_page_slot(current_page_plan_index_);
+ current_page_plan_index_++;
+ current_page_loaded_ = false;
+ current_page_field_mask_valid_ = false;
+ continue;
+ }
+ if (!single_thread_lazy && RET_FAIL(decode_output_columns_for_page(
+ current_page_plan_index_))) {
+ return ret;
+ }
}
+
const std::vector<int64_t>& times =
per_page_times_[current_page_plan_index_];
@@ -2033,7 +2181,12 @@ int AlignedChunkReader::get_next_page_multi(TsBlock* ret_tsblock,
return E_OK;
}
int64_t ts = times[page_time_cursor_];
- if (boundary_filter && !filter->satisfy_start_end_time(ts, ts)) {
+ const uint32_t field_row = page_time_cursor_ - page_info.row_begin;
+ const bool field_rejected =
+ field_filter_ != nullptr &&
+ !current_page_field_mask_.test(field_row);
+ if (field_rejected ||
+ (boundary_filter && !filter->satisfy_start_end_time(ts, ts))) {
for (uint32_t c = 0; c < num_cols; c++) {
auto* col = value_columns_[c];
auto& pps = col->per_page_state[current_page_plan_index_];
@@ -2095,6 +2248,7 @@ int AlignedChunkReader::get_next_page_multi(TsBlock* ret_tsblock,
}
current_page_plan_index_++;
current_page_loaded_ = false;
+ current_page_field_mask_valid_ = false;
}
return E_NO_MORE_DATA;
}
diff --git a/cpp/src/reader/aligned_chunk_reader.h b/cpp/src/reader/aligned_chunk_reader.h
index 44e28fda3..81311febf 100644
--- a/cpp/src/reader/aligned_chunk_reader.h
+++ b/cpp/src/reader/aligned_chunk_reader.h
@@ -25,6 +25,7 @@
#include "compress/compressor.h"
#include "encoding/decoder.h"
#include "file/random_access_read_file.h"
+#include "reader/filter/aligned_field_filter.h"
#include "reader/filter/filter.h"
#include "reader/ichunk_reader.h"
@@ -122,6 +123,12 @@ class AlignedChunkReader : public IChunkReader {
cur_value_index(-1) {}
int init(RandomAccessReadFile* read_file, common::String m_name,
common::TSDataType data_type, Filter* time_filter) override;
+ void set_field_filter(AlignedFieldFilter* field_filter) {
+ field_filter_ = field_filter;
+ }
+ bool field_filter_applied() const {
+ return field_filter_ != nullptr && page_plan_built_;
+ }
void reset() override;
void destroy() override;
~AlignedChunkReader() override = default;
@@ -257,6 +264,9 @@ class AlignedChunkReader : public IChunkReader {
common::RowAppender& row_appender, Filter* filter,
common::PageArena* pa);
int build_page_plan(Filter* filter, int& row_offset);
+ void build_filter_column_mask();
+ int evaluate_page_field_mask(size_t page_idx);
+ int decode_output_columns_for_page(size_t page_idx);
int decode_time_page_direct(const ChunkPageInfo& page_info,
std::vector<int64_t>& out_times);
int decode_time_page_with(const ChunkPageInfo& page_info,
@@ -292,6 +302,7 @@ class AlignedChunkReader : public IChunkReader {
Compressor* time_compressor_;
Compressor* value_compressor_;
Filter* time_filter_;
+ AlignedFieldFilter* field_filter_ = nullptr;
Decoder* time_decoder_;
Decoder* value_decoder_;
@@ -319,6 +330,9 @@ class AlignedChunkReader : public IChunkReader {
bool page_plan_built_ = false;
bool current_page_loaded_ = false;
size_t current_page_plan_index_ = 0;
+ std::vector<bool> filter_column_mask_;
+ SelectionMask current_page_field_mask_;
+ bool current_page_field_mask_valid_ = false;
#ifdef ENABLE_THREADS
common::ThreadPool* decode_pool_ = nullptr; // borrowed, not owned
diff --git a/cpp/src/reader/block/single_device_tsblock_reader.cc b/cpp/src/reader/block/single_device_tsblock_reader.cc
index 5586a03df..8330997de 100644
--- a/cpp/src/reader/block/single_device_tsblock_reader.cc
+++ b/cpp/src/reader/block/single_device_tsblock_reader.cc
@@ -40,7 +40,9 @@ SingleDeviceTsBlockReader::SingleDeviceTsBlockReader(
field_filter_(field_filter),
block_size_(block_size),
tuple_desc_(),
- tsfile_io_reader_(tsfile_io_reader) {}
+ tsfile_io_reader_(tsfile_io_reader) {
+ aligned_field_filter_ = dynamic_cast<AlignedFieldFilter*>(field_filter_);
+}
int SingleDeviceTsBlockReader::init(DeviceQueryTask* device_query_task,
uint32_t block_size, Filter* time_filter,
@@ -240,7 +242,8 @@ int SingleDeviceTsBlockReader::init_internal(DeviceQueryTask* device_query_task,
break;
}
}
- if (can_multi && field_filter == nullptr) {
+ if (can_multi &&
+ (field_filter == nullptr || aligned_field_filter_ != nullptr)) {
std::vector<std::string> meas_names(meas_cols.begin(),
meas_cols.end());
// Stable order by first appearance in the result schema so the
@@ -280,8 +283,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_, ssi_offset, -1);
+ int ctx_ret =
+ ctx->init(device_query_task_, meas_names, time_filter, pos_list,
+ pa_, aligned_field_filter_, ssi_offset, -1);
if (ctx_ret == common::E_OK) {
if (ssi_offset > 0) {
row_offset_pushed_to_ssi_ = true;
@@ -299,6 +303,13 @@ int SingleDeviceTsBlockReader::init_internal(DeviceQueryTask* device_query_task,
used_multi = true;
} else {
delete ctx;
+ if (ctx_ret == common::E_NO_MORE_DATA &&
+ aligned_field_filter_ != nullptr) {
+ // Page statistics can prove that no row in this device can
+ // pass the structured FIELD filter. This is a valid empty
+ // result, not a reason to fall back to the row path.
+ return common::E_OK;
+ }
if (ctx_ret != common::E_NO_MORE_DATA &&
ctx_ret != common::E_NOT_SUPPORT) {
return ctx_ret;
@@ -342,7 +353,8 @@ int SingleDeviceTsBlockReader::init_internal(DeviceQueryTask* device_query_task,
}
}
if (any_value_column_requested &&
- (field_filter == nullptr || !all_value_columns_aligned)) {
+ ((field_filter == nullptr && aligned_field_filter_ == nullptr) ||
+ !all_value_columns_aligned)) {
return common::E_OK;
}
@@ -350,9 +362,9 @@ int SingleDeviceTsBlockReader::init_internal(DeviceQueryTask* device_query_task,
std::vector<std::vector<int32_t>> empty_positions;
auto* time_only_ctx =
new VectorMeasurementColumnContext(tsfile_io_reader_);
- int time_only_ret =
- time_only_ctx->init(device_query_task_, empty_measurements,
- time_filter, empty_positions, pa_);
+ int time_only_ret = time_only_ctx->init(
+ device_query_task_, empty_measurements, time_filter,
+ empty_positions, pa_, nullptr, 0, -1);
if (common::E_OK == time_only_ret) {
field_column_contexts_.insert(
std::make_pair(kTimeOnlyContextName, time_only_ctx));
@@ -369,8 +381,9 @@ int SingleDeviceTsBlockReader::init_internal(DeviceQueryTask* device_query_task,
}
// Detect aligned fast path: every field column comes from an aligned chunk.
- if (field_filter == nullptr && !field_column_contexts_.empty() &&
- dense_row_count_ >= 0 &&
+ if ((field_filter == nullptr || aligned_field_filter_ != nullptr) &&
+ !field_column_contexts_.empty() &&
+ (dense_row_count_ >= 0 || aligned_field_filter_ != nullptr) &&
aligned_col_count_ == field_column_contexts_.size()) {
all_aligned_ = true;
aligned_vec_.reserve(field_column_contexts_.size());
@@ -386,6 +399,13 @@ int SingleDeviceTsBlockReader::init_internal(DeviceQueryTask* device_query_task,
}
}
+ if (aligned_field_filter_ != nullptr && !all_aligned_ &&
+ !field_column_contexts_.empty()) {
+ // The structured batch filter is aligned-only. Legacy Filter objects
+ // continue to use the row-oriented path.
+ return common::E_NOT_SUPPORT;
+ }
+
if (field_column_contexts_.empty()) {
return common::E_OK;
}
@@ -526,7 +546,79 @@ int SingleDeviceTsBlockReader::has_next(bool& has_next) {
return ret;
}
+int SingleDeviceTsBlockReader::has_next_aligned_filtered(
+ bool& result_has_next) {
+ int ret = common::E_OK;
+ if (aligned_vec_.empty()) {
+ result_has_next = false;
+ return common::E_OK;
+ }
+ auto* context =
+ dynamic_cast<VectorMeasurementColumnContext*>(aligned_vec_[0]);
+ if (context == nullptr) {
+ return common::E_NOT_SUPPORT;
+ }
+
+ const int time_in_query_index = tuple_desc_.get_time_column_index();
+ common::ColAppender* extra_time_appender =
+ time_in_query_index == -1 ? nullptr
+ : col_appenders_[time_in_query_index];
+
+ while (current_block_->get_row_count() < block_size_) {
+ if (remaining_limit_ == 0) break;
+
+ if (!aligned_filter_mask_valid_) {
+ if (context->available_rows() == 0) {
+ int next_ret = context->get_next_tsblock(false);
+ if (next_ret == common::E_NO_MORE_DATA) break;
+ if (next_ret != common::E_OK) return next_ret;
+ }
+ if (RET_FAIL(context->evaluate_filter(*aligned_field_filter_,
+ aligned_filter_mask_))) {
+ return ret;
+ }
+ aligned_filter_next_row_ = 0;
+ aligned_filter_mask_valid_ = true;
+ if (aligned_filter_mask_.empty()) {
+ const uint32_t skip = context->available_rows();
+ if (RET_FAIL(context->skip_rows(skip))) return ret;
+ aligned_filter_mask_valid_ = false;
+ continue;
+ }
+ }
+
+ uint32_t copied = 0;
+ bool block_exhausted = false;
+ if (RET_FAIL(context->filtered_bulk_copy_into(
+ col_appenders_, col_appenders_[time_column_index_],
+ extra_time_appender, row_appender_, aligned_filter_mask_,
+ aligned_filter_next_row_, remaining_offset_, remaining_limit_,
+ copied, block_exhausted))) {
+ return ret;
+ }
+ if (block_exhausted) {
+ aligned_filter_mask_valid_ = false;
+ }
+ if (remaining_limit_ == 0) break;
+ if (current_block_->get_row_count() >= block_size_) break;
+ if (copied == 0 && !block_exhausted) break;
+ }
+
+ if (current_block_->get_row_count() > 0) {
+ if (RET_FAIL(fill_ids())) return ret;
+ current_block_->fill_trailling_nulls();
+ last_block_returned_ = false;
+ result_has_next = true;
+ } else {
+ result_has_next = false;
+ }
+ return ret;
+}
+
int SingleDeviceTsBlockReader::has_next_aligned(bool& result_has_next) {
+ if (aligned_field_filter_ != nullptr) {
+ return has_next_aligned_filtered(result_has_next);
+ }
int ret = common::E_OK;
int time_in_query_index = tuple_desc_.get_time_column_index();
@@ -1081,13 +1173,13 @@ 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,
- int ssi_offset, int ssi_limit) {
+ AlignedFieldFilter* field_filter, int ssi_offset, int ssi_limit) {
int ret = common::E_OK;
pos_in_result_ = pos_in_result;
column_names_ = measurement_names;
if (RET_FAIL(tsfile_io_reader_->alloc_multi_ssi(
device_query_task->get_device_id(), measurement_names, ssi_, pa,
- time_filter))) {
+ time_filter, field_filter))) {
return ret;
}
ssi_->set_row_range(ssi_offset, ssi_limit);
@@ -1099,6 +1191,7 @@ int VectorMeasurementColumnContext::init(
int VectorMeasurementColumnContext::get_next_tsblock(bool alloc_mem) {
int ret = common::E_OK;
+ if (exhausted_) return common::E_NO_MORE_DATA;
if (tsblock_ != nullptr) {
if (time_iter_) {
delete time_iter_;
@@ -1111,6 +1204,7 @@ int VectorMeasurementColumnContext::get_next_tsblock(bool alloc_mem) {
tsblock_->reset();
}
if (RET_FAIL(ssi_->get_next(tsblock_, alloc_mem))) {
+ if (ret == common::E_NO_MORE_DATA) exhausted_ = true;
if (time_iter_) {
delete time_iter_;
time_iter_ = nullptr;
@@ -1199,10 +1293,95 @@ void VectorMeasurementColumnContext::remove_from(
}
uint32_t VectorMeasurementColumnContext::available_rows() const {
- if (!time_iter_ || time_iter_->end()) return 0;
+ if (exhausted_ || !time_iter_ || time_iter_->end()) return 0;
return time_iter_->remaining();
}
+int VectorMeasurementColumnContext::evaluate_filter(
+ const AlignedFieldFilter& filter, SelectionMask& mask) {
+ if (tsblock_ == nullptr) return common::E_NO_MORE_DATA;
+ if (ssi_ != nullptr && ssi_->field_filter_applied_in_chunk()) {
+ mask.reset(tsblock_->get_row_count());
+ mask.set_all();
+ return common::E_OK;
+ }
+ return filter.evaluate(*tsblock_, column_names_, mask);
+}
+
+int VectorMeasurementColumnContext::filtered_bulk_copy_into(
+ std::vector<common::ColAppender*>& col_appenders,
+ common::ColAppender* time_appender,
+ common::ColAppender* extra_time_appender, common::RowAppender* row_appender,
+ const SelectionMask& mask, uint32_t& source_row, int& remaining_offset,
+ int& remaining_limit, uint32_t& copied, bool& block_exhausted) {
+ copied = 0;
+ block_exhausted = false;
+ if (tsblock_ == nullptr || time_iter_ == nullptr || time_iter_->end()) {
+ block_exhausted = true;
+ return common::E_OK;
+ }
+
+ const uint32_t row_count = tsblock_->get_row_count();
+ while (source_row < row_count) {
+ if (remaining_limit == 0) break;
+ if (row_appender != nullptr && row_appender->remaining() == 0) break;
+
+ uint32_t time_len = 0;
+ int64_t time = *reinterpret_cast<int64_t*>(time_iter_->read(&time_len));
+ bool selected = mask.test(source_row);
+ if (selected && remaining_offset > 0) {
+ --remaining_offset;
+ selected = false;
+ }
+
+ if (selected) {
+ if (time_appender != nullptr) {
+ time_appender->add_row();
+ time_appender->append(reinterpret_cast<char*>(&time),
+ sizeof(time));
+ }
+ if (extra_time_appender != nullptr) {
+ extra_time_appender->add_row();
+ extra_time_appender->append(reinterpret_cast<char*>(&time),
+ sizeof(time));
+ }
+ for (uint32_t c = 0;
+ c < value_iters_.size() && c < pos_in_result_.size(); ++c) {
+ uint32_t len = 0;
+ bool is_null = false;
+ char* value = value_iters_[c]->read(&len, &is_null);
+ for (int32_t pos : pos_in_result_[c]) {
+ auto* appender = col_appenders[pos + 1];
+ appender->add_row();
+ if (is_null) {
+ appender->append_null();
+ } else {
+ appender->append(value, len);
+ }
+ }
+ }
+ if (row_appender != nullptr) row_appender->add_row();
+ ++copied;
+ if (remaining_limit > 0) --remaining_limit;
+ } else {
+ // Reading is required even for rejected rows because variable
+ // length cursors use the current row length when advancing.
+ for (auto* value_iter : value_iters_) {
+ uint32_t len = 0;
+ bool is_null = false;
+ value_iter->read(&len, &is_null);
+ }
+ }
+
+ time_iter_->next();
+ for (auto* value_iter : value_iters_) value_iter->next();
+ ++source_row;
+ }
+
+ if (source_row >= row_count) block_exhausted = true;
+ return common::E_OK;
+}
+
int VectorMeasurementColumnContext::bulk_copy_into(
std::vector<common::ColAppender*>& col_appenders,
common::ColAppender* time_appender, common::RowAppender* row_appender,
diff --git a/cpp/src/reader/block/single_device_tsblock_reader.h b/cpp/src/reader/block/single_device_tsblock_reader.h
index eb58d40ad..b028fe2b9 100644
--- a/cpp/src/reader/block/single_device_tsblock_reader.h
+++ b/cpp/src/reader/block/single_device_tsblock_reader.h
@@ -21,7 +21,9 @@
#include "common/tsblock/tsblock.h"
#include "reader/block/tsblock_reader.h"
+#include "reader/filter/aligned_field_filter.h"
#include "reader/filter/filter.h"
+#include "reader/filter/selection_mask.h"
#include "reader/imeta_data_querier.h"
#include "reader/task/device_query_task.h"
@@ -69,9 +71,11 @@ class SingleDeviceTsBlockReader : public TsBlockReader {
// Fast path for aligned data: all columns share the same timestamps,
// so no per-row merge-sort is needed.
int has_next_aligned(bool& has_next);
+ int has_next_aligned_filtered(bool& has_next);
DeviceQueryTask* device_query_task_;
Filter* field_filter_;
+ AlignedFieldFilter* aligned_field_filter_ = nullptr;
uint32_t block_size_;
common::TsBlock* current_block_ = nullptr;
std::vector<common::ColAppender*> col_appenders_;
@@ -93,6 +97,9 @@ class SingleDeviceTsBlockReader : public TsBlockReader {
bool all_aligned_ = false;
uint32_t aligned_col_count_ = 0;
std::vector<MeasurementColumnContext*> aligned_vec_;
+ SelectionMask aligned_filter_mask_;
+ uint32_t aligned_filter_next_row_ = 0;
+ bool aligned_filter_mask_valid_ = false;
};
class MeasurementColumnContext {
@@ -194,7 +201,8 @@ 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, int ssi_offset = 0, int ssi_limit = -1);
+ common::PageArena& pa, AlignedFieldFilter* field_filter = nullptr,
+ 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;
@@ -205,11 +213,20 @@ class VectorMeasurementColumnContext final : public MeasurementColumnContext {
common::RowAppender* row_appender,
uint32_t count) override;
int skip_rows(uint32_t count) override;
+ int evaluate_filter(const AlignedFieldFilter& filter, SelectionMask& mask);
+ int filtered_bulk_copy_into(
+ std::vector<common::ColAppender*>& col_appenders,
+ common::ColAppender* time_appender,
+ common::ColAppender* extra_time_appender,
+ common::RowAppender* row_appender, const SelectionMask& mask,
+ uint32_t& source_row, int& remaining_offset, int& remaining_limit,
+ uint32_t& copied, bool& block_exhausted);
private:
std::vector<std::string> column_names_;
std::vector<std::vector<int32_t>> pos_in_result_;
std::vector<common::ColIterator*> value_iters_;
+ bool exhausted_ = false;
};
class IdColumnContext {
diff --git a/cpp/src/reader/column_mapping.h b/cpp/src/reader/column_mapping.h
index 99e153030..33e76f47e 100644
--- a/cpp/src/reader/column_mapping.h
+++ b/cpp/src/reader/column_mapping.h
@@ -21,6 +21,7 @@
#include "common/schema.h"
#include "expression.h"
+#include "reader/filter/aligned_field_filter.h"
namespace storage {
class ColumnMapping {
public:
@@ -51,6 +52,22 @@ class ColumnMapping {
return common::E_OK;
}
+ int add_filter_columns(const AlignedFieldFilter& filter,
+ TableSchema& schema) {
+ for (const auto& column_name : filter.referenced_columns()) {
+ const int column_index = schema.find_column_index(column_name);
+ if (column_index < 0) {
+ return common::E_COLUMN_NOT_EXIST;
+ }
+ if (schema.get_column_categories()[column_index] !=
+ common::ColumnCategory::FIELD) {
+ return common::E_INVALID_ARG;
+ }
+ field_columns_.insert(column_name);
+ }
+ return common::E_OK;
+ }
+
const std::vector<int>& get_column_pos(
const std::string& column_name) const {
static const std::vector<int> empty;
diff --git a/cpp/src/reader/filter/aligned_field_filter.cc b/cpp/src/reader/filter/aligned_field_filter.cc
new file mode 100644
index 000000000..ebdcfd3f3
--- /dev/null
+++ b/cpp/src/reader/filter/aligned_field_filter.cc
@@ -0,0 +1,775 @@
+/*
+ * 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 "reader/filter/aligned_field_filter.h"
+
+#include <algorithm>
+#include <cctype>
+#include <cstring>
+#include <utility>
+
+#include "common/statistic.h"
+
+namespace storage {
+namespace {
+
+template <typename T>
+bool compare_values(T left, T right, AlignedFieldFilter::CompareOp op) {
+ switch (op) {
+ case AlignedFieldFilter::CompareOp::EQ:
+ return left == right;
+ case AlignedFieldFilter::CompareOp::NE:
+ return left != right;
+ case AlignedFieldFilter::CompareOp::LT:
+ return left < right;
+ case AlignedFieldFilter::CompareOp::LE:
+ return left <= right;
+ case AlignedFieldFilter::CompareOp::GT:
+ return left > right;
+ case AlignedFieldFilter::CompareOp::GE:
+ return left >= right;
+ }
+ return false;
+}
+
+int compare_string(const char* data, uint32_t len, const std::string& literal,
+ AlignedFieldFilter::CompareOp op) {
+ const std::string value(data, len);
+ switch (op) {
+ case AlignedFieldFilter::CompareOp::EQ:
+ return value == literal;
+ case AlignedFieldFilter::CompareOp::NE:
+ return value != literal;
+ case AlignedFieldFilter::CompareOp::LT:
+ return value < literal;
+ case AlignedFieldFilter::CompareOp::LE:
+ return value <= literal;
+ case AlignedFieldFilter::CompareOp::GT:
+ return value > literal;
+ case AlignedFieldFilter::CompareOp::GE:
+ return value >= literal;
+ }
+ return false;
+}
+
+bool is_integer_type(common::TSDataType type) {
+ return type == common::INT32 || type == common::DATE ||
+ type == common::INT64 || type == common::TIMESTAMP;
+}
+
+bool is_string_type(common::TSDataType type) {
+ return type == common::STRING || type == common::TEXT ||
+ type == common::BLOB;
+}
+
+std::string normalize_column_name(const std::string& column) {
+ std::string normalized(column);
+ for (char& ch : normalized) {
+ ch = static_cast<char>(std::tolower(static_cast<unsigned char>(ch)));
+ }
+ return normalized;
+}
+
+} // namespace
+
+AlignedFieldFilter::Literal AlignedFieldFilter::Literal::boolean(bool value) {
+ Literal literal;
+ literal.type = common::BOOLEAN;
+ literal.bool_value = value;
+ return literal;
+}
+
+AlignedFieldFilter::Literal AlignedFieldFilter::Literal::int32(int32_t value) {
+ Literal literal;
+ literal.type = common::INT32;
+ literal.int32_value = value;
+ return literal;
+}
+
+AlignedFieldFilter::Literal AlignedFieldFilter::Literal::int64(int64_t value) {
+ Literal literal;
+ literal.type = common::INT64;
+ literal.int64_value = value;
+ return literal;
+}
+
+AlignedFieldFilter::Literal AlignedFieldFilter::Literal::float32(float value) {
+ Literal literal;
+ literal.type = common::FLOAT;
+ literal.float_value = value;
+ return literal;
+}
+
+AlignedFieldFilter::Literal AlignedFieldFilter::Literal::float64(double value) {
+ Literal literal;
+ literal.type = common::DOUBLE;
+ literal.double_value = value;
+ return literal;
+}
+
+AlignedFieldFilter::Literal AlignedFieldFilter::Literal::string(
+ std::string value) {
+ Literal literal;
+ literal.type = common::STRING;
+ literal.string_value = std::move(value);
+ return literal;
+}
+
+AlignedFieldFilter::AlignedFieldFilter(Kind kind, std::string column,
+ CompareOp op, Literal literal)
+ : kind_(kind),
+ column_(std::move(column)),
+ compare_op_(op),
+ literal_(std::move(literal)) {}
+
+AlignedFieldFilter::AlignedFieldFilter(Kind kind, Ptr left, Ptr right)
+ : kind_(kind), left_(std::move(left)), right_(std::move(right)) {}
+
+AlignedFieldFilter::Ptr AlignedFieldFilter::compare(std::string column,
+ CompareOp op,
+ Literal literal) {
+ return Ptr(new AlignedFieldFilter(Kind::COMPARISON,
+ normalize_column_name(column), op,
+ std::move(literal)));
+}
+
+AlignedFieldFilter::Ptr AlignedFieldFilter::is_null(std::string column) {
+ return Ptr(new AlignedFieldFilter(Kind::IS_NULL,
+ normalize_column_name(column),
+ CompareOp::EQ, Literal()));
+}
+
+AlignedFieldFilter::Ptr AlignedFieldFilter::is_not_null(std::string column) {
+ return Ptr(new AlignedFieldFilter(Kind::IS_NOT_NULL,
+ normalize_column_name(column),
+ CompareOp::EQ, Literal()));
+}
+
+AlignedFieldFilter::Ptr AlignedFieldFilter::and_filter(Ptr left, Ptr right) {
+ return Ptr(
+ new AlignedFieldFilter(Kind::AND, std::move(left), std::move(right)));
+}
+
+AlignedFieldFilter::Ptr AlignedFieldFilter::or_filter(Ptr left, Ptr right) {
+ return Ptr(
+ new AlignedFieldFilter(Kind::OR, std::move(left), std::move(right)));
+}
+
+AlignedFieldFilter::Ptr AlignedFieldFilter::not_filter(Ptr child) {
+ return negate(std::move(child));
+}
+
+AlignedFieldFilter::Ptr AlignedFieldFilter::negate(Ptr node) {
+ if (node == nullptr) return nullptr;
+ switch (node->kind_) {
+ case Kind::COMPARISON: {
+ CompareOp neg_op = CompareOp::EQ;
+ switch (node->compare_op_) {
+ case CompareOp::EQ:
+ neg_op = CompareOp::NE;
+ break;
+ case CompareOp::NE:
+ neg_op = CompareOp::EQ;
+ break;
+ case CompareOp::LT:
+ neg_op = CompareOp::GE;
+ break;
+ case CompareOp::LE:
+ neg_op = CompareOp::GT;
+ break;
+ case CompareOp::GT:
+ neg_op = CompareOp::LE;
+ break;
+ case CompareOp::GE:
+ neg_op = CompareOp::LT;
+ break;
+ }
+ return Ptr(new AlignedFieldFilter(Kind::COMPARISON, node->column_,
+ neg_op, node->literal_));
+ }
+ case Kind::IS_NULL:
+ return is_not_null(node->column_);
+ case Kind::IS_NOT_NULL:
+ return is_null(node->column_);
+ case Kind::NOT:
+ return std::move(node->left_);
+ case Kind::AND:
+ return or_filter(negate(std::move(node->left_)),
+ negate(std::move(node->right_)));
+ case Kind::OR:
+ return and_filter(negate(std::move(node->left_)),
+ negate(std::move(node->right_)));
+ }
+ return nullptr;
+}
+
+std::vector<std::string> AlignedFieldFilter::referenced_columns() const {
+ std::vector<std::string> columns;
+ std::set<std::string> seen_columns;
+ std::set<const AlignedFieldFilter*> seen_nodes;
+ std::vector<const AlignedFieldFilter*> stack{this};
+ while (!stack.empty()) {
+ const AlignedFieldFilter* node = stack.back();
+ stack.pop_back();
+ if (node == nullptr || !seen_nodes.insert(node).second) continue;
+ if (node->kind_ == Kind::COMPARISON || node->kind_ == Kind::IS_NULL ||
+ node->kind_ == Kind::IS_NOT_NULL) {
+ if (seen_columns.insert(node->column_).second) {
+ columns.push_back(node->column_);
+ }
+ continue;
+ }
+ if (node->right_) stack.push_back(node->right_.get());
+ if (node->left_) stack.push_back(node->left_.get());
+ }
+ return columns;
+}
+
+bool AlignedFieldFilter::may_match(
+ const std::map<std::string, Statistic*>& statistics,
+ int64_t row_count) const {
+ if (kind_ == Kind::AND) {
+ return !left_ || !right_ ||
+ (left_->may_match(statistics, row_count) &&
+ right_->may_match(statistics, row_count));
+ }
+ if (kind_ == Kind::OR) {
+ return !left_ || !right_ ||
+ (left_->may_match(statistics, row_count) ||
+ right_->may_match(statistics, row_count));
+ }
+ if (kind_ == Kind::NOT) {
+ // NOT can be pushed down to leaves in a future planner. Until then,
+ // never use page statistics to prune a negated predicate.
+ return true;
+ }
+
+ const auto found = statistics.find(column_);
+ if (found == statistics.end() || found->second == nullptr) return true;
+ Statistic* statistic = found->second;
+ if (kind_ == Kind::IS_NULL) {
+ return row_count <= 0 || statistic->count_ < row_count;
+ }
+ if (kind_ == Kind::IS_NOT_NULL) {
+ return statistic->count_ > 0;
+ }
+ if (statistic->count_ <= 0) return false;
+
+ if (auto* typed = dynamic_cast<Int32Statistic*>(statistic)) {
+ if (literal_.type != common::INT32) return true;
+ switch (compare_op_) {
+ case CompareOp::EQ:
+ return typed->min_value_ <= literal_.int32_value &&
+ literal_.int32_value <= typed->max_value_;
+ case CompareOp::NE:
+ return !(typed->min_value_ == typed->max_value_ &&
+ typed->min_value_ == literal_.int32_value &&
+ typed->count_ == row_count);
+ case CompareOp::LT:
+ return typed->min_value_ < literal_.int32_value;
+ case CompareOp::LE:
+ return typed->min_value_ <= literal_.int32_value;
+ case CompareOp::GT:
+ return typed->max_value_ > literal_.int32_value;
+ case CompareOp::GE:
+ return typed->max_value_ >= literal_.int32_value;
+ }
+ }
+ if (auto* typed = dynamic_cast<Int64Statistic*>(statistic)) {
+ if (literal_.type != common::INT64) return true;
+ switch (compare_op_) {
+ case CompareOp::EQ:
+ return typed->min_value_ <= literal_.int64_value &&
+ literal_.int64_value <= typed->max_value_;
+ case CompareOp::NE:
+ return !(typed->min_value_ == typed->max_value_ &&
+ typed->min_value_ == literal_.int64_value &&
+ typed->count_ == row_count);
+ case CompareOp::LT:
+ return typed->min_value_ < literal_.int64_value;
+ case CompareOp::LE:
+ return typed->min_value_ <= literal_.int64_value;
+ case CompareOp::GT:
+ return typed->max_value_ > literal_.int64_value;
+ case CompareOp::GE:
+ return typed->max_value_ >= literal_.int64_value;
+ }
+ }
+ if (auto* typed = dynamic_cast<FloatStatistic*>(statistic)) {
+ if (literal_.type != common::FLOAT) return true;
+ switch (compare_op_) {
+ case CompareOp::EQ:
+ return typed->min_value_ <= literal_.float_value &&
+ literal_.float_value <= typed->max_value_;
+ case CompareOp::NE:
+ return !(typed->min_value_ == typed->max_value_ &&
+ typed->min_value_ == literal_.float_value &&
+ typed->count_ == row_count);
+ case CompareOp::LT:
+ return typed->min_value_ < literal_.float_value;
+ case CompareOp::LE:
+ return typed->min_value_ <= literal_.float_value;
+ case CompareOp::GT:
+ return typed->max_value_ > literal_.float_value;
+ case CompareOp::GE:
+ return typed->max_value_ >= literal_.float_value;
+ }
+ }
+ if (auto* typed = dynamic_cast<DoubleStatistic*>(statistic)) {
+ if (literal_.type != common::DOUBLE) return true;
+ switch (compare_op_) {
+ case CompareOp::EQ:
+ return typed->min_value_ <= literal_.double_value &&
+ literal_.double_value <= typed->max_value_;
+ case CompareOp::NE:
+ return !(typed->min_value_ == typed->max_value_ &&
+ typed->min_value_ == literal_.double_value &&
+ typed->count_ == row_count);
+ case CompareOp::LT:
+ return typed->min_value_ < literal_.double_value;
+ case CompareOp::LE:
+ return typed->min_value_ <= literal_.double_value;
+ case CompareOp::GT:
+ return typed->max_value_ > literal_.double_value;
+ case CompareOp::GE:
+ return typed->max_value_ >= literal_.double_value;
+ }
+ }
+ return true;
+}
+
+bool AlignedFieldFilter::all_match(
+ const std::map<std::string, Statistic*>& statistics,
+ int64_t row_count) const {
+ if (row_count <= 0) return false;
+ if (kind_ == Kind::AND) {
+ return left_ && right_ && left_->all_match(statistics, row_count) &&
+ right_->all_match(statistics, row_count);
+ }
+ if (kind_ == Kind::OR) {
+ return (left_ && left_->all_match(statistics, row_count)) ||
+ (right_ && right_->all_match(statistics, row_count));
+ }
+ if (kind_ == Kind::NOT) {
+ // Physical plans may keep a NOT node for now. Do not infer a
+ // page-wide match from its child statistics.
+ return false;
+ }
+
+ const auto found = statistics.find(column_);
+ if (found == statistics.end() || found->second == nullptr) return false;
+ Statistic* statistic = found->second;
+ if (kind_ == Kind::IS_NULL) {
+ return statistic->count_ == 0;
+ }
+ if (kind_ == Kind::IS_NOT_NULL) {
+ return statistic->count_ == row_count;
+ }
+ // Ordinary comparisons do not match NULL. A page-wide match requires that
+ // every row has a non-NULL value and that the whole min/max range passes.
+ if (statistic->count_ != row_count) return false;
+
+ if (auto* typed = dynamic_cast<Int32Statistic*>(statistic)) {
+ if (literal_.type != common::INT32) return false;
+ switch (compare_op_) {
+ case CompareOp::EQ:
+ return typed->min_value_ == literal_.int32_value &&
+ typed->max_value_ == literal_.int32_value;
+ case CompareOp::NE:
+ return literal_.int32_value < typed->min_value_ ||
+ literal_.int32_value > typed->max_value_;
+ case CompareOp::LT:
+ return typed->max_value_ < literal_.int32_value;
+ case CompareOp::LE:
+ return typed->max_value_ <= literal_.int32_value;
+ case CompareOp::GT:
+ return typed->min_value_ > literal_.int32_value;
+ case CompareOp::GE:
+ return typed->min_value_ >= literal_.int32_value;
+ }
+ }
+ if (auto* typed = dynamic_cast<Int64Statistic*>(statistic)) {
+ if (literal_.type != common::INT64) return false;
+ switch (compare_op_) {
+ case CompareOp::EQ:
+ return typed->min_value_ == literal_.int64_value &&
+ typed->max_value_ == literal_.int64_value;
+ case CompareOp::NE:
+ return literal_.int64_value < typed->min_value_ ||
+ literal_.int64_value > typed->max_value_;
+ case CompareOp::LT:
+ return typed->max_value_ < literal_.int64_value;
+ case CompareOp::LE:
+ return typed->max_value_ <= literal_.int64_value;
+ case CompareOp::GT:
+ return typed->min_value_ > literal_.int64_value;
+ case CompareOp::GE:
+ return typed->min_value_ >= literal_.int64_value;
+ }
+ }
+ if (auto* typed = dynamic_cast<FloatStatistic*>(statistic)) {
+ if (literal_.type != common::FLOAT) return false;
+ switch (compare_op_) {
+ case CompareOp::EQ:
+ return typed->min_value_ == literal_.float_value &&
+ typed->max_value_ == literal_.float_value;
+ case CompareOp::NE:
+ return literal_.float_value < typed->min_value_ ||
+ literal_.float_value > typed->max_value_;
+ case CompareOp::LT:
+ return typed->max_value_ < literal_.float_value;
+ case CompareOp::LE:
+ return typed->max_value_ <= literal_.float_value;
+ case CompareOp::GT:
+ return typed->min_value_ > literal_.float_value;
+ case CompareOp::GE:
+ return typed->min_value_ >= literal_.float_value;
+ }
+ }
+ if (auto* typed = dynamic_cast<DoubleStatistic*>(statistic)) {
+ if (literal_.type != common::DOUBLE) return false;
+ switch (compare_op_) {
+ case CompareOp::EQ:
+ return typed->min_value_ == literal_.double_value &&
+ typed->max_value_ == literal_.double_value;
+ case CompareOp::NE:
+ return literal_.double_value < typed->min_value_ ||
+ literal_.double_value > typed->max_value_;
+ case CompareOp::LT:
+ return typed->max_value_ < literal_.double_value;
+ case CompareOp::LE:
+ return typed->max_value_ <= literal_.double_value;
+ case CompareOp::GT:
+ return typed->min_value_ > literal_.double_value;
+ case CompareOp::GE:
+ return typed->min_value_ >= literal_.double_value;
+ }
+ }
+ return false;
+}
+
+int AlignedFieldFilter::evaluate(common::TsBlock& block,
+ const std::vector<std::string>& column_names,
+ SelectionMask& mask) const {
+ std::map<std::string, uint32_t> column_index;
+ for (uint32_t i = 0; i < column_names.size(); ++i) {
+ column_index[column_names[i]] = i + 1;
+ }
+
+ struct Frame {
+ const AlignedFieldFilter* node;
+ uint8_t state;
+ };
+ std::vector<Frame> traversal;
+ std::vector<const AlignedFieldFilter*> postorder;
+ traversal.push_back({this, 0});
+ while (!traversal.empty()) {
+ Frame& frame = traversal.back();
+ const AlignedFieldFilter* node = frame.node;
+ if (node->kind_ == Kind::COMPARISON || node->kind_ == Kind::IS_NULL ||
+ node->kind_ == Kind::IS_NOT_NULL) {
+ postorder.push_back(node);
+ traversal.pop_back();
+ } else if (node->kind_ == Kind::NOT) {
+ if (!node->left_) return common::E_INVALID_ARG;
+ if (frame.state == 0) {
+ frame.state = 1;
+ traversal.push_back({node->left_.get(), 0});
+ } else {
+ postorder.push_back(node);
+ traversal.pop_back();
+ }
+ } else {
+ if (!node->left_ || !node->right_) return common::E_INVALID_ARG;
+ if (frame.state == 0) {
+ frame.state = 1;
+ traversal.push_back({node->left_.get(), 0});
+ } else if (frame.state == 1) {
+ frame.state = 2;
+ traversal.push_back({node->right_.get(), 0});
+ } else {
+ postorder.push_back(node);
+ traversal.pop_back();
+ }
+ }
+ }
+
+ std::vector<SelectionMask> masks;
+ masks.reserve(postorder.size());
+ for (const auto* node : postorder) {
+ if (node->kind_ == Kind::COMPARISON || node->kind_ == Kind::IS_NULL ||
+ node->kind_ == Kind::IS_NOT_NULL) {
+ SelectionMask leaf_mask(block.get_row_count());
+ int ret = node->evaluate_leaf(block, column_index, leaf_mask);
+ if (ret != common::E_OK) return ret;
+ masks.push_back(std::move(leaf_mask));
+ } else if (node->kind_ == Kind::NOT) {
+ if (masks.empty()) return common::E_INVALID_ARG;
+ masks.back().invert();
+ } else {
+ if (masks.size() < 2) return common::E_INVALID_ARG;
+ SelectionMask right = std::move(masks.back());
+ masks.pop_back();
+ if (node->kind_ == Kind::AND) {
+ masks.back().and_with(right);
+ } else {
+ masks.back().or_with(right);
+ }
+ }
+ }
+ if (masks.size() != 1) return common::E_INVALID_ARG;
+ mask = std::move(masks.back());
+ return common::E_OK;
+}
+
+int AlignedFieldFilter::evaluate_columns(const std::vector<ColumnView>& columns,
+ uint32_t row_count,
+ SelectionMask& mask) const {
+ std::map<std::string, uint32_t> column_index;
+ for (uint32_t i = 0; i < columns.size(); ++i) {
+ column_index[columns[i].name] = i;
+ }
+
+ struct Frame {
+ const AlignedFieldFilter* node;
+ uint8_t state;
+ };
+ std::vector<Frame> traversal{{this, 0}};
+ std::vector<const AlignedFieldFilter*> postorder;
+ while (!traversal.empty()) {
+ Frame& frame = traversal.back();
+ const AlignedFieldFilter* node = frame.node;
+ if (node->kind_ == Kind::COMPARISON || node->kind_ == Kind::IS_NULL ||
+ node->kind_ == Kind::IS_NOT_NULL) {
+ postorder.push_back(node);
+ traversal.pop_back();
+ } else if (node->kind_ == Kind::NOT) {
+ if (!node->left_) return common::E_INVALID_ARG;
+ if (frame.state == 0) {
+ frame.state = 1;
+ traversal.push_back({node->left_.get(), 0});
+ } else {
+ postorder.push_back(node);
+ traversal.pop_back();
+ }
+ } else {
+ if (!node->left_ || !node->right_) return common::E_INVALID_ARG;
+ if (frame.state == 0) {
+ frame.state = 1;
+ traversal.push_back({node->left_.get(), 0});
+ } else if (frame.state == 1) {
+ frame.state = 2;
+ traversal.push_back({node->right_.get(), 0});
+ } else {
+ postorder.push_back(node);
+ traversal.pop_back();
+ }
+ }
+ }
+
+ std::vector<SelectionMask> masks;
+ masks.reserve(postorder.size());
+ for (const auto* node : postorder) {
+ if (node->kind_ == Kind::COMPARISON || node->kind_ == Kind::IS_NULL ||
+ node->kind_ == Kind::IS_NOT_NULL) {
+ const auto found = column_index.find(node->column_);
+ if (found == column_index.end()) return common::E_COLUMN_NOT_EXIST;
+ SelectionMask leaf_mask(row_count);
+ int ret = node->evaluate_leaf(columns[found->second], row_count,
+ leaf_mask);
+ if (ret != common::E_OK) return ret;
+ masks.push_back(std::move(leaf_mask));
+ } else if (node->kind_ == Kind::NOT) {
+ if (masks.empty()) return common::E_INVALID_ARG;
+ masks.back().invert();
+ } else {
+ if (masks.size() < 2) return common::E_INVALID_ARG;
+ SelectionMask right = std::move(masks.back());
+ masks.pop_back();
+ if (node->kind_ == Kind::AND) {
+ masks.back().and_with(right);
+ } else {
+ masks.back().or_with(right);
+ }
+ }
+ }
+ if (masks.size() != 1) return common::E_INVALID_ARG;
+ mask = std::move(masks.back());
+ return common::E_OK;
+}
+
+int AlignedFieldFilter::evaluate_leaf(const ColumnView& column,
+ uint32_t row_count,
+ SelectionMask& mask) const {
+ if (column.type == common::INVALID_DATATYPE) return common::E_INVALID_ARG;
+ uint32_t nonnull_index = column.nonnull_start;
+ for (uint32_t row = 0; row < row_count; ++row) {
+ const uint32_t source_row = column.row_offset + row;
+ bool is_null = true;
+ if (column.notnull_bitmap != nullptr &&
+ source_row < column.notnull_bitmap_size * 8U) {
+ is_null = (column.notnull_bitmap[source_row / 8U] &
+ (0x80U >> (source_row % 8U))) == 0;
+ }
+
+ const char* data = nullptr;
+ uint32_t len = 0;
+ if (!is_null) {
+ if (column.type == common::STRING || column.type == common::TEXT ||
+ column.type == common::BLOB) {
+ if (column.string_values == nullptr) {
+ return common::E_INVALID_ARG;
+ }
+ data = column.string_values[nonnull_index].buf_;
+ len = column.string_values[nonnull_index].len_;
+ } else {
+ if (column.fixed_values == nullptr) {
+ return common::E_INVALID_ARG;
+ }
+ const uint32_t elem_size =
+ common::get_data_type_size(column.type);
+ data = column.fixed_values +
+ static_cast<size_t>(nonnull_index) * elem_size;
+ len = elem_size;
+ }
+ ++nonnull_index;
+ }
+
+ bool satisfied = false;
+ if (kind_ == Kind::IS_NULL) {
+ satisfied = is_null;
+ } else if (kind_ == Kind::IS_NOT_NULL) {
+ satisfied = !is_null;
+ } else if (!is_null) {
+ if (column.type == common::BOOLEAN &&
+ literal_.type == common::BOOLEAN) {
+ bool value;
+ std::memcpy(&value, data, sizeof(value));
+ satisfied =
+ compare_values(value, literal_.bool_value, compare_op_);
+ } else if ((column.type == common::INT32 ||
+ column.type == common::DATE) &&
+ literal_.type == common::INT32) {
+ int32_t value;
+ std::memcpy(&value, data, sizeof(value));
+ satisfied =
+ compare_values(value, literal_.int32_value, compare_op_);
+ } else if ((column.type == common::INT64 ||
+ column.type == common::TIMESTAMP) &&
+ literal_.type == common::INT64) {
+ int64_t value;
+ std::memcpy(&value, data, sizeof(value));
+ satisfied =
+ compare_values(value, literal_.int64_value, compare_op_);
+ } else if (column.type == common::FLOAT &&
+ literal_.type == common::FLOAT) {
+ float value;
+ std::memcpy(&value, data, sizeof(value));
+ satisfied =
+ compare_values(value, literal_.float_value, compare_op_);
+ } else if (column.type == common::DOUBLE &&
+ literal_.type == common::DOUBLE) {
+ double value;
+ std::memcpy(&value, data, sizeof(value));
+ satisfied =
+ compare_values(value, literal_.double_value, compare_op_);
+ } else if (is_string_type(column.type) &&
+ is_string_type(literal_.type)) {
+ satisfied = compare_string(data, len, literal_.string_value,
+ compare_op_);
+ } else {
+ return common::E_NOT_SUPPORT;
+ }
+ }
+ if (satisfied) mask.set_one(row);
+ }
+ return common::E_OK;
+}
+
+int AlignedFieldFilter::evaluate_leaf(
+ common::TsBlock& block, const std::map<std::string, uint32_t>& column_index,
+ SelectionMask& mask) const {
+ const auto found = column_index.find(column_);
+ if (found == column_index.end()) return common::E_COLUMN_NOT_EXIST;
+
+ const uint32_t vector_index = found->second;
+ if (vector_index >= block.get_column_count()) return common::E_INVALID_ARG;
+ common::Vector* vector = block.get_vector(vector_index);
+ if (vector == nullptr) return common::E_INVALID_ARG;
+
+ common::ColIterator iter(vector_index, &block);
+ const common::TSDataType column_type = vector->get_vector_type();
+ for (uint32_t row = 0; row < block.get_row_count(); ++row) {
+ uint32_t len = 0;
+ bool is_null = false;
+ char* data = iter.read(&len, &is_null);
+
+ bool satisfied = false;
+ if (kind_ == Kind::IS_NULL) {
+ satisfied = is_null;
+ } else if (kind_ == Kind::IS_NOT_NULL) {
+ satisfied = !is_null;
+ } else if (!is_null) {
+ if (column_type == common::BOOLEAN &&
+ literal_.type == common::BOOLEAN) {
+ bool value;
+ std::memcpy(&value, data, sizeof(value));
+ satisfied =
+ compare_values(value, literal_.bool_value, compare_op_);
+ } else if ((column_type == common::INT32 ||
+ column_type == common::DATE) &&
+ literal_.type == common::INT32) {
+ int32_t value;
+ std::memcpy(&value, data, sizeof(value));
+ satisfied =
+ compare_values(value, literal_.int32_value, compare_op_);
+ } else if ((column_type == common::INT64 ||
+ column_type == common::TIMESTAMP) &&
+ literal_.type == common::INT64) {
+ int64_t value;
+ std::memcpy(&value, data, sizeof(value));
+ satisfied =
+ compare_values(value, literal_.int64_value, compare_op_);
+ } else if (column_type == common::FLOAT &&
+ literal_.type == common::FLOAT) {
+ float value;
+ std::memcpy(&value, data, sizeof(value));
+ satisfied =
+ compare_values(value, literal_.float_value, compare_op_);
+ } else if (column_type == common::DOUBLE &&
+ literal_.type == common::DOUBLE) {
+ double value;
+ std::memcpy(&value, data, sizeof(value));
+ satisfied =
+ compare_values(value, literal_.double_value, compare_op_);
+ } else if (is_string_type(column_type) &&
+ is_string_type(literal_.type)) {
+ satisfied = compare_string(data, len, literal_.string_value,
+ compare_op_);
+ } else {
+ return common::E_NOT_SUPPORT;
+ }
+ }
+ if (satisfied) mask.set_one(row);
+ iter.next();
+ }
+ return common::E_OK;
+}
+
+} // namespace storage
diff --git a/cpp/src/reader/filter/aligned_field_filter.h b/cpp/src/reader/filter/aligned_field_filter.h
new file mode 100644
index 000000000..64f571dcf
--- /dev/null
+++ b/cpp/src/reader/filter/aligned_field_filter.h
@@ -0,0 +1,129 @@
+/*
+ * 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.
+ */
+
+#ifndef READER_FILTER_ALIGNED_FIELD_FILTER_H
+#define READER_FILTER_ALIGNED_FIELD_FILTER_H
+
+#include <cstdint>
+#include <map>
+#include <memory>
+#include <set>
+#include <string>
+#include <vector>
+
+#include "common/tsblock/tsblock.h"
+#include "reader/filter/filter.h"
+#include "reader/filter/selection_mask.h"
+
+namespace storage {
+
+// Column-aware predicate used by aligned batch readers. It deliberately does
+// not implement TableFieldFilter::satisfy_row(); aligned execution evaluates
+// it against a whole TsBlock selection.
+class AlignedFieldFilter : public Filter {
+ public:
+ enum class Kind { COMPARISON, IS_NULL, IS_NOT_NULL, AND, OR, NOT };
+
+ enum class CompareOp { EQ, NE, LT, LE, GT, GE };
+
+ struct ColumnView {
+ std::string name;
+ common::TSDataType type = common::INVALID_DATATYPE;
+ const char* fixed_values = nullptr;
+ const common::String* string_values = nullptr;
+ const uint8_t* notnull_bitmap = nullptr;
+ uint32_t notnull_bitmap_size = 0;
+ uint32_t row_offset = 0;
+ uint32_t nonnull_start = 0;
+ };
+
+ struct Literal {
+ common::TSDataType type = common::INVALID_DATATYPE;
+ bool bool_value = false;
+ int32_t int32_value = 0;
+ int64_t int64_value = 0;
+ float float_value = 0.0F;
+ double double_value = 0.0;
+ std::string string_value;
+
+ static Literal boolean(bool value);
+ static Literal int32(int32_t value);
+ static Literal int64(int64_t value);
+ static Literal float32(float value);
+ static Literal float64(double value);
+ static Literal string(std::string value);
+ };
+
+ using Ptr = std::unique_ptr<AlignedFieldFilter>;
+
+ ~AlignedFieldFilter() override = default;
+
+ static Ptr compare(std::string column, CompareOp op, Literal literal);
+ static Ptr is_null(std::string column);
+ static Ptr is_not_null(std::string column);
+ static Ptr and_filter(Ptr left, Ptr right);
+ static Ptr or_filter(Ptr left, Ptr right);
+ static Ptr not_filter(Ptr child);
+
+ Kind kind() const { return kind_; }
+ CompareOp compare_op() const { return compare_op_; }
+ const std::string& column() const { return column_; }
+ const Literal& literal() const { return literal_; }
+
+ std::vector<std::string> referenced_columns() const;
+
+ // Conservatively decides whether any row in a page can satisfy the filter.
+ // Missing or unsupported statistics always return true.
+ bool may_match(const std::map<std::string, Statistic*>& statistics,
+ int64_t row_count) const;
+ bool all_match(const std::map<std::string, Statistic*>& statistics,
+ int64_t row_count) const;
+
+ // Evaluates against block.get_vector(0) as time and column_names[i] as
+ // block.get_vector(i + 1). The mask is reset to block row count.
+ int evaluate(common::TsBlock& block,
+ const std::vector<std::string>& column_names,
+ SelectionMask& mask) const;
+ int evaluate_columns(const std::vector<ColumnView>& columns,
+ uint32_t row_count, SelectionMask& mask) const;
+
+ private:
+ AlignedFieldFilter(Kind kind, std::string column, CompareOp op,
+ Literal literal);
+ AlignedFieldFilter(Kind kind, Ptr left, Ptr right);
+ // Push NOT down to the leaves so negation keeps SQL three-valued NULL
+ // semantics; a single SelectionMask bit cannot represent NULL vs FALSE.
+ static Ptr negate(Ptr node);
+
+ int evaluate_leaf(common::TsBlock& block,
+ const std::map<std::string, uint32_t>& column_index,
+ SelectionMask& mask) const;
+ int evaluate_leaf(const ColumnView& column, uint32_t row_count,
+ SelectionMask& mask) const;
+ Kind kind_;
+ std::string column_;
+ CompareOp compare_op_ = CompareOp::EQ;
+ Literal literal_;
+ Ptr left_;
+ Ptr right_;
+};
+
+} // namespace storage
+
+#endif // READER_FILTER_ALIGNED_FIELD_FILTER_H
diff --git a/cpp/src/reader/filter/selection_mask.cc b/cpp/src/reader/filter/selection_mask.cc
new file mode 100644
index 000000000..8ee93c1f3
--- /dev/null
+++ b/cpp/src/reader/filter/selection_mask.cc
@@ -0,0 +1,157 @@
+/*
+ * 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 "reader/filter/selection_mask.h"
+
+#include <algorithm>
+#include <cassert>
+
+namespace storage {
+namespace {
+uint32_t count_bits(uint64_t value) {
+#if defined(__GNUC__) || defined(__clang__)
+ return static_cast<uint32_t>(__builtin_popcountll(value));
+#else
+ uint32_t count = 0;
+ while (value != 0) {
+ value &= value - 1;
+ ++count;
+ }
+ return count;
+#endif
+}
+
+uint32_t trailing_zeros(uint64_t value) {
+ assert(value != 0);
+#if defined(__GNUC__) || defined(__clang__)
+ return static_cast<uint32_t>(__builtin_ctzll(value));
+#else
+ uint32_t count = 0;
+ while ((value & 1U) == 0) {
+ value >>= 1U;
+ ++count;
+ }
+ return count;
+#endif
+}
+} // namespace
+
+SelectionMask::SelectionMask() : row_count_(0) {}
+
+SelectionMask::SelectionMask(uint32_t row_count) { reset(row_count); }
+
+void SelectionMask::reset(uint32_t row_count) {
+ row_count_ = row_count;
+ words_.assign(word_count(row_count), 0);
+}
+
+void SelectionMask::set_all(bool value) {
+ const uint64_t fill = value ? ~static_cast<uint64_t>(0) : 0;
+ for (auto& word : words_) {
+ word = fill;
+ }
+ clear_unused_bits();
+}
+
+void SelectionMask::set_one(uint32_t row) {
+ if (row >= row_count_) return;
+ words_[row >> 6U] |= (static_cast<uint64_t>(1) << (row & 63U));
+}
+
+bool SelectionMask::test(uint32_t row) const {
+ if (row >= row_count_) return false;
+ return (words_[row >> 6U] & (static_cast<uint64_t>(1) << (row & 63U))) != 0;
+}
+
+void SelectionMask::and_with(const SelectionMask& other) {
+ assert(row_count_ == other.row_count_);
+ const size_t common = std::min(words_.size(), other.words_.size());
+ for (size_t i = 0; i < common; ++i) {
+ words_[i] &= other.words_[i];
+ }
+ // Rows past the shorter mask have no counterpart; ANDing with nothing
+ // clears them so a mismatched operand cannot leave stale bits.
+ for (size_t i = common; i < words_.size(); ++i) {
+ words_[i] = 0;
+ }
+}
+
+void SelectionMask::or_with(const SelectionMask& other) {
+ assert(row_count_ == other.row_count_);
+ const size_t common = std::min(words_.size(), other.words_.size());
+ for (size_t i = 0; i < common; ++i) {
+ words_[i] |= other.words_[i];
+ }
+ // Rows past the shorter mask keep their value (ORing with nothing).
+}
+
+void SelectionMask::invert() {
+ for (auto& word : words_) {
+ word = ~word;
+ }
+ clear_unused_bits();
+}
+
+uint32_t SelectionMask::count_ones() const {
+ uint32_t count = 0;
+ for (uint64_t word : words_) {
+ count += count_bits(word);
+ }
+ return count;
+}
+
+bool SelectionMask::next_set_bit(uint32_t from, uint32_t& next) const {
+ if (from >= row_count_) return false;
+ uint32_t word_index = from >> 6U;
+ uint64_t word =
+ words_[word_index] & (~static_cast<uint64_t>(0) << (from & 63U));
+ while (true) {
+ if (word != 0) {
+ const uint32_t candidate =
+ (word_index << 6U) + trailing_zeros(word);
+ if (candidate >= row_count_) return false;
+ next = candidate;
+ return true;
+ }
+ ++word_index;
+ if (word_index >= words_.size()) return false;
+ word = words_[word_index];
+ }
+}
+
+void SelectionMask::clear_from(uint32_t row) {
+ if (row >= row_count_) return;
+ const uint32_t first_word = row >> 6U;
+ const uint32_t bit = row & 63U;
+ words_[first_word] &=
+ bit == 0 ? 0 : ((static_cast<uint64_t>(1) << bit) - 1U);
+ for (size_t i = first_word + 1; i < words_.size(); ++i) {
+ words_[i] = 0;
+ }
+}
+
+void SelectionMask::clear_unused_bits() {
+ if (words_.empty()) return;
+ const uint32_t used = row_count_ & 63U;
+ if (used != 0) {
+ words_.back() &= ((static_cast<uint64_t>(1) << used) - 1U);
+ }
+}
+
+} // namespace storage
diff --git a/cpp/src/reader/filter/selection_mask.h b/cpp/src/reader/filter/selection_mask.h
new file mode 100644
index 000000000..ed091e9a8
--- /dev/null
+++ b/cpp/src/reader/filter/selection_mask.h
@@ -0,0 +1,65 @@
+/*
+ * 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.
+ */
+
+#ifndef READER_FILTER_SELECTION_MASK_H
+#define READER_FILTER_SELECTION_MASK_H
+
+#include <cstddef>
+#include <cstdint>
+#include <vector>
+
+namespace storage {
+
+// A compact row-selection bitmap. Bit i corresponds to row i in an aligned
+// batch. All operations require masks with the same row count.
+class SelectionMask {
+ public:
+ SelectionMask();
+ explicit SelectionMask(uint32_t row_count);
+
+ void reset(uint32_t row_count);
+ void set_all(bool value = true);
+
+ uint32_t row_count() const { return row_count_; }
+ bool empty() const { return count_ones() == 0; }
+
+ void set_one(uint32_t row);
+ bool test(uint32_t row) const;
+
+ void and_with(const SelectionMask& other);
+ void or_with(const SelectionMask& other);
+ void invert();
+
+ uint32_t count_ones() const;
+ bool next_set_bit(uint32_t from, uint32_t& next) const;
+ void clear_from(uint32_t row);
+
+ private:
+ static size_t word_count(uint32_t row_count) {
+ return (static_cast<uint64_t>(row_count) + 63U) >> 6U;
+ }
+ void clear_unused_bits();
+
+ std::vector<uint64_t> words_;
+ uint32_t row_count_;
+};
+
+} // namespace storage
+
+#endif // READER_FILTER_SELECTION_MASK_H
diff --git a/cpp/src/reader/table_query_executor.cc b/cpp/src/reader/table_query_executor.cc
index 5d708baae..084e1b363 100644
--- a/cpp/src/reader/table_query_executor.cc
+++ b/cpp/src/reader/table_query_executor.cc
@@ -31,6 +31,7 @@ int TableQueryExecutor::query(const std::string& table_name,
ret_qds = nullptr;
if (RET_FAIL(tsfile_io_reader_->get_tsfile_meta(file_metadata))) {
delete time_filter;
+ delete field_filter;
return ret;
}
common::PageArena pa;
@@ -44,6 +45,7 @@ int TableQueryExecutor::query(const std::string& table_name,
if (RET_FAIL(file_metadata->get_table_schema(table_name, table_schema))) {
ret_qds = nullptr;
delete time_filter;
+ delete field_filter;
return ret;
}
file_metadata->get_table_metaindex_node(table_name, table_root);
@@ -63,11 +65,21 @@ int TableQueryExecutor::query(const std::string& table_name,
auto ind = table_schema->find_column_index(lower_case_column_names[i]);
if (ind < 0) {
delete time_filter;
+ delete field_filter;
return common::E_COLUMN_NOT_EXIST;
}
data_types.push_back(table_schema->get_data_types()[ind]);
}
- // column_mapping.add(*measurement_filter);
+ if (auto* aligned_filter =
+ dynamic_cast<AlignedFieldFilter*>(field_filter)) {
+ ret =
+ column_mapping->add_filter_columns(*aligned_filter, *table_schema);
+ if (ret != common::E_OK) {
+ delete time_filter;
+ delete field_filter;
+ return ret;
+ }
+ }
auto device_task_iterator =
std::unique_ptr<DeviceTaskIterator>(new DeviceTaskIterator(
@@ -85,6 +97,7 @@ int TableQueryExecutor::query(const std::string& table_name,
case TableQueryOrdering::TIME:
default:
delete time_filter;
+ delete field_filter;
ret_qds = nullptr;
return common::E_UNSUPPORTED_ORDER;
}
@@ -105,6 +118,7 @@ int TableQueryExecutor::query(const std::string& table_name,
ret_qds = nullptr;
if (RET_FAIL(tsfile_io_reader_->get_tsfile_meta(file_metadata))) {
delete time_filter;
+ delete field_filter;
return ret;
}
common::PageArena pa;
@@ -114,6 +128,7 @@ int TableQueryExecutor::query(const std::string& table_name,
if (RET_FAIL(file_metadata->get_table_schema(table_name, table_schema))) {
ret_qds = nullptr;
delete time_filter;
+ delete field_filter;
return ret;
}
file_metadata->get_table_metaindex_node(table_name, table_root);
@@ -133,10 +148,21 @@ int TableQueryExecutor::query(const std::string& table_name,
auto ind = table_schema->find_column_index(lower_case_column_names[i]);
if (ind < 0) {
delete time_filter;
+ delete field_filter;
return common::E_COLUMN_NOT_EXIST;
}
data_types.push_back(table_schema->get_data_types()[ind]);
}
+ if (auto* aligned_filter =
+ dynamic_cast<AlignedFieldFilter*>(field_filter)) {
+ ret =
+ column_mapping->add_filter_columns(*aligned_filter, *table_schema);
+ if (ret != common::E_OK) {
+ delete time_filter;
+ delete field_filter;
+ return ret;
+ }
+ }
auto device_task_iterator =
std::unique_ptr<DeviceTaskIterator>(new DeviceTaskIterator(
@@ -155,6 +181,7 @@ int TableQueryExecutor::query(const std::string& table_name,
case TableQueryOrdering::TIME:
default:
delete time_filter;
+ delete field_filter;
ret_qds = nullptr;
return common::E_UNSUPPORTED_ORDER;
}
diff --git a/cpp/src/reader/tsfile_series_scan_iterator.cc b/cpp/src/reader/tsfile_series_scan_iterator.cc
index fbf8f93db..948d1a553 100644
--- a/cpp/src/reader/tsfile_series_scan_iterator.cc
+++ b/cpp/src/reader/tsfile_series_scan_iterator.cc
@@ -304,7 +304,8 @@ int TsFileSeriesScanIterator::get_next(TsBlock*& ret_tsblock, bool alloc,
if (should_skip_chunk_by_time(time_cm, min_time_hint)) {
continue;
}
- if (chunk_fully_satisfies_filter(time_cm, filter) &&
+ if (field_filter_ == nullptr &&
+ chunk_fully_satisfies_filter(time_cm, filter) &&
should_skip_multi_aligned_chunk_by_offset(time_cm,
value_cms)) {
continue;
@@ -512,6 +513,7 @@ int TsFileSeriesScanIterator::init_chunk_reader_multi() {
itimeseries_index_->get_data_type(), time_filter_))) {
return ret;
}
+ acr->set_field_filter(field_filter_);
// No chunks → nothing to load; iteration short-circuits via
// has_next_chunk() returning false.
diff --git a/cpp/src/reader/tsfile_series_scan_iterator.h b/cpp/src/reader/tsfile_series_scan_iterator.h
index 449f240b4..19c261694 100644
--- a/cpp/src/reader/tsfile_series_scan_iterator.h
+++ b/cpp/src/reader/tsfile_series_scan_iterator.h
@@ -29,6 +29,7 @@
#include "file/random_access_read_file.h"
#include "file/tsfile_io_reader.h"
#include "reader/chunk_reader.h"
+#include "reader/filter/aligned_field_filter.h"
#include "reader/filter/filter.h"
#include "utils/util_define.h"
@@ -51,6 +52,7 @@ class TsFileSeriesScanIterator {
tuple_desc_(),
tsblock_(nullptr),
time_filter_(nullptr),
+ field_filter_(nullptr),
prepared_(),
is_aligned_(false),
is_multi_value_(false),
@@ -61,12 +63,14 @@ class TsFileSeriesScanIterator {
int init(std::shared_ptr<IDeviceID> device_id,
const std::string& measurement_name,
RandomAccessReadFile* read_file, Filter* time_filter,
- common::PageArena& data_pa) {
+ common::PageArena& data_pa,
+ AlignedFieldFilter* field_filter = nullptr) {
ASSERT(read_file != nullptr);
device_id_ = device_id;
measurement_name_ = measurement_name;
read_file_ = read_file;
time_filter_ = time_filter;
+ field_filter_ = field_filter;
data_pa_ = &data_pa;
return common::E_OK;
}
@@ -115,6 +119,12 @@ class TsFileSeriesScanIterator {
int64_t min_time_hint = std::numeric_limits<int64_t>::min());
void revert_tsblock();
+ bool field_filter_applied_in_chunk() const {
+ if (!is_multi_value_ || chunk_reader_ == nullptr) return false;
+ auto* aligned = dynamic_cast<AlignedChunkReader*>(chunk_reader_);
+ return aligned != nullptr && aligned->field_filter_applied();
+ }
+
// Multi-value: number of value columns in the TsBlock
uint32_t get_value_column_count() const {
if (is_multi_value_ && chunk_reader_) {
@@ -225,6 +235,7 @@ class TsFileSeriesScanIterator {
common::TupleDesc tuple_desc_;
common::TsBlock* tsblock_;
Filter* time_filter_;
+ AlignedFieldFilter* field_filter_;
// Keeps the arena-backed index alive until the chunk reader has released
// every pointer into it. Empty for the legacy path-owned metadata arena.
std::shared_ptr<PreparedSeries> prepared_;
diff --git a/cpp/test/reader/filter/aligned_field_filter_test.cc b/cpp/test/reader/filter/aligned_field_filter_test.cc
new file mode 100644
index 000000000..880053112
--- /dev/null
+++ b/cpp/test/reader/filter/aligned_field_filter_test.cc
@@ -0,0 +1,147 @@
+/*
+ * 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 "reader/filter/aligned_field_filter.h"
+
+#include <gtest/gtest.h>
+
+#include "common/statistic.h"
+#include "reader/filter/selection_mask.h"
+
+namespace storage {
+namespace {
+
+TEST(SelectionMaskTest, BooleanOperationsAndTrailingBits) {
+ SelectionMask left(70);
+ SelectionMask right(70);
+ for (uint32_t row : {0U, 2U, 63U, 64U, 69U}) left.set_one(row);
+ for (uint32_t row : {1U, 2U, 64U}) right.set_one(row);
+
+ left.and_with(right);
+ EXPECT_EQ(left.count_ones(), 2U);
+ EXPECT_TRUE(left.test(2));
+ EXPECT_TRUE(left.test(64));
+
+ left.or_with(right);
+ EXPECT_EQ(left.count_ones(), 3U);
+ left.invert();
+ EXPECT_EQ(left.count_ones(), 67U);
+ EXPECT_FALSE(left.test(2));
+ EXPECT_TRUE(left.test(69));
+}
+
+TEST(SelectionMaskTest, NextSetBitAndClearFrom) {
+ SelectionMask mask(130);
+ for (uint32_t row : {0U, 63U, 64U, 129U}) mask.set_one(row);
+ uint32_t value = 0;
+ EXPECT_TRUE(mask.next_set_bit(0, value));
+ EXPECT_EQ(value, 0U);
+ EXPECT_TRUE(mask.next_set_bit(1, value));
+ EXPECT_EQ(value, 63U);
+ EXPECT_TRUE(mask.next_set_bit(64, value));
+ EXPECT_EQ(value, 64U);
+ EXPECT_TRUE(mask.next_set_bit(65, value));
+ EXPECT_EQ(value, 129U);
+ mask.clear_from(64);
+ EXPECT_EQ(mask.count_ones(), 2U);
+ EXPECT_FALSE(mask.test(64));
+ EXPECT_FALSE(mask.test(129));
+}
+
+TEST(AlignedFieldFilterTest, PageStatisticsAreConservative) {
+ Int64Statistic statistic;
+ statistic.count_ = 5;
+ statistic.min_value_ = 10;
+ statistic.max_value_ = 20;
+ const std::map<std::string, Statistic*> stats{{"a", &statistic}};
+
+ auto below =
+ AlignedFieldFilter::compare("a", AlignedFieldFilter::CompareOp::LT,
+ AlignedFieldFilter::Literal::int64(5));
+ EXPECT_FALSE(below->may_match(stats, 5));
+
+ auto overlap =
+ AlignedFieldFilter::compare("a", AlignedFieldFilter::CompareOp::LE,
+ AlignedFieldFilter::Literal::int64(10));
+ EXPECT_TRUE(overlap->may_match(stats, 5));
+
+ auto is_null = AlignedFieldFilter::is_null("a");
+ EXPECT_FALSE(is_null->may_match(stats, 5));
+ statistic.count_ = 4;
+ EXPECT_TRUE(is_null->may_match(stats, 5));
+
+ auto missing = AlignedFieldFilter::compare(
+ "missing", AlignedFieldFilter::CompareOp::EQ,
+ AlignedFieldFilter::Literal::int64(1));
+ EXPECT_TRUE(missing->may_match(stats, 5));
+}
+
+TEST(AlignedFieldFilterTest, PageStatisticsAllMatchIsConservative) {
+ Int64Statistic statistic;
+ statistic.count_ = 5;
+ statistic.min_value_ = 10;
+ statistic.max_value_ = 20;
+ const std::map<std::string, Statistic*> stats{{"a", &statistic}};
+
+ auto ge =
+ AlignedFieldFilter::compare("a", AlignedFieldFilter::CompareOp::GE,
+ AlignedFieldFilter::Literal::int64(10));
+ EXPECT_TRUE(ge->all_match(stats, 5));
+
+ auto lt =
+ AlignedFieldFilter::compare("a", AlignedFieldFilter::CompareOp::LT,
+ AlignedFieldFilter::Literal::int64(21));
+ EXPECT_TRUE(lt->all_match(stats, 5));
+
+ auto strict_lt =
+ AlignedFieldFilter::compare("a", AlignedFieldFilter::CompareOp::LT,
+ AlignedFieldFilter::Literal::int64(20));
+ EXPECT_FALSE(strict_lt->all_match(stats, 5));
+
+ auto not_null = AlignedFieldFilter::is_not_null("a");
+ EXPECT_TRUE(not_null->all_match(stats, 5));
+ statistic.count_ = 4;
+ EXPECT_FALSE(not_null->all_match(stats, 5));
+ auto is_null = AlignedFieldFilter::is_null("a");
+ statistic.count_ = 0;
+ EXPECT_TRUE(is_null->all_match(stats, 5));
+
+ statistic.count_ = 5;
+ auto combined = AlignedFieldFilter::and_filter(
+ AlignedFieldFilter::compare("a", AlignedFieldFilter::CompareOp::GE,
+ AlignedFieldFilter::Literal::int64(10)),
+ AlignedFieldFilter::is_not_null("a"));
+ EXPECT_TRUE(combined->all_match(stats, 5));
+}
+
+TEST(AlignedFieldFilterTest, CollectsReferencedColumnsOnce) {
+ auto left =
+ AlignedFieldFilter::compare("a", AlignedFieldFilter::CompareOp::EQ,
+ AlignedFieldFilter::Literal::int32(1));
+ auto right = AlignedFieldFilter::or_filter(
+ AlignedFieldFilter::is_null("b"),
+ AlignedFieldFilter::compare("a", AlignedFieldFilter::CompareOp::GT,
+ AlignedFieldFilter::Literal::int32(3)));
+ auto root =
+ AlignedFieldFilter::and_filter(std::move(left), std::move(right));
+ EXPECT_EQ(root->referenced_columns(), (std::vector<std::string>{"a", "b"}));
+}
+
+} // namespace
+} // namespace storage
diff --git a/cpp/test/reader/table_field_filter_test.cc b/cpp/test/reader/table_field_filter_test.cc
index 861cb0fa7..62b495a83 100644
--- a/cpp/test/reader/table_field_filter_test.cc
+++ b/cpp/test/reader/table_field_filter_test.cc
@@ -29,6 +29,7 @@
#include "common/tablet.h"
#include "file/local_random_access_read_file.h"
#include "file/write_file.h"
+#include "reader/filter/aligned_field_filter.h"
#include "reader/filter/tag_filter.h"
#include "reader/filter/time_filter.h"
#include "reader/filter/time_operator.h"
@@ -307,6 +308,99 @@ class TableFieldFilterTest
const int64_t base_ = INT64_C(1099511627776);
};
+TEST_P(TableFieldFilterTest, AlignedBatchFilterAndOrUsesFilterOnlyColumns) {
+ auto left = AlignedFieldFilter::and_filter(
+ AlignedFieldFilter::compare("i", AlignedFieldFilter::CompareOp::LE,
+ AlignedFieldFilter::Literal::int32(-4)),
+ AlignedFieldFilter::compare(
+ "flag", AlignedFieldFilter::CompareOp::EQ,
+ AlignedFieldFilter::Literal::boolean(true)));
+ auto right = AlignedFieldFilter::is_not_null("tail_null");
+ auto filter =
+ AlignedFieldFilter::or_filter(std::move(left), std::move(right));
+
+ EXPECT_EQ(scan({"v"}, filter.release()), expected([](int t) {
+ const bool and_branch = t % 3 != 0 && t >= 4 && (t % 2 == 1);
+ const bool or_branch = t % 3 == 0;
+ return and_branch || or_branch;
+ }));
+}
+
+TEST_P(TableFieldFilterTest, AlignedBatchFilterSupportsNotAndChunkNulls) {
+ auto filter = AlignedFieldFilter::not_filter(
+ AlignedFieldFilter::is_null("chunk_null"));
+ EXPECT_EQ(scan({"v"}, filter.release()),
+ expected([](int t) { return t < 7; }));
+}
+
+TEST_P(TableFieldFilterTest, AlignedBatchFilterNotComparisonExcludesNulls) {
+ // NOT over a nullable comparison must exclude NULL rows (SQL three-valued
+ // logic): NOT(i <= -5) == i > -5, and NULL > -5 is still NULL, not TRUE.
+ auto filter = AlignedFieldFilter::not_filter(
+ AlignedFieldFilter::compare("i", AlignedFieldFilter::CompareOp::LE,
+ AlignedFieldFilter::Literal::int32(-5)));
+ EXPECT_EQ(scan({"v"}, filter.release()),
+ expected([](int t) { return t % 3 != 0 && t < 5; }));
+}
+
+TEST_P(TableFieldFilterTest, AlignedBatchFilterAppliesOffsetAfterMask) {
+ for (int offset : {0, 1, 3, 8}) {
+ for (int limit : {0, 1, 4, -1}) {
+ auto filter = AlignedFieldFilter::compare(
+ "i", AlignedFieldFilter::CompareOp::LE,
+ AlignedFieldFilter::Literal::int32(-5));
+ auto all = expected([](int t) { return t % 3 != 0 && t >= 5; });
+ auto begin = all.begin() + std::min<size_t>(offset, all.size());
+ auto end = limit < 0
+ ? all.end()
+ : begin + std::min<size_t>(limit, all.end() - begin);
+ std::vector<int64_t> wanted(begin, end);
+ EXPECT_EQ(
+ scan({"v"}, filter.release(), offset, limit, nullptr, nullptr),
+ wanted);
+ }
+ }
+}
+
+TEST_P(TableFieldFilterTest, AlignedBatchFilterSharedProjectionColumn) {
+ auto filter =
+ AlignedFieldFilter::compare("i", AlignedFieldFilter::CompareOp::LE,
+ AlignedFieldFilter::Literal::int32(-5));
+ EXPECT_EQ(scan({"i"}, filter.release()),
+ expected([](int t) { return t % 3 != 0 && t >= 5; }));
+}
+
+TEST_P(TableFieldFilterTest, AlignedBatchFilterAllMatchFastPath) {
+ auto filter =
+ AlignedFieldFilter::compare("v", AlignedFieldFilter::CompareOp::GE,
+ AlignedFieldFilter::Literal::int64(0));
+ EXPECT_EQ(scan({"v"}, filter.release()),
+ expected([](int) { return true; }));
+}
+
+TEST_P(TableFieldFilterTest, AlignedBatchFilterAllNullFilterOnlyColumn) {
+ auto filter = AlignedFieldFilter::is_null("empty");
+ EXPECT_EQ(scan({"v"}, filter.release()),
+ expected([](int) { return true; }));
+}
+
+TEST_P(TableFieldFilterTest, AlignedBatchFilterCombinesTimeAndFieldMask) {
+ auto filter =
+ AlignedFieldFilter::compare("i", AlignedFieldFilter::CompareOp::LE,
+ AlignedFieldFilter::Literal::int32(-5));
+ EXPECT_EQ(scan({"v"}, filter.release(), 0, -1,
+ TimeFilter::between(base_ + 7, base_ + 10, false)),
+ expected([](int t) { return t >= 7 && t <= 10 && t % 3 != 0; }));
+}
+
+TEST_P(TableFieldFilterTest, AlignedBatchFilterSupportsStringLiterals) {
+ auto filter = AlignedFieldFilter::compare(
+ "s", AlignedFieldFilter::CompareOp::EQ,
+ AlignedFieldFilter::Literal::string(std::string("a\0b", 3)));
+ EXPECT_EQ(scan({"v"}, filter.release()),
+ expected([](int t) { return t % 2 == 1; }));
+}
+
TEST_P(TableFieldFilterTest, NullTrueFalseAndRejectWholePages) {
EXPECT_EQ(scan({"v"}, nullptr), expected([](int) { return true; }));
for (bool accept : {false, true}) {