This is an automated email from the ASF dual-hosted git repository. Gabriel39 pushed a commit to branch refactor_0702 in repository https://gitbox.apache.org/repos/asf/doris.git
commit c5f2765706ff9829f4d070956e456a3c53be691a Author: Gabriel <[email protected]> AuthorDate: Thu Jul 2 22:00:08 2026 +0800 [fix](be) Use VExpr for file scanner v2 ZoneMap pruning ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: File scanner v2 still used ColumnPredicate min/max and page-index statistics to prune Parquet row groups and pages. VExpr now owns ZoneMap evaluation, so the Parquet v2 pruning path builds ZoneMapEvalContext from row-group statistics and page indexes and evaluates localized VExpr conjuncts instead. ColumnPredicate pruning is kept only for dictionary and bloom filters. The existing row-group and page ZoneMap counters are preserved, page-index row-group pruning is expos [...] ### Release note None ### Check List (For Author) - Test: Manual test - Ran build-support/clang-format.sh - Ran git diff --check - Verified no file scanner v2 ColumnPredicate ZoneMap calls remain with rg - Verified no Chinese characters remain under be/src/core/data_type_serde with rg - Attempted BE UT filter `ParquetStatisticsPruningTest.*:ParquetScanTest.*:NewParquetReaderTest.*`; it did not compile/run because local thirdparty dependencies are missing (`thirdparty/installed/bin/protoc` and Snappy). - Behavior changed: No - Does this need documentation: No --- .../data_type_serde/data_type_number_serde.cpp | 10 +- be/src/core/data_type_serde/decoded_column_view.h | 5 +- be/src/format_v2/column_mapper.cpp | 5 +- be/src/format_v2/column_mapper.h | 7 +- be/src/format_v2/file_reader.h | 9 +- be/src/format_v2/parquet/parquet_profile.cpp | 13 + be/src/format_v2/parquet/parquet_profile.h | 4 + be/src/format_v2/parquet/parquet_reader.cpp | 4 +- be/src/format_v2/parquet/parquet_scan.cpp | 8 +- be/src/format_v2/parquet/parquet_scan.h | 4 +- be/src/format_v2/parquet/parquet_statistics.cpp | 271 +++++++++++++++------ be/src/format_v2/parquet/parquet_statistics.h | 17 +- be/test/format_v2/parquet/parquet_reader_test.cpp | 38 ++- be/test/format_v2/parquet/parquet_scan_test.cpp | 95 ++++++-- .../format_v2/parquet/parquet_statistics_test.cpp | 105 ++++++-- be/test/format_v2/table/iceberg_reader_test.cpp | 7 +- be/test/format_v2/table_reader_test.cpp | 7 +- 17 files changed, 453 insertions(+), 156 deletions(-) diff --git a/be/src/core/data_type_serde/data_type_number_serde.cpp b/be/src/core/data_type_serde/data_type_number_serde.cpp index 3c99a53b5b0..c4a6ae50ede 100644 --- a/be/src/core/data_type_serde/data_type_number_serde.cpp +++ b/be/src/core/data_type_serde/data_type_number_serde.cpp @@ -177,7 +177,7 @@ Status read_integer_decoded_values(IColumn& column, const DecodedColumnView& vie } } // namespace -// Type map的基本结构 +// Basic structure of the type map. template <typename Key, typename Value, typename... Rest> struct TypeMap { using KeyType = Key; @@ -185,21 +185,21 @@ struct TypeMap { using Next = TypeMap<Rest...>; }; -// Type map的末端 +// End marker of the type map. template <> struct TypeMap<void, void> {}; -// TypeMapLookup 前向声明 +// Forward declaration of TypeMapLookup. template <typename Key, typename Map> struct TypeMapLookup; -// Type map查找：找到匹配的键时的情况 +// Type map lookup when the key matches. template <typename Key, typename Value, typename... Rest> struct TypeMapLookup<Key, TypeMap<Key, Value, Rest...>> { using ValueType = Value; }; -// Type map查找：递归查找 +// Type map lookup by recursive search. template <typename Key, typename K, typename V, typename... Rest> struct TypeMapLookup<Key, TypeMap<K, V, Rest...>> { using ValueType = typename TypeMapLookup<Key, TypeMap<Rest...>>::ValueType; diff --git a/be/src/core/data_type_serde/decoded_column_view.h b/be/src/core/data_type_serde/decoded_column_view.h index 9f40f1513db..62efd2ca466 100644 --- a/be/src/core/data_type_serde/decoded_column_view.h +++ b/be/src/core/data_type_serde/decoded_column_view.h @@ -33,8 +33,9 @@ namespace doris { class IColumn; -// 已解码 column batch 的物理值来源类型。 -// 该枚举只描述通用内存布局，不包含 Parquet/ORC/Arrow 等格式专有类型。 +// Physical value source type for a decoded column batch. +// This enum describes only generic memory layouts, not format-specific types such as +// Parquet/ORC/Arrow. enum class DecodedValueKind { BOOL, INT32, diff --git a/be/src/format_v2/column_mapper.cpp b/be/src/format_v2/column_mapper.cpp index 4a76b112456..896f144b881 100644 --- a/be/src/format_v2/column_mapper.cpp +++ b/be/src/format_v2/column_mapper.cpp @@ -1710,8 +1710,9 @@ Status TableColumnMapper::create_scan_request( auto* mapping = _find_mapping(global_index); if (mapping != nullptr && mapping->file_local_id.has_value()) { // A file column can be read lazily as a non-predicate column only when it is not used - // by row-level expression filters. Single-column ColumnPredicate filters are pruning - // hints only and must not force row-level predicate materialization. + // by row-level expression filters. Single-column ColumnPredicate filters are + // dictionary/bloom pruning hints only and must not force row-level predicate + // materialization. bool used_by_filter = false; for (const auto& table_filter : table_filters) { const auto& global_indices = table_filter.global_indices; diff --git a/be/src/format_v2/column_mapper.h b/be/src/format_v2/column_mapper.h index 77c444ba0f1..acda5467bd5 100644 --- a/be/src/format_v2/column_mapper.h +++ b/be/src/format_v2/column_mapper.h @@ -225,7 +225,7 @@ protected: // should keep all required columns in one scan list. virtual bool enable_lazy_materialization() const { return true; } // File-layer column predicate filters are reader-specific pruning hints. Parquet consumes them - // for row-group/page-index/statistics pruning; simple delimited readers do not. + // for dictionary and bloom-filter pruning; ZoneMap pruning uses file-local VExpr conjuncts. virtual bool enable_column_predicate_filters() const { return true; } // Row-oriented readers such as CSV/Text cannot physically read only a nested child from one // delimited text field. They must scan the whole complex top-level field and let TableReader @@ -271,8 +271,9 @@ protected: std::map<std::string, Field> _partition_values; }; -// Parquet consumes the full FileScanRequest shape: predicate columns for lazy materialization and -// top-level column_predicate_filters for statistics/page-index pruning. +// Parquet consumes the full FileScanRequest shape: predicate columns for lazy materialization, +// file-local conjuncts for ZoneMap pruning, and top-level column_predicate_filters for +// dictionary/bloom pruning. class ParquetColumnMapper final : public TableColumnMapper { public: using TableColumnMapper::TableColumnMapper; diff --git a/be/src/format_v2/file_reader.h b/be/src/format_v2/file_reader.h index e4c55bd9a6b..27aceeae944 100644 --- a/be/src/format_v2/file_reader.h +++ b/be/src/format_v2/file_reader.h @@ -48,8 +48,9 @@ namespace doris::format { class TableColumnMapper; struct TableColumnMapperOptions; -// File-local single-column predicates for file-layer pruning, such as min/max, page index, -// dictionary and bloom filter. Predicates must all belong to file_column_id. +// File-local single-column predicates for file-layer pruning, such as dictionary and bloom filter. +// ZoneMap pruning uses file-local VExpr conjuncts instead. Predicates must all belong to +// file_column_id. // These predicates are pruning hints only and are not row-level conjuncts. struct FileColumnPredicateFilter { LocalColumnId file_column_id = LocalColumnId::invalid(); @@ -86,8 +87,8 @@ struct FileScanRequest { VExprContextSPtrs conjuncts; // Delete predicates converted to file-local expressions. VExprContextSPtrs delete_conjuncts; - // Single-column predicates used only for file-layer pruning, such as statistics, page index, - // dictionary and bloom filter. They must not be used for batch row-level filtering. + // Single-column predicates used only for file-layer pruning, such as dictionary and bloom + // filter. They must not be used for batch row-level filtering. std::vector<FileColumnPredicateFilter> column_predicate_filters; }; diff --git a/be/src/format_v2/parquet/parquet_profile.cpp b/be/src/format_v2/parquet/parquet_profile.cpp index 79f979ea0cf..4bf7af687f2 100644 --- a/be/src/format_v2/parquet/parquet_profile.cpp +++ b/be/src/format_v2/parquet/parquet_profile.cpp @@ -37,6 +37,8 @@ void ParquetProfile::init(RuntimeProfile* profile) { profile, "RowGroupsFilteredByDictionary", TUnit::UNIT, parquet_profile, 1); filtered_row_groups_by_bloom_filter = ADD_CHILD_COUNTER_WITH_LEVEL( profile, "RowGroupsFilteredByBloomFilter", TUnit::UNIT, parquet_profile, 1); + filtered_row_groups_by_page_index = ADD_CHILD_COUNTER_WITH_LEVEL( + profile, "RowGroupsFilteredByPageIndex", TUnit::UNIT, parquet_profile, 1); to_read_row_groups = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "RowGroupsReadNum", TUnit::UNIT, parquet_profile, 1); total_row_groups = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "RowGroupsTotalNum", TUnit::UNIT, @@ -91,6 +93,12 @@ void ParquetProfile::init(RuntimeProfile* profile) { ADD_CHILD_TIMER_WITH_LEVEL(profile, "PageIndexReadTime", parquet_profile, 1); parse_page_index_time = ADD_CHILD_TIMER_WITH_LEVEL(profile, "PageIndexParseTime", parquet_profile, 1); + expr_zonemap_unusable = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "ExprZoneMapUnusableEvals", + TUnit::UNIT, parquet_profile, 1); + in_zonemap_point_check = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "InZoneMapPointCheckCount", + TUnit::UNIT, parquet_profile, 1); + in_zonemap_range_only = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "InZoneMapRangeOnlyCount", + TUnit::UNIT, parquet_profile, 1); row_group_filter_time = ADD_CHILD_TIMER_WITH_LEVEL(profile, "RowGroupFilterTime", parquet_profile, 1); file_footer_read_calls = ADD_COUNTER_WITH_LEVEL(profile, "FileFooterReadCalls", TUnit::UNIT, 1); @@ -144,6 +152,8 @@ void ParquetProfile::update_pruning_stats(const ParquetPruningStats& pruning_sta pruning_stats.filtered_row_groups_by_dictionary); COUNTER_UPDATE(filtered_row_groups_by_bloom_filter, pruning_stats.filtered_row_groups_by_bloom_filter); + COUNTER_UPDATE(filtered_row_groups_by_page_index, + pruning_stats.filtered_row_groups_by_page_index); COUNTER_UPDATE(to_read_row_groups, pruning_stats.selected_row_groups); COUNTER_UPDATE(total_row_groups, pruning_stats.total_row_groups); COUNTER_UPDATE(selected_row_ranges, pruning_stats.selected_row_ranges); @@ -154,6 +164,9 @@ void ParquetProfile::update_pruning_stats(const ParquetPruningStats& pruning_sta COUNTER_UPDATE(row_group_filter_time, pruning_stats.row_group_filter_time); COUNTER_UPDATE(page_index_filter_time, pruning_stats.page_index_filter_time); COUNTER_UPDATE(read_page_index_time, pruning_stats.read_page_index_time); + COUNTER_UPDATE(expr_zonemap_unusable, pruning_stats.expr_zonemap_unusable_evals); + COUNTER_UPDATE(in_zonemap_point_check, pruning_stats.in_zonemap_point_check_count); + COUNTER_UPDATE(in_zonemap_range_only, pruning_stats.in_zonemap_range_only_count); } ParquetPageSkipProfile ParquetProfile::page_skip_profile() const { diff --git a/be/src/format_v2/parquet/parquet_profile.h b/be/src/format_v2/parquet/parquet_profile.h index 8f7623527ca..91c8ad89ddc 100644 --- a/be/src/format_v2/parquet/parquet_profile.h +++ b/be/src/format_v2/parquet/parquet_profile.h @@ -70,6 +70,7 @@ struct ParquetProfile { RuntimeProfile::Counter* filtered_row_groups_by_min_max = nullptr; RuntimeProfile::Counter* filtered_row_groups_by_dictionary = nullptr; RuntimeProfile::Counter* filtered_row_groups_by_bloom_filter = nullptr; + RuntimeProfile::Counter* filtered_row_groups_by_page_index = nullptr; RuntimeProfile::Counter* to_read_row_groups = nullptr; RuntimeProfile::Counter* total_row_groups = nullptr; RuntimeProfile::Counter* selected_row_ranges = nullptr; @@ -110,6 +111,9 @@ struct ParquetProfile { RuntimeProfile::Counter* page_index_filter_time = nullptr; RuntimeProfile::Counter* read_page_index_time = nullptr; RuntimeProfile::Counter* parse_page_index_time = nullptr; + RuntimeProfile::Counter* expr_zonemap_unusable = nullptr; + RuntimeProfile::Counter* in_zonemap_point_check = nullptr; + RuntimeProfile::Counter* in_zonemap_range_only = nullptr; RuntimeProfile::Counter* decompress_time = nullptr; RuntimeProfile::Counter* decompress_cnt = nullptr; diff --git a/be/src/format_v2/parquet/parquet_reader.cpp b/be/src/format_v2/parquet/parquet_reader.cpp index 3e13333e7d4..50ccc9f29d8 100644 --- a/be/src/format_v2/parquet/parquet_reader.cpp +++ b/be/src/format_v2/parquet/parquet_reader.cpp @@ -48,6 +48,7 @@ struct ParquetReaderScanState { std::vector<std::unique_ptr<ParquetColumnSchema>> file_schema; RowGroupScanPlan scan_plan; ParquetScanScheduler scheduler; + const RuntimeState* runtime_state = nullptr; const cctz::time_zone* timezone = nullptr; bool enable_bloom_filter = false; bool enable_page_cache = false; @@ -329,6 +330,7 @@ Status ParquetReader::init(RuntimeState* state) { _state->enable_page_cache = state != nullptr && state->query_options().enable_parquet_file_page_cache; if (state != nullptr) { + _state->runtime_state = state; _state->timezone = &state->timezone_obj(); _state->enable_strict_mode = state->enable_strict_mode(); _state->scheduler.set_timezone(&state->timezone_obj()); @@ -443,7 +445,7 @@ Status ParquetReader::open(std::shared_ptr<format::FileScanRequest> request) { RETURN_IF_ERROR(plan_parquet_row_groups( *_state->file_context.metadata, _state->file_context.file_reader.get(), _state->file_schema, *request_snapshot, scan_range, _state->enable_bloom_filter, - &row_group_plan, _state->timezone)); + &row_group_plan, _state->timezone, _state->runtime_state)); if (_profile != nullptr) { _parquet_profile.update_pruning_stats(row_group_plan.pruning_stats); } diff --git a/be/src/format_v2/parquet/parquet_scan.cpp b/be/src/format_v2/parquet/parquet_scan.cpp index d636f3e3f9e..bc391ba09c9 100644 --- a/be/src/format_v2/parquet/parquet_scan.cpp +++ b/be/src/format_v2/parquet/parquet_scan.cpp @@ -76,7 +76,8 @@ Status plan_parquet_row_groups(const ::parquet::FileMetaData& metadata, const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, const format::FileScanRequest& request, const ParquetScanRange& scan_range, bool enable_bloom_filter, - RowGroupScanPlan* plan, const cctz::time_zone* timezone) { + RowGroupScanPlan* plan, const cctz::time_zone* timezone, + const RuntimeState* runtime_state) { DORIS_CHECK(plan != nullptr); plan->row_groups.clear(); plan->pruning_stats = ParquetPruningStats {}; @@ -103,7 +104,8 @@ Status plan_parquet_row_groups(const ::parquet::FileMetaData& metadata, std::vector<int> statistics_selected_row_groups; RETURN_IF_ERROR(select_row_groups_by_statistics( metadata, file_reader, file_schema, request, &scan_range_selected_row_groups, - &statistics_selected_row_groups, enable_bloom_filter, &plan->pruning_stats, timezone)); + &statistics_selected_row_groups, enable_bloom_filter, &plan->pruning_stats, timezone, + runtime_state)); plan->row_groups.reserve(statistics_selected_row_groups.size()); for (const auto row_group_idx : statistics_selected_row_groups) { @@ -121,7 +123,7 @@ Status plan_parquet_row_groups(const ::parquet::FileMetaData& metadata, RETURN_IF_ERROR(select_row_group_ranges_by_page_index( file_reader, file_schema, request, row_group_idx, row_group_rows, &row_group_plan.selected_ranges, &row_group_plan.page_skip_plans, - &plan->pruning_stats, timezone)); + &plan->pruning_stats, timezone, runtime_state)); if (row_group_plan.selected_ranges.empty()) { continue; } diff --git a/be/src/format_v2/parquet/parquet_scan.h b/be/src/format_v2/parquet/parquet_scan.h index ca3c665a2e9..e6477f9f34e 100644 --- a/be/src/format_v2/parquet/parquet_scan.h +++ b/be/src/format_v2/parquet/parquet_scan.h @@ -44,6 +44,7 @@ class time_zone; namespace doris { class Block; +class RuntimeState; namespace format { struct FileScanRequest; @@ -86,7 +87,8 @@ Status plan_parquet_row_groups(const ::parquet::FileMetaData& metadata, const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, const format::FileScanRequest& request, const ParquetScanRange& scan_range, bool enable_bloom_filter, - RowGroupScanPlan* plan, const cctz::time_zone* timezone = nullptr); + RowGroupScanPlan* plan, const cctz::time_zone* timezone = nullptr, + const RuntimeState* runtime_state = nullptr); IColumn::Filter selection_to_filter(const SelectionVector& selection, uint16_t selected_rows, int64_t batch_rows); diff --git a/be/src/format_v2/parquet/parquet_statistics.cpp b/be/src/format_v2/parquet/parquet_statistics.cpp index 66af8515aba..918017f501b 100644 --- a/be/src/format_v2/parquet/parquet_statistics.cpp +++ b/be/src/format_v2/parquet/parquet_statistics.cpp @@ -30,6 +30,7 @@ #include <exception> #include <map> #include <memory> +#include <optional> #include <set> #include <string> #include <utility> @@ -40,9 +41,12 @@ #include "core/data_type/data_type_nullable.h" #include "core/data_type_serde/data_type_serde.h" #include "core/field.h" +#include "exprs/expr_zonemap_filter.h" +#include "exprs/vexpr_context.h" #include "format_v2/parquet/parquet_column_schema.h" #include "runtime/runtime_profile.h" #include "storage/index/zone_map/zone_map_index.h" +#include "storage/index/zone_map/zonemap_eval_context.h" #include "storage/predicate/accept_null_predicate.h" #include "storage/predicate/column_predicate.h" @@ -52,7 +56,6 @@ namespace { enum class ParquetRowGroupPruneReason { NONE, // cannot prune; must read - STATISTICS, // excluded by min/max statistics DICTIONARY, // excluded by dictionary BLOOM_FILTER, // excluded by bloom filter }; @@ -586,15 +589,6 @@ bool read_dictionary_words(::parquet::ParquetFileReader* file_reader, int row_gr return false; } -segment_v2::ZoneMap to_column_predicate_statistics(const ParquetColumnStatistics& statistics) { - segment_v2::ZoneMap predicate_statistics; - predicate_statistics.min_value = statistics.min_value; - predicate_statistics.max_value = statistics.max_value; - predicate_statistics.has_null = statistics.has_null; - predicate_statistics.has_not_null = statistics.has_not_null; - return predicate_statistics; -} - const ParquetColumnSchema* resolve_predicate_leaf_schema( const std::vector<std::unique_ptr<ParquetColumnSchema>>& schema, const format::FileColumnPredicateFilter& column_filter) { @@ -613,27 +607,92 @@ const ParquetColumnSchema* resolve_predicate_leaf_schema( return column_schema; } -bool check_statistics(const format::FileColumnPredicateFilter& column_filter, - const ParquetColumnStatistics& statistics) { - if (!statistics.has_any_statistics()) { - return false; +const ParquetColumnSchema* resolve_local_leaf_schema( + const std::vector<std::unique_ptr<ParquetColumnSchema>>& schema, + const format::LocalColumnId file_column_id) { + if (!file_column_id.is_valid() || file_column_id.value() >= static_cast<int>(schema.size())) { + return nullptr; } + const ParquetColumnSchema* column_schema = schema[file_column_id.value()].get(); + if (column_schema == nullptr || column_schema->kind != ParquetColumnSchemaKind::PRIMITIVE || + column_schema->leaf_column_id < 0 || column_schema->max_repetition_level > 0) { + return nullptr; + } + return column_schema; +} - for (const auto& column_predicate : column_filter.predicates) { - if (is_null_only_predicate(*column_predicate)) { - if (!statistics.has_null_count) { - continue; - } - } else if (!statistics.has_any_statistics()) { - continue; +std::optional<format::LocalColumnId> file_column_id_by_block_position( + const format::FileScanRequest& request, int block_position) { + for (const auto& [file_column_id, local_index] : request.local_positions) { + if (local_index.value() == block_position) { + return file_column_id; } - if (!column_predicate->evaluate_and(to_column_predicate_statistics(statistics))) { + } + return std::nullopt; +} + +bool has_expr_zonemap_filter(const format::FileScanRequest& request, + const RuntimeState* runtime_state) { + if (!expr_zonemap::is_expr_zonemap_filter_enabled(runtime_state)) { + return false; + } + for (const auto& conjunct : request.conjuncts) { + if (conjunct != nullptr && conjunct->root() != nullptr && + conjunct->root()->can_evaluate_zonemap_filter()) { return true; } } return false; } +std::set<int> collect_expr_zonemap_slot_indexes(const VExprContextSPtrs& conjuncts) { + std::set<int> slot_indexes; + for (const auto& conjunct : conjuncts) { + if (conjunct != nullptr && conjunct->root() != nullptr && + conjunct->root()->can_evaluate_zonemap_filter()) { + conjunct->root()->collect_slot_column_ids(slot_indexes); + } + } + return slot_indexes; +} + +std::shared_ptr<segment_v2::ZoneMap> make_zonemap_from_statistics( + const ParquetColumnStatistics& statistics) { + if (!statistics.has_null_count && !statistics.has_min_max) { + return nullptr; + } + segment_v2::ZoneMap zone_map; + zone_map.has_null = statistics.has_null; + zone_map.has_not_null = statistics.has_not_null; + if (!statistics.has_not_null) { + return std::make_shared<segment_v2::ZoneMap>(std::move(zone_map)); + } + if (!statistics.has_min_max) { + return nullptr; + } + zone_map.min_value = statistics.min_value; + zone_map.max_value = statistics.max_value; + return std::make_shared<segment_v2::ZoneMap>(std::move(zone_map)); +} + +void add_slot_zonemap(ZoneMapEvalContext* ctx, int slot_index, const DataTypePtr& data_type, + std::shared_ptr<segment_v2::ZoneMap> zone_map) { + DORIS_CHECK(ctx != nullptr); + ZoneMapEvalContext::SlotZoneMap slot_zone_map; + slot_zone_map.data_type = data_type; + slot_zone_map.zone_map = std::move(zone_map); + ctx->slots.emplace(slot_index, std::move(slot_zone_map)); +} + +void accumulate_zonemap_stats(const ZoneMapEvalContext& ctx, ParquetPruningStats* pruning_stats) { + if (pruning_stats == nullptr) { + return; + } + pruning_stats->expr_zonemap_unusable_evals += ctx.stats.unusable_zonemap_eval_count; + pruning_stats->in_zonemap_point_check_count += ctx.stats.in_zonemap_point_check_count; + pruning_stats->in_zonemap_range_only_count += ctx.stats.in_zonemap_range_only_count; +} + } // namespace ParquetColumnStatistics ParquetStatisticsUtils::TransformColumnStatistics( @@ -704,11 +763,6 @@ ParquetRowGroupPruneReason row_group_prune_reason( if (column_chunk == nullptr) { return ParquetRowGroupPruneReason::NONE; } - if (check_statistics(column_filter, - ParquetStatisticsUtils::TransformColumnStatistics( - *column_schema, column_chunk->statistics(), timezone))) { - return ParquetRowGroupPruneReason::STATISTICS; - } if (!supports_dictionary_pruning(*column_schema, *column_chunk, column_filter) || !is_dictionary_encoded_chunk(*column_chunk)) { return bloom_filter_prune_reason(row_group_idx, schema, column_filter, bloom_filter_cache, @@ -744,13 +798,51 @@ void init_bloom_filter_cache(::parquet::ParquetFileReader* file_reader, bool ena } } +bool row_group_pruned_by_expr_zonemap( + const ::parquet::RowGroupMetaData& row_group, + const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, + const format::FileScanRequest& request, ParquetPruningStats* pruning_stats, + const cctz::time_zone* timezone) { + const auto slot_indexes = collect_expr_zonemap_slot_indexes(request.conjuncts); + if (slot_indexes.empty()) { + return false; + } + + ZoneMapEvalContext ctx; + for (const int slot_index : slot_indexes) { + const auto file_column_id = file_column_id_by_block_position(request, slot_index); + if (!file_column_id.has_value()) { + continue; + } + const auto* column_schema = resolve_local_leaf_schema(file_schema, *file_column_id); + if (column_schema == nullptr || column_schema->type == nullptr) { + continue; + } + + std::shared_ptr<segment_v2::ZoneMap> zone_map; + DCHECK_LT(column_schema->leaf_column_id, row_group.num_columns()); + auto column_chunk = row_group.ColumnChunk(column_schema->leaf_column_id); + if (column_chunk != nullptr) { + zone_map = + make_zonemap_from_statistics(ParquetStatisticsUtils::TransformColumnStatistics( + *column_schema, column_chunk->statistics(), timezone)); + } + add_slot_zonemap(&ctx, slot_index, column_schema->type, std::move(zone_map)); + } + + const auto result = VExprContext::evaluate_zonemap_filter(request.conjuncts, ctx); + accumulate_zonemap_stats(ctx, pruning_stats); + return result == ZoneMapFilterResult::kNoMatch; +} + Status select_row_groups(const ::parquet::FileMetaData& metadata, ::parquet::ParquetFileReader* file_reader, const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, const format::FileScanRequest& request, const std::vector<int>* candidate_row_groups, std::vector<int>* selected_row_groups, bool enable_bloom_filter, - ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone) { + ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone, + const RuntimeState* runtime_state) { int64_t row_group_filter_time_sink = 0; SCOPED_RAW_TIMER(pruning_stats == nullptr ? &row_group_filter_time_sink : &pruning_stats->row_group_filter_time); @@ -778,6 +870,15 @@ Status select_row_groups(const ::parquet::FileMetaData& metadata, selected_row_groups->push_back(row_group_idx); continue; } + if (has_expr_zonemap_filter(request, runtime_state) && + row_group_pruned_by_expr_zonemap(*row_group, file_schema, request, pruning_stats, + timezone)) { + if (pruning_stats != nullptr) { + pruning_stats->filtered_group_rows += row_group->num_rows(); + ++pruning_stats->filtered_row_groups_by_statistics; + } + continue; + } bool drop = false; RowGroupBloomFilterCache bloom_filter_cache; init_bloom_filter_cache(file_reader, enable_bloom_filter, &bloom_filter_cache); @@ -791,9 +892,7 @@ Status select_row_groups(const ::parquet::FileMetaData& metadata, drop = true; if (pruning_stats != nullptr) { pruning_stats->filtered_group_rows += row_group->num_rows(); - if (prune_reason == ParquetRowGroupPruneReason::STATISTICS) { - ++pruning_stats->filtered_row_groups_by_statistics; - } else if (prune_reason == ParquetRowGroupPruneReason::DICTIONARY) { + if (prune_reason == ParquetRowGroupPruneReason::DICTIONARY) { ++pruning_stats->filtered_row_groups_by_dictionary; } else if (prune_reason == ParquetRowGroupPruneReason::BLOOM_FILTER) { ++pruning_stats->filtered_row_groups_by_bloom_filter; @@ -824,9 +923,11 @@ Status select_row_groups_by_statistics( const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, const format::FileScanRequest& request, const std::vector<int>* candidate_row_groups, std::vector<int>* selected_row_groups, bool enable_bloom_filter, - ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone) { + ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone, + const RuntimeState* runtime_state) { return select_row_groups(metadata, file_reader, file_schema, request, candidate_row_groups, - selected_row_groups, enable_bloom_filter, pruning_stats, timezone); + selected_row_groups, enable_bloom_filter, pruning_stats, timezone, + runtime_state); } namespace { @@ -1024,35 +1125,58 @@ void append_row_range(const RowRange& range, std::vector<RowRange>* ranges) { ranges->push_back(range); } -bool select_ranges_for_filter(const std::shared_ptr<::parquet::RowGroupPageIndexReader>& row_group, - const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, - const format::FileColumnPredicateFilter& column_filter, - int64_t row_group_rows, std::vector<RowRange>* ranges, - const cctz::time_zone* timezone) { - if (column_filter.predicates.empty()) { - return false; +std::optional< + std::pair<std::shared_ptr<::parquet::ColumnIndex>, std::shared_ptr<::parquet::OffsetIndex>>> +load_page_indexes_for_slot(const std::shared_ptr<::parquet::RowGroupPageIndexReader>& row_group, + const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, + const format::FileScanRequest& request, int slot_index, + const ParquetColumnSchema** column_schema) { + DORIS_CHECK(column_schema != nullptr); + *column_schema = nullptr; + const auto file_column_id = file_column_id_by_block_position(request, slot_index); + if (!file_column_id.has_value()) { + return std::nullopt; } - const auto* column_schema = resolve_predicate_leaf_schema(file_schema, column_filter); - if (column_schema == nullptr || column_schema->descriptor == nullptr) { - return false; + *column_schema = resolve_local_leaf_schema(file_schema, *file_column_id); + if (*column_schema == nullptr || (*column_schema)->descriptor == nullptr) { + return std::nullopt; } - std::shared_ptr<::parquet::ColumnIndex> column_index; - std::shared_ptr<::parquet::OffsetIndex> offset_index; try { - column_index = row_group->GetColumnIndex(column_schema->leaf_column_id); - offset_index = row_group->GetOffsetIndex(column_schema->leaf_column_id); + auto column_index = row_group->GetColumnIndex((*column_schema)->leaf_column_id); + auto offset_index = row_group->GetOffsetIndex((*column_schema)->leaf_column_id); + if (column_index == nullptr || offset_index == nullptr || + column_index->null_pages().size() != offset_index->page_locations().size()) { + return std::nullopt; + } + return std::make_pair(std::move(column_index), std::move(offset_index)); } catch (const ::parquet::ParquetException&) { - return false; + return std::nullopt; } catch (const std::exception&) { + return std::nullopt; + } +} + +bool select_ranges_for_expr_zonemap( + const std::shared_ptr<::parquet::RowGroupPageIndexReader>& row_group, + const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, + const format::FileScanRequest& request, int slot_index, const VExprContextSPtrs& conjuncts, + int64_t row_group_rows, std::vector<RowRange>* ranges, ParquetPruningStats* pruning_stats, + const cctz::time_zone* timezone) { + DORIS_CHECK(ranges != nullptr); + if (conjuncts.empty()) { return false; } - if (column_index == nullptr || offset_index == nullptr || - column_index->null_pages().size() != offset_index->page_locations().size()) { + const ParquetColumnSchema* column_schema = nullptr; + const auto page_indexes = + load_page_indexes_for_slot(row_group, file_schema, request, slot_index, &column_schema); + if (!page_indexes.has_value()) { return false; } + const auto& [column_index, offset_index] = *page_indexes; ranges->clear(); + ZoneMapEvalStats page_stats; const auto page_count = offset_index->page_locations().size(); for (size_t page_idx = 0; page_idx < page_count; ++page_idx) { ParquetColumnStatistics page_statistics; @@ -1061,11 +1185,21 @@ bool select_ranges_for_filter(const std::shared_ptr<::parquet::RowGroupPageIndex ranges->clear(); return false; } - const RowRange row_range = page_row_range(*offset_index, page_idx, row_group_rows); - if (check_statistics(column_filter, page_statistics)) { + + ZoneMapEvalContext ctx; + add_slot_zonemap(&ctx, slot_index, column_schema->type, + make_zonemap_from_statistics(page_statistics)); + const auto result = VExprContext::evaluate_zonemap_filter(conjuncts, ctx); + page_stats.merge_page_eval_stats(ctx.stats); + if (result == ZoneMapFilterResult::kNoMatch) { continue; } - append_row_range(row_range, ranges); + append_row_range(page_row_range(*offset_index, page_idx, row_group_rows), ranges); + } + if (pruning_stats != nullptr) { + pruning_stats->expr_zonemap_unusable_evals += page_stats.unusable_zonemap_eval_count; + pruning_stats->in_zonemap_point_check_count += page_stats.in_zonemap_point_check_count; + pruning_stats->in_zonemap_range_only_count += page_stats.in_zonemap_range_only_count; } return true; } @@ -1127,15 +1261,6 @@ void collect_request_leaf_schemas( for (const auto& projection : request.non_predicate_columns) { collect_projection(projection); } - for (const auto& column_filter : request.column_predicate_filters) { - const auto* leaf_schema = resolve_predicate_leaf_schema(file_schema, column_filter); - if (leaf_schema == nullptr) { - continue; - } - if (seen_leaf_ids.insert(leaf_schema->leaf_column_id).second) { - leaf_schemas->push_back(leaf_schema); - } - } } bool build_page_skip_plan_for_leaf( @@ -1144,8 +1269,6 @@ bool build_page_skip_plan_for_leaf( int64_t row_group_rows, ParquetPageSkipPlan* page_skip_plan) { DORIS_CHECK(page_skip_plan != nullptr); *page_skip_plan = ParquetPageSkipPlan {}; - // OffsetIndex first_row_index is row-based only for non-repeated leaves. LIST/MAP/repeated - // leaves need repetition-level-aware range mapping and are intentionally left out for now. if (column_schema.kind != ParquetColumnSchemaKind::PRIMITIVE || column_schema.descriptor == nullptr || column_schema.leaf_column_id < 0 || column_schema.descriptor->max_repetition_level() != 0) { @@ -1212,7 +1335,8 @@ Status select_row_group_ranges_by_page_index( const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, const format::FileScanRequest& request, int row_group_idx, int64_t row_group_rows, std::vector<RowRange>* selected_ranges, std::map<int, ParquetPageSkipPlan>* page_skip_plans, - ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone) { + ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone, + const RuntimeState* runtime_state) { int64_t page_index_filter_time_sink = 0; SCOPED_RAW_TIMER(pruning_stats == nullptr ? &page_index_filter_time_sink : &pruning_stats->page_index_filter_time); @@ -1225,7 +1349,7 @@ Status select_row_group_ranges_by_page_index( return Status::OK(); } selected_ranges->push_back(RowRange {0, row_group_rows}); - if (!config::enable_parquet_page_index || request.column_predicate_filters.empty() || + if (!config::enable_parquet_page_index || !has_expr_zonemap_filter(request, runtime_state) || file_reader == nullptr) { return Status::OK(); } @@ -1255,10 +1379,19 @@ Status select_row_group_ranges_by_page_index( return Status::OK(); } - for (const auto& column_filter : request.column_predicate_filters) { + std::map<int, VExprContextSPtrs> conjuncts_by_slot; + for (const auto& conjunct : request.conjuncts) { + const auto slot_index = expr_zonemap::single_slot_zonemap_index(conjunct); + if (slot_index >= 0) { + conjuncts_by_slot[slot_index].push_back(conjunct); + } + } + + for (const auto& [slot_index, conjuncts] : conjuncts_by_slot) { std::vector<RowRange> filter_ranges; - if (!select_ranges_for_filter(row_group_index_reader, file_schema, column_filter, - row_group_rows, &filter_ranges, timezone)) { + if (!select_ranges_for_expr_zonemap(row_group_index_reader, file_schema, request, + slot_index, conjuncts, row_group_rows, &filter_ranges, + pruning_stats, timezone)) { continue; } *selected_ranges = intersect_ranges(*selected_ranges, filter_ranges); diff --git a/be/src/format_v2/parquet/parquet_statistics.h b/be/src/format_v2/parquet/parquet_statistics.h index 3d4b9d35791..d9c9e45b171 100644 --- a/be/src/format_v2/parquet/parquet_statistics.h +++ b/be/src/format_v2/parquet/parquet_statistics.h @@ -38,6 +38,7 @@ class time_zone; namespace doris { class ColumnPredicate; +class RuntimeState; } // namespace doris namespace doris::format::parquet { @@ -50,7 +51,7 @@ struct ParquetColumnSchema; 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 - int64_t filtered_row_groups_by_statistics = 0; // row groups pruned by min/max statistics + int64_t filtered_row_groups_by_statistics = 0; // row groups pruned by ZoneMap statistics int64_t filtered_row_groups_by_dictionary = 0; // row groups pruned by dictionary int64_t filtered_row_groups_by_bloom_filter = 0; // row groups pruned by bloom filter int64_t filtered_row_groups_by_page_index = 0; // row groups fully pruned by page index @@ -62,6 +63,9 @@ struct ParquetPruningStats { int64_t row_group_filter_time = 0; // row-group pruning time (ns) int64_t page_index_filter_time = 0; // page-index pruning time (ns) int64_t read_page_index_time = 0; // page-index read time (ns) + int64_t expr_zonemap_unusable_evals = 0; // VExpr ZoneMap unusable evaluations + int64_t in_zonemap_point_check_count = 0; // VExpr IN ZoneMap point checks + int64_t in_zonemap_range_only_count = 0; // VExpr IN ZoneMap range-only checks }; struct ParquetColumnStatistics { @@ -77,8 +81,9 @@ struct ParquetColumnStatistics { // ============================================================================ // ============================================================================ -// statistics(TransformColumnStatistics + check_statistics) -// -> dictionary(read_dictionary_words + predicate::evaluate_and) +// VExpr ZoneMap(TransformColumnStatistics + evaluate_zonemap_filter) +// -> page-index ZoneMap(evaluate_zonemap_filter) +// dictionary(read_dictionary_words + predicate::evaluate_and) // -> bloom filter(bloom_filter_prune_reason) // ============================================================================ struct ParquetStatisticsUtils { @@ -97,13 +102,15 @@ Status select_row_groups_by_statistics( const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, const format::FileScanRequest& request, const std::vector<int>* candidate_row_groups, std::vector<int>* selected_row_groups, bool enable_bloom_filter, - ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone = nullptr); + ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone = nullptr, + const RuntimeState* runtime_state = nullptr); Status select_row_group_ranges_by_page_index( ::parquet::ParquetFileReader* file_reader, const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, const format::FileScanRequest& request, int row_group_idx, int64_t row_group_rows, std::vector<RowRange>* selected_ranges, std::map<int, ParquetPageSkipPlan>* page_skip_plans, - ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone = nullptr); + ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone = nullptr, + const RuntimeState* runtime_state = nullptr); } // namespace doris::format::parquet diff --git a/be/test/format_v2/parquet/parquet_reader_test.cpp b/be/test/format_v2/parquet/parquet_reader_test.cpp index 7fbfacd5d5d..eb984d528f7 100644 --- a/be/test/format_v2/parquet/parquet_reader_test.cpp +++ b/be/test/format_v2/parquet/parquet_reader_test.cpp @@ -30,6 +30,7 @@ #include <memory> #include <numeric> #include <optional> +#include <set> #include <string> #include <utility> #include <vector> @@ -60,6 +61,8 @@ #include "gen_cpp/Types_types.h" #include "io/io_common.h" #include "runtime/runtime_state.h" +#include "storage/index/zone_map/zonemap_eval_context.h" +#include "storage/index/zone_map/zonemap_filter_result.h" #include "storage/predicate/predicate_creator.h" #include "storage/segment/condition_cache.h" #include "storage/utils.h" @@ -114,6 +117,25 @@ public: const std::string& expr_name() const override { return _expr_name; } + bool can_evaluate_zonemap_filter() const override { return true; } + + void collect_slot_column_ids(std::set<int>& column_ids) const override { + column_ids.insert(_column_id); + } + + ZoneMapFilterResult evaluate_zonemap_filter(const ZoneMapEvalContext& ctx) const override { + auto zone_map = ctx.zone_map(_column_id); + if (zone_map == nullptr) { + return unsupported_zonemap_filter(ctx); + } + if (!zone_map->has_not_null) { + return ZoneMapFilterResult::kNoMatch; + } + const auto literal = Field::create_field<TYPE_INT>(_value); + return zone_map->max_value <= literal ? ZoneMapFilterResult::kNoMatch + : ZoneMapFilterResult::kMayMatch; + } + private: const int _column_id; const int32_t _value; @@ -1748,12 +1770,8 @@ TEST_F(NewParquetReaderTest, PlannerNarrowsRowRangesByPageIndex) { ASSERT_EQ(file_schema.size(), 1); format::FileScanRequest request; - format::FileColumnPredicateFilter column_filter; - column_filter.file_column_id = format::LocalColumnId(0); - auto id_type = std::make_shared<DataTypeInt32>(); - column_filter.predicates.push_back(create_comparison_predicate<PredicateType::GT>( - 0, "id", id_type, Field::create_field<TYPE_INT>(63), false)); - request.column_predicate_filters.push_back(std::move(column_filter)); + request.local_positions.emplace(format::LocalColumnId(0), format::LocalIndex(0)); + request.conjuncts.push_back(create_int32_greater_than_conjunct(0, 63)); format::parquet::RowGroupScanPlan plan; format::parquet::ParquetScanRange scan_range; @@ -1808,12 +1826,8 @@ TEST_F(NewParquetReaderTest, NestedStructPredicateDoesNotNarrowRowRangesByPageIn ASSERT_EQ(file_schema[0]->children[0]->name, "id"); format::FileScanRequest request; - format::FileColumnPredicateFilter column_filter; - column_filter.file_column_id = format::LocalColumnId(0); - auto id_type = std::make_shared<DataTypeInt32>(); - column_filter.predicates.push_back(create_comparison_predicate<PredicateType::GT>( - 0, "id", id_type, Field::create_field<TYPE_INT>(63), false)); - request.column_predicate_filters.push_back(std::move(column_filter)); + request.local_positions.emplace(format::LocalColumnId(0), format::LocalIndex(0)); + request.conjuncts.push_back(create_int32_greater_than_conjunct(0, 63)); format::parquet::RowGroupScanPlan plan; format::parquet::ParquetScanRange scan_range; diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp b/be/test/format_v2/parquet/parquet_scan_test.cpp index 3b381c3158f..e3efd6e1b93 100644 --- a/be/test/format_v2/parquet/parquet_scan_test.cpp +++ b/be/test/format_v2/parquet/parquet_scan_test.cpp @@ -28,6 +28,7 @@ #include <memory> #include <numeric> #include <optional> +#include <set> #include <string> #include <utility> #include <vector> @@ -41,6 +42,8 @@ #include "core/data_type/data_type_number.h" #include "core/data_type/data_type_string.h" #include "core/field.h" +#include "exprs/vexpr.h" +#include "exprs/vexpr_context.h" #include "format_v2/file_reader.h" #include "format_v2/parquet/parquet_column_schema.h" #include "format_v2/parquet/parquet_reader.h" @@ -48,6 +51,8 @@ #include "gen_cpp/Types_types.h" #include "io/io_common.h" #include "runtime/runtime_state.h" +#include "storage/index/zone_map/zonemap_eval_context.h" +#include "storage/index/zone_map/zonemap_filter_result.h" #include "storage/predicate/predicate_creator.h" #include "storage/utils.h" @@ -72,6 +77,57 @@ const ColumnString& string_data_column(const IColumn& column) { return assert_cast<const ColumnString&>(column); } +class Int32ZoneMapExpr final : public VExpr { +public: + enum class Op { GE, GT, LT }; + + Int32ZoneMapExpr(int column_id, Op op, int32_t value) + : VExpr(std::make_shared<DataTypeUInt8>(), false), + _column_id(column_id), + _op(op), + _value(value) {} + + const std::string& expr_name() const override { return _expr_name; } + + bool can_evaluate_zonemap_filter() const override { return true; } + + void collect_slot_column_ids(std::set<int>& column_ids) const override { + column_ids.insert(_column_id); + } + + ZoneMapFilterResult evaluate_zonemap_filter(const ZoneMapEvalContext& ctx) const override { + auto zone_map = ctx.zone_map(_column_id); + if (zone_map == nullptr) { + return unsupported_zonemap_filter(ctx); + } + if (!zone_map->has_not_null) { + return ZoneMapFilterResult::kNoMatch; + } + const auto literal = Field::create_field<TYPE_INT>(_value); + if (_op == Op::GE) { + return zone_map->max_value < literal ? ZoneMapFilterResult::kNoMatch + : ZoneMapFilterResult::kMayMatch; + } + if (_op == Op::GT) { + return zone_map->max_value <= literal ? ZoneMapFilterResult::kNoMatch + : ZoneMapFilterResult::kMayMatch; + } + return zone_map->min_value >= literal ? ZoneMapFilterResult::kNoMatch + : ZoneMapFilterResult::kMayMatch; + } + +private: + int _column_id; + Op _op; + int32_t _value; + const std::string _expr_name = "Int32ZoneMapExpr"; +}; + +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)); +} + std::shared_ptr<arrow::Array> finish_array(arrow::ArrayBuilder* builder) { std::shared_ptr<arrow::Array> array; EXPECT_TRUE(builder->Finish(&array).ok()); @@ -330,8 +386,8 @@ TEST_F(ParquetScanTest, PlanRowGroupsAppliesScanRangeBeforeStatistics) { auto file_schema = build_file_schema(*parquet_file_reader); format::FileScanRequest request; - request.column_predicate_filters.push_back( - int32_filter(0, "id", file_schema[0]->type, PredicateType::GE, 5)); + request.local_positions.emplace(format::LocalColumnId(0), format::LocalIndex(0)); + request.conjuncts.push_back(create_int32_zonemap_conjunct(0, Int32ZoneMapExpr::Op::GE, 5)); const auto [range_start_offset, range_size] = row_group_mid_range(_file_path, 1); format::parquet::ParquetScanRange scan_range; @@ -358,8 +414,8 @@ TEST_F(ParquetScanTest, PlanRowGroupsPreservesFirstFileRowAcrossPrunedRowGroups) auto file_schema = build_file_schema(*parquet_file_reader); format::FileScanRequest request; - request.column_predicate_filters.push_back( - int32_filter(0, "id", file_schema[0]->type, PredicateType::GE, 5)); + request.local_positions.emplace(format::LocalColumnId(0), format::LocalIndex(0)); + request.conjuncts.push_back(create_int32_zonemap_conjunct(0, Int32ZoneMapExpr::Op::GE, 5)); format::parquet::RowGroupScanPlan plan; format::parquet::ParquetScanRange scan_range; @@ -413,8 +469,9 @@ TEST_F(ParquetScanTest, PageIndexIntersectsMultipleFiltersAndBuildsSkipPlan) { auto file_schema = build_file_schema(*parquet_file_reader); format::FileScanRequest single_filter_request; - single_filter_request.column_predicate_filters.push_back( - int32_filter(0, "id", file_schema[0]->type, PredicateType::GE, 32)); + single_filter_request.local_positions.emplace(format::LocalColumnId(0), format::LocalIndex(0)); + single_filter_request.conjuncts.push_back( + create_int32_zonemap_conjunct(0, Int32ZoneMapExpr::Op::GE, 32)); format::parquet::RowGroupScanPlan single_filter_plan; format::parquet::ParquetScanRange scan_range; ASSERT_TRUE(format::parquet::plan_parquet_row_groups( @@ -426,10 +483,12 @@ TEST_F(ParquetScanTest, PageIndexIntersectsMultipleFiltersAndBuildsSkipPlan) { count_range_rows(single_filter_plan.row_groups[0].selected_ranges); format::FileScanRequest intersect_request; - intersect_request.column_predicate_filters.push_back( - int32_filter(0, "id", file_schema[0]->type, PredicateType::GE, 32)); - intersect_request.column_predicate_filters.push_back( - int32_filter(1, "score", file_schema[1]->type, PredicateType::LT, 96)); + intersect_request.local_positions.emplace(format::LocalColumnId(0), format::LocalIndex(0)); + intersect_request.local_positions.emplace(format::LocalColumnId(1), format::LocalIndex(1)); + intersect_request.conjuncts.push_back( + create_int32_zonemap_conjunct(0, Int32ZoneMapExpr::Op::GE, 32)); + intersect_request.conjuncts.push_back( + create_int32_zonemap_conjunct(1, Int32ZoneMapExpr::Op::LT, 96)); format::parquet::RowGroupScanPlan intersect_plan; ASSERT_TRUE(format::parquet::plan_parquet_row_groups( *parquet_file_reader->metadata(), parquet_file_reader.get(), file_schema, @@ -463,10 +522,9 @@ TEST_F(ParquetScanTest, PageIndexCanFullyFilterRowGroupAfterRangeIntersection) { auto file_schema = build_file_schema(*parquet_file_reader); format::FileScanRequest request; - request.column_predicate_filters.push_back( - int32_filter(0, "id", file_schema[0]->type, PredicateType::GE, 32)); - request.column_predicate_filters.push_back( - int32_filter(0, "id", file_schema[0]->type, PredicateType::LT, 32)); + request.local_positions.emplace(format::LocalColumnId(0), format::LocalIndex(0)); + request.conjuncts.push_back(create_int32_zonemap_conjunct(0, Int32ZoneMapExpr::Op::GE, 32)); + request.conjuncts.push_back(create_int32_zonemap_conjunct(0, Int32ZoneMapExpr::Op::LT, 32)); format::parquet::RowGroupScanPlan plan; format::parquet::ParquetScanRange scan_range; @@ -488,8 +546,8 @@ TEST_F(ParquetScanTest, PageIndexFullRangeWhenDisabledOrUnavailable) { auto file_schema = build_file_schema(*parquet_file_reader); format::FileScanRequest request; - request.column_predicate_filters.push_back( - int32_filter(0, "id", file_schema[0]->type, PredicateType::GT, 63)); + request.local_positions.emplace(format::LocalColumnId(0), format::LocalIndex(0)); + request.conjuncts.push_back(create_int32_zonemap_conjunct(0, Int32ZoneMapExpr::Op::GT, 63)); const bool old_enable_page_index = config::enable_parquet_page_index; config::enable_parquet_page_index = false; @@ -511,8 +569,9 @@ TEST_F(ParquetScanTest, PageIndexFullRangeWhenDisabledOrUnavailable) { auto no_index_reader = ::parquet::ParquetFileReader::OpenFile(_file_path, false); auto no_index_schema = build_file_schema(*no_index_reader); format::FileScanRequest no_index_request; - no_index_request.column_predicate_filters.push_back( - int32_filter(0, "id", no_index_schema[0]->type, PredicateType::GT, 3)); + no_index_request.local_positions.emplace(format::LocalColumnId(0), format::LocalIndex(0)); + no_index_request.conjuncts.push_back( + create_int32_zonemap_conjunct(0, Int32ZoneMapExpr::Op::GT, 3)); selected_ranges.clear(); page_skip_plans.clear(); pruning_stats = {}; diff --git a/be/test/format_v2/parquet/parquet_statistics_test.cpp b/be/test/format_v2/parquet/parquet_statistics_test.cpp index ae9ee4bf1e3..615fb479b2c 100644 --- a/be/test/format_v2/parquet/parquet_statistics_test.cpp +++ b/be/test/format_v2/parquet/parquet_statistics_test.cpp @@ -28,14 +28,19 @@ #include <memory> #include <numeric> #include <optional> +#include <set> #include <string> #include <vector> #include "core/data_type/data_type_number.h" #include "core/data_type/data_type_string.h" #include "core/field.h" +#include "exprs/vexpr.h" +#include "exprs/vexpr_context.h" #include "format_v2/file_reader.h" #include "format_v2/parquet/parquet_column_schema.h" +#include "storage/index/zone_map/zonemap_eval_context.h" +#include "storage/index/zone_map/zonemap_filter_result.h" #include "storage/predicate/accept_null_predicate.h" #include "storage/predicate/null_predicate.h" #include "storage/predicate/predicate_creator.h" @@ -146,6 +151,63 @@ format::FileScanRequest request_with_filter(format::FileColumnPredicateFilter fi return request; } +class Int32ZoneMapExpr final : public VExpr { +public: + enum class Op { GE, GT, IS_NULL, IS_NOT_NULL }; + + Int32ZoneMapExpr(int column_id, Op op, int32_t value = 0) + : VExpr(std::make_shared<DataTypeUInt8>(), false), + _column_id(column_id), + _op(op), + _value(value) {} + + const std::string& expr_name() const override { return _expr_name; } + + bool can_evaluate_zonemap_filter() const override { return true; } + + void collect_slot_column_ids(std::set<int>& column_ids) const override { + column_ids.insert(_column_id); + } + + ZoneMapFilterResult evaluate_zonemap_filter(const ZoneMapEvalContext& ctx) const override { + auto zone_map = ctx.zone_map(_column_id); + if (zone_map == nullptr) { + return unsupported_zonemap_filter(ctx); + } + if (_op == Op::IS_NULL) { + return zone_map->has_null ? ZoneMapFilterResult::kMayMatch + : ZoneMapFilterResult::kNoMatch; + } + if (_op == Op::IS_NOT_NULL) { + return zone_map->has_not_null ? ZoneMapFilterResult::kMayMatch + : ZoneMapFilterResult::kNoMatch; + } + if (!zone_map->has_not_null) { + return ZoneMapFilterResult::kNoMatch; + } + const auto literal = Field::create_field<TYPE_INT>(_value); + if (_op == Op::GE) { + return zone_map->max_value < literal ? ZoneMapFilterResult::kNoMatch + : ZoneMapFilterResult::kMayMatch; + } + return zone_map->max_value <= literal ? ZoneMapFilterResult::kNoMatch + : ZoneMapFilterResult::kMayMatch; + } + +private: + int _column_id; + Op _op; + int32_t _value; + const std::string _expr_name = "Int32ZoneMapExpr"; +}; + +format::FileScanRequest request_with_zonemap_conjunct(std::shared_ptr<VExpr> expr) { + format::FileScanRequest request; + request.local_positions.emplace(format::LocalColumnId(0), format::LocalIndex(0)); + request.conjuncts.push_back(VExprContext::create_shared(std::move(expr))); + return request; +} + ::parquet::BlockSplitBloomFilter bloom_filter_for_int32_values(const std::vector<int32_t>& values) { ::parquet::BlockSplitBloomFilter bloom_filter; bloom_filter.Init(::parquet::BlockSplitBloomFilter::kMinimumBloomFilterBytes); @@ -223,45 +285,38 @@ TEST(ParquetStatisticsTransformTest, HandlesMissingStatisticsAndAllNullChunks) { EXPECT_FALSE(all_null_stats.has_min_max); } -TEST(ParquetStatisticsPruningTest, StatisticsPredicatesAndNullPredicatesPruneRowGroups) { +TEST(ParquetStatisticsPruningTest, ExprZonemapPredicatesAndNullPredicatesPruneRowGroups) { auto table = arrow::Table::Make(arrow::schema({arrow::field("i", arrow::int32(), true)}), {int32_array({std::nullopt, std::nullopt, 3, 4, 5, 6})}); auto reader = make_reader(table, 2, false, true); auto schema = build_file_schema(*reader); - format::FileColumnPredicateFilter ge_filter; - ge_filter.file_column_id = format::LocalColumnId(0); - ge_filter.predicates.push_back(create_comparison_predicate<PredicateType::GE>( - 0, "i", schema[0]->type, Field::create_field<TYPE_INT>(5), false)); std::vector<int> selected; format::parquet::ParquetPruningStats pruning_stats; ASSERT_TRUE(format::parquet::select_row_groups_by_statistics( - *reader->metadata(), reader.get(), schema, request_with_filter(ge_filter), + *reader->metadata(), reader.get(), schema, + request_with_zonemap_conjunct( + std::make_shared<Int32ZoneMapExpr>(0, Int32ZoneMapExpr::Op::GE, 5)), nullptr, &selected, false, &pruning_stats) .ok()); EXPECT_EQ(selected, std::vector<int>({2})); EXPECT_EQ(pruning_stats.filtered_row_groups_by_statistics, 2); - format::FileColumnPredicateFilter is_not_null_filter; - is_not_null_filter.file_column_id = format::LocalColumnId(0); - is_not_null_filter.predicates.push_back( - std::make_shared<NullPredicate>(0, "i", false, TYPE_INT)); selected.clear(); ASSERT_TRUE(format::parquet::select_row_groups_by_statistics( *reader->metadata(), reader.get(), schema, - request_with_filter(is_not_null_filter), nullptr, &selected, false, - &pruning_stats) + request_with_zonemap_conjunct(std::make_shared<Int32ZoneMapExpr>( + 0, Int32ZoneMapExpr::Op::IS_NOT_NULL)), + nullptr, &selected, false, &pruning_stats) .ok()); EXPECT_EQ(selected, std::vector<int>({1, 2})); - format::FileColumnPredicateFilter is_null_filter; - is_null_filter.file_column_id = format::LocalColumnId(0); - is_null_filter.predicates.push_back(std::make_shared<NullPredicate>(0, "i", true, TYPE_INT)); selected.clear(); ASSERT_TRUE(format::parquet::select_row_groups_by_statistics( *reader->metadata(), reader.get(), schema, - request_with_filter(is_null_filter), nullptr, &selected, false, - &pruning_stats) + request_with_zonemap_conjunct(std::make_shared<Int32ZoneMapExpr>( + 0, Int32ZoneMapExpr::Op::IS_NULL)), + nullptr, &selected, false, &pruning_stats) .ok()); EXPECT_EQ(selected, std::vector<int>({0})); } @@ -313,26 +368,26 @@ TEST(ParquetStatisticsPruningTest, DictionaryPruningHandlesExcludeIncludeAndUnsu EXPECT_EQ(pruning_stats.filtered_row_groups_by_dictionary, 0); } -TEST(ParquetStatisticsPruningTest, StatisticsRunsBeforeDictionaryAndMissingBloomKeepsRows) { +TEST(ParquetStatisticsPruningTest, ColumnPredicatesUseDictionaryAndMissingBloomKeepsRows) { auto table = arrow::Table::Make(arrow::schema({arrow::field("s", arrow::utf8(), false)}), {string_array({"alpha", "beta", "gamma", "omega"})}); auto reader = make_reader(table, 2, true, true); auto schema = build_file_schema(*reader); - format::FileColumnPredicateFilter beyond_max_filter; - beyond_max_filter.file_column_id = format::LocalColumnId(0); - beyond_max_filter.predicates.push_back(create_comparison_predicate<PredicateType::GT>( - 0, "s", schema[0]->type, Field::create_field<TYPE_STRING>("zzzz"), false)); std::vector<int> selected; format::parquet::ParquetPruningStats pruning_stats; + format::FileColumnPredicateFilter absent_filter; + absent_filter.file_column_id = format::LocalColumnId(0); + absent_filter.predicates.push_back(create_comparison_predicate<PredicateType::EQ>( + 0, "s", schema[0]->type, Field::create_field<TYPE_STRING>("absent"), false)); ASSERT_TRUE(format::parquet::select_row_groups_by_statistics( *reader->metadata(), reader.get(), schema, - request_with_filter(beyond_max_filter), nullptr, &selected, true, + request_with_filter(absent_filter), nullptr, &selected, true, &pruning_stats) .ok()); EXPECT_TRUE(selected.empty()); - EXPECT_EQ(pruning_stats.filtered_row_groups_by_statistics, 2); - EXPECT_EQ(pruning_stats.filtered_row_groups_by_dictionary, 0); + EXPECT_EQ(pruning_stats.filtered_row_groups_by_statistics, 0); + EXPECT_EQ(pruning_stats.filtered_row_groups_by_dictionary, 2); EXPECT_EQ(pruning_stats.filtered_row_groups_by_bloom_filter, 0); auto no_stats_reader = make_reader(table, 2, false, false); diff --git a/be/test/format_v2/table/iceberg_reader_test.cpp b/be/test/format_v2/table/iceberg_reader_test.cpp index 84fe09bc0c5..a52f4f22eb3 100644 --- a/be/test/format_v2/table/iceberg_reader_test.cpp +++ b/be/test/format_v2/table/iceberg_reader_test.cpp @@ -1148,8 +1148,8 @@ TEST(IcebergV2ReaderTest, IcebergVirtualColumnsKeepRowLineageAfterRowGroupPredic std::filesystem::create_directories(test_dir); const auto file_path = (test_dir / "split.parquet").string(); - // ColumnPredicate is used for row-group/statistics pruning. Keep one row per row group so - // id > 2 prunes the first two row groups and leaves only the third file-local row. + // Keep one row per row group so the VExpr ZoneMap path can prune the first two row groups and + // leave only the third file-local row. write_int_pair_parquet_file(file_path, {1, 2, 3}, {10, 20, 30}, {"one", "two", "three"}, 1); std::vector<ColumnDefinition> projected_columns; @@ -1168,7 +1168,8 @@ TEST(IcebergV2ReaderTest, IcebergVirtualColumnsKeepRowLineageAfterRowGroupPredic ASSERT_TRUE(reader.init({ .projected_columns = projected_columns, .column_predicates = std::move(column_predicates), - .conjuncts = {}, + .conjuncts = {prepared_conjunct( + &state, table_int32_greater_than_expr(2, 2, 2))}, .format = FileFormat::PARQUET, .scan_params = nullptr, .io_ctx = nullptr, diff --git a/be/test/format_v2/table_reader_test.cpp b/be/test/format_v2/table_reader_test.cpp index a8659667688..57762c24202 100644 --- a/be/test/format_v2/table_reader_test.cpp +++ b/be/test/format_v2/table_reader_test.cpp @@ -2862,8 +2862,8 @@ TEST(TableReaderTest, OpenReaderBuildsColumnPredicateFilters) { std::filesystem::create_directories(test_dir); const auto file_path = (test_dir / "split.parquet").string(); - // ColumnPredicate is only used for row-group/statistics pruning. Keep one row per row - // group so the predicate can prune the first two row groups and leave only id = 3. + // Keep one row per row group so the VExpr ZoneMap path can prune the first two row groups and + // leave only id = 3. write_int_pair_parquet_file(file_path, {1, 2, 3}, {1, 5, 8}, {"one", "two", "three"}, 1); std::vector<ColumnDefinition> projected_columns; @@ -2882,7 +2882,8 @@ TEST(TableReaderTest, OpenReaderBuildsColumnPredicateFilters) { ASSERT_TRUE(reader.init({ .projected_columns = projected_columns, .column_predicates = std::move(column_predicates), - .conjuncts = {}, + .conjuncts = {prepared_conjunct( + &state, table_int32_greater_than_expr(1, 1, 2))}, .format = FileFormat::PARQUET, .scan_params = nullptr, .io_ctx = nullptr, --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
