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 73032987277 [improvement](be) Optimize Parquet predicate filtering in
format v2 (#65369)
73032987277 is described below
commit 730329872770d909d1c1e67431d2c70be97a7eab
Author: Gabriel <[email protected]>
AuthorDate: Fri Jul 10 13:22:26 2026 +0800
[improvement](be) Optimize Parquet predicate filtering in format v2 (#65369)
Problem Summary: The format_v2 Parquet reader previously materialized all
predicate columns before evaluating row-level predicates, so later predicate
columns could not reuse rows rejected by earlier single-column predicates. This
PR implements the StarRocks-style round-by-round predicate-column read path for
Doris format_v2: deterministic single-column conjuncts are scheduled with their
predicate columns, evaluated immediately after each column is read, and later
predicate columns ar [...]
The PR also adds row-level dictionary predicate filtering for
dictionary-encoded Parquet columns in the v2 path. Dictionary-capable predicate
children are evaluated against dictionary entries first, the data-page reader
filters by dictionary id, and residual conjuncts keep the normal row-level
expression path. Compound AND predicates are split so dictionary-covered
children are not re-evaluated after the dictionary prefilter, while
non-dictionary residual children still run on survivi [...]
For correctness, volatile or non-deterministic predicates stay on the old
full-batch path. Expressions such as random/rand, random_bytes, uuid, and
uuid_numeric are marked non-deterministic at the vectorized expression layer;
if any pushed conjunct is non-deterministic, the round-by-round schedule is
disabled for that batch so stateful functions see the same full input stream as
before this optimization.
---
be/src/exprs/vcast_expr.h | 4 +
be/src/exprs/vectorized_fn_call.cpp | 13 +
be/src/exprs/vectorized_fn_call.h | 2 +
be/src/exprs/vexpr.h | 11 +
be/src/format_v2/parquet/parquet_file_context.cpp | 7 +
be/src/format_v2/parquet/parquet_file_context.h | 4 +
be/src/format_v2/parquet/parquet_profile.cpp | 25 +
be/src/format_v2/parquet/parquet_profile.h | 21 +-
be/src/format_v2/parquet/parquet_scan.cpp | 707 ++++++++++++++++++++-
be/src/format_v2/parquet/parquet_scan.h | 28 +-
be/src/format_v2/parquet/parquet_statistics.cpp | 41 +-
be/src/format_v2/parquet/parquet_statistics.h | 32 +
be/src/format_v2/parquet/reader/column_reader.cpp | 51 +-
be/src/format_v2/parquet/reader/column_reader.h | 13 +-
.../parquet/reader/parquet_leaf_reader.cpp | 63 ++
.../format_v2/parquet/reader/parquet_leaf_reader.h | 3 +
.../parquet/reader/scalar_column_reader.cpp | 204 ++++++
.../parquet/reader/scalar_column_reader.h | 16 +
be/test/exprs/try_cast_expr_test.cpp | 10 +-
be/test/format_v2/parquet/parquet_reader_test.cpp | 589 +++++++++++++++++
be/test/format_v2/parquet/parquet_scan_test.cpp | 202 ++++++
.../hive/test_parquet_lazy_mat_profile.groovy | 36 ++
22 files changed, 2006 insertions(+), 76 deletions(-)
diff --git a/be/src/exprs/vcast_expr.h b/be/src/exprs/vcast_expr.h
index f0f3ead95d5..4ac5539ff28 100644
--- a/be/src/exprs/vcast_expr.h
+++ b/be/src/exprs/vcast_expr.h
@@ -55,6 +55,7 @@ public:
const std::string& expr_name() const override;
std::string debug_string() const override;
const DataTypePtr& get_target_type() const;
+ bool is_safe_to_execute_on_selected_rows() const override { return false; }
virtual std::string cast_name() const { return "CAST"; }
Status clone_node(VExprSPtr* cloned_expr) const override {
@@ -99,6 +100,9 @@ public:
size_t count, ColumnPtr& result_column) const
override;
~TryCastExpr() override = default;
std::string cast_name() const override { return "TRY CAST"; }
+ bool is_safe_to_execute_on_selected_rows() const override {
+ return VExpr::is_safe_to_execute_on_selected_rows();
+ }
Status clone_node(VExprSPtr* cloned_expr) const override {
DORIS_CHECK(cloned_expr != nullptr);
auto node = clone_texpr_node();
diff --git a/be/src/exprs/vectorized_fn_call.cpp
b/be/src/exprs/vectorized_fn_call.cpp
index a36f0da49b6..b6e5c13d111 100644
--- a/be/src/exprs/vectorized_fn_call.cpp
+++ b/be/src/exprs/vectorized_fn_call.cpp
@@ -25,6 +25,7 @@
#include <memory>
#include <ostream>
+#include <set>
#include "common/config.h"
#include "common/exception.h"
@@ -378,6 +379,18 @@ bool VectorizedFnCall::can_push_down_to_index() const {
return _function->can_push_down_to_index();
}
+bool VectorizedFnCall::is_deterministic() const {
+ static const std::set<std::string> NON_DETERMINISTIC_FUNCTIONS = {
+ "random", "rand", "random_bytes", "uuid", "uuid_numeric"};
+ return !NON_DETERMINISTIC_FUNCTIONS.contains(_function_name) &&
VExpr::is_deterministic();
+}
+
+bool VectorizedFnCall::is_safe_to_execute_on_selected_rows() const {
+ static const std::set<std::string> ERROR_PRESERVING_FUNCTIONS =
{"assert_true"};
+ return !ERROR_PRESERVING_FUNCTIONS.contains(_function_name) &&
+ VExpr::is_safe_to_execute_on_selected_rows();
+}
+
bool VectorizedFnCall::equals(const VExpr& other) {
const auto* other_ptr = dynamic_cast<const VectorizedFnCall*>(&other);
if (!other_ptr) {
diff --git a/be/src/exprs/vectorized_fn_call.h
b/be/src/exprs/vectorized_fn_call.h
index 67d5dbe158d..8343242ce31 100644
--- a/be/src/exprs/vectorized_fn_call.h
+++ b/be/src/exprs/vectorized_fn_call.h
@@ -77,6 +77,8 @@ public:
std::any_of(_children.begin(), _children.end(),
[](VExprSPtr child) { return child->is_blockable();
});
}
+ bool is_deterministic() const override;
+ bool is_safe_to_execute_on_selected_rows() const override;
bool is_constant() const override {
if (!_function->is_use_default_implementation_for_constants() ||
// udf function with no argument, can't sure it's must return
const column
diff --git a/be/src/exprs/vexpr.h b/be/src/exprs/vexpr.h
index 209de7b431f..f54af6f7819 100644
--- a/be/src/exprs/vexpr.h
+++ b/be/src/exprs/vexpr.h
@@ -180,6 +180,17 @@ public:
[](VExprSPtr child) { return child->is_blockable();
});
}
+ [[nodiscard]] virtual bool is_deterministic() const {
+ return std::ranges::all_of(
+ _children, [](const VExprSPtr& child) { return
child->is_deterministic(); });
+ }
+
+ [[nodiscard]] virtual bool is_safe_to_execute_on_selected_rows() const {
+ return is_deterministic() && std::ranges::all_of(_children, [](const
VExprSPtr& child) {
+ return child->is_safe_to_execute_on_selected_rows();
+ });
+ }
+
// execute current expr with inverted index to filter block. Given a
roaring bitmap of match rows
virtual Status evaluate_inverted_index(VExprContext* context, uint32_t
segment_num_rows) {
return Status::OK();
diff --git a/be/src/format_v2/parquet/parquet_file_context.cpp
b/be/src/format_v2/parquet/parquet_file_context.cpp
index 1df3d7b353d..d80dc58181d 100644
--- a/be/src/format_v2/parquet/parquet_file_context.cpp
+++ b/be/src/format_v2/parquet/parquet_file_context.cpp
@@ -353,6 +353,8 @@ public:
return true;
}
+ void reset_random_access_ranges() { reset_active_file_reader(); }
+
ParquetPageCacheStats page_cache_stats() const {
std::lock_guard lock(_page_cache_mutex);
return _page_cache_stats;
@@ -584,6 +586,11 @@ bool ParquetFileContext::set_random_access_ranges(const
std::vector<ParquetPageC
->set_random_access_ranges(ranges, avg_io_size, profile,
merge_read_slice_size);
}
+void ParquetFileContext::reset_random_access_ranges() {
+ DORIS_CHECK(arrow_file != nullptr);
+
static_cast<DorisRandomAccessFile*>(arrow_file.get())->reset_random_access_ranges();
+}
+
ParquetPageCacheStats ParquetFileContext::page_cache_stats() const {
if (arrow_file == nullptr) {
return {};
diff --git a/be/src/format_v2/parquet/parquet_file_context.h
b/be/src/format_v2/parquet/parquet_file_context.h
index e75293f470f..0dca5224495 100644
--- a/be/src/format_v2/parquet/parquet_file_context.h
+++ b/be/src/format_v2/parquet/parquet_file_context.h
@@ -129,6 +129,10 @@ struct ParquetFileContext {
bool set_random_access_ranges(const std::vector<ParquetPageCacheRange>&
ranges,
size_t avg_io_size, RuntimeProfile* profile,
int64_t merge_read_slice_size);
+ // Restore Arrow ReadAt() to the base Doris file reader and flush any
active merge-reader
+ // counters. Row-group setup uses this before dictionary-page probes,
because those probes are
+ // a separate pass over the column chunk from the later Arrow RecordReader
data-page stream.
+ void reset_random_access_ranges();
ParquetPageCacheStats page_cache_stats() const;
Status close();
};
diff --git a/be/src/format_v2/parquet/parquet_profile.cpp
b/be/src/format_v2/parquet/parquet_profile.cpp
index 4bf7af687f2..d41ff295ec4 100644
--- a/be/src/format_v2/parquet/parquet_profile.cpp
+++ b/be/src/format_v2/parquet/parquet_profile.cpp
@@ -139,6 +139,22 @@ void ParquetProfile::init(RuntimeProfile* profile) {
ADD_CHILD_TIMER_WITH_LEVEL(profile, "PredicateFilterTime",
parquet_profile, 1);
dict_filter_rewrite_time =
ADD_CHILD_TIMER_WITH_LEVEL(profile, "DictFilterRewriteTime",
parquet_profile, 1);
+ dict_filter_expr_rewrite_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(profile, "DictFilterExprRewriteTime",
parquet_profile, 1);
+ dict_filter_read_dict_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(profile, "DictFilterReadDictTime",
parquet_profile, 1);
+ dict_filter_build_time =
+ ADD_CHILD_TIMER_WITH_LEVEL(profile, "DictFilterBuildTime",
parquet_profile, 1);
+ dict_filter_candidate_columns = ADD_CHILD_COUNTER_WITH_LEVEL(
+ profile, "DictFilterCandidateColumns", TUnit::UNIT,
parquet_profile, 1);
+ dict_filter_columns = ADD_CHILD_COUNTER_WITH_LEVEL(profile,
"DictFilterColumns", TUnit::UNIT,
+ parquet_profile, 1);
+ dict_filter_unsupported_columns = ADD_CHILD_COUNTER_WITH_LEVEL(
+ profile, "DictFilterUnsupportedColumns", TUnit::UNIT,
parquet_profile, 1);
+ dict_filter_read_failures = ADD_CHILD_COUNTER_WITH_LEVEL(profile,
"DictFilterReadFailures",
+ TUnit::UNIT,
parquet_profile, 1);
+ rows_filtered_by_dict_filter = ADD_CHILD_COUNTER_WITH_LEVEL(profile,
"RowsFilteredByDictFilter",
+ TUnit::UNIT,
parquet_profile, 1);
convert_time = ADD_CHILD_TIMER_WITH_LEVEL(profile, "ConvertTime",
parquet_profile, 1);
bloom_filter_read_time =
ADD_CHILD_TIMER_WITH_LEVEL(profile, "BloomFilterReadTime",
parquet_profile, 1);
@@ -197,6 +213,15 @@ ParquetScanProfile ParquetProfile::scan_profile() const {
.range_gap_skipped_rows = range_gap_skipped_rows,
.column_read_time = column_read_time,
.predicate_filter_time = predicate_filter_time,
+ .dict_filter_rewrite_time = dict_filter_rewrite_time,
+ .dict_filter_expr_rewrite_time = dict_filter_expr_rewrite_time,
+ .dict_filter_read_dict_time = dict_filter_read_dict_time,
+ .dict_filter_build_time = dict_filter_build_time,
+ .dict_filter_candidate_columns = dict_filter_candidate_columns,
+ .dict_filter_columns = dict_filter_columns,
+ .dict_filter_unsupported_columns = dict_filter_unsupported_columns,
+ .dict_filter_read_failures = dict_filter_read_failures,
+ .rows_filtered_by_dict_filter = rows_filtered_by_dict_filter,
.column_reader_profile = column_reader_profile(),
};
}
diff --git a/be/src/format_v2/parquet/parquet_profile.h
b/be/src/format_v2/parquet/parquet_profile.h
index 91c8ad89ddc..27f9818f4a0 100644
--- a/be/src/format_v2/parquet/parquet_profile.h
+++ b/be/src/format_v2/parquet/parquet_profile.h
@@ -52,7 +52,18 @@ struct ParquetScanProfile {
RuntimeProfile::Counter* range_gap_skipped_rows = nullptr; // rows skipped
by range gaps
RuntimeProfile::Counter* column_read_time = nullptr; // column read
time (ns)
RuntimeProfile::Counter* predicate_filter_time = nullptr; // predicate
filter time (ns)
- ParquetColumnReaderProfile column_reader_profile; // nested
column read statistics
+ RuntimeProfile::Counter* dict_filter_rewrite_time = nullptr; // dictionary
rewrite time (ns)
+ RuntimeProfile::Counter* dict_filter_expr_rewrite_time =
+ nullptr; // expression/residual rewrite time (ns)
+ RuntimeProfile::Counter* dict_filter_read_dict_time = nullptr; //
dictionary page read time (ns)
+ RuntimeProfile::Counter* dict_filter_build_time =
+ nullptr; // dictionary entry bitmap build time (ns)
+ RuntimeProfile::Counter* dict_filter_candidate_columns = nullptr; //
candidate columns
+ RuntimeProfile::Counter* dict_filter_columns = nullptr; //
optimized columns
+ RuntimeProfile::Counter* dict_filter_unsupported_columns = nullptr; //
unsupported columns
+ RuntimeProfile::Counter* dict_filter_read_failures = nullptr; //
dictionary read failures
+ RuntimeProfile::Counter* rows_filtered_by_dict_filter = nullptr; //
rows filtered by dict
+ ParquetColumnReaderProfile column_reader_profile; // nested column read
statistics
};
// ============================================================================
@@ -137,6 +148,14 @@ struct ParquetProfile {
RuntimeProfile::Counter* predicate_filter_time = nullptr;
RuntimeProfile::Counter* dict_filter_rewrite_time = nullptr;
+ RuntimeProfile::Counter* dict_filter_expr_rewrite_time = nullptr;
+ RuntimeProfile::Counter* dict_filter_read_dict_time = nullptr;
+ RuntimeProfile::Counter* dict_filter_build_time = nullptr;
+ RuntimeProfile::Counter* dict_filter_candidate_columns = nullptr;
+ RuntimeProfile::Counter* dict_filter_columns = nullptr;
+ RuntimeProfile::Counter* dict_filter_unsupported_columns = nullptr;
+ RuntimeProfile::Counter* dict_filter_read_failures = nullptr;
+ RuntimeProfile::Counter* rows_filtered_by_dict_filter = nullptr;
RuntimeProfile::Counter* convert_time = nullptr;
RuntimeProfile::Counter* bloom_filter_read_time = nullptr;
};
diff --git a/be/src/format_v2/parquet/parquet_scan.cpp
b/be/src/format_v2/parquet/parquet_scan.cpp
index 86de8be9f92..99bfcc8e0ab 100644
--- a/be/src/format_v2/parquet/parquet_scan.cpp
+++ b/be/src/format_v2/parquet/parquet_scan.cpp
@@ -15,10 +15,13 @@
#include "format_v2/parquet/parquet_scan.h"
+#include <parquet/encoding.h>
+
#include <algorithm>
#include <limits>
#include <memory>
#include <ranges>
+#include <set>
#include <unordered_set>
#include <utility>
@@ -27,6 +30,7 @@
#include "core/assert_cast.h"
#include "core/block/block.h"
#include "core/column/column_vector.h"
+#include "exprs/vcompound_pred.h"
#include "exprs/vexpr_context.h"
#include "format_v2/parquet/parquet_column_schema.h"
#include "format_v2/parquet/parquet_file_context.h"
@@ -42,6 +46,70 @@ int64_t column_start_offset(const
::parquet::ColumnChunkMetaData& column_metadat
: cast_set<int64_t>(column_metadata.data_page_offset());
}
+bool is_dictionary_data_encoding(::parquet::Encoding::type encoding) {
+ return encoding == ::parquet::Encoding::PLAIN_DICTIONARY ||
+ encoding == ::parquet::Encoding::RLE_DICTIONARY;
+}
+
+bool is_level_encoding(::parquet::Encoding::type encoding) {
+ return encoding == ::parquet::Encoding::RLE || encoding ==
::parquet::Encoding::BIT_PACKED;
+}
+
+bool is_data_page_type(::parquet::PageType::type page_type) {
+ return page_type == ::parquet::PageType::DATA_PAGE ||
+ page_type == ::parquet::PageType::DATA_PAGE_V2;
+}
+
+bool is_fully_dictionary_encoded_chunk(const ::parquet::ColumnChunkMetaData&
column_metadata) {
+ if (!column_metadata.has_dictionary_page()) {
+ return false;
+ }
+
+ const auto& encoding_stats = column_metadata.encoding_stats();
+ if (!encoding_stats.empty()) {
+ bool has_dictionary_data_page = false;
+ for (const auto& encoding_stat : encoding_stats) {
+ if (!is_data_page_type(encoding_stat.page_type) ||
encoding_stat.count <= 0) {
+ continue;
+ }
+ if (!is_dictionary_data_encoding(encoding_stat.encoding)) {
+ return false;
+ }
+ has_dictionary_data_page = true;
+ }
+ return has_dictionary_data_page;
+ }
+
+ bool has_dictionary_encoding = false;
+ for (const auto encoding : column_metadata.encodings()) {
+ if (is_dictionary_data_encoding(encoding)) {
+ has_dictionary_encoding = true;
+ continue;
+ }
+ if (!is_level_encoding(encoding)) {
+ return false;
+ }
+ }
+ return has_dictionary_encoding;
+}
+
+bool supports_row_level_dictionary_filter(const ParquetColumnSchema&
column_schema,
+ const
::parquet::ColumnChunkMetaData& column_metadata) {
+ if (column_schema.kind != ParquetColumnSchemaKind::PRIMITIVE ||
+ column_schema.descriptor == nullptr || column_schema.type == nullptr ||
+ column_schema.max_repetition_level > 0) {
+ return false;
+ }
+ if (!column_schema.type_descriptor.is_string_like ||
+ column_metadata.type() != ::parquet::Type::BYTE_ARRAY) {
+ return false;
+ }
+ // Row-level dictionary filtering consumes dictionary ids from DATA_PAGE
payloads. It is exact
+ // only when every data page is dictionary encoded. Mixed dictionary/plain
chunks are left on
+ // the normal decoded-value path, matching the safety rule used by
StarRocks and Doris v1.
+ return is_fully_dictionary_encoded_chunk(column_metadata);
+}
+
void collect_all_leaf_column_ids(const ParquetColumnSchema& column_schema,
std::unordered_set<int>* leaf_column_ids) {
DORIS_CHECK(leaf_column_ids != nullptr);
@@ -270,6 +338,15 @@ Status plan_parquet_row_groups(const
::parquet::FileMetaData& metadata,
namespace {
+using DictionaryResidualConjunct = std::pair<VExprContextSPtr, VExprSPtr>;
+using DictionaryResidualConjuncts = std::vector<DictionaryResidualConjunct>;
+
+void update_counter_if_not_null(RuntimeProfile::Counter* counter, int64_t
value) {
+ if (counter != nullptr) {
+ COUNTER_UPDATE(counter, value);
+ }
+}
+
uint16_t apply_filter_to_selection(const IColumn::Filter& filter,
SelectionVector* selection,
uint16_t selected_rows) {
uint16_t new_selected_rows = 0;
@@ -282,6 +359,91 @@ uint16_t apply_filter_to_selection(const IColumn::Filter&
filter, SelectionVecto
return new_selected_rows;
}
+Status execute_compact_filter_conjuncts(const VExprContextSPtrs& conjuncts,
size_t rows,
+ Block* file_block, IColumn::Filter*
compact_filter,
+ bool* can_filter_all) {
+ DORIS_CHECK(compact_filter != nullptr);
+ DORIS_CHECK(can_filter_all != nullptr);
+ compact_filter->resize_fill(rows, 1);
+ *can_filter_all = false;
+ for (const auto& conjunct : conjuncts) {
+ DORIS_CHECK(conjunct != nullptr);
+ IColumn::Filter filter(rows, 1);
+ bool conjunct_can_filter_all = false;
+ RETURN_IF_ERROR(conjunct->execute_filter(file_block, filter.data(),
rows, false,
+ &conjunct_can_filter_all));
+ if (conjunct_can_filter_all) {
+ std::ranges::fill(*compact_filter, 0);
+ *can_filter_all = true;
+ break;
+ }
+ for (size_t row = 0; row < rows; ++row) {
+ (*compact_filter)[row] &= filter[row];
+ }
+ }
+ return Status::OK();
+}
+
+Status execute_compact_dictionary_residual_conjuncts(const
DictionaryResidualConjuncts& conjuncts,
+ size_t rows, Block*
file_block,
+ IColumn::Filter*
compact_filter,
+ bool* can_filter_all) {
+ DORIS_CHECK(compact_filter != nullptr);
+ DORIS_CHECK(can_filter_all != nullptr);
+ compact_filter->resize_fill(rows, 1);
+ *can_filter_all = false;
+ for (const auto& [owner_context, residual_expr] : conjuncts) {
+ DORIS_CHECK(owner_context != nullptr);
+ DORIS_CHECK(residual_expr != nullptr);
+ IColumn::Filter filter(rows, 1);
+ bool conjunct_can_filter_all = false;
+ RETURN_IF_ERROR(residual_expr->execute_filter(owner_context.get(),
file_block,
+ filter.data(), rows,
false,
+
&conjunct_can_filter_all));
+ if (conjunct_can_filter_all) {
+ std::ranges::fill(*compact_filter, 0);
+ *can_filter_all = true;
+ break;
+ }
+ for (size_t row = 0; row < rows; ++row) {
+ (*compact_filter)[row] &= filter[row];
+ }
+ }
+ return Status::OK();
+}
+
+Status execute_compact_delete_conjuncts(const VExprContextSPtrs&
delete_conjuncts, size_t rows,
+ Block* file_block, IColumn::Filter*
compact_filter,
+ bool* can_filter_all) {
+ DORIS_CHECK(compact_filter != nullptr);
+ DORIS_CHECK(can_filter_all != nullptr);
+ compact_filter->resize_fill(rows, 1);
+ *can_filter_all = false;
+ for (const auto& delete_conjunct : delete_conjuncts) {
+ DORIS_CHECK(delete_conjunct != nullptr);
+ int result_column_id = -1;
+
RETURN_IF_ERROR(delete_conjunct->root()->execute(delete_conjunct.get(),
file_block,
+ &result_column_id));
+ DORIS_CHECK(result_column_id >= 0 &&
+ result_column_id <
static_cast<int>(file_block->columns()));
+ const auto& delete_filter = assert_cast<const ColumnUInt8&>(
+
*file_block->get_by_position(result_column_id).column)
+ .get_data();
+ DORIS_CHECK(delete_filter.size() == rows);
+ bool has_kept_row = false;
+ for (size_t row = 0; row < rows; ++row) {
+ (*compact_filter)[row] &= !delete_filter[row];
+ has_kept_row |= (*compact_filter)[row] != 0;
+ }
+ file_block->erase(result_column_id);
+ if (!has_kept_row) {
+ *can_filter_all = true;
+ break;
+ }
+ }
+ return Status::OK();
+}
+
Status execute_filter_conjuncts(const format::FileScanRequest& request,
int64_t batch_rows,
Block* file_block, SelectionVector* selection,
uint16_t* selected_rows) {
@@ -334,6 +496,20 @@ Status execute_delete_conjuncts(const
format::FileScanRequest& request, int64_t
} // namespace
+uint16_t apply_compact_filter_to_selection(const IColumn::Filter& filter,
+ SelectionVector* selection,
uint16_t selected_rows) {
+ DORIS_CHECK(selection != nullptr);
+ DORIS_CHECK(filter.size() == selected_rows);
+ uint16_t new_selected_rows = 0;
+ for (uint16_t selection_idx = 0; selection_idx < selected_rows;
++selection_idx) {
+ if (filter[selection_idx] != 0) {
+ selection->set_index(new_selected_rows++,
static_cast<SelectionVector::Index>(
+
selection->get_index(selection_idx)));
+ }
+ }
+ return new_selected_rows;
+}
+
IColumn::Filter selection_to_filter(const SelectionVector& selection, uint16_t
selected_rows,
int64_t batch_rows) {
IColumn::Filter filter(static_cast<size_t>(batch_rows), 0);
@@ -462,6 +638,8 @@ void ParquetScanScheduler::reset_current_row_group() {
_current_row_group.reset();
_current_predicate_columns.clear();
_current_non_predicate_columns.clear();
+ _current_dictionary_filters.clear();
+ _current_dictionary_residual_conjuncts.clear();
_current_row_group_rows = 0;
_current_row_group_id = -1;
_current_row_group_rows_read = 0;
@@ -484,8 +662,12 @@ Status ParquetScanScheduler::open_next_row_group(
}
const RowGroupReadPlan& row_group_plan =
_row_group_plans[_next_row_group_plan_idx++];
const int row_group_idx = row_group_plan.row_group_id;
- _current_merge_range_active =
- prepare_current_row_group_reader(file_context, file_schema,
request, row_group_idx);
+ // Row-level dictionary filters read dictionary pages before Arrow
RecordReaders are created.
+ // Keep that probe on the base reader: MergeRangeFileReader expects each
registered range to be
+ // consumed as one forward pass, while the later RecordReader opens the
same column chunk again
+ // for the data-page stream.
+ file_context.reset_random_access_ranges();
+ _current_merge_range_active = false;
try {
_current_row_group = file_context.file_reader->RowGroup(row_group_idx);
} catch (const ::parquet::ParquetException& e) {
@@ -510,6 +692,11 @@ Status ParquetScanScheduler::open_next_row_group(
_current_range_rows_read = 0;
_current_predicate_columns.clear();
_current_non_predicate_columns.clear();
+ _current_dictionary_filters.clear();
+ RETURN_IF_ERROR(prepare_current_dictionary_filters(file_context,
file_schema, request,
+ row_group_idx,
*row_group_metadata));
+ _current_merge_range_active =
+ prepare_current_row_group_reader(file_context, file_schema,
request, row_group_idx);
ParquetColumnReaderFactory column_reader_factory(
_current_row_group, file_context.schema->num_columns(),
&row_group_plan.page_skip_plans,
@@ -535,7 +722,9 @@ Status ParquetScanScheduler::open_next_row_group(
const auto& column_schema = file_schema[local_id];
DORIS_CHECK(column_schema != nullptr);
std::unique_ptr<ParquetColumnReader> column_reader;
- RETURN_IF_ERROR(column_reader_factory.create(*column_schema, &col,
&column_reader));
+ RETURN_IF_ERROR(
+ column_reader_factory.create(*column_schema, &col,
&column_reader,
+
_current_dictionary_filters.contains(local_id)));
_current_predicate_columns[local_id] = std::move(column_reader);
}
// Start warming filter-column chunks as soon as their row group is
selected. Parquet v2 still
@@ -597,42 +786,517 @@ Status
ParquetScanScheduler::skip_current_row_group_rows(int64_t rows) {
return Status::OK();
}
+namespace {
+
+struct PredicateConjunctSchedule {
+ std::map<size_t, VExprContextSPtrs> single_column_conjuncts;
+ VExprContextSPtrs remaining_conjuncts;
+};
+
+PredicateConjunctSchedule build_predicate_conjunct_schedule(
+ const format::FileScanRequest& request) {
+ std::unordered_set<size_t> predicate_block_positions;
+ predicate_block_positions.reserve(request.predicate_columns.size());
+ for (const auto& col : request.predicate_columns) {
+ const auto position_it = request.local_positions.find(col.column_id());
+ DORIS_CHECK(position_it != request.local_positions.end());
+ predicate_block_positions.insert(position_it->second.value());
+ }
+
+ PredicateConjunctSchedule schedule;
+ for (const auto& conjunct : request.conjuncts) {
+ DORIS_CHECK(conjunct != nullptr);
+ DORIS_CHECK(conjunct->root() != nullptr);
+ if (!conjunct->root()->is_safe_to_execute_on_selected_rows()) {
+ // Round-by-round filtering can compact later predicate columns
before evaluating
+ // remaining expressions. Stateful functions such as random(1) and
error-preserving
+ // functions such as assert_true() must see the same full batch
they saw before this
+ // optimization, so any unsafe conjunct disables the per-column
schedule for the batch.
+ schedule.remaining_conjuncts = request.conjuncts;
+ schedule.single_column_conjuncts.clear();
+ return schedule;
+ }
+ std::set<int> referenced_positions;
+ conjunct->root()->collect_slot_column_ids(referenced_positions);
+ if (referenced_positions.size() != 1) {
+ schedule.remaining_conjuncts.push_back(conjunct);
+ continue;
+ }
+ const auto block_position =
static_cast<size_t>(*referenced_positions.begin());
+ if (!predicate_block_positions.contains(block_position)) {
+ schedule.remaining_conjuncts.push_back(conjunct);
+ continue;
+ }
+ schedule.single_column_conjuncts[block_position].push_back(conjunct);
+ }
+ return schedule;
+}
+
+bool can_evaluate_all_with_dictionary(const VExprContextSPtrs& conjuncts) {
+ if (conjuncts.empty()) {
+ return false;
+ }
+ return std::ranges::all_of(conjuncts, [](const auto& conjunct) {
+ return conjunct != nullptr && conjunct->root() != nullptr &&
+ conjunct->root()->can_evaluate_dictionary_filter();
+ });
+}
+
+bool can_evaluate_dictionary_exactly(const VExprSPtr& expr) {
+ DORIS_CHECK(expr != nullptr);
+ const auto* compound_pred = dynamic_cast<const VCompoundPred*>(expr.get());
+ if (compound_pred == nullptr) {
+ return expr->can_evaluate_dictionary_filter();
+ }
+ if (compound_pred->op() != TExprOpcode::COMPOUND_AND &&
+ compound_pred->op() != TExprOpcode::COMPOUND_OR) {
+ return false;
+ }
+ return !expr->children().empty() &&
+ std::ranges::all_of(expr->children(), [](const auto& child) {
+ return can_evaluate_dictionary_exactly(child);
+ });
+}
+
+void collect_dictionary_residual_exprs(const VExprContextSPtr& owner_context,
const VExprSPtr& expr,
+ DictionaryResidualConjuncts*
residual_conjuncts) {
+ DORIS_CHECK(owner_context != nullptr);
+ DORIS_CHECK(expr != nullptr);
+ DORIS_CHECK(residual_conjuncts != nullptr);
+
+ if (can_evaluate_dictionary_exactly(expr)) {
+ return;
+ }
+
+ // VCompoundPred dictionary evaluation is a conservative prefilter for AND
when only some
+ // children are dictionary-aware. Split AND so exact dictionary children
are not executed again
+ // on materialized rows. Do not split a non-exact OR: its branches cannot
be evaluated
+ // independently after a dictionary prefilter without changing the
original boolean semantics.
+ const auto* compound_pred = dynamic_cast<const VCompoundPred*>(expr.get());
+ if (compound_pred != nullptr && compound_pred->op() ==
TExprOpcode::COMPOUND_AND) {
+ for (const auto& child : expr->children()) {
+ collect_dictionary_residual_exprs(owner_context, child,
residual_conjuncts);
+ }
+ return;
+ }
+
+ residual_conjuncts->emplace_back(owner_context, expr);
+}
+
+DictionaryResidualConjuncts build_dictionary_residual_conjuncts(
+ const VExprContextSPtrs& conjuncts) {
+ DictionaryResidualConjuncts residual_conjuncts;
+ for (const auto& conjunct : conjuncts) {
+ DORIS_CHECK(conjunct != nullptr);
+ collect_dictionary_residual_exprs(conjunct, conjunct->root(),
&residual_conjuncts);
+ }
+ return residual_conjuncts;
+}
+
+uint16_t count_selected_rows(const IColumn::Filter& filter) {
+ uint16_t selected_rows = 0;
+ for (const auto value : filter) {
+ selected_rows += value != 0;
+ }
+ return selected_rows;
+}
+
+Status filter_read_predicate_columns(Block* file_block, const
std::vector<uint32_t>& positions,
+ const IColumn::Filter& compact_filter) {
+ if (positions.empty()) {
+ return Status::OK();
+ }
+ RETURN_IF_CATCH_EXCEPTION(Block::filter_block_internal(file_block,
positions, compact_filter));
+ return Status::OK();
+}
+
+IColumn::Filter build_dictionary_entry_filter(size_t block_position,
+ const ParquetColumnSchema&
column_schema,
+ const VExprContextSPtrs&
conjuncts,
+ const ParquetDictionaryWords&
dict_words) {
+ auto fields = dictionary_fields_from_words(dict_words);
+ IColumn::Filter dictionary_filter(fields.size(), 1);
+ DictionaryEvalContext ctx;
+ auto& slot = ctx.slots
+ .emplace(static_cast<int>(block_position),
+ DictionaryEvalContext::SlotDictionary {
+ .data_type = column_schema.type,
.values = {}})
+ .first->second;
+ slot.values.reserve(1);
+
+ for (size_t dict_idx = 0; dict_idx < fields.size(); ++dict_idx) {
+ slot.values.clear();
+ slot.values.push_back(fields[dict_idx]);
+ dictionary_filter[dict_idx] =
VExprContext::evaluate_dictionary_filter(conjuncts, ctx) ==
+
ZoneMapFilterResult::kNoMatch
+ ? 0
+ : 1;
+ }
+ return dictionary_filter;
+}
+
+} // namespace
+
+Status ParquetScanScheduler::prepare_current_dictionary_filters(
+ ParquetFileContext& file_context,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
+ const format::FileScanRequest& request, int row_group_idx,
+ const ::parquet::RowGroupMetaData& row_group_metadata) {
+ _current_dictionary_filters.clear();
+ _current_dictionary_residual_conjuncts.clear();
+ if (request.conjuncts.empty()) {
+ return Status::OK();
+ }
+ PredicateConjunctSchedule schedule;
+ {
+ SCOPED_TIMER(_scan_profile.dict_filter_expr_rewrite_time);
+ schedule = build_predicate_conjunct_schedule(request);
+ }
+ if (schedule.single_column_conjuncts.empty()) {
+ return Status::OK();
+ }
+
+ SCOPED_TIMER(_scan_profile.dict_filter_rewrite_time);
+ for (const auto& col : request.predicate_columns) {
+ const auto local_id = col.local_id();
+ if (local_id < 0 || local_id >=
static_cast<int32_t>(file_schema.size())) {
+ continue;
+ }
+ const auto position_it = request.local_positions.find(col.column_id());
+ DORIS_CHECK(position_it != request.local_positions.end());
+ const auto block_position =
static_cast<size_t>(position_it->second.value());
+ const auto conjunct_it =
schedule.single_column_conjuncts.find(block_position);
+ if (conjunct_it == schedule.single_column_conjuncts.end() ||
+ !can_evaluate_all_with_dictionary(conjunct_it->second)) {
+ continue;
+ }
+
update_counter_if_not_null(_scan_profile.dict_filter_candidate_columns, 1);
+
+ // This optimization is deliberately limited to single-column
predicates with a dictionary
+ // evaluable part. Mixed AND predicates are split so
dictionary-covered children run as a
+ // dict-id prefilter and residual children keep the normal row-level
expression path.
+ const auto& column_schema = file_schema[local_id];
+ DORIS_CHECK(column_schema != nullptr);
+ if (column_schema->leaf_column_id < 0 ||
+ column_schema->leaf_column_id >= row_group_metadata.num_columns())
{
+
update_counter_if_not_null(_scan_profile.dict_filter_unsupported_columns, 1);
+ continue;
+ }
+ auto column_chunk =
row_group_metadata.ColumnChunk(column_schema->leaf_column_id);
+ if (column_chunk == nullptr ||
+ !supports_row_level_dictionary_filter(*column_schema,
*column_chunk)) {
+
update_counter_if_not_null(_scan_profile.dict_filter_unsupported_columns, 1);
+ continue;
+ }
+
+ ParquetDictionaryWords dict_words;
+ {
+ SCOPED_TIMER(_scan_profile.dict_filter_read_dict_time);
+ if (!read_dictionary_words(file_context.file_reader.get(),
row_group_idx,
+ column_schema->leaf_column_id,
*column_schema,
+ &dict_words)) {
+
update_counter_if_not_null(_scan_profile.dict_filter_read_failures, 1);
+ continue;
+ }
+ }
+
+ // Build a safe dictionary prefilter from the dictionary-filter
interface instead of
+ // executing the row expression on a temporary dictionary block. For
compound AND,
+ // VCompoundPred intentionally evaluates only dictionary-capable
children, so residual
+ // predicates still run later on surviving rows.
+ IColumn::Filter dictionary_filter;
+ DictionaryResidualConjuncts residual_conjuncts;
+ {
+ SCOPED_TIMER(_scan_profile.dict_filter_build_time);
+ dictionary_filter = build_dictionary_entry_filter(block_position,
*column_schema,
+
conjunct_it->second, dict_words);
+ residual_conjuncts =
build_dictionary_residual_conjuncts(conjunct_it->second);
+ }
+
+ // The bitmap is keyed by Parquet dictionary id. Later data-page reads
evaluate the
+ // predicate with an integer lookup and only materialize STRING values
for surviving rows.
+ _current_dictionary_filters.emplace(local_id,
std::move(dictionary_filter));
+ _current_dictionary_residual_conjuncts.emplace(local_id,
std::move(residual_conjuncts));
+ update_counter_if_not_null(_scan_profile.dict_filter_columns, 1);
+ }
+ return Status::OK();
+}
+
Status ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
const
format::FileScanRequest& request,
Block* file_block,
SelectionVector* selection,
uint16_t* selected_rows,
- int64_t*
conjunct_filtered_rows) {
+ int64_t*
conjunct_filtered_rows,
+ bool*
predicate_columns_filtered) {
+ DORIS_CHECK(predicate_columns_filtered != nullptr);
+ *predicate_columns_filtered = false;
if (!request.conjuncts.empty() || !request.delete_conjuncts.empty()) {
selection->resize(static_cast<size_t>(batch_rows));
}
- for (const auto& [fid, column_reader] : _current_predicate_columns) {
- auto position_it =
request.local_positions.find(format::LocalColumnId(fid));
- DORIS_CHECK(position_it != request.local_positions.end());
- const auto block_position = position_it->second.value();
+ const auto schedule = build_predicate_conjunct_schedule(request);
+ const bool can_read_predicate_columns_round_by_round =
+ !schedule.single_column_conjuncts.empty();
+ std::vector<uint32_t> read_column_positions;
+ read_column_positions.reserve(request.predicate_columns.size());
+
+ auto read_predicate_column = [&](ParquetColumnReader* column_reader,
size_t block_position,
+ ColumnId local_id, bool*
used_dictionary_filter) -> Status {
+ DORIS_CHECK(used_dictionary_filter != nullptr);
+ *used_dictionary_filter = false;
DCHECK(remove_nullable(column_reader->type())
->equals(*remove_nullable(file_block->get_by_position(block_position).type)))
<< column_reader->type()->get_name() << " "
<<
file_block->get_by_position(block_position).type->get_name() << " "
<< column_reader->name() << " " <<
file_block->get_by_position(block_position).name;
auto column =
file_block->get_by_position(block_position).column->assert_mutable();
- int64_t column_rows = 0;
- {
- SCOPED_TIMER(_scan_profile.column_read_time);
- RETURN_IF_ERROR(column_reader->read(batch_rows, column,
&column_rows));
+ SCOPED_TIMER(_scan_profile.column_read_time);
+ const auto dictionary_filter_it =
_current_dictionary_filters.find(local_id);
+ if (dictionary_filter_it != _current_dictionary_filters.end()) {
+ const uint16_t selected_rows_before = *selected_rows;
+ IColumn::Filter compact_filter;
+ bool used_filter = false;
+ RETURN_IF_ERROR(column_reader->select_with_dictionary_filter(
+ *selection, *selected_rows, batch_rows,
dictionary_filter_it->second, column,
+ &compact_filter, &used_filter));
+ if (used_filter) {
+ DORIS_CHECK(compact_filter.size() == selected_rows_before);
+ const uint16_t new_selected_rows =
count_selected_rows(compact_filter);
+ const auto filtered_rows =
static_cast<int64_t>(selected_rows_before) -
+
static_cast<int64_t>(new_selected_rows);
+ if (conjunct_filtered_rows != nullptr) {
+ *conjunct_filtered_rows += filtered_rows;
+ }
+
update_counter_if_not_null(_scan_profile.rows_filtered_by_dict_filter,
+ filtered_rows);
+ if (new_selected_rows != selected_rows_before) {
+ // The dictionary reader has already appended only
surviving values for the
+ // current column. Apply the compact row filter only to
columns read before this
+ // one, then update the shared selection for later
predicate/lazy columns.
+ RETURN_IF_ERROR(filter_read_predicate_columns(file_block,
read_column_positions,
+
compact_filter));
+ *selected_rows =
apply_compact_filter_to_selection(compact_filter, selection,
+
selected_rows_before);
+ *predicate_columns_filtered = true;
+ }
+ file_block->replace_by_position(block_position,
std::move(column));
+
read_column_positions.push_back(cast_set<uint32_t>(block_position));
+ *used_dictionary_filter = true;
+ return Status::OK();
+ }
}
- if (column_rows != batch_rows) {
- return Status::Corruption("Parquet filter column {} returned {}
rows, expected {} rows",
- column_reader->name(), column_rows,
batch_rows);
+
+ if (*selected_rows == batch_rows) {
+ int64_t column_rows = 0;
+ RETURN_IF_ERROR(column_reader->read(batch_rows, column,
&column_rows));
+ if (column_rows != batch_rows) {
+ return Status::Corruption(
+ "Parquet filter column {} returned {} rows, expected
{} rows",
+ column_reader->name(), column_rows, batch_rows);
+ }
+ } else {
+ [[maybe_unused]] auto old_size = column->size();
+ RETURN_IF_ERROR(column_reader->select(*selection, *selected_rows,
batch_rows, column));
+ if (column->size() != old_size + *selected_rows) {
+ return Status::Corruption(
+ "Parquet selected filter column {} returned {} rows,
expected {} rows",
+ column_reader->name(), column->size(), old_size +
*selected_rows);
+ }
+ *predicate_columns_filtered = true;
}
file_block->replace_by_position(block_position, std::move(column));
- }
- if (_scan_profile.predicate_filter_time == nullptr) {
+ read_column_positions.push_back(cast_set<uint32_t>(block_position));
+ return Status::OK();
+ };
+
+ auto execute_scheduled_conjuncts = [&](const VExprContextSPtrs& conjuncts)
-> Status {
+ if (conjuncts.empty() || *selected_rows == 0) {
+ return Status::OK();
+ }
+ const uint16_t selected_rows_before = *selected_rows;
+ IColumn::Filter compact_filter;
+ bool can_filter_all = false;
+ RETURN_IF_ERROR(execute_compact_filter_conjuncts(
+ conjuncts, selected_rows_before, file_block, &compact_filter,
&can_filter_all));
+ if (can_filter_all) {
+ compact_filter.resize_fill(selected_rows_before, 0);
+ }
+ const uint16_t new_selected_rows = can_filter_all ? 0 :
count_selected_rows(compact_filter);
+ if (conjunct_filtered_rows != nullptr) {
+ *conjunct_filtered_rows +=
static_cast<int64_t>(selected_rows_before) -
+ static_cast<int64_t>(new_selected_rows);
+ }
+ if (new_selected_rows != selected_rows_before) {
+ // All columns read so far are already compacted to the current
selection. Apply the
+ // compact filter to those columns and the selection vector
together, so later predicate
+ // columns can read only rows that survived previous predicate
rounds.
+ RETURN_IF_ERROR(filter_read_predicate_columns(file_block,
read_column_positions,
+ compact_filter));
+ *selected_rows = can_filter_all
+ ? 0
+ :
apply_compact_filter_to_selection(compact_filter, selection,
+
selected_rows_before);
+ *predicate_columns_filtered = true;
+ }
+ return Status::OK();
+ };
+
+ auto execute_scheduled_dictionary_residual_conjuncts =
+ [&](const DictionaryResidualConjuncts& conjuncts) -> Status {
+ if (conjuncts.empty() || *selected_rows == 0) {
+ return Status::OK();
+ }
+ const uint16_t selected_rows_before = *selected_rows;
+ IColumn::Filter compact_filter;
+ bool can_filter_all = false;
+ RETURN_IF_ERROR(execute_compact_dictionary_residual_conjuncts(
+ conjuncts, selected_rows_before, file_block, &compact_filter,
&can_filter_all));
+ if (can_filter_all) {
+ compact_filter.resize_fill(selected_rows_before, 0);
+ }
+ const uint16_t new_selected_rows = can_filter_all ? 0 :
count_selected_rows(compact_filter);
+ if (conjunct_filtered_rows != nullptr) {
+ *conjunct_filtered_rows +=
static_cast<int64_t>(selected_rows_before) -
+ static_cast<int64_t>(new_selected_rows);
+ }
+ if (new_selected_rows != selected_rows_before) {
+ // Dictionary-covered children have already reduced the compact
block. Apply only the
+ // residual child filters here, then keep the same
compacted-column invariant as the
+ // normal conjunct path for later predicate rounds.
+ RETURN_IF_ERROR(filter_read_predicate_columns(file_block,
read_column_positions,
+ compact_filter));
+ *selected_rows = can_filter_all
+ ? 0
+ :
apply_compact_filter_to_selection(compact_filter, selection,
+
selected_rows_before);
+ *predicate_columns_filtered = true;
+ }
+ return Status::OK();
+ };
+
+ auto execute_scheduled_conjuncts_with_profile =
+ [&](const VExprContextSPtrs& conjuncts) -> Status {
+ if (_scan_profile.predicate_filter_time == nullptr) {
+ return execute_scheduled_conjuncts(conjuncts);
+ }
+ SCOPED_TIMER(_scan_profile.predicate_filter_time);
+ return execute_scheduled_conjuncts(conjuncts);
+ };
+
+ auto execute_scheduled_dictionary_residual_conjuncts_with_profile =
+ [&](const DictionaryResidualConjuncts& conjuncts) -> Status {
+ if (_scan_profile.predicate_filter_time == nullptr) {
+ return execute_scheduled_dictionary_residual_conjuncts(conjuncts);
+ }
+ SCOPED_TIMER(_scan_profile.predicate_filter_time);
+ return execute_scheduled_dictionary_residual_conjuncts(conjuncts);
+ };
+
+ auto execute_scheduled_delete_conjuncts = [&]() -> Status {
+ if (request.delete_conjuncts.empty() || *selected_rows == 0) {
+ return Status::OK();
+ }
+ const uint16_t selected_rows_before = *selected_rows;
+ IColumn::Filter compact_filter;
+ bool can_filter_all = false;
+
RETURN_IF_ERROR(execute_compact_delete_conjuncts(request.delete_conjuncts,
+ selected_rows_before,
file_block,
+ &compact_filter,
&can_filter_all));
+ if (can_filter_all) {
+ compact_filter.resize_fill(selected_rows_before, 0);
+ }
+ if (can_filter_all || count_selected_rows(compact_filter) !=
selected_rows_before) {
+ RETURN_IF_ERROR(filter_read_predicate_columns(file_block,
read_column_positions,
+ compact_filter));
+ *selected_rows = can_filter_all
+ ? 0
+ :
apply_compact_filter_to_selection(compact_filter, selection,
+
selected_rows_before);
+ *predicate_columns_filtered = true;
+ }
+ return Status::OK();
+ };
+
+ auto read_all_predicate_columns = [&]() -> Status {
+ for (const auto& [fid, column_reader] : _current_predicate_columns) {
+ auto position_it =
request.local_positions.find(format::LocalColumnId(fid));
+ DORIS_CHECK(position_it != request.local_positions.end());
+ bool used_dictionary_filter = false;
+ RETURN_IF_ERROR(read_predicate_column(column_reader.get(),
position_it->second.value(),
+ fid,
&used_dictionary_filter));
+ }
+ return Status::OK();
+ };
+
+ if (!can_read_predicate_columns_round_by_round) {
+ RETURN_IF_ERROR(read_all_predicate_columns());
+ if (_scan_profile.predicate_filter_time == nullptr) {
+ return execute_batch_filters(request, batch_rows, file_block,
selection, selected_rows,
+ conjunct_filtered_rows);
+ }
+ SCOPED_TIMER(_scan_profile.predicate_filter_time);
return execute_batch_filters(request, batch_rows, file_block,
selection, selected_rows,
conjunct_filtered_rows);
}
+
+ auto read_round_by_round = [&]() -> Status {
+ // Single-column conjuncts can be evaluated immediately after their
column is read. Once
+ // selection shrinks, later predicate columns use
ParquetColumnReader::select() so the
+ // reader skips rows already rejected by earlier predicates instead of
materializing them.
+ for (size_t idx = 0; idx < request.predicate_columns.size(); ++idx) {
+ const auto& col = request.predicate_columns[idx];
+ const auto fid = col.local_id();
+ auto reader_it = _current_predicate_columns.find(fid);
+ DORIS_CHECK(reader_it != _current_predicate_columns.end());
+ auto position_it = request.local_positions.find(col.column_id());
+ DORIS_CHECK(position_it != request.local_positions.end());
+ const auto block_position = position_it->second.value();
+ bool used_dictionary_filter = false;
+ RETURN_IF_ERROR(read_predicate_column(reader_it->second.get(),
block_position, fid,
+ &used_dictionary_filter));
+ if (*selected_rows == 0) {
+ for (size_t remaining_idx = idx + 1;
+ remaining_idx < request.predicate_columns.size();
++remaining_idx) {
+ const auto remaining_fid =
request.predicate_columns[remaining_idx].local_id();
+ auto remaining_reader_it =
_current_predicate_columns.find(remaining_fid);
+ DORIS_CHECK(remaining_reader_it !=
_current_predicate_columns.end());
+
RETURN_IF_ERROR(remaining_reader_it->second->skip(batch_rows));
+ }
+ return Status::OK();
+ }
+ const auto conjunct_it =
schedule.single_column_conjuncts.find(block_position);
+ if (conjunct_it == schedule.single_column_conjuncts.end()) {
+ continue;
+ }
+ if (used_dictionary_filter) {
+ const auto residual_it =
_current_dictionary_residual_conjuncts.find(fid);
+ DORIS_CHECK(residual_it !=
_current_dictionary_residual_conjuncts.end());
+
RETURN_IF_ERROR(execute_scheduled_dictionary_residual_conjuncts_with_profile(
+ residual_it->second));
+ } else {
+
RETURN_IF_ERROR(execute_scheduled_conjuncts_with_profile(conjunct_it->second));
+ }
+ if (*selected_rows != 0) {
+ continue;
+ }
+ for (size_t remaining_idx = idx + 1; remaining_idx <
request.predicate_columns.size();
+ ++remaining_idx) {
+ const auto remaining_fid =
request.predicate_columns[remaining_idx].local_id();
+ auto remaining_reader_it =
_current_predicate_columns.find(remaining_fid);
+ DORIS_CHECK(remaining_reader_it !=
_current_predicate_columns.end());
+ RETURN_IF_ERROR(remaining_reader_it->second->skip(batch_rows));
+ }
+ return Status::OK();
+ }
+ return Status::OK();
+ };
+
+ RETURN_IF_ERROR(read_round_by_round());
+
RETURN_IF_ERROR(execute_scheduled_conjuncts_with_profile(schedule.remaining_conjuncts));
+ if (_scan_profile.predicate_filter_time == nullptr) {
+ return execute_scheduled_delete_conjuncts();
+ }
SCOPED_TIMER(_scan_profile.predicate_filter_time);
- return execute_batch_filters(request, batch_rows, file_block, selection,
selected_rows,
- conjunct_filtered_rows);
+ return execute_scheduled_delete_conjuncts();
}
bool ParquetScanScheduler::prepare_current_row_group_reader(
@@ -692,8 +1356,9 @@ Status ParquetScanScheduler::read_current_row_group_batch(
DORIS_CHECK(batch_rows <= std::numeric_limits<uint16_t>::max());
uint16_t selected_rows = static_cast<uint16_t>(batch_rows);
int64_t conjunct_filtered_rows = 0;
+ bool predicate_columns_filtered = false;
RETURN_IF_ERROR(read_filter_columns(batch_rows, request, file_block,
&selection, &selected_rows,
- &conjunct_filtered_rows));
+ &conjunct_filtered_rows,
&predicate_columns_filtered));
_predicate_filtered_rows += conjunct_filtered_rows;
mark_condition_cache_granules(selection, selected_rows,
batch_first_file_row);
@@ -711,7 +1376,7 @@ Status ParquetScanScheduler::read_current_row_group_batch(
if (selected_rows == 0 && _scan_profile.empty_selection_batches !=
nullptr) {
COUNTER_UPDATE(_scan_profile.empty_selection_batches, 1);
}
- if (need_filter_output) {
+ if (need_filter_output && !predicate_columns_filtered) {
IColumn::Filter output_filter = selection_to_filter(selection,
selected_rows, batch_rows);
for (const auto& col : request.predicate_columns) {
auto position_it = request.local_positions.find(col.column_id());
diff --git a/be/src/format_v2/parquet/parquet_scan.h
b/be/src/format_v2/parquet/parquet_scan.h
index 1fb313d4f75..3fa7586fb1b 100644
--- a/be/src/format_v2/parquet/parquet_scan.h
+++ b/be/src/format_v2/parquet/parquet_scan.h
@@ -20,6 +20,7 @@
#include <map>
#include <memory>
#include <optional>
+#include <utility>
#include <vector>
#include "common/status.h"
@@ -35,6 +36,7 @@
namespace parquet {
class FileMetaData;
class ParquetFileReader;
+class RowGroupMetaData;
class RowGroupReader;
} // namespace parquet
@@ -93,6 +95,9 @@ Status plan_parquet_row_groups(const ::parquet::FileMetaData&
metadata,
IColumn::Filter selection_to_filter(const SelectionVector& selection, uint16_t
selected_rows,
int64_t batch_rows);
+uint16_t apply_compact_filter_to_selection(const IColumn::Filter& filter,
+ SelectionVector* selection,
uint16_t selected_rows);
+
Status execute_batch_filters(const format::FileScanRequest& request, int64_t
batch_rows,
Block* file_block, SelectionVector* selection,
uint16_t* selected_rows,
int64_t* conjunct_filtered_rows = nullptr);
@@ -150,7 +155,14 @@ private:
Status read_filter_columns(int64_t batch_rows, const
format::FileScanRequest& request,
Block* file_block, SelectionVector* selection,
- uint16_t* selected_rows, int64_t*
conjunct_filtered_rows);
+ uint16_t* selected_rows, int64_t*
conjunct_filtered_rows,
+ bool* predicate_columns_filtered);
+
+ Status prepare_current_dictionary_filters(
+ ParquetFileContext& file_context,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>&
file_schema,
+ const format::FileScanRequest& request, int row_group_idx,
+ const ::parquet::RowGroupMetaData& row_group_metadata);
void prefetch_current_row_group_columns(
ParquetFileContext& file_context,
@@ -178,11 +190,15 @@ private:
std::map<ColumnId, std::unique_ptr<ParquetColumnReader>>
_current_predicate_columns; // predicate ColumnReaders
std::map<ColumnId, std::unique_ptr<ParquetColumnReader>>
- _current_non_predicate_columns; // non-predicate ColumnReaders
- int64_t _current_row_group_rows = 0; // current row group row count
- int _current_row_group_id = -1; // current row group id in
parquet metadata
- int64_t _current_row_group_rows_read = 0; // rows read in the current row
group (cursor)
- int64_t _current_row_group_first_row = 0; // first file row of the current
row group
+ _current_non_predicate_columns; // non-predicate ColumnReaders
+ std::map<ColumnId, IColumn::Filter>
+ _current_dictionary_filters; // local id -> dict entry bitmap
+ std::map<ColumnId, std::vector<std::pair<VExprContextSPtr, VExprSPtr>>>
+ _current_dictionary_residual_conjuncts; // local id -> row-level
residual conjuncts
+ int64_t _current_row_group_rows = 0; // current row group row
count
+ int _current_row_group_id = -1; // current row group id in
parquet metadata
+ int64_t _current_row_group_rows_read = 0; // rows read in the
current row group (cursor)
+ int64_t _current_row_group_first_row = 0; // first file row of the
current row group
std::vector<RowRange>
_current_selected_ranges; // selected ranges for the current row
group after page-index pruning
size_t _current_range_idx = 0; // current selected_range index
diff --git a/be/src/format_v2/parquet/parquet_statistics.cpp
b/be/src/format_v2/parquet/parquet_statistics.cpp
index fa81be03e98..d0692a80205 100644
--- a/be/src/format_v2/parquet/parquet_statistics.cpp
+++ b/be/src/format_v2/parquet/parquet_statistics.cpp
@@ -498,26 +498,11 @@ bool supports_dictionary_pruning(const
ParquetColumnSchema& column_schema,
return true;
}
-struct OwnedDictionaryWords {
- std::vector<std::string> values;
- std::vector<StringRef> refs;
-
- void clear() {
- values.clear();
- refs.clear();
- }
-
- void build_refs() {
- refs.reserve(values.size());
- for (const auto& value : values) {
- refs.emplace_back(value.data(), value.size());
- }
- }
-};
+} // namespace
bool read_dictionary_words(::parquet::ParquetFileReader* file_reader, int
row_group_idx,
int leaf_column_id, const ParquetColumnSchema&
column_schema,
- OwnedDictionaryWords* dict_words) {
+ ParquetDictionaryWords* dict_words) {
DORIS_CHECK(dict_words != nullptr);
dict_words->clear();
if (file_reader == nullptr || leaf_column_id < 0) {
@@ -596,6 +581,17 @@ bool read_dictionary_words(::parquet::ParquetFileReader*
file_reader, int row_gr
return false;
}
+std::vector<Field> dictionary_fields_from_words(const ParquetDictionaryWords&
dict_words) {
+ std::vector<Field> fields;
+ fields.reserve(dict_words.refs.size());
+ for (const auto& ref : dict_words.refs) {
+ fields.push_back(Field::create_field<TYPE_STRING>(String(ref.data,
ref.size)));
+ }
+ return fields;
+}
+
+namespace {
+
const ParquetColumnSchema* resolve_local_leaf_schema(
const std::vector<std::unique_ptr<ParquetColumnSchema>>& schema,
const format::LocalColumnId file_column_id) {
@@ -658,15 +654,6 @@ std::map<int, VExprContextSPtrs>
collect_conjuncts_by_single_slot(
return conjuncts_by_slot;
}
-std::vector<Field> dictionary_fields_from_words(const OwnedDictionaryWords&
dict_words) {
- std::vector<Field> fields;
- fields.reserve(dict_words.refs.size());
- for (const auto& ref : dict_words.refs) {
- fields.push_back(Field::create_field<TYPE_STRING>(String(ref.data,
ref.size)));
- }
- return fields;
-}
-
std::shared_ptr<segment_v2::ZoneMap> make_zonemap_from_statistics(
const ParquetColumnStatistics& statistics) {
if (!statistics.has_null_count && !statistics.has_min_max) {
@@ -785,7 +772,7 @@ ParquetRowGroupPruneReason dictionary_prune_reason(
continue;
}
- OwnedDictionaryWords dict_words;
+ ParquetDictionaryWords dict_words;
if (!read_dictionary_words(file_reader, row_group_idx,
column_schema->leaf_column_id,
*column_schema, &dict_words)) {
continue;
diff --git a/be/src/format_v2/parquet/parquet_statistics.h
b/be/src/format_v2/parquet/parquet_statistics.h
index 55da5769a6c..818d645052c 100644
--- a/be/src/format_v2/parquet/parquet_statistics.h
+++ b/be/src/format_v2/parquet/parquet_statistics.h
@@ -18,10 +18,12 @@
#include <cstdint>
#include <map>
#include <memory>
+#include <string>
#include <vector>
#include "common/status.h"
#include "core/field.h"
+#include "core/string_ref.h"
#include "exprs/vexpr_fwd.h"
#include "format_v2/file_reader.h"
#include "format_v2/parquet/selection_vector.h"
@@ -48,6 +50,36 @@ struct ParquetColumnSchema;
// ============================================================================
// ============================================================================
+struct ParquetDictionaryWords {
+ std::vector<std::string> values;
+ std::vector<StringRef> refs;
+
+ void clear() {
+ values.clear();
+ refs.clear();
+ }
+
+ void build_refs() {
+ refs.clear();
+ refs.reserve(values.size());
+ for (const auto& value : values) {
+ refs.emplace_back(value.data(), value.size());
+ }
+ }
+};
+
+// Reads the PLAIN dictionary page for BYTE_ARRAY/FIXED_LEN_BYTE_ARRAY columns
and owns copied
+// dictionary bytes in `values`. Both row-group pruning and row-level
dictionary predicates use this
+// helper so they agree on dictionary id -> Doris string value mapping.
+bool read_dictionary_words(::parquet::ParquetFileReader* file_reader, int
row_group_idx,
+ int leaf_column_id, const ParquetColumnSchema&
column_schema,
+ ParquetDictionaryWords* dict_words);
+
+std::vector<Field> dictionary_fields_from_words(const ParquetDictionaryWords&
dict_words);
+
+// ============================================================================
+// ============================================================================
+
struct ParquetPruningStats {
int64_t total_row_groups = 0; // total row groups in
the file
int64_t selected_row_groups = 0; // row groups selected
after pruning
diff --git a/be/src/format_v2/parquet/reader/column_reader.cpp
b/be/src/format_v2/parquet/reader/column_reader.cpp
index 9b7577e5521..1b6e66beefe 100644
--- a/be/src/format_v2/parquet/reader/column_reader.cpp
+++ b/be/src/format_v2/parquet/reader/column_reader.cpp
@@ -199,6 +199,13 @@ Status ParquetColumnReader::select(const SelectionVector&
sel, uint16_t selected
return Status::OK();
}
+Status ParquetColumnReader::select_with_dictionary_filter(const
SelectionVector&, uint16_t, int64_t,
+ const
IColumn::Filter&, MutableColumnPtr&,
+ IColumn::Filter*,
bool*) {
+ return Status::NotSupported("Parquet dictionary filter is not implemented
for column {}",
+ name());
+}
+
ParquetColumnReaderFactory::ParquetColumnReaderFactory(
std::shared_ptr<::parquet::RowGroupReader> row_group, int
num_leaf_columns,
const std::map<int, ParquetPageSkipPlan>* page_skip_plans,
@@ -206,6 +213,7 @@ ParquetColumnReaderFactory::ParquetColumnReaderFactory(
bool enable_strict_mode, ParquetColumnReaderProfile
column_reader_profile)
: _row_group(std::move(row_group)),
_record_readers(static_cast<size_t>(num_leaf_columns)),
+ _dictionary_record_readers(static_cast<size_t>(num_leaf_columns)),
_page_skip_plans(page_skip_plans),
_page_skip_profile(page_skip_profile),
_timezone(timezone),
@@ -240,7 +248,7 @@ Status
ParquetColumnReaderFactory::make_scalar_column_reader(
}
Status ParquetColumnReaderFactory::create_scalar_column_reader(
- const ParquetColumnSchema& column_schema, bool is_nested,
+ const ParquetColumnSchema& column_schema, bool is_nested, bool
read_dictionary,
std::unique_ptr<ParquetColumnReader>* reader) const {
if (reader == nullptr) {
return Status::InvalidArgument("reader is null");
@@ -283,14 +291,15 @@ Status
ParquetColumnReaderFactory::create_scalar_column_reader(
// page filtering is also installed, those scratch reads can consume the
next selected row
// after a page-index range gap. Keep page filtering on flat scalar
readers only.
RETURN_IF_ERROR(get_record_reader(column_schema.leaf_column_id,
column_schema.descriptor,
- column_schema.name, !is_nested,
&record_reader));
+ column_schema.name, !is_nested,
read_dictionary,
+ &record_reader));
return make_scalar_column_reader(column_schema, std::move(record_reader),
!is_nested, reader);
}
// 1. RowGroupReader::GetColumnPageReader(leaf_column_id) -> Arrow PageReader
Status ParquetColumnReaderFactory::get_record_reader(
int leaf_column_id, const ::parquet::ColumnDescriptor* descriptor,
const std::string& name,
- bool install_page_filter,
+ bool install_page_filter, bool read_dictionary,
std::shared_ptr<::parquet::internal::RecordReader>* reader) const {
if (reader == nullptr) {
return Status::InvalidArgument("reader is null");
@@ -306,7 +315,8 @@ Status ParquetColumnReaderFactory::get_record_reader(
if (descriptor == nullptr) {
return Status::InvalidArgument("Parquet column descriptor is null for
column {}", name);
}
- if (_record_readers[leaf_column_id] == nullptr) {
+ auto& record_readers = read_dictionary ? _dictionary_record_readers :
_record_readers;
+ if (record_readers[leaf_column_id] == nullptr) {
try {
auto page_reader = _row_group->GetColumnPageReader(leaf_column_id);
if (install_page_filter) {
@@ -314,11 +324,11 @@ Status ParquetColumnReaderFactory::get_record_reader(
_page_skip_profile);
}
const auto level_info =
::parquet::internal::LevelInfo::ComputeLevelInfo(descriptor);
- _record_readers[leaf_column_id] =
::parquet::internal::RecordReader::Make(
+ record_readers[leaf_column_id] =
::parquet::internal::RecordReader::Make(
descriptor, level_info, ::arrow::default_memory_pool(),
- /*read_dictionary=*/false,
+ /*read_dictionary=*/read_dictionary,
/*read_dense_for_nullable=*/false);
-
_record_readers[leaf_column_id]->SetPageReader(std::move(page_reader));
+
record_readers[leaf_column_id]->SetPageReader(std::move(page_reader));
} catch (const ::parquet::ParquetException& e) {
return Status::Corruption("Failed to create parquet record reader
for column {}: {}",
name, e.what());
@@ -327,10 +337,10 @@ Status ParquetColumnReaderFactory::get_record_reader(
name, e.what());
}
}
- if (_record_readers[leaf_column_id] == nullptr) {
+ if (record_readers[leaf_column_id] == nullptr) {
return Status::Corruption("Failed to create parquet record reader for
column {}", name);
}
- *reader = _record_readers[leaf_column_id];
+ *reader = record_readers[leaf_column_id];
return Status::OK();
}
@@ -354,7 +364,8 @@ Status
ParquetColumnReaderFactory::create_struct_column_reader(
continue;
}
std::unique_ptr<ParquetColumnReader> child_reader;
- RETURN_IF_ERROR(create_column_reader(*child_schema, child_projection,
true, &child_reader));
+ RETURN_IF_ERROR(
+ create_column_reader(*child_schema, child_projection, true,
false, &child_reader));
child_output_indices.push_back(static_cast<int>(projected_child_types.size()));
projected_child_types.push_back(make_nullable(child_reader->type()));
projected_child_names.push_back(child_reader->name());
@@ -402,7 +413,7 @@ Status
ParquetColumnReaderFactory::create_list_column_reader(
column_schema.name);
}
RETURN_IF_ERROR(
- create_column_reader(element_schema, element_projection, true,
&element_reader));
+ create_column_reader(element_schema, element_projection, true,
false, &element_reader));
DataTypePtr type = column_schema.type;
if (format::is_partial_projection(element_projection)) {
type = std::make_shared<DataTypeArray>(element_reader->type());
@@ -447,9 +458,10 @@ Status
ParquetColumnReaderFactory::create_map_column_reader(
std::unique_ptr<ParquetColumnReader> key_reader;
// MAP materialization always needs the full key stream. It owns entry
existence, offsets and
// key equality semantics, so MAP projection is defined only as
value-subtree pruning.
- RETURN_IF_ERROR(create_column_reader(key_schema, nullptr, true,
&key_reader));
+ RETURN_IF_ERROR(create_column_reader(key_schema, nullptr, true, false,
&key_reader));
std::unique_ptr<ParquetColumnReader> value_reader;
- RETURN_IF_ERROR(create_column_reader(value_schema, value_projection, true,
&value_reader));
+ RETURN_IF_ERROR(
+ create_column_reader(value_schema, value_projection, true, false,
&value_reader));
DataTypePtr type = column_schema.type;
if (format::is_partial_projection(value_projection)) {
type = std::make_shared<DataTypeMap>(make_nullable(key_reader->type()),
@@ -466,8 +478,9 @@ Status ParquetColumnReaderFactory::create_map_column_reader(
Status ParquetColumnReaderFactory::create(const ParquetColumnSchema&
column_schema,
const format::LocalColumnIndex*
projection,
-
std::unique_ptr<ParquetColumnReader>* reader) const {
- return create_column_reader(column_schema, projection, false, reader);
+
std::unique_ptr<ParquetColumnReader>* reader,
+ bool read_dictionary) const {
+ return create_column_reader(column_schema, projection, false,
read_dictionary, reader);
}
Status ParquetColumnReaderFactory::create_count_shape_reader(
@@ -488,7 +501,7 @@ Status
ParquetColumnReaderFactory::create_count_shape_reader_impl(
return Status::InvalidArgument("Parquet COUNT projection is
invalid for column {}",
column_schema.name);
}
- return create_scalar_column_reader(column_schema, is_nested, reader);
+ return create_scalar_column_reader(column_schema, is_nested, false,
reader);
case ParquetColumnSchemaKind::STRUCT: {
if (column_schema.children.empty()) {
return Status::NotSupported("Parquet COUNT shape reader found
empty STRUCT column {}",
@@ -541,7 +554,7 @@ Status
ParquetColumnReaderFactory::create_count_shape_reader_impl(
Status ParquetColumnReaderFactory::create_column_reader(
const ParquetColumnSchema& column_schema, const
format::LocalColumnIndex* projection,
- bool is_nested, std::unique_ptr<ParquetColumnReader>* reader) const {
+ bool is_nested, bool read_dictionary,
std::unique_ptr<ParquetColumnReader>* reader) const {
if (reader == nullptr) {
return Status::InvalidArgument("reader is null");
}
@@ -552,9 +565,9 @@ Status ParquetColumnReaderFactory::create_column_reader(
return Status::InvalidArgument("Parquet scalar projection is
invalid for column {}",
column_schema.name);
}
- return create_scalar_column_reader(column_schema, true, reader);
+ return create_scalar_column_reader(column_schema, true, false,
reader);
}
- return create_scalar_column_reader(column_schema, false, reader);
+ return create_scalar_column_reader(column_schema, false,
read_dictionary, reader);
case ParquetColumnSchemaKind::STRUCT:
return create_struct_column_reader(column_schema, projection, reader);
case ParquetColumnSchemaKind::LIST:
diff --git a/be/src/format_v2/parquet/reader/column_reader.h
b/be/src/format_v2/parquet/reader/column_reader.h
index f439010e883..1d63e03ca9c 100644
--- a/be/src/format_v2/parquet/reader/column_reader.h
+++ b/be/src/format_v2/parquet/reader/column_reader.h
@@ -72,6 +72,12 @@ public:
virtual Status select(const SelectionVector& sel, uint16_t selected_rows,
int64_t batch_rows,
MutableColumnPtr& column);
+ virtual Status select_with_dictionary_filter(const SelectionVector& sel,
uint16_t selected_rows,
+ int64_t batch_rows,
+ const IColumn::Filter&
dictionary_filter,
+ MutableColumnPtr& column,
+ IColumn::Filter* row_filter,
bool* used_filter);
+
virtual Status load_nested_batch(int64_t rows);
// Shape-only load interface for COUNT(col). Implementations only
guarantee that
@@ -134,7 +140,7 @@ public:
Status create(const ParquetColumnSchema& column_schema,
const format::LocalColumnIndex* projection,
- std::unique_ptr<ParquetColumnReader>* reader) const;
+ std::unique_ptr<ParquetColumnReader>* reader, bool
read_dictionary = false) const;
// Create a scalar reader for one representative leaf that carries the
top-level column shape.
// This is used by COUNT(col): the caller needs definition/repetition
levels to decide whether
@@ -156,6 +162,7 @@ public:
private:
Status create_scalar_column_reader(const ParquetColumnSchema&
column_schema, bool is_nested,
+ bool read_dictionary,
std::unique_ptr<ParquetColumnReader>*
reader) const;
Status create_struct_column_reader(const ParquetColumnSchema&
column_schema,
@@ -172,6 +179,7 @@ private:
Status create_column_reader(const ParquetColumnSchema& column_schema,
const format::LocalColumnIndex* projection,
bool is_nested,
+ bool read_dictionary,
std::unique_ptr<ParquetColumnReader>* reader)
const;
Status create_count_shape_reader_impl(const ParquetColumnSchema&
column_schema,
const format::LocalColumnIndex*
projection,
@@ -180,6 +188,7 @@ private:
Status get_record_reader(int leaf_column_id, const
::parquet::ColumnDescriptor* descriptor,
const std::string& name, bool install_page_filter,
+ bool read_dictionary,
std::shared_ptr<::parquet::internal::RecordReader>* reader) const;
Status make_scalar_column_reader(
@@ -190,6 +199,8 @@ private:
std::shared_ptr<::parquet::RowGroupReader> _row_group; // Arrow RowGroup
reader
mutable std::vector<std::shared_ptr<::parquet::internal::RecordReader>>
_record_readers; // RecordReader cache by leaf_column_id
+ mutable std::vector<std::shared_ptr<::parquet::internal::RecordReader>>
+ _dictionary_record_readers; // dictionary-exposing RecordReader
cache by leaf_column_id
const std::map<int, ParquetPageSkipPlan>* _page_skip_plans =
nullptr; // page-index pruning
result
ParquetPageSkipProfile _page_skip_profile; // page skip profile
diff --git a/be/src/format_v2/parquet/reader/parquet_leaf_reader.cpp
b/be/src/format_v2/parquet/reader/parquet_leaf_reader.cpp
index c157ff84eef..cb59571d141 100644
--- a/be/src/format_v2/parquet/reader/parquet_leaf_reader.cpp
+++ b/be/src/format_v2/parquet/reader/parquet_leaf_reader.cpp
@@ -16,6 +16,7 @@
#include "format_v2/parquet/reader/parquet_leaf_reader.h"
#include <arrow/array/array_binary.h>
+#include <arrow/array/array_dict.h>
#include <parquet/api/schema.h>
#include <parquet/column_reader.h>
#include <parquet/exception.h>
@@ -91,6 +92,18 @@ Status decoded_fixed_value_size(const std::string&
column_name, DecodedValueKind
Status get_binary_chunks(const std::string& column_name,
::parquet::internal::RecordReader& record_reader,
std::vector<std::shared_ptr<::arrow::Array>>* chunks)
{
+ if (auto* dictionary_reader =
+
dynamic_cast<::parquet::internal::DictionaryRecordReader*>(&record_reader);
+ dictionary_reader != nullptr) {
+ auto chunked = dictionary_reader->GetResult();
+ if (chunked == nullptr) {
+ return Status::Corruption(
+ "Parquet dictionary record reader returned null result for
column {}",
+ column_name);
+ }
+ *chunks = chunked->chunks();
+ return Status::OK();
+ }
auto* binary_reader =
dynamic_cast<::parquet::internal::BinaryRecordReader*>(&record_reader);
if (binary_reader == nullptr) {
return Status::InternalError("Parquet binary record reader is not
available for column {}",
@@ -100,6 +113,53 @@ Status get_binary_chunks(const std::string& column_name,
return Status::OK();
}
+Status append_dictionary_binary_values(const std::string& column_name,
+ const ::arrow::DictionaryArray&
dictionary_array,
+ std::vector<StringRef>* values) {
+ DORIS_CHECK(values != nullptr);
+ const auto& dictionary = dictionary_array.dictionary();
+ if (dictionary == nullptr) {
+ return Status::Corruption("Parquet dictionary array has null
dictionary for column {}",
+ column_name);
+ }
+ auto append_value = [&](int64_t dictionary_index) -> Status {
+ if (dictionary_index < 0 || dictionary_index >= dictionary->length()) {
+ return Status::Corruption("Invalid parquet dictionary index {} for
column {}",
+ dictionary_index, column_name);
+ }
+ if (auto* binary_array =
dynamic_cast<::arrow::BinaryArray*>(dictionary.get())) {
+ if (binary_array->IsNull(dictionary_index)) {
+ values->emplace_back(static_cast<const char*>(nullptr), 0);
+ return Status::OK();
+ }
+ int32_t length = 0;
+ const uint8_t* value = binary_array->GetValue(dictionary_index,
&length);
+ values->emplace_back(reinterpret_cast<const char*>(value), length);
+ return Status::OK();
+ }
+ if (auto* fixed_array =
dynamic_cast<::arrow::FixedSizeBinaryArray*>(dictionary.get())) {
+ if (fixed_array->IsNull(dictionary_index)) {
+ values->emplace_back(static_cast<const char*>(nullptr), 0);
+ return Status::OK();
+ }
+ values->emplace_back(
+ reinterpret_cast<const
char*>(fixed_array->GetValue(dictionary_index)),
+ fixed_array->byte_width());
+ return Status::OK();
+ }
+ return Status::InternalError("Unexpected Arrow dictionary value array
type for column {}",
+ column_name);
+ };
+ for (int64_t row_idx = 0; row_idx < dictionary_array.length(); ++row_idx) {
+ if (dictionary_array.IsNull(row_idx)) {
+ values->emplace_back(static_cast<const char*>(nullptr), 0);
+ continue;
+ }
+ RETURN_IF_ERROR(append_value(dictionary_array.GetValueIndex(row_idx)));
+ }
+ return Status::OK();
+}
+
Status build_binary_values(const std::string& column_name,
const std::vector<std::shared_ptr<::arrow::Array>>&
chunks,
int64_t records_read, const NullMap* null_map,
@@ -131,6 +191,9 @@ Status build_binary_values(const std::string& column_name,
values->emplace_back(reinterpret_cast<const
char*>(fixed_array->GetValue(row_idx)),
fixed_array->byte_width());
}
+ } else if (auto* dictionary_array =
dynamic_cast<::arrow::DictionaryArray*>(chunk.get())) {
+ RETURN_IF_ERROR(
+ append_dictionary_binary_values(column_name,
*dictionary_array, values));
} else {
return Status::InternalError("Unexpected Arrow binary array type
for column {}",
column_name);
diff --git a/be/src/format_v2/parquet/reader/parquet_leaf_reader.h
b/be/src/format_v2/parquet/reader/parquet_leaf_reader.h
index 73b0a75e019..7d97d0ad698 100644
--- a/be/src/format_v2/parquet/reader/parquet_leaf_reader.h
+++ b/be/src/format_v2/parquet/reader/parquet_leaf_reader.h
@@ -73,6 +73,9 @@ public:
bool read_dense_for_nullable() const { return _read_dense_for_nullable; }
const int16_t* def_levels() const { return _def_levels; }
const int16_t* rep_levels() const { return _rep_levels; }
+ const std::vector<std::shared_ptr<::arrow::Array>>& binary_chunks() const {
+ return _binary_chunks;
+ }
private:
friend class ParquetLeafReader;
diff --git a/be/src/format_v2/parquet/reader/scalar_column_reader.cpp
b/be/src/format_v2/parquet/reader/scalar_column_reader.cpp
index 3c90279b441..6e3b1c7f4d5 100644
--- a/be/src/format_v2/parquet/reader/scalar_column_reader.cpp
+++ b/be/src/format_v2/parquet/reader/scalar_column_reader.cpp
@@ -15,6 +15,8 @@
#include "format_v2/parquet/reader/scalar_column_reader.h"
+#include <arrow/array/array_binary.h>
+#include <arrow/array/array_dict.h>
#include <parquet/api/reader.h>
#include <algorithm>
@@ -23,6 +25,8 @@
#include "core/column/column.h"
#include "core/column/column_nullable.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type_serde/decoded_column_view.h"
#include "format_v2/parquet/parquet_column_schema.h"
#include "util/simd/bits.h"
@@ -79,6 +83,38 @@ Status append_scalar_batch_value(const ScalarColumnReader&
column_reader,
return Status::OK();
}
+Status append_arrow_binary_dictionary_value(const std::string& column_name,
+ const ::arrow::Array& dictionary,
+ int64_t dictionary_index,
+ std::vector<StringRef>* values) {
+ DORIS_CHECK(values != nullptr);
+ if (dictionary_index < 0 || dictionary_index >= dictionary.length()) {
+ return Status::Corruption("Invalid parquet dictionary index {} for
column {}",
+ dictionary_index, column_name);
+ }
+ if (auto* binary_array = dynamic_cast<const
::arrow::BinaryArray*>(&dictionary)) {
+ if (binary_array->IsNull(dictionary_index)) {
+ values->emplace_back(static_cast<const char*>(nullptr), 0);
+ return Status::OK();
+ }
+ int32_t length = 0;
+ const uint8_t* value = binary_array->GetValue(dictionary_index,
&length);
+ values->emplace_back(reinterpret_cast<const char*>(value), length);
+ return Status::OK();
+ }
+ if (auto* fixed_array = dynamic_cast<const
::arrow::FixedSizeBinaryArray*>(&dictionary)) {
+ if (fixed_array->IsNull(dictionary_index)) {
+ values->emplace_back(static_cast<const char*>(nullptr), 0);
+ return Status::OK();
+ }
+ values->emplace_back(reinterpret_cast<const
char*>(fixed_array->GetValue(dictionary_index)),
+ fixed_array->byte_width());
+ return Status::OK();
+ }
+ return Status::InternalError("Unexpected Arrow dictionary value array type
for column {}",
+ column_name);
+}
+
} // namespace
ScalarColumnReader::ScalarColumnReader(
@@ -215,6 +251,174 @@ Status ScalarColumnReader::skip(int64_t rows) {
return Status::OK();
}
+Status ScalarColumnReader::select_with_dictionary_filter(const
SelectionVector& sel,
+ uint16_t
selected_rows, int64_t batch_rows,
+ const
IColumn::Filter& dictionary_filter,
+ MutableColumnPtr&
column,
+ IColumn::Filter*
row_filter,
+ bool* used_filter) {
+ DORIS_CHECK(column.get() != nullptr);
+ DORIS_CHECK(row_filter != nullptr);
+ DORIS_CHECK(used_filter != nullptr);
+ RETURN_IF_ERROR(sel.verify(selected_rows, batch_rows));
+ *used_filter = false;
+ row_filter->clear();
+ row_filter->reserve(selected_rows);
+
+ const auto ranges = selection_to_ranges(sel, selected_rows);
+ int64_t cursor = 0;
+ for (const auto& range : ranges) {
+ if (range.start < cursor || range.start + range.length > batch_rows) {
+ return Status::InvalidArgument(
+ "Invalid parquet dictionary selection range [{}, {}) for
column {}",
+ range.start, range.start + range.length, _name);
+ }
+ RETURN_IF_ERROR(skip(range.start - cursor));
+
+ int64_t range_rows_read = 0;
+ RETURN_IF_ERROR(read_range_with_dictionary_filter(range.length,
dictionary_filter, column,
+ row_filter,
&range_rows_read,
+ used_filter));
+ if (!*used_filter) {
+ return Status::OK();
+ }
+ if (range_rows_read != range.length) {
+ return Status::Corruption(
+ "Parquet dictionary selected read returned {} rows,
expected {} rows for "
+ "column {}",
+ range_rows_read, range.length, _name);
+ }
+ cursor = range.start + range.length;
+ }
+ RETURN_IF_ERROR(skip(batch_rows - cursor));
+ if (_profile.reader_select_rows != nullptr) {
+ COUNTER_UPDATE(_profile.reader_select_rows, selected_rows);
+ }
+ return Status::OK();
+}
+
+Status ScalarColumnReader::read_range_with_dictionary_filter(
+ int64_t rows, const IColumn::Filter& dictionary_filter,
MutableColumnPtr& column,
+ IColumn::Filter* row_filter, int64_t* rows_read, bool* used_filter) {
+ DORIS_CHECK(row_filter != nullptr);
+ DORIS_CHECK(rows_read != nullptr);
+ DORIS_CHECK(used_filter != nullptr);
+ DORIS_CHECK(_record_reader != nullptr);
+ if (!_record_reader->read_dictionary()) {
+ *used_filter = false;
+ return Status::OK();
+ }
+
+ ParquetLeafBatch leaf_batch;
+ RETURN_IF_ERROR(leaf_reader().read_batch(rows, &leaf_batch, rows_read));
+ int64_t matched_rows = 0;
+
RETURN_IF_ERROR(append_dictionary_filtered_values(leaf_batch.binary_chunks(),
dictionary_filter,
+ column, row_filter,
&matched_rows,
+ used_filter));
+ if (!*used_filter) {
+ return Status::Corruption(
+ "Parquet dictionary reader did not return dictionary batches
for column {}", _name);
+ }
+ if (row_filter->size() < static_cast<size_t>(*rows_read)) {
+ return Status::Corruption(
+ "Parquet dictionary filter produced too few row decisions for
column {}: "
+ "filter={}, rows={}",
+ _name, row_filter->size(), *rows_read);
+ }
+ advance_rows_read(*rows_read);
+ update_reader_read_rows(*rows_read);
+ return Status::OK();
+}
+
+Status ScalarColumnReader::append_dictionary_filtered_values(
+ const std::vector<std::shared_ptr<::arrow::Array>>& chunks,
+ const IColumn::Filter& dictionary_filter, MutableColumnPtr& column,
+ IColumn::Filter* row_filter, int64_t* matched_rows, bool* used_filter)
const {
+ DORIS_CHECK(row_filter != nullptr);
+ DORIS_CHECK(matched_rows != nullptr);
+ DORIS_CHECK(used_filter != nullptr);
+ *matched_rows = 0;
+ *used_filter = false;
+
+ std::vector<StringRef> selected_values;
+ for (const auto& chunk : chunks) {
+ DORIS_CHECK(chunk != nullptr);
+ const auto* dict_array = dynamic_cast<const
::arrow::DictionaryArray*>(chunk.get());
+ if (dict_array == nullptr) {
+ // The caller has already consumed rows from a
DictionaryRecordReader. Falling back to a
+ // normal selected read would desynchronize the Parquet stream, so
absence of a
+ // DictionaryArray is reported as corruption by
read_range_with_dictionary_filter().
+ return Status::OK();
+ }
+ *used_filter = true;
+ const auto& dictionary = dict_array->dictionary();
+ if (dictionary == nullptr) {
+ return Status::Corruption("Parquet dictionary array has null
dictionary for column {}",
+ _name);
+ }
+
+ // Dictionary predicates are evaluated once against the dictionary
page and produce a
+ // dictionary-entry bitmap. DATA_PAGE rows then only need an
integer-index lookup. NULL rows
+ // do not have a dictionary entry and cannot satisfy the supported
equality/IN predicates.
+ for (int64_t row = 0; row < dict_array->length(); ++row) {
+ bool keep = false;
+ if (!dict_array->IsNull(row)) {
+ const int64_t dictionary_index =
dict_array->GetValueIndex(row);
+ if (dictionary_index >= 0 &&
+ dictionary_index <
static_cast<int64_t>(dictionary_filter.size())) {
+ keep =
dictionary_filter[static_cast<size_t>(dictionary_index)] != 0;
+ }
+ if (keep) {
+ RETURN_IF_ERROR(append_arrow_binary_dictionary_value(
+ _name, *dictionary, dictionary_index,
&selected_values));
+ ++*matched_rows;
+ }
+ }
+ row_filter->push_back(keep ? 1 : 0);
+ }
+ }
+
+ if (!*used_filter) {
+ return Status::OK();
+ }
+ return append_decoded_binary_values(selected_values, column);
+}
+
+Status ScalarColumnReader::append_decoded_binary_values(const
std::vector<StringRef>& values,
+ MutableColumnPtr&
column) const {
+ DecodedColumnView view;
+ view.value_kind = decoded_value_kind(_type_descriptor);
+ view.row_count = static_cast<int64_t>(values.size());
+ view.logical_integer_bit_width = _type_descriptor.integer_bit_width;
+ view.logical_integer_is_signed = !_type_descriptor.is_unsigned_integer;
+ view.fixed_length = _type_descriptor.fixed_length;
+ view.binary_values = &values;
+
+ SCOPED_TIMER(_profile.materialization_time);
+ if (!_type->is_nullable()) {
+ if (auto* nullable_column =
check_and_get_column<ColumnNullable>(*column);
+ nullable_column != nullptr) {
+ auto& nested_column = nullable_column->get_nested_column();
+ auto& null_map = nullable_column->get_null_map_data();
+ const auto old_nested_size = nested_column.size();
+ const auto old_null_map_size = null_map.size();
+ auto st =
_type->get_serde()->read_column_from_decoded_values(nested_column, view);
+ if (!st.ok()) {
+ nested_column.resize(old_nested_size);
+ return st;
+ }
+ null_map.resize(old_null_map_size + nested_column.size() -
old_nested_size);
+ memset(null_map.data() + old_null_map_size, 0, null_map.size() -
old_null_map_size);
+ return Status::OK();
+ }
+ return _type->get_serde()->read_column_from_decoded_values(*column,
view);
+ }
+
+ NullMap null_map(values.size(), 0);
+ view.null_map = null_map.empty() ? nullptr : null_map.data();
+ return _type->get_serde()->read_column_from_decoded_values(*column, view);
+}
+
// The value index stream must advance on those null slots, otherwise later
payload values shift.
Status ScalarColumnReader::load_nested_batch(int64_t rows) {
DORIS_CHECK(_nested_batch != nullptr);
diff --git a/be/src/format_v2/parquet/reader/scalar_column_reader.h
b/be/src/format_v2/parquet/reader/scalar_column_reader.h
index ab7ba0d7e54..99fec69fa3a 100644
--- a/be/src/format_v2/parquet/reader/scalar_column_reader.h
+++ b/be/src/format_v2/parquet/reader/scalar_column_reader.h
@@ -17,7 +17,9 @@
#include <memory>
#include <string>
+#include <vector>
+#include "core/string_ref.h"
#include "format_v2/parquet/parquet_type.h"
#include "format_v2/parquet/reader/column_reader.h"
#include "format_v2/parquet/reader/parquet_leaf_reader.h"
@@ -53,6 +55,11 @@ public:
Status read(int64_t rows, MutableColumnPtr& column, int64_t* rows_read)
override;
Status skip(int64_t rows) override;
+ Status select_with_dictionary_filter(const SelectionVector& sel, uint16_t
selected_rows,
+ int64_t batch_rows,
+ const IColumn::Filter&
dictionary_filter,
+ MutableColumnPtr& column,
IColumn::Filter* row_filter,
+ bool* used_filter) override;
Status load_nested_batch(int64_t rows) override;
Status load_nested_levels_batch(int64_t rows) override;
@@ -65,6 +72,15 @@ public:
private:
Status append_nested_value(int64_t level_idx, MutableColumnPtr& column)
const;
+ Status read_range_with_dictionary_filter(int64_t rows, const
IColumn::Filter& dictionary_filter,
+ MutableColumnPtr& column,
IColumn::Filter* row_filter,
+ int64_t* rows_read, bool*
used_filter);
+ Status append_dictionary_filtered_values(
+ const std::vector<std::shared_ptr<::arrow::Array>>& chunks,
+ const IColumn::Filter& dictionary_filter, MutableColumnPtr& column,
+ IColumn::Filter* row_filter, int64_t* matched_rows, bool*
used_filter) const;
+ Status append_decoded_binary_values(const std::vector<StringRef>& values,
+ MutableColumnPtr& column) const;
const ::parquet::ColumnDescriptor* descriptor() const { return
_descriptor; }
diff --git a/be/test/exprs/try_cast_expr_test.cpp
b/be/test/exprs/try_cast_expr_test.cpp
index b7ad38b2eb0..d176c6cd6c4 100644
--- a/be/test/exprs/try_cast_expr_test.cpp
+++ b/be/test/exprs/try_cast_expr_test.cpp
@@ -277,4 +277,12 @@ TEST_F(TryCastExprTest, row_exec3) {
EXPECT_FALSE(st.ok()) << st.msg();
}
-} // namespace doris
\ No newline at end of file
+TEST_F(TryCastExprTest, selected_row_safety) {
+ VCastExpr cast_expr;
+ cast_expr.add_child(std::make_shared<MockVExprForTryCast>());
+
+ EXPECT_FALSE(cast_expr.is_safe_to_execute_on_selected_rows());
+ EXPECT_TRUE(try_cast_expr.is_safe_to_execute_on_selected_rows());
+}
+
+} // namespace doris
diff --git a/be/test/format_v2/parquet/parquet_reader_test.cpp
b/be/test/format_v2/parquet/parquet_reader_test.cpp
index 8ebe266f1ac..6269a9fc2a7 100644
--- a/be/test/format_v2/parquet/parquet_reader_test.cpp
+++ b/be/test/format_v2/parquet/parquet_reader_test.cpp
@@ -48,6 +48,7 @@
#include "core/data_type/data_type_struct.h"
#include "core/data_type/primitive_type.h"
#include "core/field.h"
+#include "exprs/vcompound_pred.h"
#include "exprs/vexpr.h"
#include "exprs/vexpr_context.h"
#include "exprs/vslot_ref.h"
@@ -179,6 +180,72 @@ private:
const std::string _expr_name = "Int32SumGreaterThanExpr";
};
+class NonDeterministicCountingInt32Expr final : public VExpr {
+public:
+ NonDeterministicCountingInt32Expr(int column_id, std::vector<size_t>*
executed_rows)
+ : VExpr(std::make_shared<DataTypeUInt8>(), false),
+ _column_id(column_id),
+ _executed_rows(executed_rows) {}
+
+ Status execute_column_impl(VExprContext* context, const Block* block,
const Selector* selector,
+ size_t count, ColumnPtr& result_column) const
override {
+ DORIS_CHECK(_executed_rows != nullptr);
+ DORIS_CHECK(block != nullptr);
+ (void)nullable_nested_column<ColumnInt32>(*block, _column_id);
+ _executed_rows->push_back(count);
+ auto result = ColumnUInt8::create();
+ result->get_data().resize_fill(count, 1);
+ result_column = std::move(result);
+ return Status::OK();
+ }
+
+ const std::string& expr_name() const override { return _expr_name; }
+
+ bool is_deterministic() const override { return false; }
+
+ void collect_slot_column_ids(std::set<int>& column_ids) const override {
+ column_ids.insert(_column_id);
+ }
+
+private:
+ const int _column_id;
+ std::vector<size_t>* const _executed_rows;
+ const std::string _expr_name = "NonDeterministicCountingInt32Expr";
+};
+
+class SelectedRowsUnsafeCountingInt32Expr final : public VExpr {
+public:
+ SelectedRowsUnsafeCountingInt32Expr(int column_id, std::vector<size_t>*
executed_rows)
+ : VExpr(std::make_shared<DataTypeUInt8>(), false),
+ _column_id(column_id),
+ _executed_rows(executed_rows) {}
+
+ Status execute_column_impl(VExprContext* context, const Block* block,
const Selector* selector,
+ size_t count, ColumnPtr& result_column) const
override {
+ DORIS_CHECK(_executed_rows != nullptr);
+ DORIS_CHECK(block != nullptr);
+ (void)nullable_nested_column<ColumnInt32>(*block, _column_id);
+ _executed_rows->push_back(count);
+ auto result = ColumnUInt8::create();
+ result->get_data().resize_fill(count, 1);
+ result_column = std::move(result);
+ return Status::OK();
+ }
+
+ const std::string& expr_name() const override { return _expr_name; }
+
+ bool is_safe_to_execute_on_selected_rows() const override { return false; }
+
+ void collect_slot_column_ids(std::set<int>& column_ids) const override {
+ column_ids.insert(_column_id);
+ }
+
+private:
+ const int _column_id;
+ std::vector<size_t>* const _executed_rows;
+ const std::string _expr_name = "SelectedRowsUnsafeCountingInt32Expr";
+};
+
class StringInExpr final : public VExpr {
public:
StringInExpr(int column_id, std::vector<std::string> values)
@@ -232,6 +299,75 @@ private:
const std::string _expr_name = "StringInExpr";
};
+class StringEqualsExpr final : public VExpr {
+public:
+ StringEqualsExpr(int column_id, std::string row_value)
+ : VExpr(std::make_shared<DataTypeUInt8>(), false),
+ _column_id(column_id),
+ _row_value(std::move(row_value)) {}
+
+ Status execute_column_impl(VExprContext* context, const Block* block,
const Selector* selector,
+ size_t count, ColumnPtr& result_column) const
override {
+ const auto& input = nullable_nested_column<ColumnString>(*block,
_column_id);
+ auto result = ColumnUInt8::create();
+ auto& result_data = result->get_data();
+ result_data.resize(count);
+ for (size_t row = 0; row < count; ++row) {
+ const size_t input_row = selector == nullptr ? row :
(*selector)[row];
+ result_data[row] = input.get_data_at(input_row).to_string() ==
_row_value;
+ }
+ result_column = std::move(result);
+ return Status::OK();
+ }
+
+ const std::string& expr_name() const override { return _expr_name; }
+
+ void collect_slot_column_ids(std::set<int>& column_ids) const override {
+ column_ids.insert(_column_id);
+ }
+
+private:
+ const int _column_id;
+ const std::string _row_value;
+ const std::string _expr_name = "StringEqualsExpr";
+};
+
+class StringEqualsOrLengthEqualsExpr final : public VExpr {
+public:
+ StringEqualsOrLengthEqualsExpr(int column_id, std::string row_value,
size_t length)
+ : VExpr(std::make_shared<DataTypeUInt8>(), false),
+ _column_id(column_id),
+ _row_value(std::move(row_value)),
+ _length(length) {}
+
+ Status execute_column_impl(VExprContext* context, const Block* block,
const Selector* selector,
+ size_t count, ColumnPtr& result_column) const
override {
+ const auto& input = nullable_nested_column<ColumnString>(*block,
_column_id);
+ auto result = ColumnUInt8::create();
+ auto& result_data = result->get_data();
+ result_data.resize(count);
+ for (size_t row = 0; row < count; ++row) {
+ const size_t input_row = selector == nullptr ? row :
(*selector)[row];
+ const auto value = input.get_data_at(input_row);
+ result_data[row] = value.to_string() == _row_value || value.size
== _length;
+ }
+ result_column = std::move(result);
+ return Status::OK();
+ }
+
+ const std::string& expr_name() const override { return _expr_name; }
+
+ void collect_slot_column_ids(std::set<int>& column_ids) const override {
+ column_ids.insert(_column_id);
+ }
+
+private:
+ const int _column_id;
+ const std::string _row_value;
+ const size_t _length;
+ const std::string _expr_name = "StringEqualsOrLengthEqualsExpr";
+};
+
VExprContextSPtr create_int32_greater_than_conjunct(int column_id, int32_t
value) {
auto ctx =
VExprContext::create_shared(std::make_shared<Int32GreaterThanExpr>(column_id,
value));
@@ -249,6 +385,24 @@ VExprContextSPtr
create_int32_sum_greater_than_conjunct(int left_column_id, int
return ctx;
}
+VExprContextSPtr create_non_deterministic_counting_int32_conjunct(
+ int column_id, std::vector<size_t>* executed_rows) {
+ auto ctx = VExprContext::create_shared(
+ std::make_shared<NonDeterministicCountingInt32Expr>(column_id,
executed_rows));
+ ctx->_prepared = true;
+ ctx->_opened = true;
+ return ctx;
+}
+
+VExprContextSPtr create_selected_rows_unsafe_counting_int32_conjunct(
+ int column_id, std::vector<size_t>* executed_rows) {
+ auto ctx = VExprContext::create_shared(
+ std::make_shared<SelectedRowsUnsafeCountingInt32Expr>(column_id,
executed_rows));
+ ctx->_prepared = true;
+ ctx->_opened = true;
+ return ctx;
+}
+
VExprContextSPtr create_string_in_conjunct(int column_id,
std::vector<std::string> values) {
auto ctx = VExprContext::create_shared(
std::make_shared<StringInExpr>(column_id, std::move(values)));
@@ -257,6 +411,39 @@ VExprContextSPtr create_string_in_conjunct(int column_id,
std::vector<std::strin
return ctx;
}
+TExprNode make_compound_node(TExprOpcode::type opcode, int num_children) {
+ TExprNode node;
+ node.__set_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN));
+ node.__set_node_type(TExprNodeType::COMPOUND_PRED);
+ node.__set_opcode(opcode);
+ node.__set_num_children(num_children);
+ node.__set_is_nullable(false);
+ return node;
+}
+
+VExprContextSPtr create_string_dictionary_and_residual_conjunct(
+ int column_id, std::vector<std::string> dictionary_values, std::string
row_value) {
+ auto compound =
VCompoundPred::create_shared(make_compound_node(TExprOpcode::COMPOUND_AND, 2));
+ compound->add_child(std::make_shared<StringInExpr>(column_id,
std::move(dictionary_values)));
+ compound->add_child(std::make_shared<StringEqualsExpr>(column_id,
std::move(row_value)));
+ auto ctx = VExprContext::create_shared(std::move(compound));
+ ctx->_prepared = true;
+ ctx->_opened = true;
+ return ctx;
+}
+
+VExprContextSPtr create_nested_or_dictionary_and_residual_conjunct(int
column_id) {
+ auto root =
VCompoundPred::create_shared(make_compound_node(TExprOpcode::COMPOUND_AND, 2));
+ root->add_child(
+ std::make_shared<StringInExpr>(column_id, std::vector<std::string>
{"az", "za"}));
+
root->add_child(std::make_shared<StringEqualsOrLengthEqualsExpr>(column_id,
"az", 1));
+
+ auto ctx = VExprContext::create_shared(std::move(root));
+ ctx->_prepared = true;
+ ctx->_opened = true;
+ return ctx;
+}
+
std::shared_ptr<arrow::Array> finish_array(arrow::ArrayBuilder* builder) {
std::shared_ptr<arrow::Array> array;
EXPECT_TRUE(builder->Finish(&array).ok());
@@ -567,6 +754,57 @@ void write_dictionary_filter_parquet_file(const
std::string& file_path) {
builder.build()));
}
+void write_single_row_group_dictionary_filter_parquet_file(const std::string&
file_path) {
+ auto schema = arrow::schema({
+ arrow::field("id", arrow::int32(), false),
+ arrow::field("value", arrow::utf8(), false),
+ });
+ auto table =
+ arrow::Table::Make(schema, {build_int32_array({1, 2, 3, 4, 5, 6}),
+ build_string_array({"aa", "az", "lm",
"lz", "za", "zz"})});
+
+ auto file_result = arrow::io::FileOutputStream::Open(file_path);
+ ASSERT_TRUE(file_result.ok()) << file_result.status();
+ std::shared_ptr<arrow::io::FileOutputStream> out = *file_result;
+
+ ::parquet::WriterProperties::Builder builder;
+ builder.version(::parquet::ParquetVersion::PARQUET_2_6);
+ builder.data_page_version(::parquet::ParquetDataPageVersion::V2);
+ builder.compression(::parquet::Compression::UNCOMPRESSED);
+ builder.enable_dictionary("value");
+ builder.disable_dictionary("id");
+ builder.disable_statistics();
+ PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table,
arrow::default_memory_pool(), out, 6,
+ builder.build()));
+}
+
+void write_dictionary_filter_with_trailing_column_parquet_file(const
std::string& file_path) {
+ auto schema = arrow::schema({
+ arrow::field("id", arrow::int32(), false),
+ arrow::field("value", arrow::utf8(), false),
+ arrow::field("payload", arrow::int32(), false),
+ });
+ auto table =
+ arrow::Table::Make(schema, {build_int32_array({1, 2, 3, 4, 5, 6}),
+ build_string_array({"aa", "az", "lm",
"lz", "za", "zz"}),
+ build_int32_array({10, 20, 30, 40, 50,
60})});
+
+ auto file_result = arrow::io::FileOutputStream::Open(file_path);
+ ASSERT_TRUE(file_result.ok()) << file_result.status();
+ std::shared_ptr<arrow::io::FileOutputStream> out = *file_result;
+
+ ::parquet::WriterProperties::Builder builder;
+ builder.version(::parquet::ParquetVersion::PARQUET_2_6);
+ builder.data_page_version(::parquet::ParquetDataPageVersion::V2);
+ builder.compression(::parquet::Compression::UNCOMPRESSED);
+ builder.disable_dictionary("id");
+ builder.enable_dictionary("value");
+ builder.disable_dictionary("payload");
+ builder.disable_statistics();
+ PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table,
arrow::default_memory_pool(), out, 6,
+ builder.build()));
+}
+
void write_nested_dictionary_filter_parquet_file(const std::string& file_path)
{
auto id_field = arrow::field("id", arrow::int32(), false);
auto name_field = arrow::field("name", arrow::utf8(), false);
@@ -1459,6 +1697,91 @@ TEST_F(NewParquetReaderTest,
ReadMultiPredicateColumnsBeforeExpressionFilter) {
EXPECT_EQ(scores.get_element(1), 5);
}
+TEST_F(NewParquetReaderTest,
NonDeterministicPredicateKeepsFullBatchEvaluation) {
+ write_int_pair_parquet_file(_file_path);
+ RuntimeProfile
profile("new_parquet_reader_non_deterministic_predicate_profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ Block block = build_file_block(schema);
+
+ std::vector<size_t> non_deterministic_executed_rows;
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->predicate_columns = {field_projection(0), field_projection(1)};
+ request->conjuncts.push_back(create_int32_greater_than_conjunct(0, 2));
+ request->conjuncts.push_back(
+ create_non_deterministic_counting_int32_conjunct(1,
&non_deterministic_executed_rows));
+ ASSERT_TRUE(reader->open(request).ok());
+
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ EXPECT_FALSE(eof);
+ ASSERT_EQ(rows, 3);
+
+ const auto& ids = nullable_nested_column<ColumnInt32>(block, 0);
+ const auto& scores = nullable_nested_column<ColumnInt32>(block, 1);
+ EXPECT_EQ(ids.get_element(0), 3);
+ EXPECT_EQ(ids.get_element(1), 4);
+ EXPECT_EQ(ids.get_element(2), 5);
+ EXPECT_EQ(scores.get_element(0), 3);
+ EXPECT_EQ(scores.get_element(1), 4);
+ EXPECT_EQ(scores.get_element(2), 5);
+
+ // A non-deterministic predicate must stay on the old full-batch path. If
it were left as a
+ // remaining conjunct while earlier deterministic predicates compacted
later predicate columns,
+ // this expression would only see the three surviving rows instead of the
original five.
+ EXPECT_EQ(non_deterministic_executed_rows,
+ std::vector<size_t>({static_cast<size_t>(ROW_COUNT)}));
+ ASSERT_NE(profile.get_counter("ReaderSelectRows"), nullptr);
+ EXPECT_EQ(profile.get_counter("ReaderSelectRows")->value(), 0);
+}
+
+TEST_F(NewParquetReaderTest,
SelectedRowsUnsafePredicateKeepsFullBatchEvaluation) {
+ write_int_pair_parquet_file(_file_path);
+ RuntimeProfile
profile("new_parquet_reader_selected_rows_unsafe_predicate_profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ Block block = build_file_block(schema);
+
+ std::vector<size_t> unsafe_executed_rows;
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->predicate_columns = {field_projection(0), field_projection(1)};
+ request->conjuncts.push_back(create_int32_greater_than_conjunct(0, 2));
+ request->conjuncts.push_back(
+ create_selected_rows_unsafe_counting_int32_conjunct(1,
&unsafe_executed_rows));
+ ASSERT_TRUE(reader->open(request).ok());
+
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ EXPECT_FALSE(eof);
+ ASSERT_EQ(rows, 3);
+
+ const auto& ids = nullable_nested_column<ColumnInt32>(block, 0);
+ const auto& scores = nullable_nested_column<ColumnInt32>(block, 1);
+ EXPECT_EQ(ids.get_element(0), 3);
+ EXPECT_EQ(ids.get_element(1), 4);
+ EXPECT_EQ(ids.get_element(2), 5);
+ EXPECT_EQ(scores.get_element(0), 3);
+ EXPECT_EQ(scores.get_element(1), 4);
+ EXPECT_EQ(scores.get_element(2), 5);
+
+ // Error-preserving functions such as assert_true are deterministic, but
moving them after an
+ // earlier predicate's compacted selection can hide errors from rows
filtered by that earlier
+ // predicate. Such conjuncts therefore keep the old full-batch execution
path.
+ EXPECT_EQ(unsafe_executed_rows,
std::vector<size_t>({static_cast<size_t>(ROW_COUNT)}));
+ ASSERT_NE(profile.get_counter("ReaderSelectRows"), nullptr);
+ EXPECT_EQ(profile.get_counter("ReaderSelectRows")->value(), 0);
+}
+
TEST_F(NewParquetReaderTest, PredicateColumnFiltersBeforeNonPredicateRead) {
auto reader = create_reader();
RuntimeState state {TQueryOptions(), TQueryGlobals()};
@@ -1637,6 +1960,272 @@ TEST_F(NewParquetReaderTest,
PredicateFiltersRowGroupsByDictionary) {
EXPECT_EQ(values, std::vector<std::string>({"lm"}));
}
+TEST_F(NewParquetReaderTest, DictionaryPredicateFiltersRowsInsideRowGroup) {
+ write_single_row_group_dictionary_filter_parquet_file(_file_path);
+ auto parquet_file_reader =
::parquet::ParquetFileReader::OpenFile(_file_path, false);
+ ASSERT_EQ(parquet_file_reader->metadata()->num_row_groups(), 1);
+ auto row_group = parquet_file_reader->metadata()->RowGroup(0);
+ ASSERT_NE(row_group, nullptr);
+ ASSERT_TRUE(row_group->ColumnChunk(1)->has_dictionary_page());
+
+ RuntimeProfile profile("new_parquet_reader_dictionary_filter_profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->predicate_columns = {field_projection(1)};
+ request->non_predicate_columns = {field_projection(0)};
+ request->conjuncts.push_back(create_string_in_conjunct(1, {"az", "za"}));
+ use_schema_order_positions(request.get(), schema);
+ ASSERT_TRUE(reader->open(request).ok());
+
+ std::vector<int32_t> ids;
+ std::vector<std::string> values;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ if (rows == 0) {
+ continue;
+ }
+ const auto& id_column = nullable_nested_column<ColumnInt32>(block, 0);
+ const auto& value_column = nullable_nested_column<ColumnString>(block,
1);
+ for (size_t row = 0; row < rows; ++row) {
+ ids.push_back(id_column.get_element(row));
+ values.push_back(value_column.get_data_at(row).to_string());
+ }
+ }
+
+ EXPECT_EQ(ids, std::vector<int32_t>({2, 5}));
+ EXPECT_EQ(values, std::vector<std::string>({"az", "za"}));
+ EXPECT_EQ(profile.get_counter("RowsFilteredByConjunct")->value(), 4);
+ EXPECT_EQ(profile.get_counter("RowsFilteredByDictFilter")->value(), 4);
+ EXPECT_EQ(profile.get_counter("DictFilterCandidateColumns")->value(), 1);
+ EXPECT_EQ(profile.get_counter("DictFilterColumns")->value(), 1);
+ EXPECT_EQ(profile.get_counter("DictFilterUnsupportedColumns")->value(), 0);
+ EXPECT_EQ(profile.get_counter("DictFilterReadFailures")->value(), 0);
+ ASSERT_NE(profile.get_counter("DictFilterExprRewriteTime"), nullptr);
+ ASSERT_NE(profile.get_counter("DictFilterReadDictTime"), nullptr);
+ ASSERT_NE(profile.get_counter("DictFilterBuildTime"), nullptr);
+ EXPECT_EQ(profile.get_counter("SelectedRows")->value(), 2);
+ EXPECT_GE(profile.get_counter("ReaderSelectRows")->value(), 8);
+}
+
+TEST_F(NewParquetReaderTest,
DictionaryPredicateProbeDoesNotUseMergeRangeReader) {
+ write_dictionary_filter_with_trailing_column_parquet_file(_file_path);
+
+ RuntimeProfile
profile("new_parquet_reader_dictionary_filter_merge_profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->predicate_columns = {field_projection(1)};
+ request->non_predicate_columns = {field_projection(0),
field_projection(2)};
+ request->conjuncts.push_back(create_string_in_conjunct(1, {"az", "za"}));
+ use_schema_order_positions(request.get(), schema);
+ ASSERT_TRUE(reader->open(request).ok());
+
+ std::vector<int32_t> ids;
+ std::vector<std::string> values;
+ std::vector<int32_t> payloads;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ if (rows == 0) {
+ continue;
+ }
+ const auto& id_column = nullable_nested_column<ColumnInt32>(block, 0);
+ const auto& value_column = nullable_nested_column<ColumnString>(block,
1);
+ const auto& payload_column =
nullable_nested_column<ColumnInt32>(block, 2);
+ for (size_t row = 0; row < rows; ++row) {
+ ids.push_back(id_column.get_element(row));
+ values.push_back(value_column.get_data_at(row).to_string());
+ payloads.push_back(payload_column.get_element(row));
+ }
+ }
+
+ EXPECT_EQ(ids, std::vector<int32_t>({2, 5}));
+ EXPECT_EQ(values, std::vector<std::string>({"az", "za"}));
+ EXPECT_EQ(payloads, std::vector<int32_t>({20, 50}));
+ EXPECT_EQ(profile.get_counter("RowsFilteredByDictFilter")->value(), 4);
+ ASSERT_NE(profile.get_counter("MergedIO"), nullptr);
+ ASSERT_NE(profile.get_counter("MergedBytes"), nullptr);
+}
+
+TEST_F(NewParquetReaderTest, DictionaryPredicateWorksWithoutRuntimeProfile) {
+ write_single_row_group_dictionary_filter_parquet_file(_file_path);
+
+ auto reader = create_reader();
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->predicate_columns = {field_projection(1)};
+ request->non_predicate_columns = {field_projection(0)};
+ request->conjuncts.push_back(create_string_in_conjunct(1, {"az", "za"}));
+ use_schema_order_positions(request.get(), schema);
+ ASSERT_TRUE(reader->open(request).ok());
+
+ std::vector<int32_t> ids;
+ std::vector<std::string> values;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ if (rows == 0) {
+ continue;
+ }
+ const auto& id_column = nullable_nested_column<ColumnInt32>(block, 0);
+ const auto& value_column = nullable_nested_column<ColumnString>(block,
1);
+ for (size_t row = 0; row < rows; ++row) {
+ ids.push_back(id_column.get_element(row));
+ values.push_back(value_column.get_data_at(row).to_string());
+ }
+ }
+
+ EXPECT_EQ(ids, std::vector<int32_t>({2, 5}));
+ EXPECT_EQ(values, std::vector<std::string>({"az", "za"}));
+}
+
+TEST_F(NewParquetReaderTest,
DictionaryPredicateSkipsRemainingPredicateColumnsWhenEmpty) {
+ write_single_row_group_dictionary_filter_parquet_file(_file_path);
+
+ RuntimeProfile
profile("new_parquet_reader_dictionary_filter_empty_profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->predicate_columns = {field_projection(1), field_projection(0)};
+ request->conjuncts.push_back(
+ create_string_dictionary_and_residual_conjunct(1, {"az"},
"not_present"));
+ request->conjuncts.push_back(create_int32_greater_than_conjunct(0, 0));
+ use_schema_order_positions(request.get(), schema);
+ ASSERT_TRUE(reader->open(request).ok());
+
+ bool eof = false;
+ size_t total_rows = 0;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ total_rows += rows;
+ }
+
+ EXPECT_EQ(total_rows, 0);
+ EXPECT_EQ(profile.get_counter("RowsFilteredByConjunct")->value(), 6);
+ EXPECT_EQ(profile.get_counter("RowsFilteredByDictFilter")->value(), 5);
+ EXPECT_EQ(profile.get_counter("DictFilterCandidateColumns")->value(), 1);
+ EXPECT_EQ(profile.get_counter("DictFilterColumns")->value(), 1);
+ EXPECT_EQ(profile.get_counter("DictFilterUnsupportedColumns")->value(), 0);
+ EXPECT_EQ(profile.get_counter("DictFilterReadFailures")->value(), 0);
+ EXPECT_EQ(profile.get_counter("SelectedRows")->value(), 0);
+ // The first dictionary predicate column is read once to produce a compact
row filter. The
+ // second predicate column is skipped after the selection becomes empty,
which verifies the
+ // StarRocks-style round-by-round policy: only rows surviving previous
predicates are read.
+ EXPECT_EQ(profile.get_counter("ReaderSelectRows")->value(), 6);
+ EXPECT_EQ(profile.get_counter("ReaderSkipRows")->value(), 6);
+}
+
+TEST_F(NewParquetReaderTest,
DictionaryPredicateRunsResidualConjunctOnSurvivors) {
+ write_single_row_group_dictionary_filter_parquet_file(_file_path);
+
+ RuntimeProfile
profile("new_parquet_reader_dictionary_prefilter_residual_profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->predicate_columns = {field_projection(1)};
+ request->non_predicate_columns = {field_projection(0)};
+ request->conjuncts.push_back(
+ create_string_dictionary_and_residual_conjunct(1, {"az", "za"},
"za"));
+ use_schema_order_positions(request.get(), schema);
+ ASSERT_TRUE(reader->open(request).ok());
+
+ std::vector<int32_t> ids;
+ std::vector<std::string> values;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ if (rows == 0) {
+ continue;
+ }
+ const auto& id_column = nullable_nested_column<ColumnInt32>(block, 0);
+ const auto& value_column = nullable_nested_column<ColumnString>(block,
1);
+ for (size_t row = 0; row < rows; ++row) {
+ ids.push_back(id_column.get_element(row));
+ values.push_back(value_column.get_data_at(row).to_string());
+ }
+ }
+
+ EXPECT_EQ(ids, std::vector<int32_t>({5}));
+ EXPECT_EQ(values, std::vector<std::string>({"za"}));
+ EXPECT_EQ(profile.get_counter("RowsFilteredByDictFilter")->value(), 4);
+ EXPECT_EQ(profile.get_counter("RowsFilteredByConjunct")->value(), 5);
+ EXPECT_EQ(profile.get_counter("SelectedRows")->value(), 1);
+}
+
+TEST_F(NewParquetReaderTest, DictionaryPredicateKeepsNestedOrResidualConjunct)
{
+ write_single_row_group_dictionary_filter_parquet_file(_file_path);
+
+ RuntimeProfile
profile("new_parquet_reader_dictionary_nested_or_residual_profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->predicate_columns = {field_projection(1)};
+ request->non_predicate_columns = {field_projection(0)};
+
request->conjuncts.push_back(create_nested_or_dictionary_and_residual_conjunct(1));
+ use_schema_order_positions(request.get(), schema);
+ ASSERT_TRUE(reader->open(request).ok());
+
+ std::vector<int32_t> ids;
+ std::vector<std::string> values;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ if (rows == 0) {
+ continue;
+ }
+ const auto& id_column = nullable_nested_column<ColumnInt32>(block, 0);
+ const auto& value_column = nullable_nested_column<ColumnString>(block,
1);
+ for (size_t row = 0; row < rows; ++row) {
+ ids.push_back(id_column.get_element(row));
+ values.push_back(value_column.get_data_at(row).to_string());
+ }
+ }
+
+ EXPECT_EQ(ids, std::vector<int32_t>({2}));
+ EXPECT_EQ(values, std::vector<std::string>({"az"}));
+ EXPECT_EQ(profile.get_counter("RowsFilteredByDictFilter")->value(), 4);
+ EXPECT_EQ(profile.get_counter("RowsFilteredByConjunct")->value(), 5);
+ EXPECT_EQ(profile.get_counter("SelectedRows")->value(), 1);
+}
+
TEST_F(NewParquetReaderTest, ScanRangeFiltersRowGroupsBeforeDictionaryPruning)
{
write_dictionary_filter_parquet_file(_file_path);
auto parquet_file_reader =
::parquet::ParquetFileReader::OpenFile(_file_path, false);
diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp
b/be/test/format_v2/parquet/parquet_scan_test.cpp
index 8c1ac915755..edd51c45203 100644
--- a/be/test/format_v2/parquet/parquet_scan_test.cpp
+++ b/be/test/format_v2/parquet/parquet_scan_test.cpp
@@ -146,11 +146,68 @@ private:
const std::string _expr_name = "Int32ZoneMapExpr";
};
+class Int32PairSumExpr final : public VExpr {
+public:
+ Int32PairSumExpr(int left_column_id, int right_column_id, int32_t
upper_bound)
+ : VExpr(std::make_shared<DataTypeUInt8>(), false),
+ _left_column_id(left_column_id),
+ _right_column_id(right_column_id),
+ _upper_bound(upper_bound) {}
+
+ const std::string& expr_name() const override { return _expr_name; }
+
+ Status execute_column_impl(VExprContext*, const Block* block, const
Selector* selector,
+ size_t count, ColumnPtr& result_column) const
override {
+ DORIS_CHECK(block != nullptr);
+ DORIS_CHECK(selector == nullptr);
+ DORIS_CHECK(_left_column_id >= 0 && _left_column_id <
static_cast<int>(block->columns()));
+ DORIS_CHECK(_right_column_id >= 0 && _right_column_id <
static_cast<int>(block->columns()));
+ const auto& left_column =
+
int32_data_column(*block->get_by_position(_left_column_id).column);
+ const auto& right_column =
+
int32_data_column(*block->get_by_position(_right_column_id).column);
+ DORIS_CHECK(left_column.size() >= count);
+ DORIS_CHECK(right_column.size() >= count);
+
+ auto result = ColumnUInt8::create(count, 0);
+ auto& result_data = result->get_data();
+ for (size_t row = 0; row < count; ++row) {
+ result_data[row] =
+ left_column.get_element(row) +
right_column.get_element(row) < _upper_bound;
+ }
+ result_column = std::move(result);
+ return Status::OK();
+ }
+
+ void collect_slot_column_ids(std::set<int>& column_ids) const override {
+ column_ids.insert(_left_column_id);
+ column_ids.insert(_right_column_id);
+ }
+
+private:
+ int _left_column_id;
+ int _right_column_id;
+ int32_t _upper_bound;
+ const std::string _expr_name = "Int32PairSumExpr";
+};
+
VExprContextSPtr create_int32_zonemap_conjunct(int column_id,
Int32ZoneMapExpr::Op op,
int32_t value) {
return
VExprContext::create_shared(std::make_shared<Int32ZoneMapExpr>(column_id, op,
value));
}
+VExprContextSPtr create_int32_pair_sum_conjunct(int left_column_id, int
right_column_id,
+ int32_t upper_bound) {
+ return VExprContext::create_shared(
+ std::make_shared<Int32PairSumExpr>(left_column_id,
right_column_id, upper_bound));
+}
+
+int64_t counter_value(RuntimeProfile& profile, const std::string& name) {
+ auto* counter = profile.get_counter(name);
+ DORIS_CHECK(counter != nullptr);
+ return counter->value();
+}
+
std::shared_ptr<arrow::Array> finish_array(arrow::ArrayBuilder* builder) {
std::shared_ptr<arrow::Array> array;
EXPECT_TRUE(builder->Finish(&array).ok());
@@ -380,6 +437,23 @@ protected:
std::string _file_path;
};
+TEST(ParquetScanSelectionTest, CompactFilterShrinksCurrentSelection) {
+ format::parquet::SelectionVector selection(4);
+ selection.set_index(0, 0);
+ selection.set_index(1, 2);
+ selection.set_index(2, 4);
+ selection.set_index(3, 5);
+
+ const IColumn::Filter compact_filter {1, 0, 1, 0};
+ const auto selected_rows =
+ format::parquet::apply_compact_filter_to_selection(compact_filter,
&selection, 4);
+
+ ASSERT_EQ(selected_rows, 2);
+ EXPECT_EQ(selection.get_index(0), 0);
+ EXPECT_EQ(selection.get_index(1), 4);
+ EXPECT_TRUE(selection.verify(selected_rows, 6).ok());
+}
+
TEST_F(ParquetScanTest, PlanRowGroupsAppliesScanRangeBeforeStatistics) {
write_int_pair_parquet_file(_file_path, 2);
auto parquet_file_reader =
::parquet::ParquetFileReader::OpenFile(_file_path, false);
@@ -843,6 +917,134 @@ TEST_F(ParquetScanTest,
NoRequestedColumnsReturnsRowsOnlyAcrossRowGroups) {
EXPECT_EQ(total_rows, 6);
}
+TEST_F(ParquetScanTest, PredicateColumnsFilterRoundByRound) {
+ write_int_pair_parquet_file(_file_path, 6, false);
+ RuntimeProfile profile("profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(request.get());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(0)).ok());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(1)).ok());
+ request->conjuncts.push_back(create_int32_zonemap_conjunct(0,
Int32ZoneMapExpr::Op::GT, 2));
+ request->conjuncts.push_back(create_int32_zonemap_conjunct(1,
Int32ZoneMapExpr::Op::LT, 50));
+ ASSERT_TRUE(reader->open(request).ok());
+
+ std::vector<int32_t> ids;
+ std::vector<int32_t> scores;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ if (rows == 0) {
+ continue;
+ }
+ const auto& id_column =
int32_data_column(*block.get_by_position(0).column);
+ const auto& score_column =
int32_data_column(*block.get_by_position(1).column);
+ ASSERT_EQ(id_column.size(), rows);
+ ASSERT_EQ(score_column.size(), rows);
+ for (size_t row = 0; row < rows; ++row) {
+ ids.push_back(id_column.get_element(row));
+ scores.push_back(score_column.get_element(row));
+ }
+ }
+
+ EXPECT_EQ(ids, std::vector<int32_t>({3, 4}));
+ EXPECT_EQ(scores, std::vector<int32_t>({30, 40}));
+ EXPECT_EQ(counter_value(profile, "RawRowsRead"), 6);
+ EXPECT_EQ(counter_value(profile, "SelectedRows"), 2);
+ EXPECT_EQ(counter_value(profile, "RowsFilteredByConjunct"), 4);
+ EXPECT_EQ(counter_value(profile, "ReaderReadRows"), 10);
+ EXPECT_EQ(counter_value(profile, "ReaderSelectRows"), 4);
+ EXPECT_EQ(counter_value(profile, "ReaderSkipRows"), 2);
+}
+
+TEST_F(ParquetScanTest,
PredicateColumnsSkipUnreadColumnsWhenFirstPredicateFiltersAll) {
+ write_int_pair_parquet_file(_file_path, 6, false);
+ RuntimeProfile profile("profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(request.get());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(0)).ok());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(1)).ok());
+ request->conjuncts.push_back(create_int32_zonemap_conjunct(0,
Int32ZoneMapExpr::Op::GT, 100));
+ request->conjuncts.push_back(create_int32_zonemap_conjunct(1,
Int32ZoneMapExpr::Op::LT, 50));
+ ASSERT_TRUE(reader->open(request).ok());
+
+ size_t total_rows = 0;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ total_rows += rows;
+ }
+
+ EXPECT_EQ(total_rows, 0);
+ EXPECT_EQ(counter_value(profile, "RawRowsRead"), 6);
+ EXPECT_EQ(counter_value(profile, "SelectedRows"), 0);
+ EXPECT_EQ(counter_value(profile, "RowsFilteredByConjunct"), 6);
+ EXPECT_EQ(counter_value(profile, "EmptySelectionBatches"), 1);
+ EXPECT_EQ(counter_value(profile, "ReaderReadRows"), 6);
+ EXPECT_EQ(counter_value(profile, "ReaderSelectRows"), 0);
+ EXPECT_EQ(counter_value(profile, "ReaderSkipRows"), 6);
+}
+
+TEST_F(ParquetScanTest, MultiColumnPredicateWaitsForAllPredicateColumns) {
+ write_int_pair_parquet_file(_file_path, 6, false);
+ RuntimeProfile profile("profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(request.get());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(0)).ok());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(1)).ok());
+ request->conjuncts.push_back(create_int32_pair_sum_conjunct(0, 1, 45));
+ ASSERT_TRUE(reader->open(request).ok());
+
+ std::vector<int32_t> ids;
+ std::vector<int32_t> scores;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ if (rows == 0) {
+ continue;
+ }
+ const auto& id_column =
int32_data_column(*block.get_by_position(0).column);
+ const auto& score_column =
int32_data_column(*block.get_by_position(1).column);
+ ASSERT_EQ(id_column.size(), rows);
+ ASSERT_EQ(score_column.size(), rows);
+ for (size_t row = 0; row < rows; ++row) {
+ ids.push_back(id_column.get_element(row));
+ scores.push_back(score_column.get_element(row));
+ }
+ }
+
+ EXPECT_EQ(ids, std::vector<int32_t>({1, 2, 3, 4}));
+ EXPECT_EQ(scores, std::vector<int32_t>({10, 20, 30, 40}));
+ EXPECT_EQ(counter_value(profile, "RawRowsRead"), 6);
+ EXPECT_EQ(counter_value(profile, "SelectedRows"), 4);
+ EXPECT_EQ(counter_value(profile, "RowsFilteredByConjunct"), 2);
+ EXPECT_EQ(counter_value(profile, "ReaderReadRows"), 12);
+ EXPECT_EQ(counter_value(profile, "ReaderSelectRows"), 0);
+}
+
TEST_F(ParquetScanTest, ProfileCountersReflectPageIndexAndRangeGapPruning) {
write_page_index_parquet_file(_file_path);
RuntimeProfile profile("profile");
diff --git
a/regression-test/suites/external_table_p0/hive/test_parquet_lazy_mat_profile.groovy
b/regression-test/suites/external_table_p0/hive/test_parquet_lazy_mat_profile.groovy
index 8813c96e63c..6bbde270ba2 100644
---
a/regression-test/suites/external_table_p0/hive/test_parquet_lazy_mat_profile.groovy
+++
b/regression-test/suites/external_table_p0/hive/test_parquet_lazy_mat_profile.groovy
@@ -100,6 +100,18 @@ suite("test_parquet_lazy_mat_profile", "p0,external") {
return matcher.find() ? matcher.group(1).trim() : null
}
+ def metricValueAsLong = { String value ->
+ if (value == null) {
+ return -1L
+ }
+ def formatted = value =~ /.*\((\d+)\).*/
+ if (formatted.matches()) {
+ return formatted[0][1].toLong()
+ }
+ def plain = value.replaceAll("[^0-9-]", "")
+ return plain == "" ? -1L : plain.toLong()
+ }
+
// session vars
sql "unset variable all;"
sql "set profile_level=2;"
@@ -231,6 +243,29 @@ suite("test_parquet_lazy_mat_profile", "p0,external") {
return extractProfileBlockMetrics(profileText, "ParquetReader")
}
+ def q8 = {
+ sql """ set enable_file_scanner_v2 = true; """
+ sql """ set enable_parquet_filter_by_min_max = false; """
+ sql """ set enable_parquet_lazy_materialization = true; """
+ def t1 = UUID.randomUUID().toString()
+ def sql_result = sql """
+ select *, "${t1}" from alltypes_tiny_pages_plain where id > 2
and id < 10 order by id;
+ """
+ def idColumnIndex = 7
+ assertEquals(7, sql_result.size())
+ assertEquals("3", sql_result[0][idColumnIndex].toString())
+ assertEquals("9", sql_result[6][idColumnIndex].toString())
+
+ def profileText = getProfileWithToken(t1)
+ assertTrue(profileText.contains("ParquetReader"), "Profile does
not contain ParquetReader")
+ def metrics = extractProfileBlockMetrics(profileText,
"ParquetReader")
+ logger.info("metrics = ${metrics}")
+ assertTrue(metricValueAsLong(metrics["FilteredRowsByLazyRead"]) >
0)
+ assertTrue(metricValueAsLong(metrics["RawRowsRead"]) >= 7)
+ assertTrue(metricValueAsLong(metrics["RowsFilteredByConjunct"]) >
0)
+ assertTrue(metricValueAsLong(metrics["ReaderSelectRows"]) > 0)
+ }
+
def test_true_true = {
@@ -598,6 +633,7 @@ suite("test_parquet_lazy_mat_profile", "p0,external") {
test_true_false();
test_false_false();
test_false_true();
+ q8();
sql """drop catalog ${catalog_name};"""
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
