[fix](be) Report physical new parquet skip metrics

### What problem does this PR solve?

Issue Number: close #xxx

Related PR: #64214

Problem Summary: The new parquet page skip profile mixed logical pruning information with physical reader work. Page skip counters used generic names that could be interpreted as all page-index-pruned pages, while they are updated only when Arrow's data page filter callback actually skips a page. ReaderSkipRows also counted scheduler-level logical skips, including rows already removed by page filtering, so it could overstate the actual RecordReader::SkipRecords work. This change renames the page skip counters to data-page-filter-specific names and updates ReaderSkipRows only for rows actually passed to Arrow RecordReader::SkipRecords. Parent complex readers and the synthetic row-position reader no longer add logical read/skip rows to the physical reader counters.

### 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_reader.cpp

@@ -219,8 +219,8 @@ Status ParquetReader::open(std::unique_ptr<reader::FileScanRequest>& request) {
     }
     _state->scan_plan = row_group_plan;
     _state->scheduler.set_page_skip_profile(
-            {.skipped_pages = _parquet_profile.pages_skipped_by_filter,
-             .skipped_bytes = _parquet_profile.page_skip_bytes});
+            {.skipped_pages = _parquet_profile.pages_skipped_by_data_page_filter,
+             .skipped_bytes = _parquet_profile.data_page_filter_skip_bytes});
     _state->scheduler.set_scan_profile({
             .raw_rows_read = _parquet_profile.raw_rows_read,
             .selected_rows = _parquet_profile.selected_rows,
@@ -364,10 +364,10 @@ void ParquetReader::_init_profile() {
                 _profile, "FilteredRowsByGroup", TUnit::UNIT, parquet_profile, 1);
         _parquet_profile.filtered_page_rows = ADD_CHILD_COUNTER_WITH_LEVEL(
                 _profile, "FilteredRowsByPage", TUnit::UNIT, parquet_profile, 1);
-        _parquet_profile.pages_skipped_by_filter = ADD_CHILD_COUNTER_WITH_LEVEL(
-                _profile, "PagesSkippedByFilter", TUnit::UNIT, parquet_profile, 1);
-        _parquet_profile.page_skip_bytes = ADD_CHILD_COUNTER_WITH_LEVEL(
-                _profile, "PageSkipBytes", TUnit::BYTES, parquet_profile, 1);
+        _parquet_profile.pages_skipped_by_data_page_filter = ADD_CHILD_COUNTER_WITH_LEVEL(
+                _profile, "PagesSkippedByDataPageFilter", TUnit::UNIT, parquet_profile, 1);
+        _parquet_profile.data_page_filter_skip_bytes = ADD_CHILD_COUNTER_WITH_LEVEL(
+                _profile, "DataPageFilterSkipBytes", TUnit::BYTES, parquet_profile, 1);
         _parquet_profile.selected_rows = ADD_CHILD_COUNTER_WITH_LEVEL(
                 _profile, "SelectedRows", TUnit::UNIT, parquet_profile, 1);
         _parquet_profile.rows_filtered_by_conjunct = ADD_CHILD_COUNTER_WITH_LEVEL(

be/src/format/new_parquet/parquet_reader.h

@@ -78,8 +78,8 @@ class ParquetReader : public reader::FileReader {
         RuntimeProfile::Counter* selected_row_ranges = nullptr;
         RuntimeProfile::Counter* filtered_group_rows = nullptr;
         RuntimeProfile::Counter* filtered_page_rows = nullptr;
-        RuntimeProfile::Counter* pages_skipped_by_filter = nullptr;
-        RuntimeProfile::Counter* page_skip_bytes = nullptr;
+        RuntimeProfile::Counter* pages_skipped_by_data_page_filter = nullptr;
+        RuntimeProfile::Counter* data_page_filter_skip_bytes = nullptr;
         RuntimeProfile::Counter* selected_rows = nullptr;
         RuntimeProfile::Counter* rows_filtered_by_conjunct = nullptr;
         RuntimeProfile::Counter* total_batches = nullptr;

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

@@ -345,7 +345,6 @@ Status ListColumnReader::read(int64_t rows, MutableColumnPtr& column, int64_t* r
         array_column->get_data_ptr() = std::move(nested_column);
         append_offsets(array_column->get_offsets(), entry_counts);
         append_parent_nulls(parent_null_map, parent_nulls);
-        update_reader_read_rows(*rows_read);
         return Status::OK();
     }
 
@@ -357,7 +356,6 @@ Status ListColumnReader::read(int64_t rows, MutableColumnPtr& column, int64_t* r
         array_column->get_data_ptr() = std::move(nested_column);
         append_offsets(array_column->get_offsets(), entry_counts);
         append_parent_nulls(parent_null_map, parent_nulls);
-        update_reader_read_rows(*rows_read);
         return Status::OK();
     }
 
@@ -397,7 +395,6 @@ Status ListColumnReader::read(int64_t rows, MutableColumnPtr& column, int64_t* r
     append_offsets(array_column->get_offsets(), entry_counts);
     array_column->get_data_ptr() = std::move(nested_column);
     append_parent_nulls(parent_null_map, parent_nulls);
-    update_reader_read_rows(*rows_read);
     return Status::OK();
 }
 
@@ -438,7 +435,6 @@ Status ListColumnReader::skip(int64_t rows) {
         return Status::Corruption("Failed to skip parquet LIST column {}: skipped {} of {} rows",
                                   _name, rows_read, rows);
     }
-    update_reader_skip_rows(rows);
     return Status::OK();
 }
 

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

@@ -629,7 +629,6 @@ Status MapColumnReader::read(int64_t rows, MutableColumnPtr& column, int64_t* ro
                 NestedScalarValueAppender {readers.scalar_value, "MAP", "value",
                                            value_max_definition_level},
                 rows, &context, rows_read));
-        update_reader_read_rows(*rows_read);
         return Status::OK();
     }
 
@@ -647,7 +646,6 @@ Status MapColumnReader::read(int64_t rows, MutableColumnPtr& column, int64_t* ro
                 _name, _type, _nullable_definition_level, _repeated_repetition_level, *readers.key,
                 key_max_definition_level, *readers.list_value, *scalar_list_value_reader,
                 &_key_overflow, &_value_overflow, rows, &context, rows_read));
-        update_reader_read_rows(*rows_read);
         return Status::OK();
     }
 
@@ -656,7 +654,6 @@ Status MapColumnReader::read(int64_t rows, MutableColumnPtr& column, int64_t* ro
             key_max_definition_level, *readers.struct_value, &_struct_value_overflow,
             &_key_overflow, NestedStructValueAppender {readers.struct_value}, rows, &context,
             rows_read));
-    update_reader_read_rows(*rows_read);
     return Status::OK();
 }
 
@@ -697,7 +694,6 @@ Status MapColumnReader::skip(int64_t rows) {
         return Status::Corruption("Failed to skip parquet MAP column {}: skipped {} of {} rows",
                                   _name, rows_read, rows);
     }
-    update_reader_skip_rows(rows);
     return Status::OK();
 }
 

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

@@ -66,7 +66,6 @@ Status RowPositionColumnReader::read(int64_t rows, MutableColumnPtr& column, int
     }
     _next_row_position += rows;
     *rows_read = rows;
-    update_reader_read_rows(rows);
     return Status::OK();
 }
 
@@ -75,7 +74,6 @@ Status RowPositionColumnReader::skip(int64_t rows) {
         return Status::OK();
     }
     _next_row_position += rows;
-    update_reader_skip_rows(rows);
     return Status::OK();
 }
 

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

@@ -70,6 +70,9 @@ Status ScalarColumnReader::skip_records(int64_t rows) {
         return Status::InternalError("Parquet record reader is not initialized for column {}",
                                      _name);
     }
+    if (rows <= 0) {
+        return Status::OK();
+    }
     int64_t skipped_rows = 0;
     try {
         _record_reader->Reset();
@@ -89,6 +92,7 @@ Status ScalarColumnReader::skip_records(int64_t rows) {
         return Status::InternalError("Failed to skip parquet records for column {}: {}", _name,
                                      e.what());
     }
+    update_reader_skip_rows(rows);
     return Status::OK();
 }
 
@@ -129,9 +133,9 @@ Status ScalarColumnReader::skip(int64_t rows) {
 
     const int64_t page_filtered_rows = page_filtered_rows_to_skip(rows);
     DORIS_CHECK(page_filtered_rows <= rows);
-    RETURN_IF_ERROR(skip_records(rows - page_filtered_rows));
+    const int64_t record_reader_skip_rows = rows - page_filtered_rows;
+    RETURN_IF_ERROR(skip_records(record_reader_skip_rows));
     advance_rows_read(rows);
-    update_reader_skip_rows(rows);
     return Status::OK();
 }
 

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

@@ -38,7 +38,6 @@ Status StructColumnReader::read(int64_t rows, MutableColumnPtr& column, int64_t*
     if (_children.empty()) {
         column->resize(static_cast<size_t>(rows));
         *rows_read = rows;
-        update_reader_read_rows(*rows_read);
         return Status::OK();
     }
 
@@ -153,7 +152,6 @@ Status StructColumnReader::read(int64_t rows, MutableColumnPtr& column, int64_t*
             append_parent_nulls(parent_null_map, parent_nulls);
         }
         *rows_read = expected_rows;
-        update_reader_read_rows(*rows_read);
         return Status::OK();
     }
 
@@ -269,7 +267,6 @@ Status StructColumnReader::read(int64_t rows, MutableColumnPtr& column, int64_t*
             append_parent_nulls(parent_null_map, parent_nulls);
         }
         *rows_read = expected_rows;
-        update_reader_read_rows(*rows_read);
         return Status::OK();
     }
 
@@ -305,7 +302,6 @@ Status StructColumnReader::read(int64_t rows, MutableColumnPtr& column, int64_t*
     }
 
     *rows_read = std::max<int64_t>(expected_rows, 0);
-    update_reader_read_rows(*rows_read);
     return Status::OK();
 }
 
@@ -316,7 +312,6 @@ Status StructColumnReader::skip(int64_t rows) {
     for (auto& child_reader : _children) {
         RETURN_IF_ERROR(child_reader->skip(rows));
     }
-    update_reader_skip_rows(rows);
     return Status::OK();
 }
 

be/test/format/new_parquet/parquet_reader_test.cpp

@@ -2060,19 +2060,21 @@ TEST_F(NewParquetReaderTest, PageIndexFilteredPagesDoNotDoubleSkipOutputColumns)
         }
     }
 
-    ASSERT_NE(profile.get_counter("PagesSkippedByFilter"), nullptr);
-    ASSERT_NE(profile.get_counter("PageSkipBytes"), nullptr);
+    ASSERT_NE(profile.get_counter("PagesSkippedByDataPageFilter"), nullptr);
+    ASSERT_NE(profile.get_counter("DataPageFilterSkipBytes"), nullptr);
     ASSERT_NE(profile.get_counter("RawRowsRead"), nullptr);
     ASSERT_NE(profile.get_counter("SelectedRows"), nullptr);
     ASSERT_NE(profile.get_counter("RangeGapSkippedRows"), nullptr);
+    ASSERT_NE(profile.get_counter("ReaderSkipRows"), nullptr);
     ASSERT_NE(profile.get_counter("RowGroupFilterTime"), nullptr);
     ASSERT_NE(profile.get_counter("PageIndexFilterTime"), nullptr);
     ASSERT_NE(profile.get_counter("PageIndexReadTime"), nullptr);
-    EXPECT_GT(profile.get_counter("PagesSkippedByFilter")->value(), 0);
-    EXPECT_GT(profile.get_counter("PageSkipBytes")->value(), 0);
+    EXPECT_GT(profile.get_counter("PagesSkippedByDataPageFilter")->value(), 0);
+    EXPECT_GT(profile.get_counter("DataPageFilterSkipBytes")->value(), 0);
     EXPECT_EQ(profile.get_counter("RawRowsRead")->value(), 64);
     EXPECT_EQ(profile.get_counter("SelectedRows")->value(), 64);
     EXPECT_GT(profile.get_counter("RangeGapSkippedRows")->value(), 0);
+    EXPECT_EQ(profile.get_counter("ReaderSkipRows")->value(), 0);
     EXPECT_GT(profile.get_counter("RowGroupFilterTime")->value(), 0);
     EXPECT_GT(profile.get_counter("PageIndexFilterTime")->value(), 0);
     EXPECT_GT(profile.get_counter("PageIndexReadTime")->value(), 0);