/**

* 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 "ob_table_scan_op.h"

#include "sql/engine/ob_exec_context.h"

#include "sql/executor/ob_task_spliter.h"

#include "sql/das/ob_das_group_scan_op.h"

#include "sql/das/ob_das_define.h"

#include "sql/das/ob_das_utils.h"

#include "lib/profile/ob_perf_event.h"

#include "lib/geo/ob_s2adapter.h"

#include "lib/geo/ob_geo_utils.h"

#include "share/ob_ddl_common.h"

#include "share/ob_ddl_checksum.h"

#include "storage/access/ob_table_scan_iterator.h"

#include "observer/ob_server_struct.h"

#include "observer/ob_server.h"

#include "observer/virtual_table/ob_virtual_data_access_service.h"

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

#include "observer/omt/ob_tenant_srs.h"

#include "share/external_table/ob_external_table_file_mgr.h"

#include "share/external_table/ob_external_table_utils.h"

#include "lib/container/ob_array_wrap.h"

namespace oceanbase

{

using namespace common;

using namespace storage;

using namespace share;

using namespace share::schema;

namespace sql

{

#define MY_CTDEF (MY_SPEC.tsc_ctdef_)

int FlashBackItem::set_flashback_query_info(ObEvalCtx &eval_ctx, ObDASScanRtDef &scan_rtdef) const

{

int ret = OB_SUCCESS;

ObDatum *datum = NULL;

const ObExpr *expr = flashback_query_expr_;

scan_rtdef.need_scn_ = need_scn_;

if (TableItem::NOT_USING == flashback_query_type_) {

// do nothing

} else if (OB_ISNULL(expr)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("flash back query expr is NULL", K(ret));

} else if (OB_FAIL(expr->eval(eval_ctx, datum))) {

LOG_WARN("expr evaluate failed", K(ret));

} else if (datum->is_null()) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("NULL value", K(ret));

} else {

scan_rtdef.fb_read_tx_uncommitted_ = fq_read_tx_uncommitted_;

if (TableItem::USING_TIMESTAMP == flashback_query_type_) {

if (ObTimestampTZType != expr->datum_meta_.type_) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("type not match", K(ret));

} else if (OB_FAIL(scan_rtdef.fb_snapshot_.convert_from_ts(datum->get_otimestamp_tz().time_us_))) {

LOG_WARN("failed to convert from ts", K(ret));

} else {

LOG_TRACE("fb_snapshot_ result", K(scan_rtdef.fb_snapshot_), K(*datum));

}

} else if (TableItem::USING_SCN == flashback_query_type_) {

if (ObUInt64Type != expr->datum_meta_.type_) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("type not match", K(ret));

} else if (OB_FAIL(scan_rtdef.fb_snapshot_.convert_for_sql(datum->get_int()))) {

LOG_WARN("failed to convert for gts", K(ret));

} else {

LOG_TRACE("fb_snapshot_ result", K(scan_rtdef.fb_snapshot_), K(*datum));

}

}

}

//对于同时存在hint指定的frozen_version和flashback query指定了snapshot version的情况下, 选择保留

//flashback query指定的snapshot version, 忽略hint指定的frozen_version

if (OB_SUCC(ret)) {

if (scan_rtdef.fb_snapshot_.is_valid()) {

scan_rtdef.frozen_version_ = transaction::ObTransVersion::INVALID_TRANS_VERSION;

} else {

/*do nothing*/

}

}

return ret;

}

OB_SERIALIZE_MEMBER(AgentVtAccessMeta,

vt_table_id_,

access_exprs_,

access_column_ids_,

access_row_types_,

key_types_);

OB_DEF_SERIALIZE(ObTableScanCtDef)

{

int ret = OB_SUCCESS;

bool has_lookup = (lookup_ctdef_ != nullptr);

OB_UNIS_ENCODE(pre_query_range_);

OB_UNIS_ENCODE(flashback_item_.need_scn_);

OB_UNIS_ENCODE(flashback_item_.flashback_query_expr_);

OB_UNIS_ENCODE(flashback_item_.flashback_query_type_);

OB_UNIS_ENCODE(bnlj_param_idxs_);

OB_UNIS_ENCODE(scan_flags_);

OB_UNIS_ENCODE(scan_ctdef_);

OB_UNIS_ENCODE(has_lookup);

if (OB_SUCC(ret) && has_lookup) {

OB_UNIS_ENCODE(*lookup_ctdef_);

OB_UNIS_ENCODE(*lookup_loc_meta_);

}

bool has_dppr_tbl = (das_dppr_tbl_ != nullptr);

OB_UNIS_ENCODE(has_dppr_tbl);

if (OB_SUCC(ret) && has_dppr_tbl) {

OB_UNIS_ENCODE(*das_dppr_tbl_);

}

OB_UNIS_ENCODE(calc_part_id_expr_);

OB_UNIS_ENCODE(global_index_rowkey_exprs_);

OB_UNIS_ENCODE(flashback_item_.fq_read_tx_uncommitted_);

return ret;

}

OB_DEF_SERIALIZE_SIZE(ObTableScanCtDef)

{

int64_t len = 0;

bool has_lookup = (lookup_ctdef_ != nullptr);

OB_UNIS_ADD_LEN(pre_query_range_);

OB_UNIS_ADD_LEN(flashback_item_.need_scn_);

OB_UNIS_ADD_LEN(flashback_item_.flashback_query_expr_);

OB_UNIS_ADD_LEN(flashback_item_.flashback_query_type_);

OB_UNIS_ADD_LEN(bnlj_param_idxs_);

OB_UNIS_ADD_LEN(scan_flags_);

OB_UNIS_ADD_LEN(scan_ctdef_);

OB_UNIS_ADD_LEN(has_lookup);

if (has_lookup) {

OB_UNIS_ADD_LEN(*lookup_ctdef_);

OB_UNIS_ADD_LEN(*lookup_loc_meta_);

}

bool has_dppr_tbl = (das_dppr_tbl_ != nullptr);

OB_UNIS_ADD_LEN(has_dppr_tbl);

if (has_dppr_tbl) {

OB_UNIS_ADD_LEN(*das_dppr_tbl_);

}

OB_UNIS_ADD_LEN(calc_part_id_expr_);

OB_UNIS_ADD_LEN(global_index_rowkey_exprs_);

OB_UNIS_ADD_LEN(flashback_item_.fq_read_tx_uncommitted_);

return len;

}

OB_DEF_DESERIALIZE(ObTableScanCtDef)

{

int ret = OB_SUCCESS;

bool has_lookup = false;

OB_UNIS_DECODE(pre_query_range_);

OB_UNIS_DECODE(flashback_item_.need_scn_);

OB_UNIS_DECODE(flashback_item_.flashback_query_expr_);

OB_UNIS_DECODE(flashback_item_.flashback_query_type_);

OB_UNIS_DECODE(bnlj_param_idxs_);

OB_UNIS_DECODE(scan_flags_);

OB_UNIS_DECODE(scan_ctdef_);

OB_UNIS_DECODE(has_lookup);

if (OB_SUCC(ret) && has_lookup) {

void *ctdef_buf = allocator_.alloc(sizeof(ObDASScanCtDef));

if (OB_ISNULL(ctdef_buf)) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("allocate das scan ctdef buffer failed", K(ret), K(sizeof(ObDASScanCtDef)));

} else {

lookup_ctdef_ = new(ctdef_buf) ObDASScanCtDef(allocator_);

OB_UNIS_DECODE(*lookup_ctdef_);

}

if (OB_SUCC(ret)) {

void *loc_meta_buf = allocator_.alloc(sizeof(ObDASTableLocMeta));

if (OB_ISNULL(loc_meta_buf)) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("allocate table loc meta failed", K(ret));

} else {

lookup_loc_meta_ = new(loc_meta_buf) ObDASTableLocMeta(allocator_);

OB_UNIS_DECODE(*lookup_loc_meta_);

}

}

}

bool has_dppr_tbl = (das_dppr_tbl_ != nullptr);

OB_UNIS_DECODE(has_dppr_tbl);

if (OB_SUCC(ret) && has_dppr_tbl) {

OZ(allocate_dppr_table_loc());

OB_UNIS_DECODE(*das_dppr_tbl_);

}

OB_UNIS_DECODE(calc_part_id_expr_);

OB_UNIS_DECODE(global_index_rowkey_exprs_);

OB_UNIS_DECODE(flashback_item_.fq_read_tx_uncommitted_);

return ret;

}

int ObTableScanCtDef::allocate_dppr_table_loc()

{

int ret = OB_SUCCESS;

void *buf = allocator_.alloc(sizeof(ObTableLocation));

if (OB_ISNULL(buf)) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("allocate table location buffer failed", K(ret));

} else {

das_dppr_tbl_ = new(buf) ObTableLocation(allocator_);

}

return ret;

}

OB_INLINE void ObTableScanRtDef::prepare_multi_part_limit_param()

{

/* for multi-partition scanning, */

/* the limit operation pushed down to the partition TSC needs to be adjusted */

/* its rule: */

/* TSC(limit m, n) */

/* / \ */

/* / \ */

/* DAS Scan(p0) DAS Scan(p1) */

/* (p0, limit m+n) (p1, limit m+n) */

/* each partition scans limit m+n rows of data, */

/* and TSC operator selects the offset (m) limit (n) rows in the final result */

int64_t offset = scan_rtdef_.limit_param_.offset_;

int64_t limit = scan_rtdef_.limit_param_.limit_;

scan_rtdef_.limit_param_.limit_ = offset + limit;

scan_rtdef_.limit_param_.offset_ = 0;

if (lookup_rtdef_ != nullptr) {

offset = lookup_rtdef_->limit_param_.offset_;

limit = lookup_rtdef_->limit_param_.limit_;

lookup_rtdef_->limit_param_.limit_ = offset + limit;

lookup_rtdef_->limit_param_.offset_ = 0;

}

}

ObTableScanOpInput::ObTableScanOpInput(ObExecContext &ctx, const ObOpSpec &spec)

: ObOpInput(ctx, spec),

tablet_loc_(nullptr),

not_need_extract_query_range_(false)

{

}

ObTableScanOpInput::~ObTableScanOpInput()

{

}

void ObTableScanOpInput::reset()

{

tablet_loc_ = nullptr;

key_ranges_.reset();

ss_key_ranges_.reset();

mbr_filters_.reset();

range_array_pos_.reset();

not_need_extract_query_range_ = false;

}

OB_DEF_SERIALIZE_SIZE(ObTableScanOpInput)

{

int len = 0;

LST_DO_CODE(OB_UNIS_ADD_LEN,

key_ranges_,

not_need_extract_query_range_,

ss_key_ranges_);

return len;

}

OB_DEF_SERIALIZE(ObTableScanOpInput)

{

int ret = OB_SUCCESS;

LST_DO_CODE(OB_UNIS_ENCODE,

key_ranges_,

not_need_extract_query_range_,

ss_key_ranges_);

return ret;

}

OB_DEF_DESERIALIZE(ObTableScanOpInput)

{

int ret = OB_SUCCESS;

int64_t cnt = 0;

if (OB_FAIL(serialization::decode_vi64(buf, data_len, pos, &cnt))) {

LOG_WARN("decode failed", K(ret));

} else if (OB_FAIL(key_ranges_.prepare_allocate(cnt))) {

LOG_WARN("array prepare allocate failed", K(ret));

} else {

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

if (OB_FAIL(key_ranges_.at(i).deserialize(

exec_ctx_.get_allocator(), buf, data_len, pos))) {

LOG_WARN("range deserialize failed", K(ret));

}

if (OB_FAIL(ret)) {

} else if (OB_FAIL(serialization::decode_vi64(buf, data_len, pos, &cnt))) {

LOG_WARN("decode failed", K(ret));

} else if (OB_FAIL(ss_key_ranges_.prepare_allocate(cnt))) {

LOG_WARN("array prepare allocate failed", K(ret));

}

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

if (OB_FAIL(ss_key_ranges_.at(i).deserialize(exec_ctx_.get_allocator(),

buf, data_len, pos))) {

LOG_WARN("range deserialize failed", K(ret));

}

}

}

if (OB_SUCC(ret)) {

LST_DO_CODE(OB_UNIS_DECODE, not_need_extract_query_range_);

}

}

return ret;

}

int ObTableScanOpInput::init(ObTaskInfo &task_info)

{

int ret = OB_SUCCESS;

if (PHY_FAKE_CTE_TABLE == MY_SPEC.type_) {

LOG_DEBUG("CTE TABLE do not need init", K(ret));

} else if (ObTaskSpliter::INVALID_SPLIT == task_info.get_task_split_type()) {

ret = OB_NOT_INIT;

LOG_WARN("exec type is INVALID_SPLIT", K(ret));

} else {

if (1 == task_info.get_range_location().part_locs_.count() // only one table

&& 0 < task_info.get_range_location().part_locs_.at(0).scan_ranges_.count()) {

// multi-range

ret = key_ranges_.assign(task_info.get_range_location().part_locs_.at(0).scan_ranges_);

}

}

return ret;

}

OB_INLINE int ObTableScanOp::reuse_table_rescan_allocator()

{

int ret = OB_SUCCESS;

if (OB_ISNULL(table_rescan_allocator_)) {

ObSQLSessionInfo *my_session = GET_MY_SESSION(ctx_);

lib::ContextParam param;

param.set_mem_attr(my_session->get_effective_tenant_id(),

"TableRescanCtx", ObCtxIds::DEFAULT_CTX_ID)

.set_properties(lib::USE_TL_PAGE_OPTIONAL)

.set_ablock_size(lib::INTACT_MIDDLE_AOBJECT_SIZE);

lib::MemoryContext mem_context;

if (OB_FAIL(CURRENT_CONTEXT->CREATE_CONTEXT(mem_context, param))) {

LOG_WARN("fail to create entity", K(ret));

} else if (OB_ISNULL(mem_context)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("fail to create entity ", K(ret));

} else {

table_rescan_allocator_ = &mem_context->get_arena_allocator();

}

} else {

table_rescan_allocator_->reuse();

}

return ret;

}

ObTableScanSpec::ObTableScanSpec(ObIAllocator &alloc, const ObPhyOperatorType type)

: ObOpSpec(alloc, type),

table_loc_id_(OB_INVALID_ID),

ref_table_id_(OB_INVALID_ID),

limit_(NULL),

offset_(NULL),

frozen_version_(-1),

part_level_(ObPartitionLevel::PARTITION_LEVEL_MAX),

part_type_(ObPartitionFuncType::PARTITION_FUNC_TYPE_MAX),

subpart_type_(ObPartitionFuncType::PARTITION_FUNC_TYPE_MAX),

part_expr_(NULL),

subpart_expr_(NULL),

part_range_pos_(alloc),

subpart_range_pos_(alloc),

part_dep_cols_(alloc),

subpart_dep_cols_(alloc),

table_row_count_(0),

output_row_count_(0),

phy_query_range_row_count_(0),

query_range_row_count_(0),

index_back_row_count_(0),

estimate_method_(INVALID_METHOD),

est_records_(alloc),

available_index_name_(alloc),

pruned_index_name_(alloc),

unstable_index_name_(alloc),

ddl_output_cids_(alloc),

tsc_ctdef_(alloc),

pdml_partition_id_(NULL),

agent_vt_meta_(alloc),

flags_(0),

tenant_id_col_idx_(0),

partition_id_calc_type_(0)

{

}

OB_SERIALIZE_MEMBER((ObTableScanSpec, ObOpSpec),

table_loc_id_,

ref_table_id_,

flags_,

limit_,

offset_,

frozen_version_,

part_level_,

part_type_,

subpart_type_,

part_expr_,

subpart_expr_,

part_range_pos_,

subpart_range_pos_,

part_dep_cols_,

subpart_dep_cols_,

tsc_ctdef_,

pdml_partition_id_,

agent_vt_meta_,

ddl_output_cids_,

tenant_id_col_idx_,

partition_id_calc_type_);

DEF_TO_STRING(ObTableScanSpec)

{

int64_t pos = 0;

J_OBJ_START();

J_NAME("op_spec");

J_COLON();

pos += ObOpSpec::to_string(buf + pos, buf_len - pos);

J_COMMA();

J_KV(K(table_loc_id_),

K(ref_table_id_),

K(is_index_global_),

K(limit_),

K(offset_),

K(frozen_version_),

K(force_refresh_lc_),

K(is_top_table_scan_),

K(gi_above_),

K(batch_scan_flag_),

K(use_dist_das_),

K(tsc_ctdef_),

K(report_col_checksum_),

K_(agent_vt_meta),

K_(ddl_output_cids),

K_(tenant_id_col_idx));

J_OBJ_END();

return pos;

}

int ObTableScanSpec::set_est_row_count_record(const ObIArray<ObEstRowCountRecord> &est_records)

{

int ret = OB_SUCCESS;

OZ(est_records_.init(est_records.count()));

OZ(append(est_records_, est_records));

return ret;

}

int ObTableScanSpec::set_available_index_name(const ObIArray<ObString> &idx_name,

ObIAllocator &phy_alloc)

{

int ret = OB_SUCCESS;

OZ(available_index_name_.init(idx_name.count()));

FOREACH_CNT_X(n, idx_name, OB_SUCC(ret)) {

ObString name;

OZ(ob_write_string(phy_alloc, *n, name));

OZ(available_index_name_.push_back(name));

}

return ret;

}

int ObTableScanSpec::set_unstable_index_name(const ObIArray<ObString> &idx_name,

ObIAllocator &phy_alloc)

{

int ret = OB_SUCCESS;

OZ(unstable_index_name_.init(idx_name.count()));

FOREACH_CNT_X(n, idx_name, OB_SUCC(ret)) {

ObString name;

OZ(ob_write_string(phy_alloc, *n, name));

OZ(unstable_index_name_.push_back(name));

}

return ret;

}

int ObTableScanSpec::set_pruned_index_name(const ObIArray<ObString> &idx_name,

ObIAllocator &phy_alloc)

{

int ret = OB_SUCCESS;

OZ(pruned_index_name_.init(idx_name.count()));

FOREACH_CNT_X(n, idx_name, OB_SUCC(ret)) {

ObString name;

OZ(ob_write_string(phy_alloc, *n, name));

OZ(pruned_index_name_.push_back(name));

}

return ret;

}

int ObTableScanSpec::explain_index_selection_info(

char *buf, int64_t buf_len, int64_t &pos) const

{

int ret = OB_SUCCESS;

if (OB_FAIL(BUF_PRINTF(

"table_rows:%ld, physical_range_rows:%ld, logical_range_rows:%ld, "

"index_back_rows:%ld, output_rows:%ld",

table_row_count_, phy_query_range_row_count_, query_range_row_count_,

index_back_row_count_, output_row_count_))) {

LOG_WARN("BUF_PRINTF fails", K(ret));

}

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

// print available index id

if (OB_FAIL(BUF_PRINTF(", avaiable_index_name["))) {

LOG_WARN("BUF_PRINTF fails", K(ret));

}

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

if (OB_FAIL(BUF_PRINTF("%.*s", available_index_name_.at(i).length(),

available_index_name_.at(i).ptr()))) {

LOG_WARN("BUF_PRINTF fails", K(ret));

} else if (i != available_index_name_.count() - 1) {

if (OB_FAIL(BUF_PRINTF(","))) {

LOG_WARN("BUF_PRINTF fails", K(ret));

} else { /* do nothing*/ }

} else { /* do nothing*/ }

}

if (OB_SUCC(ret)) {

if (OB_FAIL(BUF_PRINTF("]"))) {

LOG_WARN("BUF_PRINTF fails", K(ret));

} else { /* Do nothing */ }

} else { /* Do nothing */ }

}

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

if (OB_FAIL(BUF_PRINTF(", pruned_index_name["))) {

LOG_WARN("BUF_PRINTF fails", K(ret));

}

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

if (OB_FAIL(BUF_PRINTF("%.*s", pruned_index_name_.at(i).length(),

pruned_index_name_.at(i).ptr()))) {

LOG_WARN("BUF_PRINTF fails", K(ret));

} else if (i != pruned_index_name_.count() - 1) {

if (OB_FAIL(BUF_PRINTF(","))) {

LOG_WARN("BUF_PRINTF fails", K(ret));

} else { /* do nothing*/ }

} else { /* do nothing*/ }

}

if (OB_SUCC(ret)) {

if (OB_FAIL(BUF_PRINTF("]"))) {

LOG_WARN("BUF_PRINTF fails", K(ret));

} else { /* Do nothing */ }

} else { /* Do nothing */ }

}

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

if (OB_FAIL(BUF_PRINTF(", unstable_index_name["))) {

LOG_WARN("BUF_PRINTF fails", K(ret));

}

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

if (OB_FAIL(BUF_PRINTF("%.*s", unstable_index_name_.at(i).length(),

unstable_index_name_.at(i).ptr()))) {

LOG_WARN("BUF_PRINTF fails", K(ret));

} else if (i != unstable_index_name_.count() - 1) {

if (OB_FAIL(BUF_PRINTF(","))) {

LOG_WARN("BUF_PRINTF fails", K(ret));

} else { /* do nothing*/ }

} else { /* do nothing*/ }

}

if (OB_SUCC(ret)) {

if (OB_FAIL(BUF_PRINTF("]"))) {

LOG_WARN("BUF_PRINTF fails", K(ret));

} else { /* Do nothing */ }

} else { /* Do nothing */ }

}

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

// print est row count infos

if (OB_FAIL(BUF_PRINTF(", estimation info[table_id:%ld,", est_records_.at(0).table_id_))) {

LOG_WARN("BUF_PRINTF fails", K(ret));

}

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

const ObEstRowCountRecord &record = est_records_.at(i);

if (OB_FAIL(BUF_PRINTF(

" (table_type:%ld, version:%ld-%ld-%ld, logical_rc:%ld, physical_rc:%ld)%c",

record.table_type_,

record.version_range_.base_version_,

record.version_range_.multi_version_start_,

record.version_range_.snapshot_version_,

record.logical_row_count_,

record.physical_row_count_,

i == est_records_.count() - 1 ? ']' : ','))) {

LOG_WARN("BUF PRINTF fails", K(ret));

}

}

}

return ret;

}

ObTableScanOp::ObTableScanOp(ObExecContext &exec_ctx, const ObOpSpec &spec, ObOpInput *input)

: ObOperator(exec_ctx, spec, input),

das_ref_(eval_ctx_, exec_ctx),

tsc_rtdef_(exec_ctx.get_allocator()),

need_final_limit_(false),

table_rescan_allocator_(NULL),

input_row_cnt_(0),

output_row_cnt_(0),

iter_end_(false),

iterated_rows_(0),

got_feedback_(false),

vt_result_converter_(nullptr),

cur_trace_id_(nullptr),

col_need_reshape_(),

column_checksum_(),

scan_task_id_(0),

report_checksum_(false),

in_rescan_(false),

global_index_lookup_op_(NULL),

spat_index_()

{

}

ObTableScanOp::~ObTableScanOp()

{

}

bool ObTableScanOp::has_das_scan_op(const ObDASTabletLoc *tablet_loc, ObDASScanOp *&das_op)

{

if (MY_SPEC.batch_scan_flag_) {

das_op = static_cast<ObDASScanOp*>(

das_ref_.find_das_task(tablet_loc, DAS_OP_TABLE_BATCH_SCAN));

} else {

das_op = static_cast<ObDASScanOp*>(

das_ref_.find_das_task(tablet_loc, DAS_OP_TABLE_SCAN));

}

return das_op != nullptr;

}

int ObTableScanOp::init_das_group_range(const int64_t cur_group_idx, const int64_t group_size)

{

int ret = OB_SUCCESS;

if (MY_SPEC.batch_scan_flag_) {

for (DASTaskIter task_iter = das_ref_.begin_task_iter(); !task_iter.is_end(); ++task_iter) {

ObDASGroupScanOp *batch_op = static_cast<ObDASGroupScanOp*>(*task_iter);

batch_op->init_group_range(cur_group_idx, group_size);

LOG_DEBUG("init das group range", K(batch_op), K(cur_group_idx), K(group_size));

}

}

return ret;

}

OB_INLINE int ObTableScanOp::create_one_das_task(ObDASTabletLoc *tablet_loc)

{

int ret = OB_SUCCESS;

ObIDASTaskOp *task_op = nullptr;

ObDASScanOp *scan_op = nullptr;

uint64_t table_loc_id = MY_SPEC.get_table_loc_id();

ObDASOpType op_type = MY_SPEC.batch_scan_flag_ ? DAS_OP_TABLE_BATCH_SCAN : DAS_OP_TABLE_SCAN;

if (OB_LIKELY(has_das_scan_op(tablet_loc, scan_op))) {

// reuse das scan op

} else if (OB_FAIL(das_ref_.create_das_task(tablet_loc, op_type, task_op))) {

LOG_WARN("prepare das task failed", K(ret));

} else {

scan_op = static_cast<ObDASScanOp*>(task_op);

scan_op->set_scan_ctdef(&MY_CTDEF.scan_ctdef_);

scan_op->set_scan_rtdef(&tsc_rtdef_.scan_rtdef_);

scan_op->set_can_part_retry(nullptr == tsc_rtdef_.scan_rtdef_.sample_info_

&& can_partition_retry());

scan_op->set_inner_rescan(in_rescan_);

tsc_rtdef_.scan_rtdef_.table_loc_->is_reading_ = true;

if (!MY_SPEC.is_index_global_ && MY_CTDEF.lookup_ctdef_ != nullptr) {

//is local index lookup, need to set the lookup ctdef to the das scan op

ObDASTableLoc *lookup_table_loc = tsc_rtdef_.lookup_rtdef_->table_loc_;

ObDASTabletLoc *lookup_tablet_loc = ObDASUtils::get_related_tablet_loc(

*tablet_loc, lookup_table_loc->loc_meta_->ref_table_id_);

if (OB_ISNULL(lookup_tablet_loc)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("lookup tablet loc is nullptr", K(ret), KPC(tablet_loc), KPC(lookup_table_loc->loc_meta_));

} else if (OB_FAIL(scan_op->set_lookup_ctdef(MY_CTDEF.lookup_ctdef_))) {

LOG_WARN("set lookup ctdef failed", K(ret));

} else if (OB_FAIL(scan_op->set_lookup_rtdef(tsc_rtdef_.lookup_rtdef_))) {

LOG_WARN("set lookup rtdef failed", K(ret));

} else if (OB_FAIL(scan_op->set_lookup_tablet_id(lookup_tablet_loc->tablet_id_))) {

LOG_WARN("set lookup tablet id failed", K(ret), KPC(lookup_tablet_loc));

} else {

lookup_table_loc->is_reading_ = true;

}

}

}

if (OB_SUCC(ret)) {

if (OB_FAIL(cherry_pick_range_by_tablet_id(scan_op))) {

LOG_WARN("prune query range by partition id failed", K(ret), KPC(tablet_loc));

}

}

return ret;

}

int ObTableScanOp::prepare_pushdown_limit_param()

{

int ret = OB_SUCCESS;

if (!limit_param_.is_valid()) {

//ignore, do nothing

} else if (MY_SPEC.batch_scan_flag_) {

//batch scan can not pushdown limit param to storage

need_final_limit_ = true;

tsc_rtdef_.scan_rtdef_.limit_param_.offset_ = 0;

tsc_rtdef_.scan_rtdef_.limit_param_.limit_ = -1;

if (nullptr != MY_CTDEF.lookup_ctdef_) {

OB_ASSERT(nullptr != tsc_rtdef_.lookup_rtdef_);

tsc_rtdef_.lookup_rtdef_->limit_param_.offset_ = 0;

tsc_rtdef_.lookup_rtdef_->limit_param_.limit_ = -1;

}

} else if (tsc_rtdef_.has_lookup_limit() || das_ref_.get_das_task_cnt() > 1) {

//for index back, need to final limit output rows in TableScan operator,

//please see me for the reason:

/* for multi-partition scanning, */

/* the limit operation pushed down to the partition TSC needs to be adjusted */

/* its rule: */

/* TSC(limit m, n) */

/* / \ */

/* / \ */

/* DAS Scan(p0) DAS Scan(p1) */

/* (p0, limit m+n) (p1, limit m+n) */

/* each partition scans limit m+n rows of data, */

/* and TSC operator selects the offset (m) limit (n) rows in the final result */

need_final_limit_ = true;

tsc_rtdef_.prepare_multi_part_limit_param();

}

return ret;

}

int ObTableScanOp::prepare_das_task()

{

int ret = OB_SUCCESS;

ObTaskExecutorCtx &task_exec_ctx = ctx_.get_task_exec_ctx();

if (OB_LIKELY(!MY_SPEC.use_dist_das_)) {

if (OB_FAIL(create_one_das_task(MY_INPUT.tablet_loc_))) {

LOG_WARN("create one das task failed", K(ret));

}

} else if (OB_LIKELY(nullptr == MY_CTDEF.das_dppr_tbl_)) {

ObDASTableLoc *table_loc = tsc_rtdef_.scan_rtdef_.table_loc_;

for (DASTabletLocListIter node = table_loc->tablet_locs_begin();

OB_SUCC(ret) && node != table_loc->tablet_locs_end(); ++node) {

ObDASTabletLoc *tablet_loc = *node;

if (OB_FAIL(create_one_das_task(tablet_loc))) {

LOG_WARN("create one das task failed", K(ret));

}

}

} else {

// dynamic partitions

ObPhysicalPlanCtx *plan_ctx = ctx_.get_physical_plan_ctx();

ObDataTypeCastParams dtc_params = ObBasicSessionInfo::create_dtc_params(ctx_.get_my_session());

const ObTableLocation &das_location = *MY_CTDEF.das_dppr_tbl_;

ObSEArray<ObTabletID, 1> tablet_ids;

ObSEArray<ObObjectID, 1> partition_ids;

ObSEArray<ObObjectID, 1> first_level_part_ids;

if (OB_FAIL(das_location.calculate_tablet_ids(ctx_,

plan_ctx->get_param_store(),

tablet_ids,

partition_ids,

first_level_part_ids,

dtc_params))) {

LOG_WARN("calculate dynamic partitions failed", K(ret));

} else {

LOG_TRACE("dynamic partitions", K(tablet_ids), K(partition_ids), K(first_level_part_ids));

}

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

ObDASTabletLoc *tablet_loc = nullptr;

if (OB_FAIL(DAS_CTX(ctx_).extended_tablet_loc(*tsc_rtdef_.scan_rtdef_.table_loc_,

tablet_ids.at(i),

tablet_loc))) {

LOG_WARN("extended tablet loc failed", K(ret));

} else if (OB_FAIL(create_one_das_task(tablet_loc))) {

LOG_WARN("create one das task failed", K(ret));

}

}

}

return ret;

}

int ObTableScanOp::prepare_all_das_tasks()

{

int ret = OB_SUCCESS;

if (MY_SPEC.batch_scan_flag_) {

if (OB_SUCC(ret)) {

if (!tsc_rtdef_.bnlj_params_.empty()) {

tsc_rtdef_.group_size_ = tsc_rtdef_.bnlj_params_.at(0).gr_param_->count_;

if (OB_UNLIKELY(tsc_rtdef_.group_size_ > tsc_rtdef_.max_group_size_)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("The amount of data exceeds the pre allocated memory", K(ret));

}

} else {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("batch nlj params is empty", K(ret));

}

}

}

if (OB_SUCC(ret)) {

if (MY_SPEC.gi_above_ && !MY_INPUT.key_ranges_.empty()) {

if (OB_FAIL(prepare_das_task())) {

LOG_WARN("prepare das task failed", K(ret));

}

} else {

int64_t group_size = MY_SPEC.batch_scan_flag_ ? tsc_rtdef_.group_size_ : 1;

GroupRescanParamGuard grp_guard(tsc_rtdef_, GET_PHY_PLAN_CTX(ctx_)->get_param_store_for_update());

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

grp_guard.switch_group_rescan_param(i);

if (OB_FAIL(prepare_single_scan_range(i))) {

LOG_WARN("prepare single scan range failed", K(ret));

} else if (OB_FAIL(prepare_das_task())) {

LOG_WARN("prepare das task failed", K(ret));

} else {

MY_INPUT.key_ranges_.reuse();

MY_INPUT.ss_key_ranges_.reuse();

}

}

}

}

if (OB_SUCC(ret)) {

if (OB_FAIL(init_das_group_range(0, tsc_rtdef_.group_size_))) {

LOG_WARN("set group range failed", K(ret), K_(tsc_rtdef_.group_size));

}

}

return ret;

}

int ObTableScanOp::init_table_scan_rtdef()

{

int ret = OB_SUCCESS;

ObPhysicalPlanCtx *plan_ctx = GET_PHY_PLAN_CTX(ctx_);

ObSQLSessionInfo *my_session = GET_MY_SESSION(ctx_);

ObDASTaskFactory &das_factory = DAS_CTX(ctx_).get_das_factory();

ObMemAttr mem_attr;

mem_attr.tenant_id_ = my_session->get_effective_tenant_id();

mem_attr.label_ = "ScanDASCtx";

das_ref_.set_mem_attr(mem_attr);

das_ref_.set_expr_frame_info(&MY_SPEC.plan_->get_expr_frame_info());

das_ref_.set_execute_directly(!MY_SPEC.use_dist_das_);

set_cache_stat(plan_ctx->get_phy_plan()->stat_);

bool is_null_value = false;

if (OB_SUCC(ret) && NULL != MY_SPEC.limit_) {

if (OB_FAIL(calc_expr_int_value(*MY_SPEC.limit_, limit_param_.limit_, is_null_value))) {

LOG_WARN("fail get val", K(ret));

} else if (limit_param_.limit_ < 0) {

limit_param_.limit_ = 0;

}

}

if (OB_SUCC(ret) && NULL != MY_SPEC.offset_ && !is_null_value) {

if (OB_FAIL(calc_expr_int_value(*MY_SPEC.offset_, limit_param_.offset_, is_null_value))) {

LOG_WARN("fail get val", K(ret));

} else if (limit_param_.offset_ < 0) {

limit_param_.offset_ = 0;

} else if (is_null_value) {

limit_param_.limit_ = 0;

}

}

if (OB_SUCC(ret)) {

const ObDASScanCtDef &scan_ctdef = MY_CTDEF.scan_ctdef_;

ObDASScanRtDef &scan_rtdef = tsc_rtdef_.scan_rtdef_;

const ObDASTableLocMeta *loc_meta = MY_CTDEF.das_dppr_tbl_ != nullptr ?

&MY_CTDEF.das_dppr_tbl_->get_loc_meta() : nullptr;

if (OB_FAIL(init_das_scan_rtdef(scan_ctdef, scan_rtdef, loc_meta))) {

LOG_WARN("init das scan rtdef failed", K(ret));

} else if (!MY_SPEC.use_dist_das_ && !MY_SPEC.gi_above_ && !scan_rtdef.table_loc_->empty()) {

MY_INPUT.tablet_loc_ = scan_rtdef.table_loc_->get_first_tablet_loc();

}

}

if (OB_SUCC(ret) && MY_CTDEF.lookup_ctdef_ != nullptr) {

const ObDASScanCtDef &lookup_ctdef = *MY_CTDEF.lookup_ctdef_;

ObDASBaseRtDef *das_rtdef = nullptr;

if (OB_FAIL(das_factory.create_das_rtdef(DAS_OP_TABLE_SCAN, das_rtdef))) {

LOG_WARN("create das rtdef failed", K(ret));

} else {

tsc_rtdef_.lookup_rtdef_ = static_cast<ObDASScanRtDef*>(das_rtdef);

if (OB_FAIL(init_das_scan_rtdef(lookup_ctdef, *tsc_rtdef_.lookup_rtdef_, MY_CTDEF.lookup_loc_meta_))) {

LOG_WARN("init das scan rtdef failed", K(ret), K(lookup_ctdef));

}

}

}

return ret;

}

OB_INLINE int ObTableScanOp::init_das_scan_rtdef(const ObDASScanCtDef &das_ctdef,

ObDASScanRtDef &das_rtdef,

const ObDASTableLocMeta *loc_meta)

{

int ret = OB_SUCCESS;

const ObTableScanCtDef &tsc_ctdef = MY_CTDEF;

bool is_lookup = (&das_ctdef == MY_CTDEF.lookup_ctdef_);

bool is_lookup_limit = MY_SPEC.is_index_back() &&

!MY_CTDEF.lookup_ctdef_->pd_expr_spec_.pushdown_filters_.empty();

ObPhysicalPlanCtx *plan_ctx = GET_PHY_PLAN_CTX(ctx_);

ObSQLSessionInfo *my_session = GET_MY_SESSION(ctx_);

ObTaskExecutorCtx &task_exec_ctx = ctx_.get_task_exec_ctx();

das_rtdef.timeout_ts_ = plan_ctx->get_ps_timeout_timestamp();

das_rtdef.tx_lock_timeout_ = my_session->get_trx_lock_timeout();

das_rtdef.scan_flag_ = MY_CTDEF.scan_flags_;

das_rtdef.scan_flag_.is_show_seed_ = plan_ctx->get_show_seed();

if(is_foreign_check_nested_session()) {

das_rtdef.is_for_foreign_check_ = true;

if (plan_ctx->get_phy_plan()->has_for_update() && ObSQLUtils::is_iter_uncommitted_row(&ctx_)) {

das_rtdef.scan_flag_.set_iter_uncommitted_row();

}

}

if (MY_SPEC.batch_scan_flag_) {

// if tsc enable batch rescan, the output order of tsc is determined by group id

if (das_rtdef.scan_flag_.scan_order_ == ObQueryFlag::Reverse) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("Scan order is not supported in batch rescan", K(ret), K(das_rtdef.scan_flag_.scan_order_));

} else {

das_rtdef.scan_flag_.scan_order_ = ObQueryFlag::KeepOrder;

}

}

if (is_lookup) {

das_rtdef.scan_flag_.scan_order_ = ObQueryFlag::KeepOrder;

}

das_rtdef.scan_flag_.is_lookup_for_4377_ = is_lookup;

das_rtdef.need_check_output_datum_ = MY_SPEC.need_check_output_datum_;

das_rtdef.sql_mode_ = my_session->get_sql_mode();

das_rtdef.stmt_allocator_.set_alloc(&das_ref_.get_das_alloc());

das_rtdef.scan_allocator_.set_alloc(&das_ref_.get_das_alloc());

das_rtdef.eval_ctx_ = &get_eval_ctx();

if ((is_lookup_limit && is_lookup) || (!is_lookup_limit && !is_lookup)) {

//when is_lookup_limit = true means that the limit param should pushdown to the lookup rtdef

//so is_lookup = true means that the das_rtdef is the lookup rtdef

//when is_lookup_limit = false means that the limit param should pushdown to the scan rtdef

//so is_lookup = false means that the das_rtdef is the scan rtdef

das_rtdef.limit_param_ = limit_param_;

}

das_rtdef.frozen_version_ = MY_SPEC.frozen_version_;

das_rtdef.force_refresh_lc_ = MY_SPEC.force_refresh_lc_;

if (OB_SUCC(ret)) {

if (OB_FAIL(das_rtdef.init_pd_op(ctx_, das_ctdef))) {

LOG_WARN("init pushdown storage filter failed", K(ret));

}

}

if (OB_SUCC(ret)) {

int64_t schema_version = task_exec_ctx.get_query_tenant_begin_schema_version();

das_rtdef.tenant_schema_version_ = schema_version;

}

if (OB_SUCC(ret)) {

if (OB_FAIL(tsc_ctdef.flashback_item_.set_flashback_query_info(eval_ctx_, das_rtdef))) {

LOG_WARN("failed to set flashback query snapshot version", K(ret));

} else if (MY_SPEC.ref_table_id_ != das_ctdef.ref_table_id_) {

//only data table scan need to set row scn flag

das_rtdef.need_scn_ = false;

}

}

if (OB_SUCC(ret)) {

ObTableID table_loc_id = MY_SPEC.get_table_loc_id();

das_rtdef.table_loc_ = DAS_CTX(ctx_).get_table_loc_by_id(table_loc_id, das_ctdef.ref_table_id_);

if (OB_ISNULL(das_rtdef.table_loc_)) {

if (OB_ISNULL(loc_meta)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("get table loc by id failed", K(ret), K(table_loc_id), K(das_ctdef.ref_table_id_),

K(DAS_CTX(ctx_).get_table_loc_list()));

} else if (OB_FAIL(DAS_CTX(ctx_).extended_table_loc(*loc_meta, das_rtdef.table_loc_))) {

LOG_WARN("extended table location failed", K(ret), KPC(loc_meta));

}

}

}

return ret;

}

int ObTableScanOp::update_output_tablet_id()

{

int ret = OB_SUCCESS;

if (NULL != MY_SPEC.pdml_partition_id_) {

const ObDASTabletLoc *data_tablet_loc = nullptr;

int64_t output_id = OB_INVALID_ID;

if (MY_SPEC.partition_id_calc_type_ > 0) {

// partition id for gather statistics, index scan should output index partition id

data_tablet_loc = scan_result_.get_tablet_loc();

} else if (MY_SPEC.should_scan_index()) {

data_tablet_loc = ObDASUtils::get_related_tablet_loc(*scan_result_.get_tablet_loc(), MY_SPEC.ref_table_id_);

} else {

data_tablet_loc = scan_result_.get_tablet_loc();

}

if (OB_ISNULL(data_tablet_loc)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("data tablet loc is null, value of pdml partition id will not be set", K(ret),

K(MY_SPEC.should_scan_index()), K(MY_SPEC.ref_table_id_));

} else {

if (MY_SPEC.partition_id_calc_type_ == 0) {

output_id = data_tablet_loc->tablet_id_.id();

} else if (MY_SPEC.partition_id_calc_type_ == 1) {

output_id = data_tablet_loc->first_level_part_id_ != OB_INVALID_ID ?

data_tablet_loc->first_level_part_id_ : data_tablet_loc->partition_id_;

} else if (MY_SPEC.partition_id_calc_type_ == 2) {

output_id = data_tablet_loc->partition_id_;

} else {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("get invalid partition id cacl type", K(ret));

}

if (OB_FAIL(ret)) {

} else if (is_vectorized()) {

const int64_t batch_size = MY_SPEC.max_batch_size_;

ObExpr *expr = MY_SPEC.pdml_partition_id_;

ObDatum *datums = expr->locate_datums_for_update(eval_ctx_, batch_size);

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

datums[i].set_int(output_id);

}

expr->set_evaluated_projected(eval_ctx_);

LOG_TRACE("find the partition id expr in pdml table scan", K(ret), KPC(expr), KPC(data_tablet_loc), K(output_id));

} else {