[refactor](be) Clean up new parquet page skip filter

### What problem does this PR solve?

Issue Number: close #xxx

Related PR: #64214

Problem Summary: The new parquet data page filter setup was implemented as an inline closure inside RecordReader creation, which made the page ordinal tracking, profile updates and page skip plan lookup hard to read. This refactors the filter into a small DataPageSkipFilter helper, centralizes page skip plan lookup/filter installation, and documents the important page-index invariants around data-page ordinals, non-repeated leaves, and double-skip accounting. The remaining touched reader files are formatting-only changes from the project clang-format script.

### Release note

None

### Check List (For Author)

- Test: Unit Test
    - Pending: NewParquetReaderTest.*
- Behavior changed: No
- Does this need documentation: No

Author: suxiaogang223

be/src/format/new_parquet/parquet_statistics.cpp

@@ -1116,6 +1116,8 @@ 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) {

be/src/format/new_parquet/reader/column_reader.cpp

@@ -25,6 +25,7 @@
 #include <cstddef>
 #include <cstdint>
 #include <exception>
+#include <map>
 #include <memory>
 #include <string>
 #include <utility>
@@ -47,6 +48,63 @@
 namespace doris::parquet {
 namespace {
 
+class DataPageSkipFilter {
+public:
+    DataPageSkipFilter(const ParquetPageSkipPlan* page_skip_plan,
+                       ParquetPageSkipProfile page_skip_profile)
+            : _page_skip_plan(page_skip_plan), _page_skip_profile(page_skip_profile) {
+        DORIS_CHECK(_page_skip_plan != nullptr);
+    }
+
+    bool operator()(const ::parquet::DataPageStats&) {
+        // Arrow invokes this callback once for each DATA_PAGE/DATA_PAGE_V2 and never for
+        // dictionary pages, so this ordinal matches Parquet OffsetIndex page locations.
+        const size_t page_idx = _next_data_page_idx++;
+        const bool skip = _page_skip_plan->should_skip_page(page_idx);
+        if (!skip) {
+            return false;
+        }
+        update_skip_profile(page_idx);
+        return true;
+    }
+
+private:
+    void update_skip_profile(size_t page_idx) const {
+        if (_page_skip_profile.skipped_pages != nullptr) {
+            COUNTER_UPDATE(_page_skip_profile.skipped_pages, 1);
+        }
+        if (_page_skip_profile.skipped_bytes != nullptr) {
+            COUNTER_UPDATE(_page_skip_profile.skipped_bytes,
+                           _page_skip_plan->skipped_page_compressed_size(page_idx));
+        }
+    }
+
+    const ParquetPageSkipPlan* _page_skip_plan = nullptr;
+    ParquetPageSkipProfile _page_skip_profile;
+    size_t _next_data_page_idx = 0;
+};
+
+const ParquetPageSkipPlan* find_page_skip_plan(
+        const std::map<int, ParquetPageSkipPlan>* page_skip_plans, int leaf_column_id) {
+    if (page_skip_plans == nullptr) {
+        return nullptr;
+    }
+    const auto plan_it = page_skip_plans->find(leaf_column_id);
+    return plan_it == page_skip_plans->end() ? nullptr : &plan_it->second;
+}
+
+void install_data_page_filter(std::unique_ptr<::parquet::PageReader>& page_reader,
+                              const std::map<int, ParquetPageSkipPlan>* page_skip_plans,
+                              int leaf_column_id, ParquetPageSkipProfile page_skip_profile) {
+    DORIS_CHECK(page_reader != nullptr);
+    const ParquetPageSkipPlan* page_skip_plan =
+            find_page_skip_plan(page_skip_plans, leaf_column_id);
+    if (page_skip_plan == nullptr) {
+        return;
+    }
+    page_reader->set_data_page_filter(DataPageSkipFilter(page_skip_plan, page_skip_profile));
+}
+
 bool supports_nested_scalar_record_reader(const ParquetColumnSchema& column_schema) {
     if (supports_record_reader(column_schema.type_descriptor)) {
         return true;
@@ -156,13 +214,7 @@ Status ParquetColumnReaderFactory::create_scalar_reader(
     if (reader == nullptr) {
         return Status::InvalidArgument("reader is null");
     }
-    const ParquetPageSkipPlan* page_skip_plan = nullptr;
-    if (_page_skip_plans != nullptr) {
-        auto plan_it = _page_skip_plans->find(column_schema.leaf_column_id);
-        if (plan_it != _page_skip_plans->end()) {
-            page_skip_plan = &plan_it->second;
-        }
-    }
+    const auto* page_skip_plan = find_page_skip_plan(_page_skip_plans, column_schema.leaf_column_id);
     *reader = std::make_unique<ScalarColumnReader>(column_schema, std::move(record_reader),
                                                    page_skip_plan, _column_reader_profile);
     return Status::OK();
@@ -246,29 +298,8 @@ Status ParquetColumnReaderFactory::get_record_reader(
     if (_record_readers[leaf_column_id] == nullptr) {
         try {
             auto page_reader = _row_group->GetColumnPageReader(leaf_column_id);
-            if (_page_skip_plans != nullptr) {
-                auto plan_it = _page_skip_plans->find(leaf_column_id);
-                if (plan_it != _page_skip_plans->end()) {
-                    const ParquetPageSkipPlan* page_skip_plan = &plan_it->second;
-                    page_reader->set_data_page_filter(
-                            [page_skip_plan, page_skip_profile = _page_skip_profile,
-                             page_idx = size_t {0}](const ::parquet::DataPageStats&) mutable {
-                                const bool skip = page_skip_plan->should_skip_page(page_idx);
-                                if (skip) {
-                                    if (page_skip_profile.skipped_pages != nullptr) {
-                                        COUNTER_UPDATE(page_skip_profile.skipped_pages, 1);
-                                    }
-                                    if (page_skip_profile.skipped_bytes != nullptr) {
-                                        COUNTER_UPDATE(page_skip_profile.skipped_bytes,
-                                                       page_skip_plan->skipped_page_compressed_size(
-                                                               page_idx));
-                                    }
-                                }
-                                ++page_idx;
-                                return skip;
-                            });
-                }
-            }
+            install_data_page_filter(page_reader, _page_skip_plans, leaf_column_id,
+                                     _page_skip_profile);
             const auto level_info = ::parquet::internal::LevelInfo::ComputeLevelInfo(descriptor);
             _record_readers[leaf_column_id] = ::parquet::internal::RecordReader::Make(
                     descriptor, level_info, ::arrow::default_memory_pool(),

be/src/format/new_parquet/reader/scalar_column_reader.cpp

@@ -113,6 +113,9 @@ int64_t ScalarColumnReader::page_filtered_rows_to_skip(int64_t rows) const {
         const int64_t start = std::max(range.start, _row_group_rows_read);
         const int64_t end = std::min(range_end, skip_end);
         if (start < end) {
+            // Scheduler gap skips are derived from page-index selected_ranges. A page-filtered
+            // range can only overlap such a gap when the whole data page is outside every selected
+            // range, so partial overlap would mean the planner and scheduler are out of sync.
             DORIS_CHECK(start == range.start);
             DORIS_CHECK(end == range_end);
             filtered_rows += end - start;

be/src/format/new_parquet/selection_vector.h

@@ -33,8 +33,12 @@ struct RowRange {
 
 struct ParquetPageSkipPlan {
     int leaf_column_id = -1;
+    // Page ordinal is the data-page ordinal in the column chunk. It intentionally excludes
+    // dictionary pages, matching Arrow PageReader::set_data_page_filter().
     std::vector<uint8_t> skipped_pages;
     std::vector<int64_t> skipped_page_compressed_sizes;
+    // Row ranges covered by skipped data pages. ScalarColumnReader uses these ranges to avoid
+    // calling RecordReader::SkipRecords() again for pages already skipped by Arrow.
     std::vector<RowRange> skipped_ranges;
 
     bool empty() const { return skipped_ranges.empty(); }