github-actions[bot] commented on code in PR #30268:
URL: https://github.com/apache/doris/pull/30268#discussion_r1466157679
##########
be/src/runtime/runtime_filter_mgr.cpp:
##########
@@ -54,21 +54,15 @@ struct AsyncRPCContext {
RuntimeFilterMgr::RuntimeFilterMgr(const UniqueId& query_id,
RuntimeFilterParamsContext* state) {
_state = state;
_state->runtime_filter_mgr = this;
-}
-
-Status RuntimeFilterMgr::init() {
_tracker = std::make_unique<MemTracker>("RuntimeFilterMgr",
ExecEnv::GetInstance()->experimental_mem_tracker());
- return Status::OK();
}
Status RuntimeFilterMgr::get_producer_filter(const int filter_id,
IRuntimeFilter** target) {
- int32_t key = filter_id;
-
std::lock_guard<std::mutex> l(_lock);
- auto iter = _producer_map.find(key);
+ auto iter = _producer_map.find(filter_id);
if (iter == _producer_map.end()) {
- return Status::InvalidArgument("unknown filter: {}, role: PRODUCER",
key);
+ return Status::InvalidArgument("unknown filter: {}, role: PRODUCER",
filter_id);
Review Comment:
warning: method 'get_producer_filter' can be made static
[readability-convert-member-functions-to-static]
be/src/runtime/runtime_filter_mgr.h:80:
```diff
- Status get_producer_filter(const int filter_id, IRuntimeFilter**
producer_filter);
+ static Status get_producer_filter(const int filter_id,
IRuntimeFilter** producer_filter);
```
##########
be/src/runtime/runtime_filter_mgr.h:
##########
@@ -187,13 +184,37 @@ class RuntimeFilterMergeController {
// add a query-id -> entity
// If a query-id -> entity already exists
// add_entity will return a exists entity
- Status add_entity(const TExecPlanFragmentParams& params,
- std::shared_ptr<RuntimeFilterMergeControllerEntity>*
handle,
- RuntimeFilterParamsContext* state);
- Status add_entity(const TPipelineFragmentParams& params,
- const TPipelineInstanceParams& local_params,
+ Status add_entity(const auto& params, UniqueId query_id, const
TQueryOptions& query_options,
std::shared_ptr<RuntimeFilterMergeControllerEntity>*
handle,
- RuntimeFilterParamsContext* state);
+ RuntimeFilterParamsContext* state) {
+ if (!params.__isset.runtime_filter_params ||
+ params.runtime_filter_params.rid_to_runtime_filter.size() == 0) {
+ return Status::OK();
+ }
+
+ // TODO: why we need string, direct use UniqueId
+ std::string query_id_str = query_id.to_string();
+ UniqueId fragment_instance_id = UniqueId(params.fragment_instance_id);
+ uint32_t shard = _get_controller_shard_idx(query_id);
+ std::lock_guard<std::mutex> guard(_controller_mutex[shard]);
+ auto iter = _filter_controller_map[shard].find(query_id_str);
+ if (iter == _filter_controller_map[shard].end()) {
+ *handle = std::shared_ptr<RuntimeFilterMergeControllerEntity>(
+ new RuntimeFilterMergeControllerEntity(state),
+ [this](RuntimeFilterMergeControllerEntity* entity) {
+ static_cast<void>(remove_entity(entity->query_id()));
+ delete entity;
+ });
+ _filter_controller_map[shard][query_id_str] = *handle;
+ const TRuntimeFilterParams& filter_params =
params.runtime_filter_params;
+ RETURN_IF_ERROR(handle->get()->init(query_id,
fragment_instance_id, filter_params,
+ query_options));
Review Comment:
warning: method '_get_controller_shard_idx' can be made static
[readability-convert-member-functions-to-static]
```suggestion
static uint32_t _get_controller_shard_idx(UniqueId& query_id) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
