/**

* Copyright (c) 2021 OceanBase

* OceanBase CE is licensed under Mulan PubL v2.

* You can use this software according to the terms and conditions of the Mulan PubL v2.

* You may obtain a copy of Mulan PubL v2 at:

* http://license.coscl.org.cn/MulanPubL-2.0

* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,

* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,

* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.

* See the Mulan PubL v2 for more details.

*/

#define USING_LOG_PREFIX SQL_EXE

#include "sql/executor/ob_task_control.h"

#include "sql/executor/ob_task_event.h"

#include "sql/engine/ob_exec_context.h"

#include "share/ob_define.h"

#include "share/schema/ob_schema_getter_guard.h"

#include "observer/ob_server_struct.h"

using namespace oceanbase::common;

using namespace oceanbase::sql;

using namespace oceanbase::share;

int ObTaskResult::init(const ObTaskLocation &task_location,

const common::ObIArray<ObSliceEvent> &slice_events)

{

int ret = OB_SUCCESS;

if (OB_UNLIKELY(is_valid())) {

ret = OB_INIT_TWICE;

LOG_ERROR("init twice", K(ret), K(task_location_), K(slice_events_));

} else if (OB_UNLIKELY(!task_location.is_valid())) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid args", K(ret), K(task_location));

} else {

task_location_ = task_location;

slice_events_ = &slice_events;

}

return ret;

}

bool ObTaskResult::has_empty_data() const

{

bool ret = false;

if (NULL != slice_events_ && slice_events_->count() > 0) {

ret = slice_events_->at(0).has_empty_data();

}

return ret;

}

int64_t ObTaskResult::get_found_rows() const

{

int64_t found_rows = 0;

if (NULL != slice_events_ && slice_events_->count() > 0) {

found_rows = slice_events_->at(0).get_found_rows();

}

return found_rows;

}

int64_t ObTaskResult::get_affected_rows() const

{

int64_t affected_rows = 0;

if (NULL != slice_events_ && slice_events_->count() > 0) {

affected_rows = slice_events_->at(0).get_affected_rows();

}

return affected_rows;

}

int64_t ObTaskResult::get_duplicated_rows() const

{

int64_t duplicated_rows = 0;

if (NULL != slice_events_ && slice_events_->count() > 0) {

duplicated_rows = slice_events_->at(0).get_duplicated_rows();

}

return duplicated_rows;

}

int64_t ObTaskResult::get_matched_rows() const

{

int64_t matched_rows = 0;

if (NULL != slice_events_ && slice_events_->count() > 0) {

matched_rows = slice_events_->at(0).get_matched_rows();

}

return matched_rows;

}

ObTaskControl::ObTaskControl()

: tasks_(),

// is_scan_job_(false),

is_root_job_(false),

is_select_plan_(false)

{

}

ObTaskControl::~ObTaskControl()

{}

void ObTaskControl::reset()

{

tasks_.reset();

// is_scan_job_ = false;

is_root_job_ = false;

is_select_plan_ = false;

}

int ObTaskControl::find_task(uint64_t task_id, ObTaskInfo *&task) const

{

int ret = OB_ENTRY_NOT_EXIST;

for (int64_t i = 0; OB_ENTRY_NOT_EXIST == ret && i < tasks_.count(); ++i) {

ObTaskInfo *task_info = tasks_.at(i);

if (OB_ISNULL(task_info)) {

ret = OB_ERR_UNEXPECTED;

} else if (task_info->get_task_location().get_task_id() == task_id) {

task = task_info;

ret = OB_SUCCESS;

}

}

return ret;

}

int ObTaskControl::get_task_location(uint64_t task_id, ObTaskLocation &task_location) const

{

int ret = OB_SUCCESS;

ObTaskInfo *task_info = NULL;

if (OB_FAIL(find_task(task_id, task_info))) {

LOG_WARN("fail to find task", K(ret), K(task_id));

} else if (OB_ISNULL(task_info)) {

ret = OB_ERR_UNEXPECTED;

LOG_ERROR("succ to find task info, but task info is null", K(ret));

} else {

task_location = task_info->get_task_location();

}

return ret;

}

int ObTaskControl::prepare(int64_t job_parallel_degree)

{

int ret = OB_SUCCESS;

if (!is_root_job_) {

int64_t parallel_degree = MIN(tasks_.count(), job_parallel_degree);

for (int64_t i = 0; OB_SUCC(ret) && i < parallel_degree; ++i) {

ObTaskInfo *task = tasks_.at(i);

if (OB_ISNULL(task)) {

ret = OB_ERR_UNEXPECTED;

LOG_ERROR("task is NULL", K(ret), K(tasks_.count()),

K(parallel_degree), K(job_parallel_degree));

} else {

task->set_state(OB_TASK_STATE_INITED);

}

}

} else {

for (int64_t i = 0; OB_SUCC(ret) && i < tasks_.count(); i++) {

ObTaskInfo *task = tasks_.at(i);

if (OB_ISNULL(task)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("task is NULL", K(ret));

} else {

ObTaskLocation dummy_task_loc;

task->set_task_location(dummy_task_loc);

task->set_state(OB_TASK_STATE_INITED);

}

}

}

return ret;

}

int ObTaskControl::get_ready_tasks(common::ObIArray<ObTaskInfo *> &tasks) const

{

int ret = OB_SUCCESS;

bool iter_end = true;

const int64_t count = tasks_.count();

tasks.reset();

for (int64_t i = 0; OB_SUCC(ret) && i < count; ++i) {

ObTaskInfo *task = tasks_.at(i);

if (OB_ISNULL(task)) {

ret = OB_ERR_UNEXPECTED;

} else if (OB_TASK_STATE_NOT_INIT == task->get_state()) {

iter_end = false;

} else if (OB_TASK_STATE_INITED == task->get_state()) {

iter_end = false;

ret = tasks.push_back(tasks_.at(i));

}

}

if (iter_end && OB_SUCC(ret)) {

ret = OB_ITER_END;

}

return ret;

}

int ObTaskControl::get_running_tasks(common::ObIArray<ObTaskInfo *> &tasks) const

{

return get_task_by_state(tasks, OB_TASK_STATE_RUNNING);

}

int ObTaskControl::get_finished_tasks(common::ObIArray<ObTaskInfo *> &tasks) const

{

return get_task_by_state(tasks, OB_TASK_STATE_FINISHED);

}

int ObTaskControl::get_skipped_tasks(common::ObIArray<ObTaskInfo *> &tasks) const

{

return get_task_by_state(tasks, OB_TASK_STATE_SKIPPED);

}

int ObTaskControl::get_failed_tasks(common::ObIArray<ObTaskInfo *> &tasks) const

{

return get_task_by_state(tasks, OB_TASK_STATE_FAILED);

}

int ObTaskControl::get_begin_running_tasks(common::ObIArray<ObTaskInfo *> &tasks) const

{

int ret = OB_SUCCESS;

const int64_t count = tasks_.count();

tasks.reset();

for (int64_t i = 0; OB_SUCC(ret) && i < count; i++) {

if (tasks_.at(i)->get_state() >= OB_TASK_STATE_RUNNING) {

ret = tasks.push_back(tasks_.at(i));

}

}

return ret;

}

int ObTaskControl::get_all_tasks(common::ObIArray<ObTaskInfo *> &tasks) const

{

int ret = OB_SUCCESS;

tasks.reset();

if (OB_FAIL(tasks.assign(tasks_))) {

LOG_WARN("fail to assign task array", K(ret), K(tasks_.count()));

}

return ret;

}

int ObTaskControl::get_task_by_state(common::ObIArray<ObTaskInfo *> &tasks_out, int state) const

{

int ret = OB_SUCCESS;

const int64_t count = tasks_.count();

tasks_out.reset();

for (int64_t i = 0; OB_SUCC(ret) && i < count; i++) {

if (state == tasks_.at(i)->get_state()) {

ret = tasks_out.push_back(tasks_.at(i));

}

}

return ret;

}