This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 31da4c64087 [improvement](file reader) Limit dictionary filter
blocking to refere… (#66110)
31da4c64087 is described below
commit 31da4c640875158450b3ea2d250f82befc407e15
Author: foxtail463 <[email protected]>
AuthorDate: Wed Jul 29 11:13:48 2026 +0800
[improvement](file reader) Limit dictionary filter blocking to refere…
(#66110)
Problem:
A multi-slot conjunct (e.g. a = b) disabled dictionary filtering
globally in both ORC and Parquet readers, so unrelated single-slot
columns lost the dict-filter optimization.
Solution:
Replace the global disable flag with a per-slot blocked set that only
blocks slots referenced by multi-slot conjuncts, keeping other columns
eligible.
Co-authored-by: yangtao555 <[email protected]>
---
be/src/format/orc/vorc_reader.cpp | 25 +++-
be/src/format/orc/vorc_reader.h | 3 +-
be/src/format/parquet/vparquet_group_reader.cpp | 26 +++-
be/src/format/parquet/vparquet_group_reader.h | 3 +
be/test/format/orc/orc_reader_test.cpp | 121 +++++++++++++++-
be/test/format/parquet/parquet_expr_test.cpp | 184 ++++++++++++++++++++++++
6 files changed, 354 insertions(+), 8 deletions(-)
diff --git a/be/src/format/orc/vorc_reader.cpp
b/be/src/format/orc/vorc_reader.cpp
index 728595bdad1..b31e3207257 100644
--- a/be/src/format/orc/vorc_reader.cpp
+++ b/be/src/format/orc/vorc_reader.cpp
@@ -519,6 +519,9 @@ Status OrcReader::_do_init_reader(ReaderInitContext*
base_ctx) {
_not_single_slot_filter_conjuncts.insert(_not_single_slot_filter_conjuncts.end(),
ctx->not_single_slot_filter_conjuncts->begin(),
ctx->not_single_slot_filter_conjuncts->end());
+ for (const auto& conjunct : _not_single_slot_filter_conjuncts) {
+ _block_dict_filter_for_slots(conjunct->root());
+ }
}
_slot_id_to_filter_conjuncts = ctx->slot_id_to_filter_conjuncts;
_obj_pool = std::make_unique<ObjectPool>();
@@ -587,7 +590,6 @@ Status OrcReader::_do_init_reader(ReaderInitContext*
base_ctx) {
if (!_not_single_slot_filter_conjuncts.empty()) {
_filter_conjuncts.insert(_filter_conjuncts.end(),
_not_single_slot_filter_conjuncts.begin(),
_not_single_slot_filter_conjuncts.end());
- _disable_dict_filter = true;
}
if (_slot_id_to_filter_conjuncts &&
!_slot_id_to_filter_conjuncts->empty()) {
for (auto& kv : _lazy_read_ctx.predicate_partition_columns) {
@@ -2994,8 +2996,7 @@ Status OrcReader::fill_dict_filter_column_names(
int i = 0;
for (const auto& predicate_col_name : predicate_col_names) {
int slot_id = predicate_col_slot_ids[i];
- if (!_disable_dict_filter &&
- has_column_optimization(predicate_col_name,
ColumnOptimizationTypes::DICT_FILTER) &&
+ if (has_column_optimization(predicate_col_name,
ColumnOptimizationTypes::DICT_FILTER) &&
_can_filter_by_dict(slot_id)) {
_dict_filter_cols.emplace_back(predicate_col_name, slot_id);
column_names.emplace_back(
@@ -3013,7 +3014,25 @@ Status OrcReader::fill_dict_filter_column_names(
return Status::OK();
}
+void OrcReader::_block_dict_filter_for_slots(const VExprSPtr& expr) {
+ DORIS_CHECK(expr != nullptr);
+ if (auto impl = expr->get_impl()) {
+ _block_dict_filter_for_slots(impl);
+ return;
+ }
+ if (expr->is_slot_ref()) {
+ _dict_filter_blocked_slot_ids.insert(static_cast<const
VSlotRef*>(expr.get())->slot_id());
+ return;
+ }
+ for (const auto& child : expr->children()) {
+ _block_dict_filter_for_slots(child);
+ }
+}
+
bool OrcReader::_can_filter_by_dict(int slot_id) {
+ if (_dict_filter_blocked_slot_ids.contains(slot_id)) {
+ return false;
+ }
SlotDescriptor* slot = nullptr;
const std::vector<SlotDescriptor*>& slots = _tuple_descriptor->slots();
for (auto* each : slots) {
diff --git a/be/src/format/orc/vorc_reader.h b/be/src/format/orc/vorc_reader.h
index 6d64f604cde..9e21b0cc901 100644
--- a/be/src/format/orc/vorc_reader.h
+++ b/be/src/format/orc/vorc_reader.h
@@ -677,6 +677,7 @@ private:
const orc::DataBuffer<int64_t>&
orc_offsets, size_t num_values,
size_t* element_size);
+ void _block_dict_filter_for_slots(const VExprSPtr& expr);
bool _can_filter_by_dict(int slot_id);
Status _rewrite_dict_conjuncts(std::vector<int32_t>& dict_codes, int
slot_id, bool is_nullable);
@@ -825,7 +826,7 @@ private:
VExprContextSPtrs _dict_filter_conjuncts;
VExprContextSPtrs _non_dict_filter_conjuncts;
VExprContextSPtrs _filter_conjuncts;
- bool _disable_dict_filter = false;
+ std::unordered_set<int> _dict_filter_blocked_slot_ids;
// std::pair<col_name, slot_id>
std::vector<std::pair<std::string, int>> _dict_filter_cols;
std::unique_ptr<ObjectPool> _obj_pool;
diff --git a/be/src/format/parquet/vparquet_group_reader.cpp
b/be/src/format/parquet/vparquet_group_reader.cpp
index 33179f2ddac..158f7d9ade7 100644
--- a/be/src/format/parquet/vparquet_group_reader.cpp
+++ b/be/src/format/parquet/vparquet_group_reader.cpp
@@ -29,6 +29,7 @@
#include <numeric>
#include <ostream>
+#include "common/check.h"
#include "common/config.h"
#include "common/logging.h"
#include "common/object_pool.h"
@@ -144,9 +145,10 @@ Status RowGroupReader::init(
_column_readers[read_table_col] = std::move(reader);
}
- bool disable_dict_filter = false;
if (not_single_slot_filter_conjuncts != nullptr &&
!not_single_slot_filter_conjuncts->empty()) {
- disable_dict_filter = true;
+ for (const auto& conjunct : *not_single_slot_filter_conjuncts) {
+ _block_dict_filter_for_slots(conjunct->root());
+ }
_filter_conjuncts.insert(_filter_conjuncts.end(),
not_single_slot_filter_conjuncts->begin(),
not_single_slot_filter_conjuncts->end());
}
@@ -176,7 +178,7 @@ Status RowGroupReader::init(
auto predicate_file_col_name =
_table_info_node_ptr->children_file_column_name(predicate_col_name);
auto field = schema.get_column(predicate_file_col_name);
- if (!disable_dict_filter && !_lazy_read_ctx.has_complex_type &&
+ if (!_lazy_read_ctx.has_complex_type &&
_can_filter_by_dict(
slot_id,
_row_group_meta.columns[field->physical_column_index].meta_data)) {
_dict_filter_cols.emplace_back(std::make_pair(predicate_col_name, slot_id));
@@ -215,8 +217,26 @@ Status RowGroupReader::init(
return Status::OK();
}
+void RowGroupReader::_block_dict_filter_for_slots(const VExprSPtr& expr) {
+ DORIS_CHECK(expr != nullptr);
+ if (auto impl = expr->get_impl()) {
+ _block_dict_filter_for_slots(impl);
+ return;
+ }
+ if (expr->is_slot_ref()) {
+ _dict_filter_blocked_slot_ids.insert(static_cast<const
VSlotRef*>(expr.get())->slot_id());
+ return;
+ }
+ for (const auto& child : expr->children()) {
+ _block_dict_filter_for_slots(child);
+ }
+}
+
bool RowGroupReader::_can_filter_by_dict(int slot_id,
const tparquet::ColumnMetaData&
column_metadata) {
+ if (_dict_filter_blocked_slot_ids.contains(slot_id)) {
+ return false;
+ }
SlotDescriptor* slot = nullptr;
const std::vector<SlotDescriptor*>& slots = _tuple_descriptor->slots();
for (auto each : slots) {
diff --git a/be/src/format/parquet/vparquet_group_reader.h
b/be/src/format/parquet/vparquet_group_reader.h
index 79b171ca064..cfec15f8a10 100644
--- a/be/src/format/parquet/vparquet_group_reader.h
+++ b/be/src/format/parquet/vparquet_group_reader.h
@@ -23,6 +23,7 @@
#include <string>
#include <tuple>
#include <unordered_map>
+#include <unordered_set>
#include <utility>
#include <vector>
@@ -253,6 +254,7 @@ private:
Status _filter_block_internal(Block* block, const std::vector<uint32_t>&
columns_to_filter,
const IColumn::Filter& filter);
+ void _block_dict_filter_for_slots(const VExprSPtr& expr);
bool _can_filter_by_dict(int slot_id, const tparquet::ColumnMetaData&
column_metadata);
bool _need_current_batch_row_positions() const;
bool is_dictionary_encoded(const tparquet::ColumnMetaData&
column_metadata);
@@ -299,6 +301,7 @@ private:
VExprContextSPtrs _filter_conjuncts;
// std::pair<col_name, slot_id>
std::vector<std::pair<std::string, int>> _dict_filter_cols;
+ std::unordered_set<int> _dict_filter_blocked_slot_ids;
RuntimeState* _state = nullptr;
std::shared_ptr<ObjectPool> _obj_pool;
const std::set<uint64_t>& _column_ids;
diff --git a/be/test/format/orc/orc_reader_test.cpp
b/be/test/format/orc/orc_reader_test.cpp
index b678f16c014..c8099b286f0 100644
--- a/be/test/format/orc/orc_reader_test.cpp
+++ b/be/test/format/orc/orc_reader_test.cpp
@@ -18,9 +18,11 @@
#include <glog/logging.h>
#include <gtest/gtest.h>
+#include <list>
#include <memory>
#include <string>
#include <tuple>
+#include <unordered_map>
#include <vector>
#include "core/assert_cast.h"
@@ -29,8 +31,11 @@
#include "core/column/column_vector.h"
#include "core/data_type/define_primitive_type.h"
#include "exec/common/util.hpp"
+#include "exprs/vectorized_fn_call.h"
#include "exprs/vexpr_context.h"
#include "exprs/vexpr_fwd.h"
+#include "exprs/vliteral.h"
+#include "exprs/vslot_ref.h"
#include "format/orc/orc_memory_pool.h"
#include "format/orc/vorc_reader.h"
#include "io/fs/file_meta_cache.h"
@@ -47,8 +52,51 @@ public:
FileMetaCache cache;
-private:
+protected:
static constexpr const char* CANNOT_PUSH_DOWN_ERROR = "can't push down";
+
+ VExprContextSPtr create_string_equal(RuntimeState* state, const
RowDescriptor& row_desc,
+ VExprSPtr left, VExprSPtr right) {
+ TFunction fn;
+ TFunctionName fn_name;
+ fn_name.__set_db_name("");
+ fn_name.__set_function_name("eq");
+ fn.__set_name(fn_name);
+ fn.__set_binary_type(TFunctionBinaryType::BUILTIN);
+ fn.__set_arg_types({create_type_desc(TYPE_STRING),
create_type_desc(TYPE_STRING)});
+ fn.__set_ret_type(create_type_desc(TYPE_BOOLEAN));
+ fn.__set_has_var_args(false);
+
+ TExprNode expr_node;
+ expr_node.__set_type(create_type_desc(TYPE_BOOLEAN));
+ expr_node.__set_node_type(TExprNodeType::BINARY_PRED);
+ expr_node.__set_opcode(TExprOpcode::EQ);
+ expr_node.__set_fn(fn);
+ expr_node.__set_num_children(2);
+ expr_node.__set_is_nullable(true);
+ auto root = VectorizedFnCall::create_shared(expr_node);
+ root->add_child(std::move(left));
+ root->add_child(std::move(right));
+
+ auto context = VExprContext::create_shared(root);
+ auto status = context->prepare(state, row_desc);
+ EXPECT_TRUE(status.ok()) << status;
+ status = context->open(state);
+ EXPECT_TRUE(status.ok()) << status;
+ return context;
+ }
+
+ VExprSPtr create_string_literal(const std::string& value) {
+ TExprNode literal_node;
+ literal_node.__set_node_type(TExprNodeType::STRING_LITERAL);
+ literal_node.__set_type(create_type_desc(TYPE_STRING));
+ TStringLiteral literal;
+ literal.__set_value(value);
+ literal_node.__set_string_literal(literal);
+ literal_node.__set_is_nullable(false);
+ return VLiteral::create_shared(literal_node);
+ }
+
std::string build_search_argument(const std::string& expr) {
// build orc_reader for table orders
std::vector<std::string> column_names = {
@@ -251,6 +299,77 @@ TEST_F(OrcReaderTest,
set_batch_size_without_row_reader_is_safe) {
EXPECT_EQ(reader->_batch, nullptr);
}
+TEST_F(OrcReaderTest,
dict_filter_is_blocked_only_for_slots_in_multi_slot_conjuncts) {
+ ObjectPool object_pool;
+ DescriptorTblBuilder builder(&object_pool);
+ builder.declare_tuple()
+ <<
std::make_tuple(DataTypeFactory::instance().create_data_type(TYPE_STRING,
false),
+ "o_orderstatus")
+ <<
std::make_tuple(DataTypeFactory::instance().create_data_type(TYPE_STRING,
false),
+ "o_orderpriority")
+ <<
std::make_tuple(DataTypeFactory::instance().create_data_type(TYPE_STRING,
false),
+ "o_clerk");
+ DescriptorTbl* desc_tbl = builder.build();
+ auto* tuple_desc =
const_cast<TupleDescriptor*>(desc_tbl->get_tuple_descriptor(0));
+ RowDescriptor row_desc(tuple_desc);
+
+ RuntimeState state;
+ state.set_desc_tbl(desc_tbl);
+ auto multi_slot_conjunct =
+ create_string_equal(&state, row_desc,
VSlotRef::create_shared(tuple_desc->slots()[0]),
+
VSlotRef::create_shared(tuple_desc->slots()[1]));
+ auto order_status_conjunct =
+ create_string_equal(&state, row_desc,
VSlotRef::create_shared(tuple_desc->slots()[0]),
+ create_string_literal("F"));
+ auto order_priority_conjunct =
+ create_string_equal(&state, row_desc,
VSlotRef::create_shared(tuple_desc->slots()[1]),
+ create_string_literal("5-LOW"));
+ auto clerk_conjunct =
+ create_string_equal(&state, row_desc,
VSlotRef::create_shared(tuple_desc->slots()[2]),
+ create_string_literal("Clerk#000000951"));
+
+ VExprContextSPtrs conjuncts = {multi_slot_conjunct, order_status_conjunct,
+ order_priority_conjunct, clerk_conjunct};
+ VExprContextSPtrs not_single_slot_filter_conjuncts = {multi_slot_conjunct};
+ std::unordered_map<int, VExprContextSPtrs> slot_id_to_filter_conjuncts = {
+ {tuple_desc->slots()[0]->id(), {order_status_conjunct}},
+ {tuple_desc->slots()[1]->id(), {order_priority_conjunct}},
+ {tuple_desc->slots()[2]->id(), {clerk_conjunct}}};
+
+ TFileScanRangeParams params;
+ params.__set_file_type(TFileType::FILE_LOCAL);
+ params.__set_format_type(TFileFormatType::FORMAT_ORC);
+ TFileRangeDesc range;
+ range.__set_path("./be/test/exec/test_data/orc_scanner/orders.orc");
+ range.__set_start_offset(0);
+ range.__set_size(1293);
+ RuntimeProfile profile("dict_filter_with_multi_slot_conjunct");
+ io::IOContext io_ctx;
+ auto reader = std::make_unique<OrcReader>(&profile, &state, params, range,
64, "UTC", &io_ctx,
+ &cache, true);
+ std::vector<std::string> column_names = {"o_orderstatus",
"o_orderpriority", "o_clerk"};
+ std::unordered_map<std::string, uint32_t> col_name_to_block_idx = {
+ {"o_orderstatus", 0}, {"o_orderpriority", 1}, {"o_clerk", 2}};
+ OrcInitContext orc_ctx;
+ orc_ctx.column_names = column_names;
+ orc_ctx.col_name_to_block_idx = &col_name_to_block_idx;
+ orc_ctx.tuple_descriptor = tuple_desc;
+ orc_ctx.row_descriptor = &row_desc;
+ orc_ctx.params = ¶ms;
+ orc_ctx.range = ⦥
+ orc_ctx.conjuncts = &conjuncts;
+ orc_ctx.not_single_slot_filter_conjuncts =
¬_single_slot_filter_conjuncts;
+ orc_ctx.slot_id_to_filter_conjuncts = &slot_id_to_filter_conjuncts;
+
+ auto status = reader->init_reader(&orc_ctx);
+ ASSERT_TRUE(status.ok()) << status;
+
+ std::list<std::string> dict_filter_columns;
+ status = reader->fill_dict_filter_column_names(nullptr,
dict_filter_columns);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(dict_filter_columns, std::list<std::string>({"o_clerk"}));
+}
+
TEST_F(OrcReaderTest, deletion_vector_filters_rows_without_query_conjuncts) {
auto read_order_keys = [&](const roaring::Roaring64Map* deletion_vector,
std::vector<int32_t>* order_keys) -> Status {
diff --git a/be/test/format/parquet/parquet_expr_test.cpp
b/be/test/format/parquet/parquet_expr_test.cpp
index b7dbb759f5f..1b9087479bb 100644
--- a/be/test/format/parquet/parquet_expr_test.cpp
+++ b/be/test/format/parquet/parquet_expr_test.cpp
@@ -60,6 +60,10 @@
#include "exprs/hybrid_set.h"
#include "exprs/vcompound_pred.h"
#include "exprs/vdirect_in_predicate.h"
+#include "exprs/vectorized_fn_call.h"
+#include "exprs/vliteral.h"
+#include "exprs/vslot_ref.h"
+#include "format/column_descriptor.h"
#include "format/parquet/parquet_block_split_bloom_filter.h"
#include "format/parquet/parquet_thrift_util.h"
#include "format/parquet/schema_desc.h"
@@ -200,6 +204,48 @@ class ParquetExprTest : public testing::Test {
public:
ParquetExprTest() {}
+ VExprSPtr create_string_literal(const std::string& value) {
+ TExprNode literal_node;
+ literal_node.__set_node_type(TExprNodeType::STRING_LITERAL);
+ literal_node.__set_type(create_type_desc(TYPE_STRING));
+ TStringLiteral literal;
+ literal.__set_value(value);
+ literal_node.__set_string_literal(literal);
+ literal_node.__set_is_nullable(false);
+ return VLiteral::create_shared(literal_node);
+ }
+
+ VExprContextSPtr create_string_equal(RuntimeState* state, const
RowDescriptor& row_desc,
+ VExprSPtr left, VExprSPtr right) {
+ TFunction fn;
+ TFunctionName fn_name;
+ fn_name.__set_db_name("");
+ fn_name.__set_function_name("eq");
+ fn.__set_name(fn_name);
+ fn.__set_binary_type(TFunctionBinaryType::BUILTIN);
+ fn.__set_arg_types({create_type_desc(TYPE_STRING),
create_type_desc(TYPE_STRING)});
+ fn.__set_ret_type(create_type_desc(TYPE_BOOLEAN));
+ fn.__set_has_var_args(false);
+
+ TExprNode expr_node;
+ expr_node.__set_type(create_type_desc(TYPE_BOOLEAN));
+ expr_node.__set_node_type(TExprNodeType::BINARY_PRED);
+ expr_node.__set_opcode(TExprOpcode::EQ);
+ expr_node.__set_fn(fn);
+ expr_node.__set_num_children(2);
+ expr_node.__set_is_nullable(true);
+ auto root = VectorizedFnCall::create_shared(expr_node);
+ root->add_child(std::move(left));
+ root->add_child(std::move(right));
+
+ auto context = VExprContext::create_shared(root);
+ auto status = context->prepare(state, row_desc);
+ EXPECT_TRUE(status.ok()) << status;
+ status = context->open(state);
+ EXPECT_TRUE(status.ok()) << status;
+ return context;
+ }
+
void SetUp() override {
std::string test_dir = "ut_dir/test_parquet_expr";
Status st;
@@ -617,6 +663,144 @@ public:
std::unordered_map<std::string, int> colname_to_slot_id;
};
+TEST_F(ParquetExprTest,
dict_filter_is_blocked_only_for_slots_in_multi_slot_conjuncts) {
+ arrow::StringBuilder left_builder;
+ arrow::StringBuilder right_builder;
+ arrow::StringBuilder filter_builder;
+ for (const auto& value : {"x", "x", "x", "y"}) {
+ ASSERT_TRUE(left_builder.Append(value).ok());
+ }
+ for (const auto& value : {"x", "x", "y", "y"}) {
+ ASSERT_TRUE(right_builder.Append(value).ok());
+ }
+ for (const auto& value : {"keep", "drop", "keep", "keep"}) {
+ ASSERT_TRUE(filter_builder.Append(value).ok());
+ }
+
+ std::shared_ptr<arrow::Array> left_array;
+ std::shared_ptr<arrow::Array> right_array;
+ std::shared_ptr<arrow::Array> filter_array;
+ ASSERT_TRUE(left_builder.Finish(&left_array).ok());
+ ASSERT_TRUE(right_builder.Finish(&right_array).ok());
+ ASSERT_TRUE(filter_builder.Finish(&filter_array).ok());
+
+ auto schema = arrow::schema({arrow::field("left_col", arrow::utf8(),
false),
+ arrow::field("right_col", arrow::utf8(),
false),
+ arrow::field("filter_col", arrow::utf8(),
false)});
+ auto table = arrow::Table::Make(schema, {left_array, right_array,
filter_array});
+ const std::string dict_filter_file =
"ut_dir/test_parquet_expr/dict_filter.parquet";
+ auto output_result = arrow::io::FileOutputStream::Open(dict_filter_file);
+ ASSERT_TRUE(output_result.ok()) << output_result.status();
+ auto output = std::move(output_result).ValueUnsafe();
+ ::parquet::WriterProperties::Builder properties_builder;
+ properties_builder.enable_dictionary();
+ auto properties = properties_builder.build();
+ PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table,
arrow::default_memory_pool(), output,
+ table->num_rows(),
properties));
+ ASSERT_TRUE(output->Close().ok());
+
+ TDescriptorTable local_t_desc_table;
+ TTableDescriptor local_t_table_desc;
+ create_table_desc(local_t_desc_table, local_t_table_desc,
+ {"left_col", "right_col", "filter_col"},
+ {TPrimitiveType::STRING, TPrimitiveType::STRING,
TPrimitiveType::STRING});
+ ObjectPool local_obj_pool;
+ DescriptorTbl* local_desc_tbl = nullptr;
+ auto status = DescriptorTbl::create(&local_obj_pool, local_t_desc_table,
&local_desc_tbl);
+ ASSERT_TRUE(status.ok()) << status;
+ auto* tuple_desc = local_desc_tbl->get_tuple_descriptor(0);
+ RowDescriptor row_desc(tuple_desc);
+
+ RuntimeState state = RuntimeState(TQueryOptions(), TQueryGlobals());
+ state.set_desc_tbl(local_desc_tbl);
+ auto multi_slot_conjunct =
+ create_string_equal(&state, row_desc,
VSlotRef::create_shared(tuple_desc->slots()[0]),
+
VSlotRef::create_shared(tuple_desc->slots()[1]));
+ auto left_conjunct =
+ create_string_equal(&state, row_desc,
VSlotRef::create_shared(tuple_desc->slots()[0]),
+ create_string_literal("x"));
+ auto right_conjunct =
+ create_string_equal(&state, row_desc,
VSlotRef::create_shared(tuple_desc->slots()[1]),
+ create_string_literal("x"));
+ auto filter_conjunct =
+ create_string_equal(&state, row_desc,
VSlotRef::create_shared(tuple_desc->slots()[2]),
+ create_string_literal("keep"));
+
+ VExprContextSPtrs conjuncts = {multi_slot_conjunct, left_conjunct,
right_conjunct,
+ filter_conjunct};
+ VExprContextSPtrs not_single_slot_filter_conjuncts = {multi_slot_conjunct};
+ std::unordered_map<int, VExprContextSPtrs> slot_id_to_filter_conjuncts = {
+ {tuple_desc->slots()[0]->id(), {left_conjunct}},
+ {tuple_desc->slots()[1]->id(), {right_conjunct}},
+ {tuple_desc->slots()[2]->id(), {filter_conjunct}}};
+
+ auto local_fs = io::global_local_filesystem();
+ io::FileReaderSPtr file_reader;
+ status = local_fs->open_file(dict_filter_file, &file_reader);
+ ASSERT_TRUE(status.ok()) << status;
+
+ TFileScanRangeParams scan_params;
+ TFileRangeDesc scan_range;
+ scan_range.__set_path(dict_filter_file);
+ scan_range.__set_start_offset(0);
+ scan_range.__set_size(file_reader->size());
+ auto parquet_reader = ParquetReader::create_unique(nullptr, scan_params,
scan_range, 64, &ctz,
+ nullptr, &state,
nullptr, true);
+ parquet_reader->set_file_reader(file_reader);
+
+ std::vector<std::string> column_names = {"left_col", "right_col",
"filter_col"};
+ std::vector<ColumnDescriptor> column_descs;
+ for (const auto& column_name : column_names) {
+ column_descs.push_back({column_name, nullptr, ColumnCategory::REGULAR,
nullptr});
+ }
+ std::unordered_map<std::string, uint32_t> col_name_to_block_idx = {
+ {"left_col", 0}, {"right_col", 1}, {"filter_col", 2}};
+ std::unordered_map<std::string, int> col_name_to_slot_id = {
+ {"left_col", tuple_desc->slots()[0]->id()},
+ {"right_col", tuple_desc->slots()[1]->id()},
+ {"filter_col", tuple_desc->slots()[2]->id()}};
+ ParquetInitContext parquet_context;
+ parquet_context.column_descs = &column_descs;
+ parquet_context.col_name_to_block_idx = &col_name_to_block_idx;
+ parquet_context.tuple_descriptor = tuple_desc;
+ parquet_context.row_descriptor = &row_desc;
+ parquet_context.params = &scan_params;
+ parquet_context.range = &scan_range;
+ parquet_context.conjuncts = &conjuncts;
+ parquet_context.colname_to_slot_id = &col_name_to_slot_id;
+ parquet_context.not_single_slot_filter_conjuncts =
¬_single_slot_filter_conjuncts;
+ parquet_context.slot_id_to_filter_conjuncts = &slot_id_to_filter_conjuncts;
+ status = parquet_reader->init_reader(&parquet_context);
+ ASSERT_TRUE(status.ok()) << status;
+
+ Block block;
+ for (const auto* slot_desc : tuple_desc->slots()) {
+ block.insert(
+ {slot_desc->type()->create_column(), slot_desc->type(),
slot_desc->col_name()});
+ }
+ size_t read_rows = 0;
+ bool eof = false;
+ status = parquet_reader->get_next_block(&block, &read_rows, &eof);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_NE(parquet_reader->_current_group_reader, nullptr);
+
+ const auto& blocked_slot_ids =
+
parquet_reader->_current_group_reader->_dict_filter_blocked_slot_ids;
+ EXPECT_EQ(blocked_slot_ids.size(), 2);
+ EXPECT_TRUE(blocked_slot_ids.contains(tuple_desc->slots()[0]->id()));
+ EXPECT_TRUE(blocked_slot_ids.contains(tuple_desc->slots()[1]->id()));
+ const auto& dict_filter_columns =
parquet_reader->_current_group_reader->_dict_filter_cols;
+ ASSERT_EQ(dict_filter_columns.size(), 1);
+ EXPECT_EQ(dict_filter_columns.front().first, "filter_col");
+
+ EXPECT_TRUE(eof);
+ ASSERT_EQ(read_rows, 1);
+ ASSERT_EQ(block.rows(), 1);
+ EXPECT_EQ(block.get_by_position(0).column->get_data_at(0).to_string(),
"x");
+ EXPECT_EQ(block.get_by_position(1).column->get_data_at(0).to_string(),
"x");
+ EXPECT_EQ(block.get_by_position(2).column->get_data_at(0).to_string(),
"keep");
+}
+
TEST_F(ParquetExprTest, test_min_max) {
// open parquet with parquet's API
std::unique_ptr<::parquet::ParquetFileReader> arrow_reader =
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
