/** * 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/basic/ob_function_table_op.h" #include "share/object/ob_obj_cast.h" #include "common/sql_mode/ob_sql_mode_utils.h" #include "sql/ob_sql_utils.h" #include "sql/session/ob_sql_session_info.h" #include "sql/engine/ob_physical_plan.h" #include "sql/engine/ob_exec_context.h" #include "pl/ob_pl_user_type.h" #include "sql/engine/expr/ob_expr.h" #include "sql/engine/expr/ob_expr_lob_utils.h" namespace oceanbase { using namespace common; namespace sql { OB_SERIALIZE_MEMBER((ObFunctionTableSpec, ObOpSpec), value_expr_, column_exprs_, has_correlated_expr_); int ObFunctionTableOp::inner_open() { int ret = OB_SUCCESS; node_idx_ = 0; already_calc_ = false; if (OB_ISNULL(MY_SPEC.value_expr_)) { ret = OB_ERR_UNEXPECTED; LOG_WARN("value expr is not init", K(ret)); } else if (ObExtendType == MY_SPEC.value_expr_->datum_meta_.type_) { next_row_func_ = &ObFunctionTableOp::inner_get_next_row_udf; } else { next_row_func_ = &ObFunctionTableOp::inner_get_next_row_sys_func; } return ret; } int ObFunctionTableOp::inner_rescan() { int ret = OB_SUCCESS; if (OB_FAIL(ObOperator::inner_rescan())) { LOG_WARN("failed to inner rescan", K(ret)); } else { node_idx_ = 0; if (MY_SPEC.has_correlated_expr_) { row_count_ = 0; col_count_ = 0; value_table_ = NULL; already_calc_ = false; } } return ret; } int ObFunctionTableOp::inner_close() { int ret = OB_SUCCESS; node_idx_ = 0; row_count_ = 0; col_count_ = 0; value_table_ = NULL; return ret; } //ObFunctionTableOp has its own switch_iterator int ObFunctionTableOp::switch_iterator() { int ret = OB_SUCCESS; if (OB_FAIL(ObOperator::inner_switch_iterator())) { LOG_WARN("failed to switch iterator", K(ret)); } else if (OB_ISNULL(ctx_.get_my_session())) { ret = OB_ERR_UNEXPECTED; LOG_WARN("failed to get session", K(ret)); } else if (NULL == ctx_.get_my_session()->get_pl_implicit_cursor() || !ctx_.get_my_session()->get_pl_implicit_cursor()->get_in_forall()) { ret = OB_ITER_END; } else { node_idx_ = 0; } return ret; } void ObFunctionTableOp::destroy() { ObOperator::destroy(); } int ObFunctionTableOp::get_current_result(ObObj &result) { int ret = OB_SUCCESS; void *data = NULL; CK (already_calc_); if (node_idx_ < 0 || node_idx_ >= row_count_ * col_count_) { ret = OB_ERR_UNEXPECTED; LOG_WARN("fail to get current result in table function", K(node_idx_), K(row_count_), K(col_count_)); } do { CK (node_idx_ >= 0); if (OB_SUCC(ret) && node_idx_ >= row_count_) { ret = OB_ITER_END; } CK (OB_NOT_NULL(value_table_)); OX (data = value_table_->get_data()); CK (OB_NOT_NULL(data)); OX (result = (static_cast(data))[node_idx_++]); } while (OB_SUCC(ret) && result.get_meta().get_type() == ObMaxType); return ret; } int ObFunctionTableOp::inner_get_next_row() { return (this->*next_row_func_)(); } int ObFunctionTableOp::inner_get_next_row_udf() { int ret = OB_SUCCESS; ObPhysicalPlanCtx *plan_ctx = nullptr; clear_evaluated_flag(); if (OB_ISNULL(plan_ctx = ctx_.get_physical_plan_ctx())) { ret = OB_ERR_UNEXPECTED; LOG_WARN("failed to get plan ctx", K(ret), K(plan_ctx)); } else if (OB_ISNULL(MY_SPEC.value_expr_)) { ret = OB_ERR_UNEXPECTED; LOG_WARN("value expr is not init", K(ret)); } else if (ObExtendType != MY_SPEC.value_expr_->datum_meta_.type_) { ret = OB_ERR_UNEXPECTED; LOG_WARN("get unexpected value", K(ret), K(MY_SPEC.value_expr_->datum_meta_.type_)); } else if (FALSE_IT(plan_ctx->set_autoinc_id_tmp(0))) { } else if (OB_FAIL(ctx_.check_status())) { LOG_WARN("failed to check status ", K(ret)); } else { ObDatum *value = nullptr; if (!already_calc_) { if (OB_FAIL(MY_SPEC.value_expr_->eval(eval_ctx_, value))) { LOG_WARN("failed to eval value expr", K(ret)); } else if (value->is_null()) { //do nothing } else if (OB_ISNULL(value_table_ = reinterpret_cast(value->get_ext()))) { ret = OB_ERR_UNEXPECTED; LOG_WARN("failed to get value table", K(ret)); } else { row_count_ = value_table_->is_inited() ? value_table_->get_count() : 0; col_count_ = value_table_->get_column_count(); } already_calc_ = true; } if (OB_SUCC(ret) && OB_UNLIKELY(node_idx_ >= row_count_)) { ret = OB_ITER_END; } CK (MY_SPEC.column_exprs_.count() >= col_count_); ObObj obj_stack[col_count_]; if (OB_SUCC(ret)) { if (nullptr != value_table_ && ObExtendType == value_table_->get_element_type().get_obj_type()) { pl::ObPLComposite *composite = NULL; pl::ObPLRecord *record = NULL; ObObj record_obj; OZ (get_current_result(record_obj)); if (OB_SUCC(ret) && record_obj.is_pl_extend()) { CK (OB_NOT_NULL(composite = reinterpret_cast(record_obj.get_ext()))); CK (composite->is_record()); OX (record = static_cast(composite)); CK (record->get_count() == col_count_); for (int64_t i = 0; OB_SUCC(ret) && i < col_count_; ++i) { OZ (record->get_element(i, obj_stack[i])); } } else { ret = OB_ERR_UNEXPECTED; LOG_ERROR("unexpected here", K(ret), K(record_obj), K(record_obj.meta_)); } } else { for (int64_t i = 0; OB_SUCC(ret) && i < col_count_; ++i) { if (OB_FAIL(get_current_result(obj_stack[i]))) { LOG_WARN("failed to get current result", K(ret), K(i)); } } } } if (OB_SUCC(ret)) { for (int64_t i = 0; OB_SUCC(ret) && i < col_count_; ++i) { if (obj_stack[i].is_null()) { MY_SPEC.column_exprs_.at(i)->locate_datum_for_write(eval_ctx_).set_null(); } else { const ObObjDatumMapType &datum_map = MY_SPEC.column_exprs_.at(i)->obj_datum_map_; ObExpr * const &expr = MY_SPEC.column_exprs_.at(i); ObDatum &datum = expr->locate_datum_for_write(eval_ctx_); if (OB_FAIL(datum.from_obj(obj_stack[i], datum_map))) { LOG_WARN("failed to convert datum", K(ret)); } else if (is_lob_storage(obj_stack[i].get_type()) && OB_FAIL(ob_adjust_lob_datum(obj_stack[i], expr->obj_meta_, datum_map, get_exec_ctx().get_allocator(), datum))) { LOG_WARN("adjust lob datum failed", K(ret), K(obj_stack[i].get_meta()), K(expr->obj_meta_)); } } if (OB_SUCC(ret)) { MY_SPEC.column_exprs_.at(i)->set_evaluated_projected(eval_ctx_); } } } } return ret; } int ObFunctionTableOp::inner_get_next_row_sys_func() { int ret = OB_SUCCESS; ObPhysicalPlanCtx *plan_ctx = nullptr; ObDatum *value = nullptr; clear_evaluated_flag(); if (OB_ISNULL(plan_ctx = ctx_.get_physical_plan_ctx())) { ret = OB_ERR_UNEXPECTED; LOG_WARN("failed to get plan ctx", K(ret), K(plan_ctx)); } else if (OB_FAIL(ctx_.check_status())) { LOG_WARN("failed to check status ", K(ret)); } else if (OB_FAIL(MY_SPEC.value_expr_->eval(eval_ctx_, value))) { if (OB_ITER_END != ret) { LOG_WARN("failed to eval value expr", K(ret)); } } else { MY_SPEC.column_exprs_.at(0)->locate_datum_for_write(eval_ctx_).set_datum(*value); MY_SPEC.column_exprs_.at(0)->set_evaluated_projected(eval_ctx_); } return ret; } } // end namespace sql } // end namespace oceanbase