chaoyli closed pull request #505: Change reference to pointer in task
URL: https://github.com/apache/incubator-doris/pull/505
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/be/src/agent/task_worker_pool.cpp
b/be/src/agent/task_worker_pool.cpp
index 3c6c0148..a25611fb 100644
--- a/be/src/agent/task_worker_pool.cpp
+++ b/be/src/agent/task_worker_pool.cpp
@@ -41,6 +41,7 @@
#include "olap/data_dir.h"
#include "olap/snapshot_manager.h"
#include "olap/task/engine_checksum_task.h"
+#include "olap/task/engine_clear_alter_task.h"
#include "olap/task/engine_clone_task.h"
#include "olap/task/engine_cancel_delete_task.h"
#include "olap/task/engine_schema_change_task.h"
@@ -68,16 +69,11 @@ using std::vector;
namespace doris {
-const uint32_t DOWNLOAD_FILE_MAX_RETRY = 3;
const uint32_t TASK_FINISH_MAX_RETRY = 3;
const uint32_t PUBLISH_VERSION_MAX_RETRY = 3;
const uint32_t REPORT_TASK_WORKER_COUNT = 1;
const uint32_t REPORT_DISK_STATE_WORKER_COUNT = 1;
const uint32_t REPORT_OLAP_TABLE_WORKER_COUNT = 1;
-const uint32_t LIST_REMOTE_FILE_TIMEOUT = 15;
-const std::string HTTP_REQUEST_PREFIX = "/api/_tablet/_download?";
-const std::string HTTP_REQUEST_TOKEN_PARAM = "token=";
-const std::string HTTP_REQUEST_FILE_PARAM = "&file=";
std::atomic_ulong TaskWorkerPool::_s_report_version(time(NULL) * 10000);
Mutex TaskWorkerPool::_s_task_signatures_lock;
@@ -590,8 +586,13 @@ void TaskWorkerPool::_alter_tablet(
// Do not need to adjust delete success or not
// Because if delete failed create rollup will failed
if (status == DORIS_SUCCESS) {
- EngineSchemaChangeTask engine_task(alter_tablet_request, signature,
task_type, error_msgs, process_name);
- status = engine_task.execute();
+ EngineSchemaChangeTask engine_task(alter_tablet_request, signature,
task_type, &error_msgs, process_name);
+ OLAPStatus sc_status = engine_task.execute();
+ if (sc_status != OLAP_SUCCESS) {
+ status = DORIS_ERROR;
+ } else {
+ status = DORIS_SUCCESS;
+ }
}
if (status == DORIS_SUCCESS) {
@@ -709,8 +710,8 @@ void* TaskWorkerPool::_push_worker_thread_callback(void*
arg_this) {
<< " user: " << user << " priority: " << priority;
vector<TTabletInfo> tablet_infos;
- EngineBatchLoadTask task(push_req, &tablet_infos,
agent_task_req.signature);
- status = task.execute();
+ EngineBatchLoadTask task(push_req, &tablet_infos,
agent_task_req.signature, &status);
+ task.execute();
#ifndef BE_TEST
if (status == DORIS_PUSH_HAD_LOADED) {
@@ -857,9 +858,8 @@ void*
TaskWorkerPool::_clear_alter_task_worker_thread_callback(void* arg_this) {
TStatusCode::type status_code = TStatusCode::OK;
vector<string> error_msgs;
TStatus task_status;
-
- OLAPStatus clear_status = worker_pool_this->_env->olap_engine()->
- clear_alter_task(clear_alter_task_req.tablet_id,
clear_alter_task_req.schema_hash);
+ EngineClearAlterTask engine_task(clear_alter_task_req);
+ OLAPStatus clear_status = engine_task.execute();
if (clear_status != OLAPStatus::OLAP_SUCCESS) {
OLAP_LOG_WARNING("clear alter task failed. [signature: %ld
status=%d]",
agent_task_req.signature, clear_status);
@@ -961,8 +961,10 @@ void* TaskWorkerPool::_clone_worker_thread_callback(void*
arg_this) {
vector<string> error_msgs;
vector<TTabletInfo> tablet_infos;
- EngineCloneTask engine_task(clone_req, error_msgs, tablet_infos);
- status = engine_task.execute();
+ EngineCloneTask engine_task(clone_req, &error_msgs, &tablet_infos,
+ &status, agent_task_req.signature,
+ worker_pool_this->_master_info);
+ engine_task.execute();
// Return result to fe
TStatus task_status;
TFinishTaskRequest finish_task_request;
@@ -1020,8 +1022,8 @@ void*
TaskWorkerPool::_storage_medium_migrate_worker_thread_callback(void* arg_t
vector<string> error_msgs;
TStatus task_status;
EngineStorageMigrationTask task(storage_medium_migrate_req);
- AgentStatus res = task.execute();
- if (res != DORIS_SUCCESS) {
+ OLAPStatus res = task.execute();
+ if (res != OLAP_SUCCESS) {
OLAP_LOG_WARNING("storage media migrate failed. status: %d,
signature: %ld",
res, agent_task_req.signature);
status_code = TStatusCode::RUNTIME_ERROR;
@@ -1072,8 +1074,8 @@ void*
TaskWorkerPool::_cancel_delete_data_worker_thread_callback(void* arg_this)
TStatus task_status;
EngineCancelDeleteTask engine_task(cancel_delete_data_req);
- AgentStatus cancel_delete_data_status = engine_task.execute();
- if (cancel_delete_data_status != DORIS_SUCCESS) {
+ OLAPStatus cancel_delete_data_status = engine_task.execute();
+ if (cancel_delete_data_status != OLAP_SUCCESS) {
OLAP_LOG_WARNING("cancel delete data failed. statusta: %d,
signature: %ld",
cancel_delete_data_status,
agent_task_req.signature);
status_code = TStatusCode::RUNTIME_ERROR;
@@ -1131,8 +1133,8 @@ void*
TaskWorkerPool::_check_consistency_worker_thread_callback(void* arg_this)
check_consistency_req.version,
check_consistency_req.version_hash,
&checksum);
- AgentStatus res = engine_task.execute();
- if (res != DORIS_SUCCESS) {
+ OLAPStatus res = engine_task.execute();
+ if (res != OLAP_SUCCESS) {
OLAP_LOG_WARNING("check consistency failed. status: %d, signature:
%ld",
res, agent_task_req.signature);
status_code = TStatusCode::RUNTIME_ERROR;
diff --git a/be/src/http/action/checksum_action.cpp
b/be/src/http/action/checksum_action.cpp
index 747aa6fe..603512e6 100644
--- a/be/src/http/action/checksum_action.cpp
+++ b/be/src/http/action/checksum_action.cpp
@@ -126,11 +126,11 @@ void ChecksumAction::handle(HttpRequest *req) {
int64_t ChecksumAction::do_checksum(int64_t tablet_id, int64_t version,
int64_t version_hash,
int32_t schema_hash, HttpRequest *req) {
- AgentStatus res = DORIS_SUCCESS;
+ OLAPStatus res = OLAP_SUCCESS;
uint32_t checksum;
EngineChecksumTask engine_task(tablet_id, schema_hash, version,
version_hash, &checksum);
res = engine_task.execute();
- if (res != DORIS_SUCCESS) {
+ if (res != OLAP_SUCCESS) {
LOG(WARNING) << "checksum failed. status: " << res
<< ", signature: " << tablet_id;
return -1L;
diff --git a/be/src/olap/CMakeLists.txt b/be/src/olap/CMakeLists.txt
index f3a7204d..ddaf685e 100644
--- a/be/src/olap/CMakeLists.txt
+++ b/be/src/olap/CMakeLists.txt
@@ -72,8 +72,12 @@ add_library(Olap STATIC
types.cpp
utils.cpp
wrapper_field.cpp
- task/engine_schema_change_task.cpp
task/engine_batch_load_task.cpp
+ task/engine_cancel_delete_task.cpp
+ task/engine_checksum_task.cpp
+ task/engine_clear_alter_task.cpp
+ task/engine_clone_task.cpp
+ task/engine_schema_change_task.cpp
task/engine_storage_migration_task.cpp
)
diff --git a/be/src/olap/snapshot_manager.cpp b/be/src/olap/snapshot_manager.cpp
index bd284c50..d9e84ede 100644
--- a/be/src/olap/snapshot_manager.cpp
+++ b/be/src/olap/snapshot_manager.cpp
@@ -200,7 +200,7 @@ OLAPStatus SnapshotManager::_link_index_and_data_files(
int32_t segment_group_id = segment_group_entity.segment_group_id;
for (int seg_id = 0; seg_id < segment_group_entity.num_segments;
++seg_id) {
std::string index_path =
- _construct_index_file_path(tablet_path_prefix, version,
v_hash, segment_group_id, seg_id);
+ construct_index_file_path(tablet_path_prefix, version,
v_hash, segment_group_id, seg_id);
std::string ref_tablet_index_path =
ref_tablet->construct_index_file_path(version, v_hash,
segment_group_id, seg_id);
res = _create_hard_link(ref_tablet_index_path, index_path);
@@ -213,7 +213,7 @@ OLAPStatus SnapshotManager::_link_index_and_data_files(
}
std:: string data_path =
- _construct_data_file_path(tablet_path_prefix, version,
v_hash, segment_group_id, seg_id);
+ construct_data_file_path(tablet_path_prefix, version,
v_hash, segment_group_id, seg_id);
std::string ref_tablet_data_path =
ref_tablet->construct_data_file_path(version, v_hash,
segment_group_id, seg_id);
res = _create_hard_link(ref_tablet_data_path, data_path);
diff --git a/be/src/olap/storage_engine.cpp b/be/src/olap/storage_engine.cpp
index 84e17068..e80c78c2 100644
--- a/be/src/olap/storage_engine.cpp
+++ b/be/src/olap/storage_engine.cpp
@@ -72,12 +72,6 @@ using std::vector;
namespace doris {
StorageEngine* StorageEngine::_s_instance = nullptr;
-const std::string HTTP_REQUEST_PREFIX = "/api/_tablet/_download?";
-const std::string HTTP_REQUEST_TOKEN_PARAM = "token=";
-const std::string HTTP_REQUEST_FILE_PARAM = "&file=";
-
-const uint32_t DOWNLOAD_FILE_MAX_RETRY = 3;
-const uint32_t LIST_REMOTE_FILE_TIMEOUT = 15;
static Status _validate_options(const EngineOptions& options) {
if (options.store_paths.empty()) {
@@ -986,70 +980,4 @@ OLAPStatus StorageEngine::load_header(
return res;
}
-OLAPStatus StorageEngine::clear_alter_task(const TTabletId tablet_id,
- const TSchemaHash schema_hash) {
- LOG(INFO) << "begin to process clear alter task. tablet_id=" << tablet_id
- << ", schema_hash=" << schema_hash;
- TabletSharedPtr tablet = TabletManager::instance()->get_tablet(tablet_id,
schema_hash);
- if (tablet.get() == NULL) {
- OLAP_LOG_WARNING("can't find tablet when process clear alter task. ",
- "[tablet_id=%ld, schema_hash=%d]", tablet_id,
schema_hash);
- return OLAP_SUCCESS;
- }
-
- // get schema change info
- AlterTabletType type;
- TTabletId related_tablet_id;
- TSchemaHash related_schema_hash;
- vector<Version> schema_change_versions;
- tablet->obtain_header_rdlock();
- bool ret = tablet->get_schema_change_request(
- &related_tablet_id, &related_schema_hash, &schema_change_versions,
&type);
- tablet->release_header_lock();
- if (!ret) {
- return OLAP_SUCCESS;
- } else if (!schema_change_versions.empty()) {
- OLAP_LOG_WARNING("find alter task unfinished when process clear alter
task. ",
- "[tablet=%s versions_to_change_size=%d]",
- tablet->full_name().c_str(),
schema_change_versions.size());
- return OLAP_ERR_PREVIOUS_SCHEMA_CHANGE_NOT_FINISHED;
- }
-
- // clear schema change info
- tablet->obtain_header_wrlock();
- tablet->clear_schema_change_request();
- OLAPStatus res = tablet->save_tablet_meta();
- if (res != OLAP_SUCCESS) {
- LOG(FATAL) << "fail to save header. [res=" << res << " tablet='" <<
tablet->full_name() << "']";
- } else {
- LOG(INFO) << "clear alter task on tablet. [tablet='" <<
tablet->full_name() << "']";
- }
- tablet->release_header_lock();
-
- // clear related tablet's schema change info
- TabletSharedPtr related_tablet =
TabletManager::instance()->get_tablet(related_tablet_id, related_schema_hash);
- if (related_tablet.get() == NULL) {
- OLAP_LOG_WARNING("related tablet not found when process clear alter
task. "
- "[tablet_id=%ld schema_hash=%d "
- "related_tablet_id=%ld related_schema_hash=%d]",
- tablet_id, schema_hash, related_tablet_id,
related_schema_hash);
- } else {
- related_tablet->obtain_header_wrlock();
- related_tablet->clear_schema_change_request();
- res = related_tablet->save_tablet_meta();
- if (res != OLAP_SUCCESS) {
- LOG(FATAL) << "fail to save header. [res=" << res << " tablet='"
- << related_tablet->full_name() << "']";
- } else {
- LOG(INFO) << "clear alter task on tablet. [tablet='" <<
related_tablet->full_name() << "']";
- }
- related_tablet->release_header_lock();
- }
-
- LOG(INFO) << "finish to process clear alter task."
- << "tablet_id=" << related_tablet_id
- << ", schema_hash=" << related_schema_hash;
- return OLAP_SUCCESS;
-}
-
} // namespace doris
diff --git a/be/src/olap/storage_engine.h b/be/src/olap/storage_engine.h
index 1b4a6a12..ed315387 100644
--- a/be/src/olap/storage_engine.h
+++ b/be/src/olap/storage_engine.h
@@ -182,9 +182,6 @@ class StorageEngine {
TTabletId tablet_id,
TSchemaHash schema_hash);
- OLAPStatus clear_alter_task(const TTabletId tablet_id,
- const TSchemaHash schema_hash);
-
// call this if you want to trigger a disk and tablet report
void report_notify(bool is_all) {
is_all ? _report_cv.notify_all() : _report_cv.notify_one();
diff --git a/be/src/olap/task/engine_batch_load_task.cpp
b/be/src/olap/task/engine_batch_load_task.cpp
index dc489228..a05a0f58 100644
--- a/be/src/olap/task/engine_batch_load_task.cpp
+++ b/be/src/olap/task/engine_batch_load_task.cpp
@@ -46,15 +46,19 @@ namespace doris {
EngineBatchLoadTask::EngineBatchLoadTask(TPushReq& push_req,
std::vector<TTabletInfo>* tablet_infos,
- int64_t signature) :
- _push_req(push_req), _tablet_infos(tablet_infos),
_signature(signature) {
+ int64_t signature,
+ AgentStatus* res_status) :
+ _push_req(push_req),
+ _tablet_infos(tablet_infos),
+ _signature(signature),
+ _res_status(res_status) {
_download_status = DORIS_SUCCESS;
}
EngineBatchLoadTask::~EngineBatchLoadTask() {
}
-AgentStatus EngineBatchLoadTask::execute() {
+OLAPStatus EngineBatchLoadTask::execute() {
AgentStatus status = DORIS_SUCCESS;
if (_push_req.push_type == TPushType::LOAD || _push_req.push_type ==
TPushType::LOAD_DELETE) {
status = _init();
@@ -89,7 +93,8 @@ AgentStatus EngineBatchLoadTask::execute() {
} else {
status = DORIS_TASK_REQUEST_ERROR;
}
- return status;
+ *_res_status = status;
+ return OLAP_SUCCESS;
}
AgentStatus EngineBatchLoadTask::_init() {
diff --git a/be/src/olap/task/engine_batch_load_task.h
b/be/src/olap/task/engine_batch_load_task.h
index 6fc231d2..610369f8 100644
--- a/be/src/olap/task/engine_batch_load_task.h
+++ b/be/src/olap/task/engine_batch_load_task.h
@@ -39,10 +39,11 @@ class StorageEngine;
class EngineBatchLoadTask : public EngineTask{
public:
- EngineBatchLoadTask(TPushReq& push_req, std::vector<TTabletInfo>*
tablet_infos, int64_t signature);
+ EngineBatchLoadTask(TPushReq& push_req, std::vector<TTabletInfo>*
tablet_infos,
+ int64_t signature, AgentStatus* res_status);
virtual ~EngineBatchLoadTask();
- virtual AgentStatus execute();
+ virtual OLAPStatus execute();
private:
// The initial function of pusher
@@ -77,7 +78,7 @@ class EngineBatchLoadTask : public EngineTask{
FileDownloader::FileDownloaderParam _downloader_param;
FileDownloader* _file_downloader;
AgentStatus _download_status;
-
+ AgentStatus* _res_status;
int64_t _signature;
}; // class Pusher
} // namespace doris
diff --git a/be/src/olap/task/engine_cancel_delete_task.cpp
b/be/src/olap/task/engine_cancel_delete_task.cpp
index 240dbe30..a48012af 100644
--- a/be/src/olap/task/engine_cancel_delete_task.cpp
+++ b/be/src/olap/task/engine_cancel_delete_task.cpp
@@ -17,19 +17,19 @@
#include "olap/task/engine_cancel_delete_task.h"
+#include <list>
+
namespace doris {
-EngineCancelDeleteTask(const TCancelDeleteDataReq& request):_request(request) {
+using std::list;
+
+EngineCancelDeleteTask::EngineCancelDeleteTask(const TCancelDeleteDataReq&
request):_request(request) {
}
-AgentStatus EngineCancelDeleteTask::execute() {
+OLAPStatus EngineCancelDeleteTask::execute() {
OLAPStatus res = _cancel_delete();
- if (res != OLAP_SUCCESS) {
- return DORIS_ERROR;
- } else {
- return DORIS_SUCCESS;
- }
+ return res;
} // execute
OLAPStatus EngineCancelDeleteTask::_cancel_delete() {
diff --git a/be/src/olap/task/engine_cancel_delete_task.h
b/be/src/olap/task/engine_cancel_delete_task.h
index 10e340e7..cac6bfba 100644
--- a/be/src/olap/task/engine_cancel_delete_task.h
+++ b/be/src/olap/task/engine_cancel_delete_task.h
@@ -29,14 +29,14 @@ namespace doris {
class EngineCancelDeleteTask : public EngineTask {
public:
- virtual AgentStatus execute();
+ virtual OLAPStatus execute();
public:
EngineCancelDeleteTask(const TCancelDeleteDataReq& request);
~EngineCancelDeleteTask() {}
private:
- OLAPStatus _cancel_delete(const TCancelDeleteDataReq& request);
+ OLAPStatus _cancel_delete();
private:
const TCancelDeleteDataReq& _request;
diff --git a/be/src/olap/task/engine_checksum_task.cpp
b/be/src/olap/task/engine_checksum_task.cpp
index 76c289c1..72f62a3c 100644
--- a/be/src/olap/task/engine_checksum_task.cpp
+++ b/be/src/olap/task/engine_checksum_task.cpp
@@ -17,9 +17,11 @@
#include "olap/task/engine_checksum_task.h"
+#include "olap/reader.h"
+
namespace doris {
-EngineChecksumTask(TTabletId tablet_id, TSchemaHash schema_hash,
+EngineChecksumTask::EngineChecksumTask(TTabletId tablet_id, TSchemaHash
schema_hash,
TVersion version, TVersionHash version_hash, uint32_t* checksum)
:_tablet_id(tablet_id),
_schema_hash(schema_hash),
@@ -29,17 +31,13 @@ EngineChecksumTask(TTabletId tablet_id, TSchemaHash
schema_hash,
}
-AgentStatus EngineChecksumTask::execute() {
+OLAPStatus EngineChecksumTask::execute() {
OLAPStatus res = _compute_checksum();
- if (res != OLAP_SUCCESS) {
- return DORIS_ERROR;
- } else {
- return DORIS_SUCCESS;
- }
+ return res;
} // execute
-OLAPStatus StorageEngine::_compute_checksum() {
+OLAPStatus EngineChecksumTask::_compute_checksum() {
LOG(INFO) << "begin to process compute checksum."
<< "tablet_id=" << _tablet_id
<< ", schema_hash=" << _schema_hash
diff --git a/be/src/olap/task/engine_checksum_task.h
b/be/src/olap/task/engine_checksum_task.h
index cfec30cd..9a114435 100644
--- a/be/src/olap/task/engine_checksum_task.h
+++ b/be/src/olap/task/engine_checksum_task.h
@@ -29,7 +29,7 @@ namespace doris {
class EngineChecksumTask : public EngineTask {
public:
- virtual AgentStatus execute();
+ virtual OLAPStatus execute();
public:
EngineChecksumTask(TTabletId tablet_id, TSchemaHash schema_hash,
diff --git a/be/src/olap/task/engine_clear_alter_task.cpp
b/be/src/olap/task/engine_clear_alter_task.cpp
new file mode 100644
index 00000000..2422f192
--- /dev/null
+++ b/be/src/olap/task/engine_clear_alter_task.cpp
@@ -0,0 +1,99 @@
+// 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/task/engine_clear_alter_task.h"
+
+namespace doris {
+
+EngineClearAlterTask::EngineClearAlterTask(const TClearAlterTaskRequest&
request)
+ :_clear_alter_task_req(request) {
+
+}
+
+OLAPStatus EngineClearAlterTask::execute() {
+ OLAPStatus status = _clear_alter_task(_clear_alter_task_req.tablet_id,
_clear_alter_task_req.schema_hash);
+
+ return status;
+}
+
+OLAPStatus EngineClearAlterTask::_clear_alter_task(const TTabletId tablet_id,
+ const TSchemaHash schema_hash) {
+ LOG(INFO) << "begin to process clear alter task. tablet_id=" << tablet_id
+ << ", schema_hash=" << schema_hash;
+ TabletSharedPtr tablet = TabletManager::instance()->get_tablet(tablet_id,
schema_hash);
+ if (tablet.get() == NULL) {
+ OLAP_LOG_WARNING("can't find tablet when process clear alter task. ",
+ "[tablet_id=%ld, schema_hash=%d]", tablet_id,
schema_hash);
+ return OLAP_SUCCESS;
+ }
+
+ // get schema change info
+ AlterTabletType type;
+ TTabletId related_tablet_id;
+ TSchemaHash related_schema_hash;
+ vector<Version> schema_change_versions;
+ tablet->obtain_header_rdlock();
+ bool ret = tablet->get_schema_change_request(
+ &related_tablet_id, &related_schema_hash, &schema_change_versions,
&type);
+ tablet->release_header_lock();
+ if (!ret) {
+ return OLAP_SUCCESS;
+ } else if (!schema_change_versions.empty()) {
+ OLAP_LOG_WARNING("find alter task unfinished when process clear alter
task. ",
+ "[tablet=%s versions_to_change_size=%d]",
+ tablet->full_name().c_str(),
schema_change_versions.size());
+ return OLAP_ERR_PREVIOUS_SCHEMA_CHANGE_NOT_FINISHED;
+ }
+
+ // clear schema change info
+ tablet->obtain_header_wrlock();
+ tablet->clear_schema_change_request();
+ OLAPStatus res = tablet->save_tablet_meta();
+ if (res != OLAP_SUCCESS) {
+ LOG(FATAL) << "fail to save header. [res=" << res << " tablet='" <<
tablet->full_name() << "']";
+ } else {
+ LOG(INFO) << "clear alter task on tablet. [tablet='" <<
tablet->full_name() << "']";
+ }
+ tablet->release_header_lock();
+
+ // clear related tablet's schema change info
+ TabletSharedPtr related_tablet =
TabletManager::instance()->get_tablet(related_tablet_id, related_schema_hash);
+ if (related_tablet.get() == NULL) {
+ OLAP_LOG_WARNING("related tablet not found when process clear alter
task. "
+ "[tablet_id=%ld schema_hash=%d "
+ "related_tablet_id=%ld related_schema_hash=%d]",
+ tablet_id, schema_hash, related_tablet_id,
related_schema_hash);
+ } else {
+ related_tablet->obtain_header_wrlock();
+ related_tablet->clear_schema_change_request();
+ res = related_tablet->save_tablet_meta();
+ if (res != OLAP_SUCCESS) {
+ LOG(FATAL) << "fail to save header. [res=" << res << " tablet='"
+ << related_tablet->full_name() << "']";
+ } else {
+ LOG(INFO) << "clear alter task on tablet. [tablet='" <<
related_tablet->full_name() << "']";
+ }
+ related_tablet->release_header_lock();
+ }
+
+ LOG(INFO) << "finish to process clear alter task."
+ << "tablet_id=" << related_tablet_id
+ << ", schema_hash=" << related_schema_hash;
+ return OLAP_SUCCESS;
+}
+
+} // doris
\ No newline at end of file
diff --git a/be/src/olap/task/engine_clear_alter_task.h
b/be/src/olap/task/engine_clear_alter_task.h
new file mode 100644
index 00000000..bb45b4bf
--- /dev/null
+++ b/be/src/olap/task/engine_clear_alter_task.h
@@ -0,0 +1,48 @@
+// 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.
+
+#ifndef DORIS_BE_SRC_OLAP_TASK_ENGINE_CLEAR_ALTER_TASK_H
+#define DORIS_BE_SRC_OLAP_TASK_ENGINE_CLEAR_ALTER_TASK_H
+
+#include "gen_cpp/AgentService_types.h"
+#include "olap/olap_define.h"
+#include "olap/task/engine_task.h"
+
+namespace doris {
+
+// base class for storage engine
+// add "Engine" as task prefix to prevent duplicate name with agent task
+class EngineClearAlterTask : public EngineTask {
+
+public:
+ virtual OLAPStatus execute();
+
+public:
+ EngineClearAlterTask(const TClearAlterTaskRequest& request);
+ ~EngineClearAlterTask() {}
+
+private:
+ OLAPStatus _clear_alter_task(const TTabletId tablet_id,
+ const TSchemaHash schema_hash);
+
+private:
+ const TClearAlterTaskRequest& _clear_alter_task_req;
+
+}; // EngineTask
+
+} // doris
+#endif //DORIS_BE_SRC_OLAP_TASK_ENGINE_CLEAR_ALTER_TASK_H
\ No newline at end of file
diff --git a/be/src/olap/task/engine_clone_task.cpp
b/be/src/olap/task/engine_clone_task.cpp
index 7dc003c5..816e7454 100644
--- a/be/src/olap/task/engine_clone_task.cpp
+++ b/be/src/olap/task/engine_clone_task.cpp
@@ -17,14 +17,32 @@
#include "olap/task/engine_clone_task.h"
+#include <set>
+
+using std::set;
+using std::stringstream;
+
namespace doris {
-EngineCloneTask(TCloneReq& _clone_req, vector<string>& error_msgs) :
+const std::string HTTP_REQUEST_PREFIX = "/api/_tablet/_download?";
+const std::string HTTP_REQUEST_TOKEN_PARAM = "token=";
+const std::string HTTP_REQUEST_FILE_PARAM = "&file=";
+const uint32_t DOWNLOAD_FILE_MAX_RETRY = 3;
+const uint32_t LIST_REMOTE_FILE_TIMEOUT = 15;
+
+EngineCloneTask::EngineCloneTask(const TCloneReq& clone_req, vector<string>*
error_msgs,
+ vector<TTabletInfo>* tablet_infos,
+ AgentStatus* res_status,
+ int64_t signature,
+ const TMasterInfo& master_info) :
_clone_req(clone_req),
_error_msgs(error_msgs),
- _tablet_infos(tablet_infos) {}
+ _tablet_infos(tablet_infos),
+ _res_status(res_status),
+ _signature(signature),
+ _master_info(master_info) {}
-AgentStatus EngineCloneTask::execute() {
+OLAPStatus EngineCloneTask::execute() {
AgentStatus status = DORIS_SUCCESS;
string src_file_path;
TBackend src_host;
@@ -34,7 +52,7 @@ AgentStatus EngineCloneTask::execute() {
_clone_req.tablet_id, _clone_req.schema_hash);
if (tablet.get() != NULL) {
LOG(INFO) << "clone tablet exist yet, begin to incremental clone. "
- << "signature:" << agent_task_req.signature
+ << "signature:" << _signature
<< ", tablet_id:" << _clone_req.tablet_id
<< ", schema_hash:" << _clone_req.schema_hash
<< ", committed_version:" << _clone_req.committed_version;
@@ -44,12 +62,12 @@ AgentStatus EngineCloneTask::execute() {
string local_data_path = _get_info_before_incremental_clone(tablet,
_clone_req.committed_version, &missing_versions);
bool allow_incremental_clone = false;
- status = _clone_copy(clone_req,
- agent_task_req.signature,
+ status = _clone_copy(_clone_req,
+ _signature,
local_data_path,
&src_host,
&src_file_path,
- &_error_msgs,
+ _error_msgs,
&missing_versions,
&allow_incremental_clone);
if (status == DORIS_SUCCESS) {
@@ -57,18 +75,18 @@ AgentStatus EngineCloneTask::execute() {
if (olap_status != OLAP_SUCCESS) {
LOG(WARNING) << "failed to finish incremental clone. [table="
<< tablet->full_name()
<< " res=" << olap_status << "]";
- _error_msgs.push_back("incremental clone error.");
+ _error_msgs->push_back("incremental clone error.");
status = DORIS_ERROR;
}
} else {
// begin to full clone if incremental failed
LOG(INFO) << "begin to full clone. [table=" << tablet->full_name();
- status = _clone_copy(clone_req,
- agent_task_req.signature,
+ status = _clone_copy(_clone_req,
+ _signature,
local_data_path,
&src_host,
&src_file_path,
- &_error_msgs,
+ _error_msgs,
NULL, NULL);
if (status == DORIS_SUCCESS) {
LOG(INFO) << "download successfully when full clone. [table="
<< tablet->full_name()
@@ -80,7 +98,7 @@ AgentStatus EngineCloneTask::execute() {
if (olap_status != OLAP_SUCCESS) {
LOG(WARNING) << "fail to finish full clone. [table=" <<
tablet->full_name()
<< " res=" << olap_status << "]";
- _error_msgs.push_back("full clone error.");
+ _error_msgs->push_back("full clone error.");
status = DORIS_ERROR;
}
}
@@ -90,12 +108,12 @@ AgentStatus EngineCloneTask::execute() {
// Get local disk from olap
string local_shard_root_path;
DataDir* store = nullptr;
- OLAPStatus olap_status =
worker_pool_this->_env->olap_engine()->obtain_shard_path(
+ OLAPStatus olap_status =
StorageEngine::get_instance()->obtain_shard_path(
_clone_req.storage_medium, &local_shard_root_path, &store);
if (olap_status != OLAP_SUCCESS) {
OLAP_LOG_WARNING("clone get local root path failed. signature:
%ld",
- agent_task_req.signature);
- _error_msgs.push_back("clone get local root path failed.");
+ _signature);
+ _error_msgs->push_back("clone get local root path failed.");
status = DORIS_ERROR;
}
@@ -104,12 +122,12 @@ AgentStatus EngineCloneTask::execute() {
tablet_dir_stream << local_shard_root_path
<< "/" << _clone_req.tablet_id
<< "/" << _clone_req.schema_hash;
- status = _clone_copy(clone_req,
- agent_task_req.signature,
+ status = _clone_copy(_clone_req,
+ _signature,
tablet_dir_stream.str(),
&src_host,
&src_file_path,
- &_error_msgs,
+ _error_msgs,
NULL, NULL);
}
@@ -118,7 +136,7 @@ AgentStatus EngineCloneTask::execute() {
<< " src_file_path: " << src_file_path;
// Load header
OLAPStatus load_header_status =
- worker_pool_this->_env->olap_engine()->load_header(
+ StorageEngine::get_instance()->load_header(
store,
local_shard_root_path,
_clone_req.tablet_id,
@@ -126,8 +144,8 @@ AgentStatus EngineCloneTask::execute() {
if (load_header_status != OLAP_SUCCESS) {
LOG(WARNING) << "load header failed. local_shard_root_path: '"
<< local_shard_root_path
<< "' schema_hash: " << _clone_req.schema_hash
<< ". status: " << load_header_status
- << ". signature: " << agent_task_req.signature;
- _error_msgs.push_back("load header failed.");
+ << ". signature: " << _signature;
+ _error_msgs->push_back("load header failed.");
status = DORIS_ERROR;
}
}
@@ -140,7 +158,7 @@ AgentStatus EngineCloneTask::execute() {
<< "/" << _clone_req.tablet_id;
string local_data_path = local_data_path_stream.str();
LOG(INFO) << "clone failed. want to delete local dir: " <<
local_data_path
- << ". signature: " << agent_task_req.signature;
+ << ". signature: " << _signature;
try {
boost::filesystem::path local_path(local_data_path);
if (boost::filesystem::exists(local_path)) {
@@ -151,7 +169,7 @@ AgentStatus EngineCloneTask::execute() {
OLAP_LOG_WARNING("clone delete useless dir failed. "
"error: %s, local dir: %s, signature: %ld",
e.what(), local_data_path.c_str(),
- agent_task_req.signature);
+ _signature);
}
}
#endif
@@ -160,18 +178,18 @@ AgentStatus EngineCloneTask::execute() {
// Get clone tablet info
if (status == DORIS_SUCCESS || status == DORIS_CREATE_TABLE_EXIST) {
TTabletInfo tablet_info;
- tablet_info.__set_tablet_id(tablet_id);
- tablet_info.__set_schema_hash(schema_hash);
- OLAPStatus get_tablet_info_status =
TabletManager::instance()->report_tablet_info(tablet_info);
+ tablet_info.__set_tablet_id(_clone_req.tablet_id);
+ tablet_info.__set_schema_hash(_clone_req.schema_hash);
+ OLAPStatus get_tablet_info_status =
TabletManager::instance()->report_tablet_info(&tablet_info);
if (get_tablet_info_status != OLAP_SUCCESS) {
OLAP_LOG_WARNING("clone success, but get tablet info failed."
"tablet id: %ld, schema hash: %ld, signature:
%ld",
_clone_req.tablet_id, _clone_req.schema_hash,
- agent_task_req.signature);
- _error_msgs.push_back("clone success, but get tablet info
failed.");
+ _signature);
+ _error_msgs->push_back("clone success, but get tablet info
failed.");
status = DORIS_ERROR;
} else if (
- (clone_req.__isset.committed_version
+ (_clone_req.__isset.committed_version
&& _clone_req.__isset.committed_version_hash)
&& (tablet_info.version < _clone_req.committed_version ||
(tablet_info.version == _clone_req.committed_version
@@ -182,7 +200,7 @@ AgentStatus EngineCloneTask::execute() {
// we drop it.
LOG(INFO) << "begin to drop the stale table. tablet_id:" <<
_clone_req.tablet_id
<< ", schema_hash:" << _clone_req.schema_hash
- << ", signature:" << agent_task_req.signature
+ << ", signature:" << _signature
<< ", version:" << tablet_info.version
<< ", version_hash:" << tablet_info.version_hash
<< ", expected_version: " <<
_clone_req.committed_version
@@ -199,13 +217,14 @@ AgentStatus EngineCloneTask::execute() {
} else {
LOG(INFO) << "clone get tablet info success. tablet_id:" <<
_clone_req.tablet_id
<< ", schema_hash:" << _clone_req.schema_hash
- << ", signature:" << agent_task_req.signature
+ << ", signature:" << _signature
<< ", version:" << tablet_info.version
<< ", version_hash:" << tablet_info.version_hash;
- _tablet_infos.push_back(tablet_info);
+ _tablet_infos->push_back(tablet_info);
}
}
- return status;
+ *_res_status = status;
+ return OLAP_SUCCESS;
}
string EngineCloneTask::_get_info_before_incremental_clone(TabletSharedPtr
tablet,
@@ -234,7 +253,7 @@ string
EngineCloneTask::_get_info_before_incremental_clone(TabletSharedPtr table
return tablet->tablet_path() + CLONE_PREFIX;
}
-OLAPStatus EngineCloneTask::_clone_copy(
+AgentStatus EngineCloneTask::_clone_copy(
const TCloneReq& clone_req,
int64_t signature,
const string& local_data_path,
@@ -243,7 +262,7 @@ OLAPStatus EngineCloneTask::_clone_copy(
vector<string>* error_msgs,
const vector<Version>* missing_versions,
bool* allow_incremental_clone) {
- OLAPStatus status = OLAP_SUCCESS;
+ AgentStatus status = DORIS_SUCCESS;
std::string token = _master_info.token;
for (auto src_backend : clone_req.src_backends) {
@@ -252,9 +271,7 @@ OLAPStatus EngineCloneTask::_clone_copy(
string http_host = http_host_stream.str();
// Make snapshot in remote olap engine
*src_host = src_backend;
-#ifndef BE_TEST
AgentServerClient agent_client(*src_host);
-#endif
TAgentResult make_snapshot_result;
status = DORIS_SUCCESS;
@@ -271,15 +288,9 @@ OLAPStatus EngineCloneTask::_clone_copy(
}
snapshot_request.__set_missing_version(snapshot_versions);
}
-#ifndef BE_TEST
agent_client.make_snapshot(
snapshot_request,
&make_snapshot_result);
-#else
- _agent_client->make_snapshot(
- snapshot_request,
- &make_snapshot_result);
-#endif
if (make_snapshot_result.__isset.allow_incremental_clone) {
// During upgrading, some BE nodes still be installed an old
previous old.
@@ -549,15 +560,9 @@ OLAPStatus EngineCloneTask::_clone_copy(
// Release snapshot, if failed, ignore it. OLAP engine will drop
useless snapshot
TAgentResult release_snapshot_result;
-#ifndef BE_TEST
agent_client.release_snapshot(
make_snapshot_result.snapshot_path,
&release_snapshot_result);
-#else
- _agent_client->release_snapshot(
- make_snapshot_result.snapshot_path,
- &release_snapshot_result);
-#endif
if (release_snapshot_result.status.status_code != TStatusCode::OK) {
LOG(WARNING) << "release snapshot failed. src_file_path: " <<
*src_file_path
<< ". signature: " << signature;
@@ -571,7 +576,7 @@ OLAPStatus EngineCloneTask::_clone_copy(
}
-OLAPStatus StorageEngine::_finish_clone(TabletSharedPtr tablet, const string&
clone_dir,
+OLAPStatus EngineCloneTask::_finish_clone(TabletSharedPtr tablet, const
string& clone_dir,
int64_t committed_version, bool
is_incremental_clone) {
OLAPStatus res = OLAP_SUCCESS;
vector<string> linked_success_files;
diff --git a/be/src/olap/task/engine_clone_task.h
b/be/src/olap/task/engine_clone_task.h
index b83483a2..bc11d887 100644
--- a/be/src/olap/task/engine_clone_task.h
+++ b/be/src/olap/task/engine_clone_task.h
@@ -18,7 +18,11 @@
#ifndef DORIS_BE_SRC_OLAP_TASK_ENGINE_CLONE_TASK_H
#define DORIS_BE_SRC_OLAP_TASK_ENGINE_CLONE_TASK_H
+#include "agent/file_downloader.h"
+#include "agent/utils.h"
#include "gen_cpp/AgentService_types.h"
+#include "gen_cpp/HeartbeatService.h"
+#include "gen_cpp/MasterService_types.h"
#include "olap/olap_define.h"
#include "olap/task/engine_task.h"
@@ -29,10 +33,14 @@ namespace doris {
class EngineCloneTask : public EngineTask {
public:
- virtual AgentStatus execute();
+ virtual OLAPStatus execute();
public:
- EngineCloneTask(TCloneReq& _clone_req, vector<string>& error_msgs,
vector<TTabletInfo> tablet_infos);
+ EngineCloneTask(const TCloneReq& _clone_req, vector<string>* error_msgs,
+ vector<TTabletInfo>* tablet_infos,
+ AgentStatus* _res_status,
+ int64_t _signature,
+ const TMasterInfo& _master_info);
~EngineCloneTask() {}
private:
@@ -53,10 +61,27 @@ class EngineCloneTask : public EngineTask {
OLAPStatus _clone_full_data(TabletSharedPtr tablet, TabletMeta&
clone_header);
+ AgentStatus _clone_copy(
+ const TCloneReq& clone_req,
+ int64_t signature,
+ const string& local_data_path,
+ TBackend* src_host,
+ string* src_file_path,
+ vector<string>* error_msgs,
+ const vector<Version>* missing_versions,
+ bool* allow_incremental_clone);
+
private:
const TCloneReq& _clone_req;
- vector<string>& _error_msgs;
- vector<TTabletInfo> _tablet_infos;
+ vector<string>* _error_msgs;
+ vector<TTabletInfo>* _tablet_infos;
+ AgentStatus* _res_status;
+ int64_t _signature;
+ const TMasterInfo& _master_info;
+#ifdef BE_TEST
+ AgentServerClient* _agent_client;
+ FileDownloader* _file_downloader_ptr;
+#endif
}; // EngineTask
} // doris
diff --git a/be/src/olap/task/engine_schema_change_task.cpp
b/be/src/olap/task/engine_schema_change_task.cpp
index fa4a6406..009cdbf7 100644
--- a/be/src/olap/task/engine_schema_change_task.cpp
+++ b/be/src/olap/task/engine_schema_change_task.cpp
@@ -24,8 +24,8 @@ namespace doris {
using std::to_string;
EngineSchemaChangeTask::EngineSchemaChangeTask(const TAlterTabletReq&
alter_tablet_request,
- int64_t signature, const TTaskType::type task_type, vector<string>&
error_msgs,
- string& process_name):
+ int64_t signature, const TTaskType::type task_type, vector<string>*
error_msgs,
+ const string& process_name):
_alter_tablet_req(alter_tablet_request),
_signature(signature),
_task_type(task_type),
@@ -34,7 +34,7 @@ EngineSchemaChangeTask::EngineSchemaChangeTask(const
TAlterTabletReq& alter_tabl
}
-AgentStatus EngineSchemaChangeTask::execute() {
+OLAPStatus EngineSchemaChangeTask::execute() {
OLAPStatus status = OLAP_SUCCESS;
TTabletId base_tablet_id = _alter_tablet_req.base_tablet_id;
TSchemaHash base_schema_hash = _alter_tablet_req.base_schema_hash;
@@ -61,7 +61,7 @@ AgentStatus EngineSchemaChangeTask::execute() {
OLAP_LOG_WARNING("delete failed rollup file failed, status: %d, "
"signature: %ld.",
status, _signature);
- _error_msgs.push_back("delete failed rollup file failed, "
+ _error_msgs->push_back("delete failed rollup file failed, "
"signature: " + to_string(_signature));
}
}
@@ -89,11 +89,7 @@ AgentStatus EngineSchemaChangeTask::execute() {
}
}
- if (status != OLAP_SUCCESS) {
- return DORIS_ERROR;
- } else {
- return DORIS_SUCCESS;
- }
+ return status;
} // execute
OLAPStatus EngineSchemaChangeTask::_create_rollup_tablet(const
TAlterTabletReq& request) {
diff --git a/be/src/olap/task/engine_schema_change_task.h
b/be/src/olap/task/engine_schema_change_task.h
index 9d827a4c..016b6148 100644
--- a/be/src/olap/task/engine_schema_change_task.h
+++ b/be/src/olap/task/engine_schema_change_task.h
@@ -29,11 +29,11 @@ namespace doris {
class EngineSchemaChangeTask : public EngineTask {
public:
- virtual AgentStatus execute();
+ virtual OLAPStatus execute();
public:
EngineSchemaChangeTask(const TAlterTabletReq& alter_tablet_request,
int64_t signature,
- const TTaskType::type task_type, vector<string>& error_msgs, string&
process_name);
+ const TTaskType::type task_type, vector<string>* error_msgs, const
string& process_name);
~EngineSchemaChangeTask() {}
private:
@@ -62,8 +62,8 @@ class EngineSchemaChangeTask : public EngineTask {
const TAlterTabletReq& _alter_tablet_req;
int64_t _signature;
const TTaskType::type _task_type;
- vector<string>& _error_msgs;
- string& _process_name;
+ vector<string>* _error_msgs;
+ const string& _process_name;
}; // EngineTask
diff --git a/be/src/olap/task/engine_storage_migration_task.cpp
b/be/src/olap/task/engine_storage_migration_task.cpp
index 7cc095cd..d387e39d 100644
--- a/be/src/olap/task/engine_storage_migration_task.cpp
+++ b/be/src/olap/task/engine_storage_migration_task.cpp
@@ -29,17 +29,13 @@
EngineStorageMigrationTask::EngineStorageMigrationTask(TStorageMediumMigrateReq&
}
-AgentStatus EngineStorageMigrationTask::execute() {
+OLAPStatus EngineStorageMigrationTask::execute() {
OLAPStatus res = OLAP_SUCCESS;
res = _storage_medium_migrate(
_storage_medium_migrate_req.tablet_id,
_storage_medium_migrate_req.schema_hash,
_storage_medium_migrate_req.storage_medium);
- if (res != OLAP_SUCCESS) {
- return DORIS_ERROR;
- } else {
- return DORIS_SUCCESS;
- }
+ return res;
}
OLAPStatus EngineStorageMigrationTask::_storage_medium_migrate(
diff --git a/be/src/olap/task/engine_storage_migration_task.h
b/be/src/olap/task/engine_storage_migration_task.h
index 249113aa..903b3997 100644
--- a/be/src/olap/task/engine_storage_migration_task.h
+++ b/be/src/olap/task/engine_storage_migration_task.h
@@ -29,7 +29,7 @@ namespace doris {
class EngineStorageMigrationTask : public EngineTask {
public:
- virtual AgentStatus execute();
+ virtual OLAPStatus execute();
public:
EngineStorageMigrationTask(TStorageMediumMigrateReq&
storage_medium_migrate_req);
diff --git a/be/src/olap/task/engine_task.h b/be/src/olap/task/engine_task.h
index 6fc39122..2710a466 100644
--- a/be/src/olap/task/engine_task.h
+++ b/be/src/olap/task/engine_task.h
@@ -18,7 +18,6 @@
#ifndef DORIS_BE_SRC_OLAP_TASK_ENGINE_TASK_H
#define DORIS_BE_SRC_OLAP_TASK_ENGINE_TASK_H
-#include "agent/status.h"
#include "olap/olap_common.h"
#include "olap/olap_define.h"
#include "olap/storage_engine.h"
@@ -34,10 +33,10 @@ class EngineTask {
public:
// use agent_status not olap_status, because the task is very close to
engine
- virtual AgentStatus prepare() { return DORIS_SUCCESS; }
- virtual AgentStatus execute() { return DORIS_SUCCESS; }
- virtual AgentStatus finish() { return DORIS_SUCCESS; }
- virtual AgentStatus clean() { return DORIS_SUCCESS; }
+ virtual OLAPStatus prepare() { return OLAP_SUCCESS; }
+ virtual OLAPStatus execute() { return OLAP_SUCCESS; }
+ virtual OLAPStatus finish() { return OLAP_SUCCESS; }
+ virtual OLAPStatus cancel() { return OLAP_SUCCESS; }
}; // EngineTask
} // doris
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
