Repository: StarRocks/starrocks (language: python)

Issue:
[Enhancement] Support PredicateTree for parquet page index
## Why I'm doing:

## What I'm doing:

Fixes #issue

## What type of PR is this:

- [ ] BugFix
- [ ] Feature
- [x] Enhancement
- [ ] Refactor
- [ ] UT
- [ ] Doc
- [ ] Tool

Does this PR entail a change in behavior?

- [ ] Yes, this PR will result in a change in behavior.
- [x] No, this PR will not result in a change in behavior.

If yes, please specify the type of change:

- [ ] Interface/UI changes: syntax, type conversion, expression evaluation, display information
- [ ] Parameter changes: default values, similar parameters but with different default values
- [ ] Policy changes: use new policy to replace old one, functionality automatically enabled
- [ ] Feature removed
- [ ] Miscellaneous: upgrade & downgrade compatibility, etc.

## Checklist:

- [x] I have added test cases for my bug fix or my new feature
- [ ] This pr needs user documentation (for new or modified features or behaviors)
  - [ ] I have added documentation for my new feature or new function
- [ ] This is a backport pr

## Bugfix cherry-pick branch check:
- [x] I have checked the version labels which the pr will be auto-backported to the target branch
  - [x] 3.4
  - [ ] 3.3
  - [ ] 3.2
  - [ ] 3.1
  - [ ] 3.0


Discussion / hints:
#### [Java-Extensions Incremental Coverage Report]


:white_check_mark: **pass** : 0 / 0 (0%)




#### [FE Incremental Coverage Report]


:white_check_mark: **pass** : 0 / 0 (0%)




#### [BE Incremental Coverage Report]

:white_check_mark: **pass** : 82 / 88 (93.18%)




#### file detail

|   |path|covered_line|new_line|coverage|not_covered_line_detail|
|----|----|----|----|----|----|
|:large_blue_circle:|be/src/formats/parquet/scalar_column_reader.cpp|53|58|91.38%|[366, 367, 374, 375, 387]|
|:large_blue_circle:|be/src/formats/parquet/group_reader.cpp|16|17|94.12%|[89]|
|:large_blue_circle:|be/src/formats/parquet/file_reader.cpp|1|1|100.00%|[]|
|:large_blue_circle:|be/src/formats/parquet/zone_map_filter_evaluator.h|9|9|100.00%|[]|
|:large_blue_circle:|be/src/formats/parquet/column_reader.h|3|3|100.00%|[]|




> could you check if all the data has been filtered, will the predicate tree return immediately?

Of course it can

Fix patch:
diff --git a/be/src/formats/parquet/column_reader.h b/be/src/formats/parquet/column_reader.h
index cc1413125fc1d..9af20fa8bcda7 100644
--- a/be/src/formats/parquet/column_reader.h
+++ b/be/src/formats/parquet/column_reader.h
@@ -161,6 +161,15 @@ class ColumnReader {
         return true;
     }
 
+    // return true means page index filter happened
+    // return false means no page index filter happened
+    virtual StatusOr<bool> page_index_zone_map_filter(const std::vector<const ColumnPredicate*>& predicates,
+                                                      SparseRange<uint64_t>* row_ranges, CompoundNodeType pred_relation,
+                                                      const uint64_t rg_first_row, const uint64_t rg_num_rows) {
+        DCHECK(row_ranges->empty());
+        return false;
+    }
+
 private:
     // _parquet_field is generated by parquet format, so ParquetField's children order may different from ColumnReader's children.
     const ParquetField* _parquet_field = nullptr;
diff --git a/be/src/formats/parquet/file_reader.cpp b/be/src/formats/parquet/file_reader.cpp
index 5987a6e7dafc5..b60f51a7e40e1 100644
--- a/be/src/formats/parquet/file_reader.cpp
+++ b/be/src/formats/parquet/file_reader.cpp
@@ -453,6 +453,7 @@ Status FileReader::_init_group_readers() {
     _group_reader_param.not_existed_slots = &fd_scanner_ctx.not_existed_slots;
     // for pageIndex
     _group_reader_param.min_max_conjunct_ctxs = fd_scanner_ctx.min_max_conjunct_ctxs;
+    _group_reader_param.predicate_tree = &fd_scanner_ctx.predicate_tree;
 
     int64_t row_group_first_row = 0;
     // select and create row group readers.
diff --git a/be/src/formats/parquet/group_reader.cpp b/be/src/formats/parquet/group_reader.cpp
index 1dec3b0bdbfc4..ae6bc30126833 100644
--- a/be/src/formats/parquet/group_reader.cpp
+++ b/be/src/formats/parquet/group_reader.cpp
@@ -32,6 +32,7 @@
 #include "formats/parquet/page_index_reader.h"
 #include "formats/parquet/scalar_column_reader.h"
 #include "formats/parquet/schema.h"
+#include "formats/parquet/zone_map_filter_evaluator.h"
 #include "gutil/strings/substitute.h"
 #include "runtime/types.h"
 #include "simd/simd.h"
@@ -79,12 +80,29 @@ Status GroupReader::_deal_with_pageindex() {
     if (config::parquet_page_index_enable) {
         SCOPED_RAW_TIMER(&_param.stats->page_index_ns);
         _param.stats->rows_before_page_index += _row_group_metadata->num_rows;
-        auto page_index_reader =
-                std::make_unique<PageIndexReader>(this, _param.file, _column_readers, _row_group_metadata,
-                                                  _param.min_max_conjunct_ctxs, _param.conjunct_ctxs_by_slot);
-        ASSIGN_OR_RETURN(bool flag, page_index_reader->generate_read_range(_range));
-        if (flag && !_is_group_filtered) {
-            page_index_reader->select_column_offset_index();
+        if (config::parquet_advance_zonemap_filter) {
+            ASSIGN_OR_RETURN(auto sparse_range, _param.predicate_tree->visit(ZoneMapEvaluator<FilterLevel::PAGE_INDEX>{
+                                                        *_param.predicate_tree, this}));
+            if (sparse_range.has_value()) {
+                if (sparse_range.value().empty()) {
+                    // the whole row group has been filtered
+                    _is_group_filtered = true;
+                } else if (sparse_range->span_size() < _row_group_metadata->num_rows) {
+                    // some pages have been filtered
+                    _range = sparse_range.value();
+                    for (const auto& pair : _column_readers) {
+                        pair.second->select_offset_index(_range, _row_group_first_row);
+                    }
+                }
+            }
+        } else {
+            auto page_index_reader =
+                    std::make_unique<PageIndexReader>(this, _param.file, _column_readers, _row_group_metadata,
+                                                      _param.min_max_conjunct_ctxs, _param.conjunct_ctxs_by_slot);
+            ASSIGN_OR_RETURN(bool flag, page_index_reader->generate_read_range(_range));
+            if (flag && !_is_group_filtered) {
+                page_index_reader->select_column_offset_index();
+            }
         }
     }
 
diff --git a/be/src/formats/parquet/group_reader.h b/be/src/formats/parquet/group_reader.h
index d9021c8a47c52..55b6ee7c6176d 100644
--- a/be/src/formats/parquet/group_reader.h
+++ b/be/src/formats/parquet/group_reader.h
@@ -94,6 +94,7 @@ struct GroupReaderParam {
 
     // used for pageIndex
     std::vector<ExprContext*> min_max_conjunct_ctxs;
+    const PredicateTree* predicate_tree = nullptr;
 
     // partition column
     const std::vector<HdfsScannerContext::ColumnInfo>* partition_columns = nullptr;
diff --git a/be/src/formats/parquet/scalar_column_reader.cpp b/be/src/formats/parquet/scalar_column_reader.cpp
index 2658f12b97a2c..e41a88bb670a3 100644
--- a/be/src/formats/parquet/scalar_column_reader.cpp
+++ b/be/src/formats/parquet/scalar_column_reader.cpp
@@ -15,11 +15,12 @@
 #include "formats/parquet/scalar_column_reader.h"
 
 #include "formats/parquet/stored_column_reader_with_index.h"
+#include "formats/parquet/utils.h"
+#include "formats/parquet/zone_map_filter_evaluator.h"
 #include "gutil/casts.h"
 #include "io/shared_buffered_input_stream.h"
 #include "simd/simd.h"
 #include "statistics_helper.h"
-#include "utils.h"
 
 namespace starrocks::parquet {
 
@@ -28,16 +29,20 @@ StatusOr<bool> FixedValueColumnReader::row_group_zone_map_filter(const std::vect
                                                                  const uint64_t rg_first_row,
                                                                  const uint64_t rg_num_rows) const {
     ZoneMapDetail zone_map{_fixed_value, _fixed_value, _fixed_value.is_null()};
-    auto is_satisfy = [&](const ZoneMapDetail& detail) {
-        if (pred_relation == CompoundNodeType::AND) {
-            return std::ranges::all_of(predicates, [&](const auto* pred) { return pred->zone_map_filter(detail); });
-        } else {
-            return predicates.empty() ||
-                   std::ranges::any_of(predicates, [&](const auto* pred) { return pred->zone_map_filter(detail); });
-        }
-    };
+    return ZoneMapEvaluatorUtils::is_satisfy(predicates, zone_map, pred_relation);
+}
+
+StatusOr<bool> FixedValueColumnReader::page_index_zone_map_filter(const std::vector<const ColumnPredicate*>& predicates,
+                                                                  SparseRange<uint64_t>* row_ranges,
+                                                                  CompoundNodeType pred_relation,
+                                                                  const uint64_t rg_first_row,
+                                                                  const uint64_t rg_num_rows) {
+    DCHECK(row_ranges->empty());
+    ZoneMapDetail zone_map{_fixed_value, _fixed_value, _fixed_value.is_null()};
 
-    return is_satisfy(zone_map);
+    // is_satisfy = true means no filter happened, return false
+    // is_satisfy = false means entire row group can be filtered, filter happened, return true
+    return !ZoneMapEvaluatorUtils::is_satisfy(predicates, zone_map, pred_relation);
 }
 
 Status ScalarColumnReader::read_range(const Range<uint64_t>& range, const Filter* filter, ColumnPtr& dst) {
@@ -320,16 +325,96 @@ StatusOr<bool> ScalarColumnReader::row_group_zone_map_filter(const std::vector<c
         return true;
     }
 
-    auto is_satisfy = [&](const ZoneMapDetail& detail) {
-        if (pred_relation == CompoundNodeType::AND) {
-            return std::ranges::all_of(predicates, [&](const auto* pred) { return pred->zone_map_filter(detail); });
+    return ZoneMapEvaluatorUtils::is_satisfy(predicates, zone_map_detail.value(), pred_relation);
+}
+
+StatusOr<bool> ScalarColumnReader::page_index_zone_map_filter(const std::vector<const ColumnPredicate*>& predicates,
+                                                              SparseRange<uint64_t>* row_ranges,
+                                                              CompoundNodeType pred_relation,
+                                                              const uint64_t rg_first_row, const uint64_t rg_num_rows) {
+    DCHECK(row_ranges->empty());
+    const tparquet::ColumnChunk* chunk_meta = get_chunk_metadata();
+    if (!chunk_meta->__isset.column_index_offset || !chunk_meta->__isset.offset_index_offset ||
+        !chunk_meta->__isset.meta_data) {
+        // no page index, dont filter
+        return false;
+    }
+
+    // get column index
+    int64_t column_index_offset = chunk_meta->column_index_offset;
+    uint32_t column_index_length = chunk_meta->column_index_length;
+
+    std::vector<uint8_t> page_index_data;
+    page_index_data.reserve(column_index_length);
+    RETURN_IF_ERROR(_opts.file->read_at_fully(column_index_offset, page_index_data.data(), column_index_length));
+
+    tparquet::ColumnIndex column_index;
+    RETURN_IF_ERROR(deserialize_thrift_msg(page_index_data.data(), &column_index_length, TProtocolType::COMPACT,
+                                           &column_index));
+
+    ASSIGN_OR_RETURN(const tparquet::OffsetIndex* offset_index, get_offset_index(rg_first_row));
+
+    const size_t page_num = column_index.min_values.size();
+
+    ColumnPtr min_column = ColumnHelper::create_column(*_col_type, true);
+    ColumnPtr max_column = ColumnHelper::create_column(*_col_type, true);
+    // deal with min_values
+    auto st = StatisticsHelper::decode_value_into_column(min_column, column_index.min_values, *_col_type,
+                                                         get_column_parquet_field(), _opts.timezone);
+    if (!st.ok()) {
+        // swallow error status
+        LOG(INFO) << "Error when decode min/max statistics, type " << _col_type->debug_string();
+        return false;
+    }
+    // deal with max_values
+    st = StatisticsHelper::decode_value_into_column(max_column, column_index.max_values, *_col_type,
+                                                    get_column_parquet_field(), _opts.timezone);
+    if (!st.ok()) {
+        // swallow error status
+        LOG(INFO) << "Error when decode min/max statistics, type " << _col_type->debug_string();
+        return false;
+    }
+
+    DCHECK_EQ(page_num, min_column->size());
+    DCHECK_EQ(page_num, max_column->size());
+
+    // fill ZoneMapDetail
+    const std::vector<bool> null_pages = column_index.null_pages;
+    std::vector<ZoneMapDetail> zone_map_details{};
+    for (size_t i = 0; i < page_num; i++) {
+        if (null_pages[i]) {
+            // all null
+            zone_map_details.emplace_back(Datum{}, Datum{}, true);
         } else {
-            return predicates.empty() ||
-                   std::ranges::any_of(predicates, [&](const auto* pred) { return pred->zone_map_filter(detail); });
+            bool has_null = column_index.null_counts[i] > 0;
+            zone_map_details.emplace_back(min_column->get(i), max_column->get(i), has_null);
         }
-    };
+    }
+
+    // select all pages by default
+    Filter page_filter(page_num, 1);
+    for (size_t i = 0; i < page_num; i++) {
+        page_filter[i] = ZoneMapEvaluatorUtils::is_satisfy(predicates, zone_map_details[i], pred_relation);
+    }
 
-    return is_satisfy(zone_map_detail.value());
+    if (!SIMD::contain_zero(page_filter)) {
+        // no page has been filtered
+        return false;
+    }
+
+    for (int i = 0; i < page_num; i++) {
+        if (page_filter[i]) {
+            int64_t first_row = offset_index->page_locations[i].first_row_index + rg_first_row;
+            int64_t end_row = first_row;
+            if (i != page_num - 1) {
+                end_row = offset_index->page_locations[i + 1].first_row_index + rg_first_row;
+            } else {
+                end_row = rg_first_row + rg_num_rows;
+            }
+            row_ranges->add(Range<uint64_t>(first_row, end_row));
+        }
+    }
+    return true;
 }
 
 } // namespace starrocks::parquet
\ No newline at end of file
diff --git a/be/src/formats/parquet/scalar_column_reader.h b/be/src/formats/parquet/scalar_column_reader.h
index 8778804375210..063a0036cd60a 100644
--- a/be/src/formats/parquet/scalar_column_reader.h
+++ b/be/src/formats/parquet/scalar_column_reader.h
@@ -47,6 +47,10 @@ class FixedValueColumnReader final : public ColumnReader {
                                              CompoundNodeType pred_relation, const uint64_t rg_first_row,
                                              const uint64_t rg_num_rows) const override;
 
+    StatusOr<bool> page_index_zone_map_filter(const std::vector<const ColumnPredicate*>& predicates,
+                                              SparseRange<uint64_t>* row_ranges, CompoundNodeType pred_relation,
+                                              const uint64_t rg_first_row, const uint64_t rg_num_rows) override;
+
 private:
     const Datum _fixed_value;
 };
@@ -121,6 +125,10 @@ class ScalarColumnReader final : public ColumnReader {
                                              CompoundNodeType pred_relation, uint64_t rg_first_row,
                                              uint64_t rg_num_rows) const override;
 
+    StatusOr<bool> page_index_zone_map_filter(const std::vector<const ColumnPredicate*>& predicates,
+                                              SparseRange<uint64_t>* row_ranges, CompoundNodeType pred_relation,
+                                              const uint64_t rg_first_row, const uint64_t rg_num_rows) override;
+
 private:
     // Returns true if all of the data pages in the column chunk are dict encoded
     bool _column_all_pages_dict_encoded();
diff --git a/be/src/formats/parquet/zone_map_filter_evaluator.h b/be/src/formats/parquet/zone_map_filter_evaluator.h
index 633cf15de1440..519c5ee1c55aa 100644
--- a/be/src/formats/parquet/zone_map_filter_evaluator.h
+++ b/be/src/formats/parquet/zone_map_filter_evaluator.h
@@ -21,6 +21,19 @@ namespace starrocks::parquet {
 
 enum class FilterLevel { ROW_GROUP = 0, PAGE_INDEX };
 
+class ZoneMapEvaluatorUtils {
+public:
+    static bool is_satisfy(const std::vector<const ColumnPredicate*>& predicates, const ZoneMapDetail& detail,
+                           const CompoundNodeType pred_relation) {
+        if (pred_relation == CompoundNodeType::AND) {
+            return std::ranges::all_of(predicates, [&](const auto* pred) { return pred->zone_map_filter(detail); });
+        } else {
+            return predicates.empty() ||
+                   std::ranges::any_of(predicates, [&](const auto* pred) { return pred->zone_map_filter(detail); });
+        }
+    }
+};
+
 template <FilterLevel level>
 struct ZoneMapEvaluator {
     template <CompoundNodeType Type>
@@ -35,7 +48,7 @@ struct ZoneMapEvaluator {
         for (const auto& [cid, col_preds] : cid_to_col_preds) {
             SparseRange<uint64_t> cur_row_ranges;
 
-            const auto* column_reader = group_reader->get_column_reader(cid);
+            auto* column_reader = group_reader->get_column_reader(cid);
 
             if (column_reader == nullptr) {
                 // ColumnReader not found, select all by default
@@ -47,7 +60,14 @@ struct ZoneMapEvaluator {
                     cur_row_ranges.add({rg_first_row, rg_first_row + rg_num_rows});
                 }
             } else {
-                return Status::InternalError("not supported yet");
+                ASSIGN_OR_RETURN(bool has_filtered,
+                                 column_reader->page_index_zone_map_filter(
+                                         col_preds, &cur_row_ranges, Type, group_reader->get_row_group_first_row(),
+                                         group_reader->get_row_group_metadata()->num_rows));
+                if (!has_filtered) {
+                    // no filter happened, select the whole row group by default
+                    cur_row_ranges.add({rg_first_row, rg_first_row + rg_num_rows});
+                }
             }
 
             merge_row_ranges<Type>(row_ranges, cur_row_ranges);
diff --git a/be/src/storage/rowset/segment_iterator.cpp b/be/src/storage/rowset/segment_iterator.cpp
index 1293ffbaaef72..d90903309dc65 100644
--- a/be/src/storage/rowset/segment_iterator.cpp
+++ b/be/src/storage/rowset/segment_iterator.cpp
@@ -46,6 +46,7 @@
 #include "storage/index/vector/vector_index_reader_factory.h"
 #include "storage/index/vector/vector_search_option.h"
 #include "storage/lake/update_manager.h"
+#include "storage/olap_runtime_range_pruner.h"
 #include "storage/olap_runtime_range_pruner.hpp"
 #include "storage/projection_iterator.h"
 #include "storage/range.h"


Test patch:
diff --git a/be/test/exec/hdfs_scanner_test.cpp b/be/test/exec/hdfs_scanner_test.cpp
index 40ae634e0777e..c85baecacabaa 100644
--- a/be/test/exec/hdfs_scanner_test.cpp
+++ b/be/test/exec/hdfs_scanner_test.cpp
@@ -2720,6 +2720,7 @@ TEST_F(HdfsScannerTest, TestMinMaxFilterWhenContainsComplexTypes) {
         ExprContext* ctx = create_expr_context(&_pool, nodes);
         param->min_max_conjunct_ctxs.push_back(ctx);
         param->scanner_conjunct_ctxs.push_back(ctx);
+        param->all_conjunct_ctxs.push_back(ctx);
     }
     {
         std::vector<TExprNode> nodes;
@@ -2729,6 +2730,7 @@ TEST_F(HdfsScannerTest, TestMinMaxFilterWhenContainsComplexTypes) {
         ExprContext* ctx = create_expr_context(&_pool, nodes);
         param->min_max_conjunct_ctxs.push_back(ctx);
         param->scanner_conjunct_ctxs.push_back(ctx);
+        param->all_conjunct_ctxs.push_back(ctx);
     }
 
     ASSERT_OK(Expr::prepare(param->min_max_conjunct_ctxs, _runtime_state));
diff --git a/be/test/formats/parquet/page_index_test.cpp b/be/test/formats/parquet/page_index_test.cpp
index ade15f2b59445..e09b67a933c28 100644
--- a/be/test/formats/parquet/page_index_test.cpp
+++ b/be/test/formats/parquet/page_index_test.cpp
@@ -338,166 +338,205 @@ TEST_F(PageIndexTest, TestRandomReadWith2PageSize) {
 }
 
 TEST_F(PageIndexTest, TestCollectIORangeWithPageIndex) {
-    auto chunk = std::make_shared<Chunk>();
-    chunk->append_column(ColumnHelper::create_column(TypeDescriptor::from_logical_type(LogicalType::TYPE_INT), true),
-                         chunk->num_columns());
-
-    const std::string small_page_file = "./be/test/formats/parquet/test_data/page_index_small_page.parquet";
-
-    Utils::SlotDesc min_max_slots[] = {
-            {"c0", TypeDescriptor::from_logical_type(LogicalType::TYPE_INT), 0},
-            {""},
-    };
+    auto test = [&](bool enable_advanced_zone_map) {
+        config::parquet_advance_zonemap_filter = enable_advanced_zone_map;
+        auto chunk = std::make_shared<Chunk>();
+        chunk->append_column(
+                ColumnHelper::create_column(TypeDescriptor::from_logical_type(LogicalType::TYPE_INT), true),
+                chunk->num_columns());
+
+        const std::string small_page_file = "./be/test/formats/parquet/test_data/page_index_small_page.parquet";
+
+        Utils::SlotDesc min_max_slots[] = {
+                {"c0", TypeDescriptor::from_logical_type(LogicalType::TYPE_INT), 0},
+                {""},
+        };
 
-    auto ctx = _create_file_only_c0_context(small_page_file);
-    auto file = _create_file(small_page_file);
-    ctx->conjunct_ctxs_by_slot[0].clear();
-    ctx->min_max_conjunct_ctxs.clear();
-    ctx->min_max_tuple_desc = Utils::create_tuple_descriptor(_runtime_state, &_pool, min_max_slots);
+        auto ctx = _create_file_only_c0_context(small_page_file);
+        auto file = _create_file(small_page_file);
+        ctx->conjunct_ctxs_by_slot[0].clear();
+        ctx->min_max_conjunct_ctxs.clear();
+        ctx->min_max_tuple_desc = Utils::create_tuple_descriptor(_runtime_state, &_pool, min_max_slots);
 
-    std::vector<TExpr> t_conjuncts;
-    ParquetUTBase::append_int_conjunct(TExprOpcode::GT, 0, 5500, &t_conjuncts);
-    ParquetUTBase::append_int_conjunct(TExprOpcode::LT, 0, 7500, &t_conjuncts);
+        std::vector<TExpr> t_conjuncts;
+        ParquetUTBase::append_int_conjunct(TExprOpcode::GT, 0, 5500, &t_conjuncts);
+        ParquetUTBase::append_int_conjunct(TExprOpcode::LT, 0, 7500, &t_conjuncts);
 
-    ParquetUTBase::create_conjunct_ctxs(&_pool, _runtime_state, &t_conjuncts, &ctx->min_max_conjunct_ctxs);
-    ParquetUTBase::create_conjunct_ctxs(&_pool, _runtime_state, &t_conjuncts, &ctx->conjunct_ctxs_by_slot[0]);
+        ParquetUTBase::create_conjunct_ctxs(&_pool, _runtime_state, &t_conjuncts, &ctx->min_max_conjunct_ctxs);
+        ParquetUTBase::create_conjunct_ctxs(&_pool, _runtime_state, &t_conjuncts, &ctx->conjunct_ctxs_by_slot[0]);
 
-    // tuple desc
-    Utils::SlotDesc slot_descs[] = {
-            {"c0", TypeDescriptor::from_logical_type(LogicalType::TYPE_INT)},
-            {""},
+        // tuple desc
+        Utils::SlotDesc slot_descs[] = {
+                {"c0", TypeDescriptor::from_logical_type(LogicalType::TYPE_INT)},
+                {""},
+        };
+        TupleDescriptor* tuple_desc = Utils::create_tuple_descriptor(_runtime_state, &_pool, slot_descs);
+        std::vector<ExprContext*> all_conjuncts{};
+        for (auto* expr : ctx->min_max_conjunct_ctxs) {
+            all_conjuncts.push_back(expr);
+        }
+        for (auto* expr : ctx->conjunct_ctxs_by_slot[0]) {
+            all_conjuncts.push_back(expr);
+        }
+        ParquetUTBase::setup_conjuncts_manager(all_conjuncts, tuple_desc, _runtime_state, ctx);
+
+        auto file_reader = std::make_shared<FileReader>(config::vector_chunk_size, file.get(),
+                                                        std::filesystem::file_size(small_page_file));
+
+        Status status = file_reader->init(ctx);
+        ASSERT_TRUE(status.ok());
+
+        // two row groups, but one is filtered.
+        EXPECT_EQ(file_reader->_row_group_readers.size(), 1);
+        std::vector<io::SharedBufferedInputStream::IORange> ranges;
+        int64_t end_offset = 0;
+
+        file_reader->_row_group_readers[0]->collect_io_ranges(&ranges, &end_offset, ColumnIOType::PAGE_INDEX);
+        // collect io of column index and offset index for active column.
+        EXPECT_EQ(ranges.size(), 2);
+        // offset_index_offset = 293436, offset_index_length = 113, column_index_offset = 291196, column_index_length = 211
+        EXPECT_EQ(ranges[1].offset, 293436);
+        EXPECT_EQ(ranges[1].size, 113);
+
+        ranges.clear();
+        end_offset = 0;
+
+        file_reader->_row_group_readers[0]->collect_io_ranges(&ranges, &end_offset);
+        // 3 pages, 1 range 5000-8000
+        EXPECT_EQ(file_reader->_row_group_readers[0]->_range.size(), 1);
+        // only collect io of 3 pages, 5001-6000, 6001-7000, 7001-8000 and a dict page.
+        EXPECT_EQ(ranges.size(), 4);
+        // page 7001-8000: offset 50814, size 1660
+        EXPECT_EQ(end_offset, 52474);
+
+        size_t total_row_nums = 0;
+        while (!status.is_end_of_file()) {
+            chunk->reset();
+            status = file_reader->get_next(&chunk);
+            chunk->check_or_die();
+            total_row_nums += chunk->num_rows();
+        }
+        EXPECT_EQ(total_row_nums, 1999);
     };
-    TupleDescriptor* tuple_desc = Utils::create_tuple_descriptor(_runtime_state, &_pool, slot_descs);
-    std::vector<ExprContext*> all_conjuncts{};
-    for (auto* expr : ctx->min_max_conjunct_ctxs) {
-        all_conjuncts.push_back(expr);
-    }
-    for (auto* expr : ctx->conjunct_ctxs_by_slot[0]) {
-        all_conjuncts.push_back(expr);
-    }
-    ParquetUTBase::setup_conjuncts_manager(all_conjuncts, tuple_desc, _runtime_state, ctx);
-
-    auto file_reader = std::make_shared<FileReader>(config::vector_chunk_size, file.get(),
-                                                    std::filesystem::file_size(small_page_file));
 
-    Status status = file_reader->init(ctx);
-    ASSERT_TRUE(status.ok());
-
-    // two row groups, but one is filtered.
-    EXPECT_EQ(file_reader->_row_group_readers.size(), 1);
-    std::vector<io::SharedBufferedInputStream::IORange> ranges;
-    int64_t end_offset = 0;
-
-    file_reader->_row_group_readers[0]->collect_io_ranges(&ranges, &end_offset, ColumnIOType::PAGE_INDEX);
-    // collect io of column index and offset index for active column.
-    EXPECT_EQ(ranges.size(), 2);
-    // offset_index_offset = 293436, offset_index_length = 113, column_index_offset = 291196, column_index_length = 211
-    EXPECT_EQ(ranges[1].offset, 293436);
-    EXPECT_EQ(ranges[1].size, 113);
-
-    ranges.clear();
-    end_offset = 0;
-
-    file_reader->_row_group_readers[0]->collect_io_ranges(&ranges, &end_offset);
-    // 3 pages, 1 range 5000-8000
-    EXPECT_EQ(file_reader->_row_group_readers[0]->_range.size(), 1);
-    // only collect io of 3 pages, 5001-6000, 6001-7000, 7001-8000 and a dict page.
-    EXPECT_EQ(ranges.size(), 4);
-    // page 7001-8000: offset 50814, size 1660
-    EXPECT_EQ(end_offset, 52474);
-
-    size_t total_row_nums = 0;
-    while (!status.is_end_of_file()) {
-        chunk->reset();
-        status = file_reader->get_next(&chunk);
-        chunk->check_or_die();
-        total_row_nums += chunk->num_rows();
-    }
-    EXPECT_EQ(total_row_nums, 1999);
+    bool origin_value = config::parquet_advance_zonemap_filter;
+    test(true);
+    test(false);
+    config::parquet_advance_zonemap_filter = origin_value;
 }
 
 TEST_F(PageIndexTest, TestTwoColumnIntersectPageIndex) {
-    auto chunk = std::make_shared<Chunk>();
-    chunk->append_column(ColumnHelper::create_column(TypeDescriptor::from_logical_type(LogicalType::TYPE_INT), true),
-                         chunk->num_columns());
-    chunk->append_column(ColumnHelper::create_column(TypeDescriptor::from_logical_type(LogicalType::TYPE_INT), true),
-                         chunk->num_columns());
-    chunk->append_column(
-            ColumnHelper::create_column(TypeDescriptor::from_logical_type(LogicalType::TYPE_VARCHAR), true),
-            chunk->num_columns());
-
-    const std::string small_page_file = "./be/test/formats/parquet/test_data/page_index_small_page.parquet";
-
-    Utils::SlotDesc min_max_slots[] = {
-            {"c0", TypeDescriptor::from_logical_type(LogicalType::TYPE_INT), 0},
-            {"c1", TypeDescriptor::from_logical_type(LogicalType::TYPE_INT), 1},
-            {""},
-    };
-
-    auto ctx = _create_file_c0_c1_c2_context(small_page_file);
-    auto file = _create_file(small_page_file);
-    ctx->conjunct_ctxs_by_slot[0].clear();
-    ctx->conjunct_ctxs_by_slot[1].clear();
-    ctx->min_max_conjunct_ctxs.clear();
-    ctx->min_max_tuple_desc = Utils::create_tuple_descriptor(_runtime_state, &_pool, min_max_slots);
-
-    std::vector<TExpr> t_conjuncts;
-    // c0: 1->20000, c0 > 5000
-    ParquetUTBase::append_int_conjunct(TExprOpcode::GT, 0, 5000, &t_conjuncts);
-    // c1: 20000->1, c1 > 5000
-    ParquetUTBase::append_int_conjunct(TExprOpcode::GT, 1, 5000, &t_conjuncts);
-
-    ParquetUTBase::create_conjunct_ctxs(&_pool, _runtime_state, &t_conjuncts, &ctx->min_max_conjunct_ctxs);
+    auto test = [&](bool enable_advanced_zone_map) {
+        config::parquet_advance_zonemap_filter = enable_advanced_zone_map;
+        auto chunk = std::make_shared<Chunk>();
+        chunk->append_column(
+                ColumnHelper::create_column(TypeDescriptor::from_logical_type(LogicalType::TYPE_INT), true),
+                chunk->num_columns());
+        chunk->append_column(
+                ColumnHelper::create_column(TypeDescriptor::from_logical_type(LogicalType::TYPE_INT), true),
+                chunk->num_columns());
+        chunk->append_column(
+                ColumnHelper::create_column(TypeDescriptor::from_logical_type(LogicalType::TYPE_VARCHAR), true),
+                chunk->num_columns());
+
+        const std::string small_page_file = "./be/test/formats/parquet/test_data/page_index_small_page.parquet";
+
+        Utils::SlotDesc min_max_slots[] = {
+                {"c0", TypeDescriptor::from_logical_type(LogicalType::TYPE_INT), 0},
+                {"c1", TypeDescriptor::from_logical_type(LogicalType::TYPE_INT), 1},
+                {""},
+        };
 
-    std::vector<TExpr> t_conjuncts_slot0{t_conjuncts[0]};
-    ParquetUTBase::create_conjunct_ctxs(&_pool, _runtime_state, &t_conjuncts_slot0, &ctx->conjunct_ctxs_by_slot[0]);
+        auto ctx = _create_file_c0_c1_c2_context(small_page_file);
+        auto file = _create_file(small_page_file);
+        ctx->conjunct_ctxs_by_slot[0].clear();
+        ctx->conjunct_ctxs_by_slot[1].clear();
+        ctx->min_max_conjunct_ctxs.clear();
+        ctx->min_max_tuple_desc = Utils::create_tuple_descriptor(_runtime_state, &_pool, min_max_slots);
+
+        std::vector<TExpr> t_conjuncts;
+        // c0: 1->20000, c0 > 5000
+        ParquetUTBase::append_int_conjunct(TExprOpcode::GT, 0, 5000, &t_conjuncts);
+        // c1: 20000->1, c1 > 5000
+        ParquetUTBase::append_int_conjunct(TExprOpcode::GT, 1, 5000, &t_conjuncts);
+
+        ParquetUTBase::create_conjunct_ctxs(&_pool, _runtime_state, &t_conjuncts, &ctx->min_max_conjunct_ctxs);
+
+        std::vector<TExpr> t_conjuncts_slot0{t_conjuncts[0]};
+        ParquetUTBase::create_conjunct_ctxs(&_pool, _runtime_state, &t_conjuncts_slot0, &ctx->conjunct_ctxs_by_slot[0]);
+
+        std::vector<TExpr> t_conjuncts_slot1{t_conjuncts[1]};
+        ParquetUTBase::create_conjunct_ctxs(&_pool, _runtime_state, &t_conjuncts_slot1, &ctx->conjunct_ctxs_by_slot[1]);
+
+        // tuple desc
+        Utils::SlotDesc slot_descs[] = {
+                {"c0", TypeDescriptor::from_logical_type(LogicalType::TYPE_INT)},
+                {"c1", TypeDescriptor::from_logical_type(LogicalType::TYPE_INT)},
+                {"c2", TypeDescriptor::from_logical_type(LogicalType::TYPE_VARCHAR)},
+                {""},
+        };
+        TupleDescriptor* tuple_desc = Utils::create_tuple_descriptor(_runtime_state, &_pool, slot_descs);
+        std::vector<ExprContext*> all_conjuncts{};
+        for (auto* expr : ctx->min_max_conjunct_ctxs) {
+            all_conjuncts.push_back(expr);
+        }
+        for (auto* expr : ctx->conjunct_ctxs_by_slot[0]) {
+            all_conjuncts.push_back(expr);
+        }
+        for (auto* expr : ctx->conjunct_ctxs_by_slot[1]) {
+            all_conjuncts.push_back(expr);
+        }
+        ParquetUTBase::setup_conjuncts_manager(all_conjuncts, tuple_desc, _runtime_state, ctx);
 
-    std::vector<TExpr> t_conjuncts_slot1{t_conjuncts[1]};
-    ParquetUTBase::create_conjunct_ctxs(&_pool, _runtime_state, &t_conjuncts_slot1, &ctx->conjunct_ctxs_by_slot[1]);
+        auto shared_buffer = std::make_shared<io::SharedBufferedInputStream>(
+                file->stream(), small_page_file, std::filesystem::file_size(small_page_file));
+        auto file_reader = std::make_shared<FileReader>(config::vector_chunk_size, file.get(),
+                                                        std::filesystem::file_size(small_page_file), DataCacheOptions(),
+                                                        shared_buffer.get());
 
-    auto shared_buffer = std::make_shared<io::SharedBufferedInputStream>(file->stream(), small_page_file,
-                                                                         std::filesystem::file_size(small_page_file));
-    auto file_reader = std::make_shared<FileReader>(config::vector_chunk_size, file.get(),
-                                                    std::filesystem::file_size(small_page_file), DataCacheOptions(),
-                                                    shared_buffer.get());
+        Status status = file_reader->init(ctx);
+        ASSERT_TRUE(status.ok());
 
-    Status status = file_reader->init(ctx);
-    ASSERT_TRUE(status.ok());
+        // two row groups.
+        EXPECT_EQ(file_reader->_row_group_readers.size(), 2);
+        std::vector<io::SharedBufferedInputStream::IORange> ranges;
+        int64_t end_offset = 0;
 
-    // two row groups.
-    EXPECT_EQ(file_reader->_row_group_readers.size(), 2);
-    std::vector<io::SharedBufferedInputStream::IORange> ranges;
-    int64_t end_offset = 0;
+        for (auto& r : file_reader->_row_group_readers) {
+            r->collect_io_ranges(&ranges, &end_offset, ColumnIOType::PAGE_INDEX);
+        }
 
-    for (auto& r : file_reader->_row_group_readers) {
-        r->collect_io_ranges(&ranges, &end_offset, ColumnIOType::PAGE_INDEX);
-    }
+        // collect io of column index and offset index for active column,
+        // and offset index for lazy column
+        // and two group collect together. (2 + 2 + 1) * 2 = 10
+        EXPECT_EQ(ranges.size(), 10);
 
-    // collect io of column index and offset index for active column,
-    // and offset index for lazy column
-    // and two group collect together. (2 + 2 + 1) * 2 = 10
-    EXPECT_EQ(ranges.size(), 10);
+        ranges.clear();
+        end_offset = 0;
 
-    ranges.clear();
-    end_offset = 0;
+        file_reader->_row_group_readers[0]->collect_io_ranges(&ranges, &end_offset);
+        // only collect io of 5 pages, 5001-6000, 6001-7000, 7001-8000, 8001-9000, 9001-10000 and a dict page.
+        // three columns, (5 + 1) * 3 = 18
+        EXPECT_EQ(ranges.size(), 18);
 
-    file_reader->_row_group_readers[0]->collect_io_ranges(&ranges, &end_offset);
-    // only collect io of 5 pages, 5001-6000, 6001-7000, 7001-8000, 8001-9000, 9001-10000 and a dict page.
-    // three columns, (5 + 1) * 3 = 18
-    EXPECT_EQ(ranges.size(), 18);
+        EXPECT_EQ(shared_buffer->current_range_ref_sum(), 28);
 
-    EXPECT_EQ(shared_buffer->current_range_ref_sum(), 28);
+        // The second row group is not prepare yet
 
-    // The second row group is not prepare yet
+        size_t total_row_nums = 0;
+        while (!status.is_end_of_file()) {
+            chunk->reset();
+            status = file_reader->get_next(&chunk);
+            chunk->check_or_die();
+            total_row_nums += chunk->num_rows();
+        }
+        EXPECT_EQ(total_row_nums, 10000);
+    };
 
-    size_t total_row_nums = 0;
-    while (!status.is_end_of_file()) {
-        chunk->reset();
-        status = file_reader->get_next(&chunk);
-        chunk->check_or_die();
-        total_row_nums += chunk->num_rows();
-    }
-    EXPECT_EQ(total_row_nums, 10000);
+    bool origin_value = config::parquet_advance_zonemap_filter;
+    test(true);
+    test(false);
+    config::parquet_advance_zonemap_filter = origin_value;
 }
 
 TEST_F(PageIndexTest, TestPageIndexNoPageFiltered) {