add profile in agg node (apache#33)

Author: stdpain

be/src/vec/exec/vaggregation_node.cpp

@@ -41,7 +41,10 @@ AggregationNode::AggregationNode(ObjectPool* pool, const TPlanNode& tnode,
           _output_tuple_desc(NULL),
           _needs_finalize(tnode.agg_node.need_finalize),
           _is_merge(false),
-          _agg_data() {}
+          _agg_data(),
+          _build_timer(nullptr),
+          _exec_timer(nullptr),
+          _merge_timer(nullptr) {}
 
 AggregationNode::~AggregationNode() = default;
 
@@ -68,7 +71,9 @@ Status AggregationNode::init(const TPlanNode& tnode, RuntimeState* state) {
 
 Status AggregationNode::prepare(RuntimeState* state) {
     RETURN_IF_ERROR(ExecNode::prepare(state));
-
+    _build_timer = ADD_TIMER(runtime_profile(), "BuildTime");
+    _exec_timer = ADD_TIMER(runtime_profile(), "ExecTime");
+    _merge_timer = ADD_TIMER(runtime_profile(), "MergeTime");
     SCOPED_TIMER(_runtime_profile->total_time_counter());
     _intermediate_tuple_desc = state->desc_tbl().get_tuple_descriptor(_intermediate_tuple_id);
     _output_tuple_desc = state->desc_tbl().get_tuple_descriptor(_output_tuple_id);
@@ -87,6 +92,11 @@ Status AggregationNode::prepare(RuntimeState* state) {
                                                           output_slot_desc, mem_tracker()));
     }
 
+    // set profile timer to evaluators
+    for (auto& evaluator : _aggregate_evaluators) {
+        evaluator->set_timer(_exec_timer, _merge_timer);
+    }
+
     _offsets_of_aggregate_states.resize(_aggregate_evaluators.size());
 
     for (size_t i = 0; i < _aggregate_evaluators.size(); ++i) {
@@ -300,6 +310,8 @@ Status AggregationNode::_serialize_without_key(RuntimeState* state, Block* block
 
 Status AggregationNode::_execute_without_key(Block* block) {
     DCHECK(_agg_data.without_key != nullptr);
+    LOG(WARNING) << "block rows:" << block->rows();
+    SCOPED_TIMER(_build_timer);
     for (int i = 0; i < _aggregate_evaluators.size(); ++i) {
         _aggregate_evaluators[i]->execute_single_add(
                 block, _agg_data.without_key + _offsets_of_aggregate_states[i]);

be/src/vec/exec/vaggregation_node.h

@@ -147,6 +147,10 @@ class AggregationNode : public ::doris::ExecNode {
     // add tracker here
     Arena _agg_arena_pool;
 
+    RuntimeProfile::Counter* _build_timer;
+    RuntimeProfile::Counter* _exec_timer;
+    RuntimeProfile::Counter* _merge_timer;
+
 private:
     Status _create_agg_status(AggregateDataPtr data);
     Status _destory_agg_status(AggregateDataPtr data);

be/src/vec/exprs/vectorized_agg_fn.cpp

@@ -29,8 +29,10 @@ AggFnEvaluator::AggFnEvaluator(const TExprNode& desc)
         : _fn(desc.fn),
           _return_type(TypeDescriptor::from_thrift(desc.fn.ret_type)),
           _intermediate_type(TypeDescriptor::from_thrift(desc.fn.aggregate_fn.intermediate_type)),
-          _intermediate_slot_desc(NULL),
-          _output_slot_desc(NULL) {}
+          _intermediate_slot_desc(nullptr),
+          _output_slot_desc(nullptr),
+          _exec_timer(nullptr),
+          _merge_timer(nullptr) {}
 
 Status AggFnEvaluator::create(ObjectPool* pool, const TExpr& desc, AggFnEvaluator** result) {
     *result = pool->add(new AggFnEvaluator(desc.nodes[0]));
@@ -111,6 +113,7 @@ void AggFnEvaluator::execute_single_add(Block* block, AggregateDataPtr place, Ar
     std::vector<const IColumn*> column_arguments(columns.size());
     std::transform(columns.cbegin(), columns.cend(), column_arguments.begin(),
                    [](const auto& ptr) { return ptr.get(); });
+    SCOPED_TIMER(_exec_timer);
     _function->addBatchSinglePlace(block->rows(), place, column_arguments.data(), nullptr);
 }
 
@@ -124,11 +127,13 @@ void AggFnEvaluator::execute_batch_add(Block* block, size_t offset, AggregateDat
         column_arguments[i] =
                 block->getByPosition(column_id).column->convertToFullColumnIfConst().get();
     }
+    SCOPED_TIMER(_exec_timer);
     _function->addBatch(block->rows(), places, offset, column_arguments.data(), arena);
 }
 
 void AggFnEvaluator::execute_single_merge(AggregateDataPtr place, ConstAggregateDataPtr rhs,
                                           Arena* arena) {
+    SCOPED_TIMER(_merge_timer);
     _function->merge(place, rhs, arena);
 }
 

be/src/vec/exprs/vectorized_agg_fn.h

@@ -17,6 +17,7 @@
 
 #pragma once
 #include "runtime/types.h"
+#include "util/runtime_profile.h"
 #include "vec/aggregate_functions/aggregate_function.h"
 #include "vec/core/block.h"
 #include "vec/data_types/data_type.h"
@@ -35,6 +36,11 @@ class AggFnEvaluator {
                    const SlotDescriptor* output_slot_desc,
                    const std::shared_ptr<MemTracker>& mem_tracker);
 
+    void set_timer(RuntimeProfile::Counter* exec_timer, RuntimeProfile::Counter* merge_timer) {
+        _exec_timer = exec_timer;
+        _merge_timer = merge_timer;
+    }
+
     Status open(RuntimeState* state);
 
     void close(RuntimeState* state);
@@ -71,6 +77,9 @@ class AggFnEvaluator {
     const SlotDescriptor* _intermediate_slot_desc;
     const SlotDescriptor* _output_slot_desc;
 
+    RuntimeProfile::Counter* _exec_timer;
+    RuntimeProfile::Counter* _merge_timer;
+
     // input context
     std::vector<VExprContext*> _input_exprs_ctxs;