Coverage Report

Created: 2025-03-10 17:42

/root/doris/be/src/runtime/fragment_mgr.h
Line
Count
Source (jump to first uncovered line)
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#pragma once
19
20
#include <gen_cpp/FrontendService_types.h>
21
#include <gen_cpp/QueryPlanExtra_types.h>
22
#include <gen_cpp/Types_types.h>
23
#include <gen_cpp/types.pb.h>
24
25
#include <cstdint>
26
#include <functional>
27
#include <iosfwd>
28
#include <memory>
29
#include <mutex>
30
#include <string>
31
#include <unordered_map>
32
#include <vector>
33
34
#include "common/be_mock_util.h"
35
#include "common/status.h"
36
#include "gutil/ref_counted.h"
37
#include "http/rest_monitor_iface.h"
38
#include "runtime/query_context.h"
39
#include "runtime_filter_mgr.h"
40
#include "util/countdown_latch.h"
41
#include "util/hash_util.hpp" // IWYU pragma: keep
42
#include "util/metrics.h"
43
44
namespace butil {
45
class IOBufAsZeroCopyInputStream;
46
}
47
48
namespace doris {
49
extern bvar::Adder<uint64_t> g_fragment_executing_count;
50
extern bvar::Status<uint64_t> g_fragment_last_active_time;
51
52
namespace pipeline {
53
class PipelineFragmentContext;
54
} // namespace pipeline
55
class QueryContext;
56
class ExecEnv;
57
class ThreadPool;
58
class TExecPlanFragmentParams;
59
class PExecPlanFragmentStartRequest;
60
class PMergeFilterRequest;
61
class PPublishFilterRequest;
62
class RuntimeProfile;
63
class RuntimeState;
64
class TPipelineFragmentParams;
65
class TPipelineInstanceParams;
66
class TScanColumnDesc;
67
class TScanOpenParams;
68
class Thread;
69
class WorkloadQueryInfo;
70
71
std::string to_load_error_http_path(const std::string& file_name);
72
73
template <typename Key, typename Value, typename ValueType>
74
class ConcurrentContextMap {
75
public:
76
    using ApplyFunction = std::function<Status(phmap::flat_hash_map<Key, Value>&)>;
77
    ConcurrentContextMap();
78
    Value find(const Key& query_id);
79
    void insert(const Key& query_id, std::shared_ptr<ValueType>);
80
    void clear();
81
    void erase(const Key& query_id);
82
6.81k
    size_t num_items() const {
83
6.81k
        size_t n = 0;
84
872k
        for (auto& pair : _internal_map) {
85
872k
            std::shared_lock lock(*pair.first);
86
872k
            auto& map = pair.second;
87
872k
            n += map.size();
88
872k
        }
89
6.81k
        return n;
90
6.81k
    }
_ZNK5doris20ConcurrentContextMapINS_9TUniqueIdESt8weak_ptrINS_12QueryContextEES3_E9num_itemsEv
Line
Count
Source
82
6.81k
    size_t num_items() const {
83
6.81k
        size_t n = 0;
84
872k
        for (auto& pair : _internal_map) {
85
872k
            std::shared_lock lock(*pair.first);
86
872k
            auto& map = pair.second;
87
872k
            n += map.size();
88
872k
        }
89
6.81k
        return n;
90
6.81k
    }
Unexecuted instantiation: _ZNK5doris20ConcurrentContextMapISt4pairINS_9TUniqueIdEiESt10shared_ptrINS_8pipeline23PipelineFragmentContextEES6_E9num_itemsEv
91
20.4k
    void apply(ApplyFunction&& function) {
92
2.61M
        for (auto& pair : _internal_map) {
93
            // TODO: Now only the cancel worker do the GC the _query_ctx_map. each query must
94
            // do erase the finish query unless in _query_ctx_map. Rethink the logic is ok
95
2.61M
            std::unique_lock lock(*pair.first);
96
2.61M
            static_cast<void>(function(pair.second));
97
2.61M
        }
98
20.4k
    }
_ZN5doris20ConcurrentContextMapISt4pairINS_9TUniqueIdEiESt10shared_ptrINS_8pipeline23PipelineFragmentContextEES6_E5applyEOSt8functionIFNS_6StatusERN5phmap13flat_hash_mapIS3_S7_NSB_4HashIS3_EENSB_7EqualToIS3_EESaIS1_IKS3_S7_EEEEEE
Line
Count
Source
91
6.81k
    void apply(ApplyFunction&& function) {
92
872k
        for (auto& pair : _internal_map) {
93
            // TODO: Now only the cancel worker do the GC the _query_ctx_map. each query must
94
            // do erase the finish query unless in _query_ctx_map. Rethink the logic is ok
95
872k
            std::unique_lock lock(*pair.first);
96
872k
            static_cast<void>(function(pair.second));
97
872k
        }
98
6.81k
    }
_ZN5doris20ConcurrentContextMapINS_9TUniqueIdESt8weak_ptrINS_12QueryContextEES3_E5applyEOSt8functionIFNS_6StatusERN5phmap13flat_hash_mapIS1_S4_NS8_4HashIS1_EENS8_7EqualToIS1_EESaISt4pairIKS1_S4_EEEEEE
Line
Count
Source
91
13.6k
    void apply(ApplyFunction&& function) {
92
1.74M
        for (auto& pair : _internal_map) {
93
            // TODO: Now only the cancel worker do the GC the _query_ctx_map. each query must
94
            // do erase the finish query unless in _query_ctx_map. Rethink the logic is ok
95
1.74M
            std::unique_lock lock(*pair.first);
96
1.74M
            static_cast<void>(function(pair.second));
97
1.74M
        }
98
13.6k
    }
99
100
    Status apply_if_not_exists(const Key& query_id, std::shared_ptr<ValueType>& query_ctx,
101
                               ApplyFunction&& function);
102
103
private:
104
    // The lock should only be used to protect the structures in fragment manager. Has to be
105
    // used in a very small scope because it may dead lock. For example, if the _lock is used
106
    // in prepare stage, the call path is  prepare --> expr prepare --> may call allocator
107
    // when allocate failed, allocator may call query_is_cancelled, query is callced will also
108
    // call _lock, so that there is dead lock.
109
    std::vector<std::pair<std::unique_ptr<std::shared_mutex>, phmap::flat_hash_map<Key, Value>>>
110
            _internal_map;
111
};
112
113
// This class used to manage all the fragment execute in this instance
114
class FragmentMgr : public RestMonitorIface {
115
public:
116
    using FinishCallback = std::function<void(RuntimeState*, Status*)>;
117
118
    FragmentMgr(ExecEnv* exec_env);
119
    ~FragmentMgr() override;
120
121
    void stop();
122
123
    // execute one plan fragment
124
    Status exec_plan_fragment(const TExecPlanFragmentParams& params, const QuerySource query_type);
125
126
    Status exec_plan_fragment(const TPipelineFragmentParams& params, const QuerySource query_type);
127
128
    void remove_pipeline_context(std::pair<TUniqueId, int> key);
129
130
    // TODO(zc): report this is over
131
    Status exec_plan_fragment(const TExecPlanFragmentParams& params, const QuerySource query_type,
132
                              const FinishCallback& cb);
133
134
    Status exec_plan_fragment(const TPipelineFragmentParams& params, const QuerySource query_type,
135
                              const FinishCallback& cb);
136
137
    Status start_query_execution(const PExecPlanFragmentStartRequest* request);
138
139
    Status trigger_pipeline_context_report(const ReportStatusRequest,
140
                                           std::shared_ptr<pipeline::PipelineFragmentContext>&&);
141
142
    // Can be used in both version.
143
    MOCK_FUNCTION void cancel_query(const TUniqueId query_id, const Status reason);
144
145
    void cancel_worker();
146
147
    void debug(std::stringstream& ss) override;
148
149
    // input: TQueryPlanInfo fragment_instance_id
150
    // output: selected_columns
151
    // execute external query, all query info are packed in TScanOpenParams
152
    Status exec_external_plan_fragment(const TScanOpenParams& params,
153
                                       const TQueryPlanInfo& t_query_plan_info,
154
                                       const TUniqueId& query_id,
155
                                       const TUniqueId& fragment_instance_id,
156
                                       std::vector<TScanColumnDesc>* selected_columns);
157
158
    Status apply_filterv2(const PPublishFilterRequestV2* request,
159
                          butil::IOBufAsZeroCopyInputStream* attach_data);
160
161
    Status merge_filter(const PMergeFilterRequest* request,
162
                        butil::IOBufAsZeroCopyInputStream* attach_data);
163
164
    Status send_filter_size(const PSendFilterSizeRequest* request);
165
166
    Status sync_filter_size(const PSyncFilterSizeRequest* request);
167
168
    std::string to_http_path(const std::string& file_name);
169
170
    void coordinator_callback(const ReportStatusRequest& req);
171
172
0
    ThreadPool* get_thread_pool() { return _thread_pool.get(); }
173
174
0
    int32_t running_query_num() { return _query_ctx_map.num_items(); }
175
176
    std::string dump_pipeline_tasks(int64_t duration = 0);
177
    std::string dump_pipeline_tasks(TUniqueId& query_id);
178
179
    void get_runtime_query_info(std::vector<std::weak_ptr<ResourceContext>>* _resource_ctx_list);
180
181
    Status get_realtime_exec_status(const TUniqueId& query_id,
182
                                    TReportExecStatusParams* exec_status);
183
184
    std::shared_ptr<QueryContext> get_query_ctx(const TUniqueId& query_id);
185
186
private:
187
    struct BrpcItem {
188
        TNetworkAddress network_address;
189
        std::vector<std::weak_ptr<QueryContext>> queries;
190
    };
191
192
    template <typename Param>
193
    void _set_scan_concurrency(const Param& params, QueryContext* query_ctx);
194
195
    Status _get_or_create_query_ctx(const TPipelineFragmentParams& params, TUniqueId query_id,
196
                                    bool pipeline, QuerySource query_type,
197
                                    std::shared_ptr<QueryContext>& query_ctx);
198
199
    void _check_brpc_available(const std::shared_ptr<PBackendService_Stub>& brpc_stub,
200
                               const BrpcItem& brpc_item);
201
202
    // This is input params
203
    ExecEnv* _exec_env = nullptr;
204
205
    // (QueryID, FragmentID) -> PipelineFragmentContext
206
    ConcurrentContextMap<std::pair<TUniqueId, int>,
207
                         std::shared_ptr<pipeline::PipelineFragmentContext>,
208
                         pipeline::PipelineFragmentContext>
209
            _pipeline_map;
210
211
    // query id -> QueryContext
212
    ConcurrentContextMap<TUniqueId, std::weak_ptr<QueryContext>, QueryContext> _query_ctx_map;
213
    std::unordered_map<TUniqueId, std::unordered_map<int, int64_t>> _bf_size_map;
214
215
    CountDownLatch _stop_background_threads_latch;
216
    scoped_refptr<Thread> _cancel_thread;
217
    // every job is a pool
218
    std::unique_ptr<ThreadPool> _thread_pool;
219
220
    std::shared_ptr<MetricEntity> _entity;
221
    UIntGauge* timeout_canceled_fragment_count = nullptr;
222
223
    RuntimeFilterMergeController _runtimefilter_controller;
224
};
225
226
uint64_t get_fragment_executing_count();
227
uint64_t get_fragment_last_active_time();
228
229
} // namespace doris