[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

@@ -1032,6 +1032,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(); }

be/src/format/reader/column_mapper.cpp

@@ -34,9 +34,9 @@
 #include "exprs/create_predicate_function.h"
 #include "exprs/vcompound_pred.h"
 #include "exprs/vdirect_in_predicate.h"
+#include "exprs/vectorized_fn_call.h"
 #include "exprs/vexpr_context.h"
 #include "exprs/vin_predicate.h"
-#include "exprs/vectorized_fn_call.h"
 #include "format/reader/expr/cast.h"
 #include "format/reader/expr/literal.h"
 #include "format/reader/expr/slot_ref.h"
@@ -310,8 +310,8 @@ std::string TableColumnMapperOptions::debug_string() const {
 
 std::string TableColumnMapper::debug_string(const ColumnDefinition& column) {
     std::ostringstream out;
-    out << "ColumnDefinition{name=" << column.name << ", identifier="
-        << field_debug_string(column.identifier)
+    out << "ColumnDefinition{name=" << column.name
+        << ", identifier=" << field_debug_string(column.identifier)
         << ", local_id=" << column.local_id << ", type=" << data_type_debug_string(column.type)
         << ", children="
         << join_debug_strings(column.children,
@@ -519,10 +519,10 @@ Status clone_table_expr_tree(const VExprSPtr& expr, VExprSPtr* cloned_expr) {
 
     VExprSPtr cloned;
     if (const auto* table_slot_ref = dynamic_cast<const TableSlotRef*>(expr.get())) {
-        cloned = TableSlotRef::create_shared(
-                table_slot_ref->slot_id(), table_slot_ref->column_id(),
-                table_slot_ref->column_uniq_id(), table_slot_ref->data_type(),
-                table_slot_ref->column_name());
+        cloned = TableSlotRef::create_shared(table_slot_ref->slot_id(), table_slot_ref->column_id(),
+                                             table_slot_ref->column_uniq_id(),
+                                             table_slot_ref->data_type(),
+                                             table_slot_ref->column_name());
     } else if (const auto* vslot_ref = dynamic_cast<const VSlotRef*>(expr.get())) {
         cloned = TableSlotRef::create_shared(vslot_ref->slot_id(), vslot_ref->column_id(),
                                              vslot_ref->column_uniq_id(), vslot_ref->data_type(),
@@ -1258,8 +1258,8 @@ static bool rewrite_binary_slot_literal_predicate(
         return false;
     }
 
-    auto rewritten_literal = rewrite_literal_to_file_type(literal_expr, *rewrite_info,
-                                                         rewrite_context);
+    auto rewritten_literal =
+            rewrite_literal_to_file_type(literal_expr, *rewrite_info, rewrite_context);
     if (rewritten_literal == nullptr) {
         children[literal_child_idx] = original_table_literal(literal_expr, rewrite_context);
         expr->set_children(std::move(children));
@@ -1295,15 +1295,15 @@ static bool rewrite_in_slot_literal_predicate(
         if (literal_expr == nullptr) {
             return false;
         }
-        auto rewritten_literal = rewrite_literal_to_file_type(literal_expr, *rewrite_info,
-                                                             rewrite_context);
+        auto rewritten_literal =
+                rewrite_literal_to_file_type(literal_expr, *rewrite_info, rewrite_context);
         if (rewritten_literal == nullptr) {
             for (size_t restore_idx = 1; restore_idx < children.size(); ++restore_idx) {
                 auto restore_literal = unwrap_literal_for_file_cast(children[restore_idx],
                                                                     rewrite_info->table_type);
                 if (restore_literal != nullptr) {
-                    children[restore_idx] = original_table_literal(restore_literal,
-                                                                   rewrite_context);
+                    children[restore_idx] =
+                            original_table_literal(restore_literal, rewrite_context);
                 }
             }
             expr->set_children(std::move(children));
@@ -1344,8 +1344,7 @@ static VExprSPtr rewrite_table_expr_to_file_expr(
                 // struct_element must see the actual file struct layout. Casting the parent struct
                 // to the output projection can hide filter-only children such as `s.id` in
                 // `SELECT s.name WHERE s.id > 5`.
-                children[0] = create_file_slot_ref(*slot_ref, rewrite_it->second,
-                                                   rewrite_context);
+                children[0] = create_file_slot_ref(*slot_ref, rewrite_it->second, rewrite_context);
                 expr->set_children(std::move(children));
                 return expr;
             }
@@ -1383,8 +1382,8 @@ static VExprSPtr rewrite_table_expr_to_file_expr(
                     global_to_file_slot.find(GlobalIndex(cast_set<size_t>(slot_ref->slot_id())));
             if (rewrite_it != global_to_file_slot.end() &&
                 expr->data_type()->equals(*rewrite_it->second.table_type)) {
-                auto rewritten_child = create_file_slot_ref(*slot_ref, rewrite_it->second,
-                                                            rewrite_context);
+                auto rewritten_child =
+                        create_file_slot_ref(*slot_ref, rewrite_it->second, rewrite_context);
                 if (rewrite_it->second.file_type->equals(*rewrite_it->second.table_type)) {
                     return rewritten_child;
                 }
@@ -1973,9 +1972,8 @@ Status TableColumnMapper::localize_filters(const std::vector<TableFilter>& table
             if (!clone_status.ok()) {
                 continue;
             }
-            auto localized_root =
-                    rewrite_table_expr_to_file_expr(rewrite_root, global_to_file_slot,
-                                                    &rewrite_context);
+            auto localized_root = rewrite_table_expr_to_file_expr(rewrite_root, global_to_file_slot,
+                                                                  &rewrite_context);
             auto localized_conjunct = VExprContext::create_shared(std::move(localized_root));
             RETURN_IF_ERROR(rewrite_context.prepare_created_exprs(localized_conjunct.get()));
             file_request->conjuncts.push_back(std::move(localized_conjunct));

be/src/format/reader/column_mapper.h

@@ -66,7 +66,7 @@ enum TableVirtualColumnType {
 
 enum class FilterConversionType {
     COPY_DIRECTLY, // filter can be copied directly from file layer without any change, e.g. column type and table type are the same and no complex nested projection is involved.
-    CAST_FILTER,   // filter can be converted from file layer by adding a cast, e.g. column type is nullable but table type is not, or file column has a trivial nested projection but table column has a complex nested projection.
+    CAST_FILTER, // filter can be converted from file layer by adding a cast, e.g. column type is nullable but table type is not, or file column has a trivial nested projection but table column has a complex nested projection.
     READER_EXPRESSION,
     FINALIZE_ONLY, // filter cannot be converted to file layer and should be evaluated at table reader finalize phase, e.g. a child column of a nested column is null in file schema.
     CONSTANT,

be/src/format/reader/table/paimon_reader.cpp

@@ -121,8 +121,8 @@ Status PaimonReader::prepare_split(const reader::SplitReadOptions& options) {
 }
 
 reader::TableColumnMappingMode PaimonReader::mapping_mode() const {
-    if (_split_schema_id < 0 || _scan_params == nullptr || !_scan_params->__isset.current_schema_id ||
-        !_scan_params->__isset.history_schema_info) {
+    if (_split_schema_id < 0 || _scan_params == nullptr ||
+        !_scan_params->__isset.current_schema_id || !_scan_params->__isset.history_schema_info) {
         return reader::TableColumnMappingMode::BY_NAME;
     }
     return find_schema(_scan_params, _split_schema_id) == nullptr

be/src/format/reader/table_reader.cpp

@@ -119,8 +119,8 @@ std::string expr_context_debug_string(const VExprContextSPtr& context) {
         return "VExprContext{root=null}";
     }
     std::ostringstream out;
-    out << "VExprContext{root_name=" << root->expr_name()
-        << ", root_debug=" << root->debug_string() << "}";
+    out << "VExprContext{root_name=" << root->expr_name() << ", root_debug=" << root->debug_string()
+        << "}";
     return out.str();
 }
 
@@ -285,11 +285,10 @@ std::string TableReader::debug_string() const {
         << ", table_column_predicates="
         << table_column_predicates_debug_string(_table_column_predicates)
         << ", conjunct_count=" << _conjuncts.size() << ", conjuncts="
-        << join_table_reader_debug_strings(
-                   _conjuncts,
-                   [](const VExprContextSPtr& conjunct) {
-                       return expr_context_debug_string(conjunct);
-                   })
+        << join_table_reader_debug_strings(_conjuncts,
+                                           [](const VExprContextSPtr& conjunct) {
+                                               return expr_context_debug_string(conjunct);
+                                           })
         << ", file_schema="
         << join_table_reader_debug_strings(_data_reader.file_schema,
                                            [](const ColumnDefinition& field) {

be/src/format/reader/table_reader.h

@@ -215,9 +215,7 @@ class TableReader {
     // 切换到下一个 reader 的通用流程。
     // 该方法先关闭当前 reader，再打开下一个具体 reader；子类不应重复实现这个循环。
     Status create_next_reader(bool* eos);
-    virtual TableColumnMappingMode mapping_mode() const {
-        return TableColumnMappingMode::BY_NAME;
-    }
+    virtual TableColumnMappingMode mapping_mode() const { return TableColumnMappingMode::BY_NAME; }
     virtual Status annotate_file_schema(std::vector<ColumnDefinition>* file_schema) {
         DORIS_CHECK(file_schema != nullptr);
         return Status::OK();

be/test/format/reader/expr/cast_test.cpp

@@ -681,8 +681,8 @@ TEST_F(CastTest, ColumnMapperDoesNotLeakRewrittenInPredicateLiteralAcrossSplits)
     ASSERT_TRUE(bigint_mapper.create_mapping(projected_columns, {}, {bigint_file_field}).ok());
     reader::FileScanRequest bigint_request;
     ASSERT_TRUE(bigint_mapper
-                        .create_scan_request({table_filter}, {}, projected_columns,
-                                             &bigint_request, &state)
+                        .create_scan_request({table_filter}, {}, projected_columns, &bigint_request,
+                                             &state)
                         .ok());
     ASSERT_EQ(bigint_request.conjuncts.size(), 1);
     const auto& bigint_localized_expr = bigint_request.conjuncts[0]->root();
@@ -781,8 +781,8 @@ TEST_F(CastTest, ColumnMapperDoesNotLeakRewrittenLiteralAcrossSplits) {
     ASSERT_TRUE(bigint_mapper.create_mapping(projected_columns, {}, {bigint_file_field}).ok());
     reader::FileScanRequest bigint_request;
     ASSERT_TRUE(bigint_mapper
-                        .create_scan_request({table_filter}, {}, projected_columns,
-                                             &bigint_request, &state)
+                        .create_scan_request({table_filter}, {}, projected_columns, &bigint_request,
+                                             &state)
                         .ok());
     ASSERT_EQ(bigint_request.conjuncts.size(), 1);
     const auto& bigint_localized_expr = bigint_request.conjuncts[0]->root();
@@ -864,13 +864,12 @@ TEST_F(CastTest, ColumnMapperDoesNotNestCastFilterAcrossScanRequests) {
     table_filter.global_indices = {reader::GlobalIndex(0)};
 
     reader::FileScanRequest first_request;
-    ASSERT_TRUE(
-            mapper.create_scan_request({table_filter}, {}, projected_columns, &first_request, &state)
-                    .ok());
+    ASSERT_TRUE(mapper.create_scan_request({table_filter}, {}, projected_columns, &first_request,
+                                           &state)
+                        .ok());
     reader::FileScanRequest second_request;
-    ASSERT_TRUE(mapper
-                        .create_scan_request({table_filter}, {}, projected_columns,
-                                             &second_request, &state)
+    ASSERT_TRUE(mapper.create_scan_request({table_filter}, {}, projected_columns, &second_request,
+                                           &state)
                         .ok());
 
     ASSERT_EQ(second_request.conjuncts.size(), 1);
@@ -923,8 +922,8 @@ TEST_F(CastTest, ColumnMapperRewritesPreviousCastFilterToMatchingSplitType) {
     ASSERT_TRUE(bigint_mapper.create_mapping(projected_columns, {}, {bigint_file_field}).ok());
     reader::FileScanRequest bigint_request;
     ASSERT_TRUE(bigint_mapper
-                        .create_scan_request({table_filter}, {}, projected_columns,
-                                             &bigint_request, &state)
+                        .create_scan_request({table_filter}, {}, projected_columns, &bigint_request,
+                                             &state)
                         .ok());
 
     const auto& bigint_localized_expr = bigint_request.conjuncts[0]->root();