github-actions[bot] commented on code in PR #29277: URL: https://github.com/apache/doris/pull/29277#discussion_r1437991891
########## be/src/olap/wal_info.cpp: ########## @@ -0,0 +1,41 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "olap/wal_info.h" +namespace doris { +WalInfo::WalInfo(int64_t wal_id, std::string wal_path, int64_t retry_num, int64_t start_time_ms) Review Comment: warning: pass by value and use std::move [modernize-pass-by-value] be/src/olap/wal_info.cpp:18: ```diff - namespace doris { + + #include <utility> + namespace doris { ``` be/src/olap/wal_info.cpp:21: ```diff - _wal_path(wal_path), + _wal_path(std::move(wal_path)), ``` ########## be/src/olap/wal_info.cpp: ########## @@ -0,0 +1,41 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "olap/wal_info.h" +namespace doris { +WalInfo::WalInfo(int64_t wal_id, std::string wal_path, int64_t retry_num, int64_t start_time_ms) + : _wal_id(wal_id), + _wal_path(wal_path), + _retry_num(retry_num), + _start_time_ms(start_time_ms) {} +WalInfo::~WalInfo() {} +int64_t WalInfo::get_wal_id() { Review Comment: warning: method 'get_wal_id' can be made const [readability-make-member-function-const] ```suggestion int64_t WalInfo::get_wal_id() const { ``` be/src/olap/wal_info.h:24: ```diff - int64_t get_wal_id(); + int64_t get_wal_id() const; ``` ########## be/src/olap/wal_info.cpp: ########## @@ -0,0 +1,41 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "olap/wal_info.h" +namespace doris { +WalInfo::WalInfo(int64_t wal_id, std::string wal_path, int64_t retry_num, int64_t start_time_ms) + : _wal_id(wal_id), + _wal_path(wal_path), + _retry_num(retry_num), + _start_time_ms(start_time_ms) {} +WalInfo::~WalInfo() {} Review Comment: warning: use '= default' to define a trivial destructor [modernize-use-equals-default] ```suggestion WalInfo::~WalInfo() = default; ``` ########## be/src/olap/wal_table.cpp: ########## @@ -41,125 +38,158 @@ namespace doris { WalTable::WalTable(ExecEnv* exec_env, int64_t db_id, int64_t table_id) - : _exec_env(exec_env), _db_id(db_id), _table_id(table_id), _stop(false) {} + : _exec_env(exec_env), _db_id(db_id), _table_id(table_id), _stop(false) { + _http_stream_action = std::make_shared<HttpStreamAction>(exec_env); +} WalTable::~WalTable() {} #ifdef BE_TEST -std::string k_request_line; +Status k_stream_load_exec_status; #endif -bool retry = false; - -void WalTable::add_wals(std::vector<std::string> wals) { +void WalTable::add_wal(int64_t wal_id, std::string wal) { std::lock_guard<std::mutex> lock(_replay_wal_lock); - for (const auto& wal : wals) { - LOG(INFO) << "add replay wal " << wal; - _replay_wal_map.emplace(wal, replay_wal_info {0, UnixMillis(), false}); - } + LOG(INFO) << "add replay wal " << wal; + auto wal_info = std::make_shared<WalInfo>(wal_id, wal, 0, UnixMillis()); + _replay_wal_map.emplace(wal, wal_info); } -Status WalTable::replay_wals() { +void WalTable::pick_relay_wals() { + std::lock_guard<std::mutex> lock(_replay_wal_lock); std::vector<std::string> need_replay_wals; std::vector<std::string> need_erase_wals; - { - std::lock_guard<std::mutex> lock(_replay_wal_lock); - if (_replay_wal_map.empty()) { - return Status::OK(); - } - VLOG_DEBUG << "Start replay wals for db=" << _db_id << ", table=" << _table_id - << ", wal size=" << _replay_wal_map.size(); - for (auto& [wal, info] : _replay_wal_map) { - auto& [retry_num, start_ts, replaying] = info; - if (replaying) { - LOG(INFO) << wal << " is replaying, skip this round"; - return Status::OK(); - } - if (retry_num >= config::group_commit_replay_wal_retry_num) { - LOG(WARNING) << "All replay wal failed, db=" << _db_id << ", table=" << _table_id - << ", wal=" << wal - << ", retry_num=" << config::group_commit_replay_wal_retry_num; - std::string rename_path = _get_tmp_path(wal); - LOG(INFO) << "rename wal from " << wal << " to " << rename_path; - std::rename(wal.c_str(), rename_path.c_str()); - need_erase_wals.push_back(wal); - continue; + for (auto it = _replay_wal_map.begin(); it != _replay_wal_map.end(); it++) { + auto wal_info = it->second; + if (wal_info->get_retry_num() >= config::group_commit_replay_wal_retry_num) { + LOG(WARNING) << "All replay wal failed, db=" << _db_id << ", table=" << _table_id + << ", wal=" << it->first << ", retry_num=" << wal_info->get_retry_num(); + auto st = _rename_to_tmp_path(it->first); + if (!st.ok()) { + LOG(WARNING) << "rename " << it->first << " fail" + << ",st:" << st.to_string(); } - if (_need_replay(info)) { - need_replay_wals.push_back(wal); + need_erase_wals.push_back(it->first); + if (config::wait_relay_wal_finish) { + auto notify_st = _exec_env->wal_mgr()->notify(it->second->get_wal_id()); + if (!notify_st.ok()) { + LOG(WARNING) << "notify wal " << it->second->get_wal_id() << " fail"; + } } + continue; } - std::sort(need_replay_wals.begin(), need_replay_wals.end()); - for (const auto& wal : need_erase_wals) { - if (_replay_wal_map.erase(wal)) { - LOG(INFO) << "erase wal " << wal << " from _replay_wal_map"; - } else { - LOG(WARNING) << "fail to erase wal " << wal << " from _replay_wal_map"; - } + if (_need_replay(wal_info)) { + need_replay_wals.push_back(it->first); } } + for (const auto& wal : need_erase_wals) { + _replay_wal_map.erase(wal); + } + std::sort(need_replay_wals.begin(), need_replay_wals.end()); for (const auto& wal : need_replay_wals) { + _replaying_queue.emplace_back(_replay_wal_map[wal]); + _replay_wal_map.erase(wal); + } +} + +Status WalTable::relay_wal_one_by_one() { + std::vector<std::shared_ptr<WalInfo>> need_retry_wals; + std::vector<std::shared_ptr<WalInfo>> need_delete_wals; + while (!_replaying_queue.empty()) { + std::shared_ptr<WalInfo> wal_info = nullptr; { std::lock_guard<std::mutex> lock(_replay_wal_lock); - if (_stop.load()) { - break; + wal_info = _replaying_queue.front(); + _replaying_queue.pop_front(); + } + wal_info->add_retry_num(); + auto st = _replay_wal_internal(wal_info->get_wal_path()); + if (!st.ok()) { + LOG(WARNING) << "failed replay wal, db=" << _db_id << ", table=" << _table_id + << ", wal=" << wal_info->get_wal_path() << ", st=" << st.to_string(); + if (!st.is<ErrorCode::NOT_FOUND>()) { + need_retry_wals.push_back(wal_info); } else { - auto it = _replay_wal_map.find(wal); - if (it != _replay_wal_map.end()) { - auto& [retry_num, start_time, replaying] = it->second; - replaying = true; - } + need_delete_wals.push_back(wal_info); } + } else { + need_delete_wals.push_back(wal_info); + } + VLOG_NOTICE << "replay wal, db=" << _db_id << ", table=" << _table_id + << ", wal=" << wal_info->get_wal_path() << ", st=" << st.to_string(); + } + { + std::lock_guard<std::mutex> lock(_replay_wal_lock); + for (auto retry_wal_info : need_retry_wals) { + _replay_wal_map.emplace(retry_wal_info->get_wal_path(), retry_wal_info); } - auto st = _replay_wal_internal(wal); + } + for (auto delete_wal_info : need_delete_wals) { + auto st = _delete_wal(delete_wal_info->get_wal_id()); if (!st.ok()) { - std::lock_guard<std::mutex> lock(_replay_wal_lock); - auto it = _replay_wal_map.find(wal); - if (it != _replay_wal_map.end()) { - auto& [retry_num, start_time, replaying] = it->second; - replaying = false; - } - LOG(WARNING) << "failed replay wal, drop this round, db=" << _db_id - << ", table=" << _table_id << ", wal=" << wal << ", st=" << st.to_string(); - break; + LOG(WARNING) << "fail to delete wal " << delete_wal_info->get_wal_path(); } - VLOG_NOTICE << "replay wal, db=" << _db_id << ", table=" << _table_id << ", label=" << wal - << ", st=" << st.to_string(); } return Status::OK(); } - -std::string WalTable::_get_tmp_path(const std::string wal) { - std::vector<std::string> path_element; - doris::vectorized::WalReader::string_split(wal, "/", path_element); - std::stringstream ss; - int index = 0; - while (index < path_element.size() - 3) { - ss << path_element[index] << "/"; - index++; +Status WalTable::replay_wals() { + { + std::lock_guard<std::mutex> lock(_replay_wal_lock); + if (_replay_wal_map.empty()) { + LOG(INFO) << "_replay_wal_map is empty,skip relaying"; + return Status::OK(); + } + if (!_replaying_queue.empty()) { + LOG(INFO) << "_replaying_queue is not empty,skip relaying"; + return Status::OK(); + } } - ss << "tmp/"; - while (index < path_element.size()) { - if (index != path_element.size() - 1) { - ss << path_element[index] << "_"; - } else { - ss << path_element[index]; + VLOG_DEBUG << "Start replay wals for db=" << _db_id << ", table=" << _table_id + << ", wal size=" << _replay_wal_map.size(); + pick_relay_wals(); + RETURN_IF_ERROR(relay_wal_one_by_one()); + return Status::OK(); +} + +Status WalTable::_rename_to_tmp_path(const std::string wal) { + io::Path wal_path = wal; + std::list<std::string> path_element; + for (int i = 0; i < 3; ++i) { + if (!wal_path.has_parent_path()) { + return Status::InternalError("parent path is not enough when rename " + wal); } - index++; + path_element.push_front(wal_path.filename().string()); + wal_path = wal_path.parent_path(); + } + wal_path.append(_exec_env->wal_mgr()->tmp); + for (auto path : path_element) { + wal_path.append(path); } - return ss.str(); + bool exists = false; + RETURN_IF_ERROR(io::global_local_filesystem()->exists(wal_path.parent_path(), &exists)); + if (!exists) { + RETURN_IF_ERROR(io::global_local_filesystem()->create_directory(wal_path.parent_path())); + } + auto res = std::rename(wal.c_str(), wal_path.string().c_str()); + if (res < 0) { + return Status::InternalError("rename fail on path " + wal); + } + LOG(INFO) << "rename wal from " << wal << " to " << wal_path.string(); + return Status::OK(); } -bool WalTable::_need_replay(const doris::WalTable::replay_wal_info& info) { +bool WalTable::_need_replay(std::shared_ptr<WalInfo> wal_info) { Review Comment: warning: method '_need_replay' can be made static [readability-convert-member-functions-to-static] be/src/olap/wal_table.h:56: ```diff - bool _need_replay(std::shared_ptr<WalInfo>); + static bool _need_replay(std::shared_ptr<WalInfo>); ``` ########## be/src/olap/wal_info.cpp: ########## @@ -0,0 +1,41 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "olap/wal_info.h" +namespace doris { +WalInfo::WalInfo(int64_t wal_id, std::string wal_path, int64_t retry_num, int64_t start_time_ms) + : _wal_id(wal_id), + _wal_path(wal_path), + _retry_num(retry_num), + _start_time_ms(start_time_ms) {} +WalInfo::~WalInfo() {} +int64_t WalInfo::get_wal_id() { + return _wal_id; +} +std::string WalInfo::get_wal_path() { + return _wal_path; +} +int64_t WalInfo::get_retry_num() { Review Comment: warning: method 'get_retry_num' can be made const [readability-make-member-function-const] ```suggestion int64_t WalInfo::get_retry_num() const { ``` be/src/olap/wal_info.h:25: ```diff - int64_t get_retry_num(); + int64_t get_retry_num() const; ``` ########## be/src/olap/wal_info.cpp: ########## @@ -0,0 +1,41 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "olap/wal_info.h" +namespace doris { +WalInfo::WalInfo(int64_t wal_id, std::string wal_path, int64_t retry_num, int64_t start_time_ms) + : _wal_id(wal_id), + _wal_path(wal_path), + _retry_num(retry_num), + _start_time_ms(start_time_ms) {} +WalInfo::~WalInfo() {} +int64_t WalInfo::get_wal_id() { + return _wal_id; +} +std::string WalInfo::get_wal_path() { + return _wal_path; +} +int64_t WalInfo::get_retry_num() { + return _retry_num; +} +int64_t WalInfo::get_start_time_ms() { Review Comment: warning: method 'get_start_time_ms' can be made const [readability-make-member-function-const] ```suggestion int64_t WalInfo::get_start_time_ms() const { ``` be/src/olap/wal_info.h:26: ```diff - int64_t get_start_time_ms(); + int64_t get_start_time_ms() const; ``` -- 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]
