/**

* 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_ENG

#include "sql/engine/ob_physical_plan.h"

#include "share/ob_define.h"

#include "lib/utility/utility.h"

#include "lib/utility/serialization.h"

#include "lib/alloc/malloc_hook.h"

#include "share/inner_table/ob_inner_table_schema.h"

#include "share/schema/ob_schema_getter_guard.h"

#include "common/sql_mode/ob_sql_mode_utils.h"

#include "sql/ob_result_set.h"

#include "sql/ob_sql_utils.h"

#include "sql/session/ob_sql_session_info.h"

#include "sql/engine/ob_exec_context.h"

#include "sql/engine/expr/ob_sql_expression.h"

#include "sql/engine/ob_phy_operator_type.h"

#include "observer/mysql/ob_mysql_request_manager.h"

#include "share/diagnosis/ob_sql_plan_monitor_node_list.h"

#include "sql/engine/ob_operator.h"

#include "sql/engine/ob_operator_factory.h"

#include "share/stat/ob_opt_stat_manager.h"

#include "share/ob_truncated_string.h"

#include "sql/spm/ob_spm_evolution_plan.h"

#include "sql/engine/ob_exec_feedback_info.h"

namespace oceanbase

{

using namespace common;

using namespace common::serialization;

using namespace share::schema;

using namespace lib;

namespace sql

{

ObPhysicalPlan::ObPhysicalPlan(MemoryContext &mem_context /* = CURRENT_CONTEXT */)

: ObPlanCacheObject(ObLibCacheNameSpace::NS_CRSR, mem_context),

phy_hint_(),

root_op_spec_(NULL),

param_count_(0),

signature_(0),

field_columns_(mem_context->get_arena_allocator()),

param_columns_(mem_context->get_arena_allocator()),

returning_param_columns_(mem_context->get_arena_allocator()),

autoinc_params_(allocator_),

stmt_need_privs_(allocator_),

stmt_ora_need_privs_(allocator_),

audit_units_(allocator_),

vars_(allocator_),

sql_expression_factory_(allocator_),

expr_op_factory_(allocator_),

literal_stmt_type_(stmt::T_NONE),

plan_type_(OB_PHY_PLAN_UNINITIALIZED),

location_type_(OB_PHY_PLAN_UNINITIALIZED),

require_local_execution_(false),

use_px_(false),

px_dop_(0),

next_phy_operator_id_(0),

next_expr_operator_id_(0),

regexp_op_count_(0),

like_op_count_(0),

px_exchange_out_op_count_(0),

is_sfu_(false),

is_contains_assignment_(false),

affected_last_insert_id_(false),

is_affect_found_row_(false),

has_top_limit_(false),

is_wise_join_(false),

contain_table_scan_(false),

has_nested_sql_(false),

session_id_(0),

contain_oracle_trx_level_temporary_table_(false),

contain_oracle_session_level_temporary_table_(false),

gtt_session_scope_ids_(allocator_),

gtt_trans_scope_ids_(allocator_),

concurrent_num_(0),

max_concurrent_num_(ObMaxConcurrentParam::UNLIMITED),

table_locations_(allocator_),

das_table_locations_(allocator_),

row_param_map_(allocator_),

is_update_uniq_index_(false),

contain_index_location_(false),

base_constraints_(allocator_),

strict_constrinats_(allocator_),

non_strict_constrinats_(allocator_),

expr_frame_info_(allocator_),

stat_(),

op_stats_(),

need_drive_dml_query_(false),

tx_id_(-1),

tm_sessid_(-1),

var_init_exprs_(allocator_),

is_returning_(false),

is_late_materialized_(false),

is_dep_base_table_(false),

is_insert_select_(false),

is_plain_insert_(false),

flashback_query_items_(allocator_),

contain_paramed_column_field_(false),

first_array_index_(OB_INVALID_INDEX),

need_consistent_snapshot_(true),

is_batched_multi_stmt_(false),

encrypt_meta_array_(allocator_),

is_new_engine_(false),

use_pdml_(false),

use_temp_table_(false),

has_link_table_(false),

has_link_sfd_(false),

need_serial_exec_(false),

temp_sql_can_prepare_(false),

is_need_trans_(false),

batch_size_(0),

contain_pl_udf_or_trigger_(false),

ddl_schema_version_(0),

ddl_table_id_(0),

ddl_execution_id_(-1),

ddl_task_id_(0),

is_packed_(false),

has_instead_of_trigger_(false),

min_cluster_version_(GET_MIN_CLUSTER_VERSION()),

need_record_plan_info_(false),

enable_append_(false),

append_table_id_(0),

logical_plan_(),

is_enable_px_fast_reclaim_(false),

subschema_ctx_(allocator_)

{

}

ObPhysicalPlan::~ObPhysicalPlan()

{

destroy();

}

void ObPhysicalPlan::reset()

{

ObPlanCacheObject::reset();

phy_hint_.reset();

root_op_spec_ = NULL;

param_count_ = 0;

signature_ = 0;

field_columns_.reset();

param_columns_.reset();

returning_param_columns_.reset();

autoinc_params_.reset();

stmt_need_privs_.reset();

stmt_ora_need_privs_.reset();

audit_units_.reset();

vars_.reset();

sql_expression_factory_.destroy();

expr_op_factory_.destroy();

literal_stmt_type_ = stmt::T_NONE;

plan_type_ = OB_PHY_PLAN_UNINITIALIZED;

location_type_ = OB_PHY_PLAN_UNINITIALIZED;

require_local_execution_ = false;

use_px_ = false;

px_dop_ = 0;

next_phy_operator_id_ = 0;

next_expr_operator_id_ = 0;

regexp_op_count_ = 0;

like_op_count_ = 0;

px_exchange_out_op_count_ = 0;

is_sfu_ = false;

is_contain_virtual_table_ = false;

is_contain_inner_table_ = false;

is_contains_assignment_ = false;

affected_last_insert_id_ = false;

is_affect_found_row_ = false;

has_top_limit_ = false;

is_wise_join_ = false;

contain_table_scan_ = false;

has_nested_sql_ = false;

session_id_ = 0;

contain_oracle_trx_level_temporary_table_ = false;

contain_oracle_session_level_temporary_table_ = false;

gtt_session_scope_ids_.reset();

gtt_trans_scope_ids_.reset();

concurrent_num_ = 0;

max_concurrent_num_ = ObMaxConcurrentParam::UNLIMITED;

is_update_uniq_index_ = false;

contain_index_location_ = false;

ObPlanCacheObject::reset();

is_returning_ = false;

is_late_materialized_ = false;

is_dep_base_table_ = false;

is_insert_select_ = false;

is_plain_insert_ = false;

base_constraints_.reset();

strict_constrinats_.reset();

non_strict_constrinats_.reset();

flashback_query_items_.reset();

contain_paramed_column_field_ = false;

first_array_index_ = OB_INVALID_INDEX;

need_consistent_snapshot_ = true;

is_batched_multi_stmt_ = false;

temp_sql_can_prepare_ = false;

is_new_engine_ = false;

#ifndef NDEBUG

bit_set_.reset();

#endif

use_pdml_ = false;

use_temp_table_ = false;

has_link_table_ = false;

has_link_sfd_ = false;

encrypt_meta_array_.reset();

need_serial_exec_ = false;

batch_size_ = 0;

contain_pl_udf_or_trigger_ = false;

is_packed_ = false;

has_instead_of_trigger_ = false;

enable_append_ = false;

append_table_id_ = 0;

stat_.expected_worker_map_.destroy();

stat_.minimal_worker_map_.destroy();

tx_id_ = -1;

tm_sessid_ = -1;

var_init_exprs_.reset();

need_record_plan_info_ = false;

logical_plan_.reset();

is_enable_px_fast_reclaim_ = false;

subschema_ctx_.reset();

}

void ObPhysicalPlan::destroy()

{

#ifndef NDEBUG

bit_set_.reset();

#endif

sql_expression_factory_.destroy();

expr_op_factory_.destroy();

stat_.expected_worker_map_.destroy();

stat_.minimal_worker_map_.destroy();

subschema_ctx_.destroy();

}

int ObPhysicalPlan::copy_common_info(ObPhysicalPlan &src)

{

int ret = OB_SUCCESS;

//copy regexp_op_count_

set_regexp_op_count(src.regexp_op_count_);

set_like_op_count(src.like_op_count_);

set_px_exchange_out_op_count(src.px_exchange_out_op_count_);

//copy others

set_fetch_cur_time(src.get_fetch_cur_time());

set_stmt_type(src.get_stmt_type());

set_literal_stmt_type(src.get_literal_stmt_type());

//copy plan_id/hint/privs

object_id_ = src.object_id_;

min_cluster_version_ = src.min_cluster_version_;

if (OB_FAIL(set_phy_plan_hint(src.get_phy_plan_hint()))) {

LOG_WARN("Failed to copy query hint", K(ret));

} else if (OB_FAIL(set_stmt_need_privs(src.get_stmt_need_privs()))) {

LOG_WARN("Failed to deep copy", K(src.get_stmt_need_privs()), K(ret));

} else { }//do nothing

return ret;

}

int ObPhysicalPlan::set_vars(const common::ObIArray<ObVarInfo> &vars)

{

int ret = OB_SUCCESS;

int64_t N = vars.count();

if (N > 0 && OB_FAIL(vars_.reserve(N))) {

OB_LOG(WARN, "fail to reserve vars", K(ret));

}

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

const ObVarInfo &var_info = vars.at(i);

ObVarInfo clone_var_info;

if (OB_FAIL(var_info.deep_copy(allocator_, clone_var_info))) {

LOG_WARN("fail to deep copy var info", K(ret), K(var_info));

} else if (OB_FAIL(vars_.push_back(clone_var_info))) {

// deep_copy时只写了ObString对象，ObString对象可以在allocator_析构的时候完全释放掉，因此这里不用调ObString的析构函数

LOG_WARN("fail to push back vars", K(ret), K(clone_var_info));

}

}

return ret;

}

int ObPhysicalPlan::init_params_info_str()

{

int ret = common::OB_SUCCESS;

int64_t N = params_info_.count();

int64_t buf_len = N * ObParamInfo::MAX_STR_DES_LEN + 1;

int64_t pos = 0;

char *buf = (char *)allocator_.alloc(buf_len);

if (OB_ISNULL(buf)) {

ret = OB_ALLOCATE_MEMORY_FAILED;

SQL_PC_LOG(WARN, "fail to alloc memory for param info", K(ret));

} else {

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

if (N - 1 != i) {

if (OB_FAIL(databuff_printf(buf, buf_len, pos, "{%d,%d,%d,%d,%d},",

params_info_.at(i).flag_.need_to_check_type_,

params_info_.at(i).flag_.need_to_check_bool_value_,

params_info_.at(i).flag_.expected_bool_value_,

params_info_.at(i).scale_,

params_info_.at(i).type_))) {

SQL_PC_LOG(WARN, "fail to buff_print param info", K(ret));

}

} else {

if (OB_FAIL(databuff_printf(buf, buf_len, pos, "{%d,%d,%d,%d,%d}",

params_info_.at(i).flag_.need_to_check_type_,

params_info_.at(i).flag_.need_to_check_bool_value_,

params_info_.at(i).flag_.expected_bool_value_,

params_info_.at(i).scale_,

params_info_.at(i).type_))) {

SQL_PC_LOG(WARN, "fail to buff_print param info", K(ret));

}

}

}

}

if (OB_SUCC(ret)) {

if (OB_FAIL(ob_write_string(allocator_, ObString(pos, buf), stat_.param_infos_))) {

SQL_PC_LOG(WARN, "fail to deep copy param infos", K(ret));

}

}

return ret;

}

int ObPhysicalPlan::set_field_columns(const ColumnsFieldArray &fields)

{

int ret = OB_SUCCESS;

ObField field;

WITH_CONTEXT(mem_context_) {

int64_t N = fields.count();

if (N > 0 && OB_FAIL(field_columns_.reserve(N))) {

OB_LOG(WARN, "fail to reserve field column", K(ret));

}

for (int i = 0; OB_SUCC(ret) && i < N; ++i) {

const ObField &ofield = fields.at(i);

LOG_DEBUG("ofield info", K(ofield));

if (!contain_paramed_column_field_ && ofield.is_paramed_select_item_) {

if (OB_ISNULL(ofield.paramed_ctx_)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid paramed ctx", K(ofield.paramed_ctx_), K(i));

} else if (ofield.paramed_ctx_->param_idxs_.count() > 0) {

contain_paramed_column_field_ = true;

}

}

if (OB_FAIL(ret)) {

// do nothing

} else if (OB_FAIL(field.deep_copy(ofield, &allocator_))) {

LOG_WARN("deep copy field failed", K(ret));

} else if (OB_FAIL(field_columns_.push_back(field))) {

LOG_WARN("push back field columns failed", K(ret));

} else {

LOG_DEBUG("succ to push back field columns", K(field));

}

}

}

return ret;

}

int ObPhysicalPlan::set_param_fields(const common::ParamsFieldArray &params)

{

int ret = OB_SUCCESS;

int64_t N = params.count();

WITH_CONTEXT(mem_context_) {

if(N > 0 && OB_FAIL(param_columns_.reserve(N))) {

LOG_WARN("failed to reserved param field", K(ret));

}

ObField tmp_field;

for (int i = 0; OB_SUCC(ret) && i < N; ++i) {

const ObField &param_field = params.at(i);

if (OB_FAIL(tmp_field.deep_copy(param_field, &allocator_))) {

LOG_WARN("deep copy field failed", K(ret));

} else if (OB_FAIL(param_columns_.push_back(tmp_field))) {

LOG_WARN("push back field columns failed", K(ret));

}

}

}

return ret;

}

int ObPhysicalPlan::set_returning_param_fields(const common::ParamsFieldArray &params)

{

int ret = OB_SUCCESS;

int64_t N = params.count();

WITH_CONTEXT(mem_context_) {

if(N > 0 && OB_FAIL(returning_param_columns_.reserve(N))) {

LOG_WARN("failed to reserved returning param field", K(ret));

}

ObField tmp_field;

for (int i = 0; OB_SUCC(ret) && i < N; ++i) {

const ObField &param_field = params.at(i);

if (OB_FAIL(tmp_field.deep_copy(param_field, &allocator_))) {

LOG_WARN("deep copy field failed", K(ret));

} else if (OB_FAIL(returning_param_columns_.push_back(tmp_field))) {

LOG_WARN("push back field columns failed", K(ret));

}

}

}

return ret;

}

int ObPhysicalPlan::set_autoinc_params(const ObIArray<share::AutoincParam> &autoinc_params)

{

return autoinc_params_.assign(autoinc_params);

}

int ObPhysicalPlan::set_stmt_need_privs(const ObStmtNeedPrivs& stmt_need_privs)

{

int ret = OB_SUCCESS;

stmt_need_privs_.reset();

if (OB_FAIL(stmt_need_privs_.deep_copy(stmt_need_privs, allocator_))) {

LOG_WARN("Failed to deep copy ObStmtNeedPrivs", K_(stmt_need_privs));

}

return ret;

}

int ObPhysicalPlan::set_stmt_ora_need_privs(const ObStmtOraNeedPrivs& stmt_ora_need_privs)

{

int ret = OB_SUCCESS;

stmt_ora_need_privs_.reset();

if (OB_FAIL(stmt_ora_need_privs_.deep_copy(stmt_ora_need_privs, allocator_))) {

LOG_WARN("Failed to deep copy ObStmtNeedPrivs", K_(stmt_ora_need_privs));

}

return ret;

}

int ObPhysicalPlan::set_audit_units(const common::ObIArray<ObAuditUnit>& audit_units)

{

return audit_units_.assign(audit_units);

}

void ObPhysicalPlan::inc_large_querys()

{

ATOMIC_INC(&(stat_.large_querys_));

}

void ObPhysicalPlan::inc_delayed_large_querys()

{

ATOMIC_INC(&(stat_.delayed_large_querys_));

}

void ObPhysicalPlan::inc_delayed_px_querys()

{

ATOMIC_INC(&(stat_.delayed_px_querys_));

}

bool ObPhysicalPlan::is_stmt_modify_trans() const

{

return is_sfu_ || ObStmt::is_dml_write_stmt(stmt_type_);

}

int ObPhysicalPlan::init_operator_stats()

{

int ret = OB_SUCCESS;

if (OB_FAIL(op_stats_.init(&allocator_, next_phy_operator_id_))) {

LOG_WARN("fail to init op_stats", K(ret));

}

return ret;

}

void ObPhysicalPlan::update_plan_stat(const ObAuditRecordData &record,

const bool is_first,

const bool is_evolution,

const ObIArray<ObTableRowCount> *table_row_count_list)

{

const int64_t current_time = ObClockGenerator::getClock();

int64_t execute_count = 0;

if (record.is_timeout()) {

ATOMIC_INC(&(stat_.timeout_count_));

ATOMIC_AAF(&(stat_.total_process_time_), record.get_process_time());

} else if (!GCONF.enable_perf_event) { // short route

ATOMIC_AAF(&(stat_.elapsed_time_), record.get_elapsed_time());

ATOMIC_AAF(&(stat_.cpu_time_), record.get_elapsed_time() - record.exec_record_.wait_time_end_

- (record.exec_timestamp_.run_ts_ - record.exec_timestamp_.receive_ts_));

if (is_first) {

ATOMIC_STORE(&(stat_.hit_count_), 0);

} else {

ATOMIC_INC(&(stat_.hit_count_));

}

} else { // long route stat begin

execute_count = ATOMIC_AAF(&stat_.execute_times_, 1);

ATOMIC_AAF(&(stat_.total_process_time_), record.get_process_time());

ATOMIC_AAF(&(stat_.disk_reads_), record.exec_record_.get_io_read_count());

ATOMIC_AAF(&(stat_.direct_writes_), record.exec_record_.get_io_write_count());

ATOMIC_AAF(&(stat_.buffer_gets_), 2 * record.exec_record_.get_row_cache_hit()

+ 2 * record.exec_record_.get_fuse_row_cache_hit()

+ 2 * record.exec_record_.get_bloom_filter_filts()

+ record.exec_record_.get_block_cache_hit()

+ record.exec_record_.get_io_read_count());

ATOMIC_AAF(&(stat_.application_wait_time_), record.exec_record_.get_application_time());

ATOMIC_AAF(&(stat_.concurrency_wait_time_), record.exec_record_.get_concurrency_time());

ATOMIC_AAF(&(stat_.user_io_wait_time_), record.exec_record_.get_user_io_time());

ATOMIC_AAF(&(stat_.rows_processed_), record.return_rows_ + record.affected_rows_);

ATOMIC_AAF(&(stat_.elapsed_time_), record.get_elapsed_time());

ATOMIC_AAF(&(stat_.cpu_time_), record.get_elapsed_time() - record.exec_record_.wait_time_end_

- (record.exec_timestamp_.run_ts_ - record.exec_timestamp_.receive_ts_));

// ATOMIC_STORE(&(stat_.expected_worker_count_), record.expected_worker_cnt_);

if (is_first) {

ATOMIC_STORE(&(stat_.hit_count_), 0);

} else {

ATOMIC_INC(&(stat_.hit_count_));

}

if (record.get_elapsed_time() > GCONF.trace_log_slow_query_watermark) {

ATOMIC_INC(&(stat_.slow_count_));

}

int64_t slowest_usec = ATOMIC_LOAD(&stat_.slowest_exec_usec_);

if (slowest_usec < record.get_elapsed_time()) {

ATOMIC_STORE(&(stat_.slowest_exec_usec_), record.get_elapsed_time());

ATOMIC_STORE(&(stat_.slowest_exec_time_), current_time);

}

if (stat_.table_row_count_first_exec_ != NULL && table_row_count_list != NULL) {

int64_t access_table_num = stat_.access_table_num_;

int64_t max_index = std::min(access_table_num,

std::min(table_row_count_list->count(),

OB_MAX_TABLE_NUM_PER_STMT));

if (is_first) {

for (int64_t i = 0; i < max_index; ++i) {

ATOMIC_STORE(&(stat_.table_row_count_first_exec_[i].op_id_),

table_row_count_list->at(i).op_id_);

ATOMIC_STORE(&(stat_.table_row_count_first_exec_[i].row_count_),

table_row_count_list->at(i).row_count_);

LOG_DEBUG("first add row stat", K(table_row_count_list->at(i)));

} // for end

} else if (record.get_elapsed_time() > SLOW_QUERY_TIME_FOR_PLAN_EXPIRE) {

for (int64_t i = 0; !is_expired() && i < max_index; ++i) {

for (int64_t j = 0; !is_expired() && j < max_index; ++j) {

// 一些场景比如并行执行时，不同次执行表的行信息存储的顺序可能不同

if (table_row_count_list->at(i).op_id_ ==

stat_.table_row_count_first_exec_[j].op_id_) {

int64_t first_exec_row_count = ATOMIC_LOAD(&stat_.table_row_count_first_exec_[j]

.row_count_);

if (first_exec_row_count == -1) {

// do nothing

} else if (check_if_is_expired(first_exec_row_count,

table_row_count_list->at(i).row_count_)) {

set_is_expired(true);

LOG_INFO("plan is expired", K(first_exec_row_count),

K(table_row_count_list->at(i)),

"current_elapsed_time", record.get_elapsed_time(),

"plan_stat", stat_);

}

}

} // for max_index end

} // for max_index end

}

}

ATOMIC_STORE(&(stat_.last_active_time_), current_time);

if (ATOMIC_LOAD(&stat_.is_evolution_)) { //for spm

ATOMIC_INC(&(stat_.evolution_stat_.executions_));

// ATOMIC_AAF(&(stat_.evolution_stat_.cpu_time_),

// record.get_elapsed_time() - record.exec_record_.wait_time_end_

// - (record.exec_timestamp_.run_ts_ - record.exec_timestamp_.receive_ts_));

ATOMIC_AAF(&(stat_.evolution_stat_.cpu_time_), record.exec_timestamp_.executor_t_);

ATOMIC_AAF(&(stat_.evolution_stat_.elapsed_time_), record.get_elapsed_time());

ATOMIC_STORE(&(stat_.evolution_stat_.last_exec_ts_), record.exec_timestamp_.executor_end_ts_);

}

if (stat_.is_bind_sensitive_ && execute_count > 0) {

int64_t pos = execute_count % ObPlanStat::MAX_SCAN_STAT_SIZE;

ATOMIC_STORE(&(stat_.table_scan_stat_[pos].query_range_row_count_),

record.table_scan_stat_.query_range_row_count_);

ATOMIC_STORE(&(stat_.table_scan_stat_[pos].indexback_row_count_),

record.table_scan_stat_.indexback_row_count_);

ATOMIC_STORE(&(stat_.table_scan_stat_[pos].output_row_count_),

record.table_scan_stat_.output_row_count_);

}

} // long route stat ends

if (!is_expired() && stat_.enable_plan_expiration_) {

if (record.is_timeout() || record.status_ == OB_SESSION_KILLED) {

set_is_expired(true);

LOG_INFO("query plan is expired due to execution timeout", K(stat_));

} else if (is_first) {

ATOMIC_STORE(&(stat_.sample_times_), 0);

ATOMIC_STORE(&(stat_.first_exec_row_count_),

record.exec_record_.get_memstore_read_row_count() +

record.exec_record_.get_ssstore_read_row_count());

ATOMIC_STORE(&(stat_.first_exec_usec_), record.get_elapsed_time() - record.exec_record_.wait_time_end_

- (record.exec_timestamp_.run_ts_ - record.exec_timestamp_.receive_ts_));

} else if (0 == stat_.sample_times_) { // first sample query

ATOMIC_INC(&(stat_.sample_times_));

ATOMIC_STORE(&(stat_.sample_exec_row_count_),

record.exec_record_.get_memstore_read_row_count() +

record.exec_record_.get_ssstore_read_row_count());

ATOMIC_STORE(&(stat_.sample_exec_usec_), record.get_elapsed_time() - record.exec_record_.wait_time_end_

- (record.exec_timestamp_.run_ts_ - record.exec_timestamp_.receive_ts_));

} else {

int64_t sample_count = ATOMIC_AAF(&(stat_.sample_times_), 1);

int64_t sample_exec_row_count = ATOMIC_AAF(&(stat_.sample_exec_row_count_),

record.exec_record_.get_memstore_read_row_count() +

record.exec_record_.get_ssstore_read_row_count());

int64_t sample_exec_usec = ATOMIC_AAF(&(stat_.sample_exec_usec_),

record.get_elapsed_time() - record.exec_record_.wait_time_end_

- (record.exec_timestamp_.run_ts_ - record.exec_timestamp_.receive_ts_));

if (sample_count < SLOW_QUERY_SAMPLE_SIZE) {

// do nothing when query execution samples are not enough

} else {

if (stat_.cpu_time_ <= SLOW_QUERY_TIME_FOR_PLAN_EXPIRE * stat_.execute_times_) {

// do nothing for fast query

} else if (is_plan_unstable(sample_count, sample_exec_row_count, sample_exec_usec)) {

set_is_expired(true);

}

ATOMIC_STORE(&(stat_.sample_times_), 0);

}

}

}

}

bool ObPhysicalPlan::is_plan_unstable(const int64_t sample_count,

const int64_t sample_exec_row_count,

const int64_t sample_exec_usec)

{

bool bret = false;

if (sample_exec_usec <= SLOW_QUERY_TIME_FOR_PLAN_EXPIRE * sample_count) {

// sample query is fast query in the average

} else if (OB_PHY_PLAN_LOCAL == plan_type_) {

int64_t first_query_range_rows = ATOMIC_LOAD(&stat_.first_exec_row_count_);

if (sample_exec_row_count <= SLOW_QUERY_ROW_COUNT_THRESOLD * sample_count) {

// the sample query does not accesses too many rows in the average

} else if (sample_exec_row_count / sample_count > first_query_range_rows * 10) {

// the average sample query range row count increases great

bret = true;

LOG_INFO("local query plan is expired due to unstable performance",

K(bret), K(stat_.execute_times_),

K(first_query_range_rows), K(sample_exec_row_count), K(sample_count));

}

} else if ( OB_PHY_PLAN_DISTRIBUTED == plan_type_) {

int64_t first_exec_usec = ATOMIC_LOAD(&stat_.first_exec_usec_);

if (sample_exec_usec / sample_count > first_exec_usec * 2) {

// the average sample query execute time increases great

bret = true;

LOG_INFO("distribute query plan is expired due to unstable performance",

K(bret), K(stat_.execute_times_), K(first_exec_usec),

K(sample_exec_usec), K(sample_count));

}

} else {

// do nothing

}

return bret;

}

int64_t ObPhysicalPlan::get_evo_perf() const {

int64_t v = 0;

if (0 == stat_.evolution_stat_.executions_) {

v = 0;

} else {

v = stat_.evolution_stat_.cpu_time_/ stat_.evolution_stat_.executions_;

}

return v;

}

/**

* 目前采3个指标来淘汰计划，只有同时满足这3个条件，才会触发计划的淘汰：

* 1. 当前行数超过阈值（100行）

* 2. 执行时间超过阈值（5ms）

* 3. 表扫描行数与原扫描行数比值超过阈值（2倍）

*

* 设置当前行数阈值的原因是因为表扫描函数并不能保证递增，在频繁插入删除的情况下，原来的表扫描函数

* 可能会保持在一个较低的值，这时候计划淘汰会很频繁，

* 设置一个阈值的可以在很大程度上缓解计划淘汰的频率

*/

inline bool ObPhysicalPlan::check_if_is_expired(const int64_t first_exec_row_count,

const int64_t current_row_count) const

{

bool ret_bool = false;

if (current_row_count <= EXPIRED_PLAN_TABLE_ROW_THRESHOLD) { // 100 行

ret_bool = false;

} else {

ret_bool = ((first_exec_row_count == 0 && current_row_count > 0)

|| (first_exec_row_count > 0 && current_row_count / first_exec_row_count > TABLE_ROW_CHANGE_THRESHOLD));

}

return ret_bool;

}

int ObPhysicalPlan::inc_concurrent_num()

{

int ret = OB_SUCCESS;

int64_t concurrent_num = 0;

int64_t new_num = 0;

bool is_succ = false;

if (max_concurrent_num_ == ObMaxConcurrentParam::UNLIMITED) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("current physical plan is unlimit", K(ret), K(max_concurrent_num_));

} else {

while(OB_SUCC(ret) && false == is_succ) {

concurrent_num = ATOMIC_LOAD(&concurrent_num_);

if (concurrent_num >= max_concurrent_num_) {

ret = OB_REACH_MAX_CONCURRENT_NUM;

} else {

new_num = concurrent_num + 1;

is_succ = ATOMIC_BCAS(&concurrent_num_, concurrent_num, new_num);

}

}

}

return ret;

}

void ObPhysicalPlan::dec_concurrent_num()

{

ATOMIC_DEC(&concurrent_num_);

}

int ObPhysicalPlan::set_max_concurrent_num(int64_t max_concurrent_num)

{

int ret = OB_SUCCESS;

if (max_concurrent_num < 0) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid max_concurrent_num", K(ret), K(max_concurrent_num));

} else {

ATOMIC_STORE(&max_concurrent_num_, max_concurrent_num);

}

return ret;

}

int64_t ObPhysicalPlan::get_max_concurrent_num()

{

return ATOMIC_LOAD(&max_concurrent_num_);

}

OB_SERIALIZE_MEMBER(FlashBackQueryItem,

table_id_,

time_val_);

//因为还没看到远程执行对force_trace_log的处理，所以暂时不序列化

OB_SERIALIZE_MEMBER(ObPhysicalPlan,

tenant_schema_version_, //该字段执行期没被使用

phy_hint_.query_timeout_,

phy_hint_.read_consistency_,

is_sfu_,

dependency_tables_,

param_count_,

plan_type_,

signature_,

stmt_type_,

regexp_op_count_,

literal_stmt_type_,

like_op_count_,

is_ignore_stmt_,

object_id_,

stat_.sql_id_,

is_contain_inner_table_,

is_update_uniq_index_,

dummy_string_,

is_returning_,

location_type_,

use_px_,

vars_,

px_dop_,

has_nested_sql_,

flashback_query_items_,

stat_.enable_early_lock_release_,

encrypt_meta_array_,

use_pdml_,

is_new_engine_,

use_temp_table_,

batch_size_,

need_drive_dml_query_,

is_need_trans_,

contain_oracle_trx_level_temporary_table_,

ddl_schema_version_,

ddl_table_id_,

phy_hint_.monitor_,

need_serial_exec_,

contain_pl_udf_or_trigger_,

is_packed_,

has_instead_of_trigger_,

is_plain_insert_,

ddl_execution_id_,

ddl_task_id_,

stat_.plan_id_,

min_cluster_version_,

need_record_plan_info_,

enable_append_,

append_table_id_,

is_enable_px_fast_reclaim_,

gtt_session_scope_ids_,

gtt_trans_scope_ids_,

subschema_ctx_);

int ObPhysicalPlan::set_table_locations(const ObTablePartitionInfoArray &infos,

ObSchemaGetterGuard &schema_guard)

{

int ret = OB_SUCCESS;

table_locations_.reset();

das_table_locations_.reset();

if (OB_FAIL(table_locations_.prepare_allocate_and_keep_count(infos.count(),

allocator_))) {

LOG_WARN("fail to init table location count", K(ret));

} else if (OB_FAIL(das_table_locations_.prepare_allocate_and_keep_count(infos.count(),

allocator_))) {

LOG_WARN("fail to init das table location count", K(ret));

}

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

ObTableLocation &tl = infos.at(i)->get_table_location();

const ObTableSchema *table_schema = nullptr;

if (tl.use_das()) {

if (OB_FAIL(das_table_locations_.push_back(tl))) {

LOG_WARN("fail to push das table location", K(ret), K(i));

}

} else if (OB_FAIL(table_locations_.push_back(tl))) {

LOG_WARN("fail to push table location", K(ret), K(i));

} else if (OB_FAIL(schema_guard.get_table_schema(MTL_ID(), tl.get_ref_table_id(), table_schema))) {

LOG_WARN("get table schema failed", K(ret), K(tl.get_ref_table_id()));

} else {

contain_index_location_ |= table_schema->is_index_table();

}

LOG_DEBUG("set table location", K(tl), K(tl.use_das()));

}

return ret;

}

int ObPhysicalPlan::set_location_constraints(const ObIArray<LocationConstraint> &base_constraints,

const ObIArray<ObPwjConstraint *> &strict_constraints,

const ObIArray<ObPwjConstraint *> &non_strict_constraints,

const ObIArray<ObDupTabConstraint> &dup_table_replica_cons)

{

// deep copy location constraints

int ret = OB_SUCCESS;

if (base_constraints.count() > 0) {

base_constraints_.reset();

base_constraints_.set_allocator(&allocator_);

if (OB_FAIL(base_constraints_.init(base_constraints.count()))) {

LOG_WARN("failed to init base constraints", K(ret));

}

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

if (OB_FAIL(base_constraints_.push_back(base_constraints.at(i)))) {

LOG_WARN("failed to push back element", K(ret), K(base_constraints.at(i)));

} else { /*do nothing*/ }

}

}

if (OB_SUCC(ret) && strict_constraints.count() > 0) {

strict_constrinats_.reset();

strict_constrinats_.set_allocator(&allocator_);

if (OB_FAIL(strict_constrinats_.init(strict_constraints.count()))) {

LOG_WARN("failed to init strict constraints", K(ret));

} else if (OB_FAIL(strict_constrinats_.prepare_allocate(strict_constraints.count()))) {

LOG_WARN("failed to prepare allocate location constraints", K(ret));

} else {

ObPwjConstraint *cur_cons;

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

if (OB_ISNULL(cur_cons = strict_constraints.at(i))) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("get unexpected null", K(ret), K(i));

} else if (cur_cons->count() <= 0) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected empty array", K(ret));

} else {

strict_constrinats_.at(i).reset();

strict_constrinats_.at(i).set_allocator(&allocator_);

if (OB_FAIL(strict_constrinats_.at(i).init(cur_cons->count()))) {

LOG_WARN("failed to init fixed array", K(ret));

}

for (int64_t j = 0; OB_SUCC(ret) && j < cur_cons->count(); ++j) {

if (OB_FAIL(strict_constrinats_.at(i).push_back(cur_cons->at(j)))) {

LOG_WARN("failed to push back element", K(ret), K(cur_cons->at(j)));

} else { /*do nothing*/ }

}

}

}

}

}

if (OB_SUCC(ret) && non_strict_constraints.count() > 0) {

non_strict_constrinats_.reset();

non_strict_constrinats_.set_allocator(&allocator_);

if (OB_FAIL(non_strict_constrinats_.init(non_strict_constraints.count()))) {

LOG_WARN("failed to init strict constraints", K(ret));

} else if (OB_FAIL(non_strict_constrinats_.prepare_allocate(non_strict_constraints.count()))) {

LOG_WARN("failed to prepare allocate location constraints", K(ret));

} else {

ObPwjConstraint *cur_cons;

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

if (OB_ISNULL(cur_cons = non_strict_constraints.at(i))) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("get unexpected null", K(ret), K(i));

} else if (cur_cons->count() <= 0) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("unexpected empty array", K(ret));

} else {

non_strict_constrinats_.at(i).reset();

non_strict_constrinats_.at(i).set_allocator(&allocator_);

if (OB_FAIL(non_strict_constrinats_.at(i).init(cur_cons->count()))) {

LOG_WARN("failed to init fixed array", K(ret));

}

for (int64_t j = 0; OB_SUCC(ret) && j < cur_cons->count(); ++j) {

if (OB_FAIL(non_strict_constrinats_.at(i).push_back(cur_cons->at(j)))) {

LOG_WARN("failed to push back element", K(ret), K(cur_cons->at(j)));

} else { /*do nothing*/ }

}

}

}

}

}

if (OB_SUCC(ret) && dup_table_replica_cons.count() > 0) {

dup_table_replica_cons_.reset();

dup_table_replica_cons_.set_allocator(&allocator_);

if (OB_FAIL(dup_table_replica_cons_.init(dup_table_replica_cons.count()))) {

LOG_WARN("failed to init duplicate table constraints", K(ret));

} else {

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

if(OB_FAIL(dup_table_replica_cons_.push_back(dup_table_replica_cons.at(i)))) {

LOG_WARN("failed to assign element", K(ret), K(dup_table_replica_cons.at(i)));

} else { /*do nothing*/ }

}

}

}

if (OB_FAIL(ret)) {

base_constraints_.reset();

strict_constrinats_.reset();

non_strict_constrinats_.reset();

dup_table_replica_cons_.reset();

} else {

LOG_TRACE("deep copied location constraints", K(base_constraints_), K(strict_constrinats_),

K(non_strict_constrinats_), K(dup_table_replica_cons_));

}

return ret;

}

bool ObPhysicalPlan::has_same_location_constraints(const ObPhysicalPlan &r) const

{

bool is_same = true;

const ObIArray<LocationConstraint>& l_base_cons = get_base_constraints();

const ObIArray<LocationConstraint>& r_base_cons = r.get_base_constraints();

const ObIArray<ObPlanPwjConstraint>& l_non_strict_cons = get_non_strict_constraints();

const ObIArray<ObPlanPwjConstraint>& r_non_strict_cons = r.get_non_strict_constraints();

const ObIArray<ObPlanPwjConstraint>& l_strict_cons = get_strict_constraints();

const ObIArray<ObPlanPwjConstraint>& r_strict_cons = r.get_strict_constraints();

const ObIArray<ObDupTabConstraint>& l_dup_rep_cons = get_dup_table_replica_constraints();

const ObIArray<ObDupTabConstraint>& r_dup_rep_cons = r.get_dup_table_replica_constraints();

if (l_base_cons.count() != r_base_cons.count() ||

l_strict_cons.count() != r_strict_cons.count() ||

l_non_strict_cons.count() != r_non_strict_cons.count()||

l_dup_rep_cons.count() != r_dup_rep_cons.count()) {

is_same = false;

} else {

for (int64_t i = 0; is_same && i < l_base_cons.count(); i++) {

is_same = is_same && (l_base_cons.at(i) == r_base_cons.at(i));

}

for (int64_t i = 0; is_same && i < l_strict_cons.count(); i++) {

if (l_strict_cons.at(i).count() != r_strict_cons.at(i).count()) {

is_same = false;

} else {

for (int64_t j = 0; is_same && j < l_strict_cons.at(i).count(); j++) {

is_same = (l_strict_cons.at(i).at(j) == (r_strict_cons.at(i)).at(j));

}

}

}

for (int64_t i = 0; is_same && i < l_non_strict_cons.count(); i++) {

if (l_non_strict_cons.at(i).count() != r_non_strict_cons.at(i).count()) {

is_same = false;

} else {

for (int64_t j = 0; is_same && j < l_non_strict_cons.at(i).count(); j++) {

is_same = (l_non_strict_cons.at(i).at(j) == r_non_strict_cons.at(i).at(j));

}

}

}

for(int64_t i = 0; is_same && i < l_dup_rep_cons.count(); i++) {

is_same = is_same && (l_dup_rep_cons.at(i) == r_dup_rep_cons.at(i));

}

}

return is_same;

}

DEF_TO_STRING(FlashBackQueryItem)

{

int64_t pos = 0;

J_OBJ_START();

J_KV(K_(table_id));

J_KV(K_(time_val));

J_OBJ_END();

return pos;

}

int ObPhysicalPlan::alloc_op_spec(const ObPhyOperatorType type,

const int64_t child_cnt,

ObOpSpec *&op,

const uint64_t op_id)

{

int ret = OB_SUCCESS;

UNUSED(op_id);

if (type >= PHY_END || child_cnt < 0) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(ret), K(type), K(child_cnt));

} else if (OB_FAIL(ObOperatorFactory::alloc_op_spec(

allocator_, type, child_cnt, op))) {

LOG_WARN("allocate operator spec failed", K(ret));

} else if (OB_ISNULL(op)) {