[Exec](profile) add max row count one backend in materialization profile (#59728)

- Add `_backend_rows_count` to `MaterializationSharedState` to count the
number of rows assigned to each backend during block distribution.
- Introduce `_max_rows_per_backend` to record the maximum row count
handled by any single backend in the current batch.
- Update `merge_multi_response()` to compute `_max_rows_per_backend`
after row assignment.
- Expose this metric via a new runtime profile counter
`MaxRowsPerBackend` in `MaterializationLocalState`.
- This helps monitor data skew across backends and aids in performance
debugging for materialization-heavy queries.

Author: HappenLee

be/src/pipeline/exec/materialization_opertor.cpp

@@ -56,6 +56,7 @@ void MaterializationSharedState::get_block(vectorized::Block* block) {
 
 Status MaterializationSharedState::merge_multi_response() {
     std::unordered_map<int64_t, std::pair<vectorized::Block, int>> block_maps;
+
     for (int i = 0; i < block_order_results.size(); ++i) {
         for (auto& [backend_id, rpc_struct] : rpc_struct_map) {
             vectorized::Block partial_block;
@@ -175,12 +176,22 @@ Status MaterializationSharedState::create_muiltget_result(const vectorized::Colu
                 rpc_struct->second.request.mutable_request_block_descs(i)->add_file_id(
                         row_location.file_id);
                 block_order[j] = row_location.backend_id;
+
+                // Count rows per backend
+                _backend_rows_count[row_location.backend_id]++;
             } else {
                 block_order[j] = 0;
             }
         }
     }
 
+    // Update max rows per backend
+    for (const auto& [_, row_count] : _backend_rows_count) {
+        if (row_count > _max_rows_per_backend) {
+            _max_rows_per_backend = row_count;
+        }
+    }
+
     eos = child_eos;
     if (eos && gc_id_map) {
         for (auto& [_, rpc_struct] : rpc_struct_map) {
@@ -361,6 +372,8 @@ Status MaterializationOperator::push(RuntimeState* state, vectorized::Block* in_
         if (local_state._materialization_state.need_merge_block) {
             SCOPED_TIMER(local_state._merge_response_timer);
             RETURN_IF_ERROR(local_state._materialization_state.merge_multi_response());
+            local_state._max_rows_per_backend_counter->set(
+                    (int64_t)local_state._materialization_state._max_rows_per_backend);
         }
     }
 

be/src/pipeline/exec/materialization_opertor.h

@@ -66,6 +66,11 @@ struct MaterializationSharedState {
     std::vector<std::vector<int64_t>> block_order_results;
     // backend id => <rpc profile info string key, rpc profile info string value>.
     std::map<int64_t, std::map<std::string, fmt::memory_buffer>> backend_profile_info_string;
+
+    // Store the maximum number of rows processed by a single backend in the current batch
+    uint32_t _max_rows_per_backend = 0;
+    // Store the number of rows processed by each backend
+    std::unordered_map<int64_t, uint32_t> _backend_rows_count; // backend_id => rows_count
 };
 
 class MaterializationLocalState final : public PipelineXLocalState<FakeSharedState> {
@@ -80,6 +85,8 @@ class MaterializationLocalState final : public PipelineXLocalState<FakeSharedSta
         RETURN_IF_ERROR(Base::init(state, info));
         _max_rpc_timer = ADD_TIMER_WITH_LEVEL(custom_profile(), "MaxRpcTime", 2);
         _merge_response_timer = ADD_TIMER_WITH_LEVEL(custom_profile(), "MergeResponseTime", 2);
+        _max_rows_per_backend_counter =
+                ADD_COUNTER_WITH_LEVEL(custom_profile(), "MaxRowsPerBackend", TUnit::UNIT, 2);
         return Status::OK();
     }
 
@@ -93,6 +100,7 @@ class MaterializationLocalState final : public PipelineXLocalState<FakeSharedSta
     MaterializationSharedState _materialization_state;
     RuntimeProfile::Counter* _max_rpc_timer = nullptr;
     RuntimeProfile::Counter* _merge_response_timer = nullptr;
+    RuntimeProfile::Counter* _max_rows_per_backend_counter = nullptr;
 };
 
 class MaterializationOperator final : public StatefulOperatorX<MaterializationLocalState> {