hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix osg bugs

Browse Source
This commit is contained in:
Larry955
2023-05-06 10:38:37 +00:00
committed by ob-robot
parent fab24a1afb
commit 03695f6cb2
13 changed files with 231 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files

333
src/sql/optimizer/ob_insert_log_plan.cpp
View File

@ -40,10 +40,11 @@ int ObInsertLogPlan::generate_normal_raw_plan()
{
int ret = OB_SUCCESS;
const ObInsertStmt *insert_stmt = get_stmt();
if (OB_ISNULL(insert_stmt)) {
if (OB_ISNULL(insert_stmt) || OB_ISNULL(get_optimizer_context().get_query_ctx())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get unexpected error");
LOG_WARN("get unexpected error", K(insert_stmt), K(ret));
} else {
LOG_TRACE("start to allocate operators for ", "sql", get_optimizer_context().get_query_ctx()->get_sql_stmt());
OPT_TRACE("generate plan for ", get_stmt());
if (!insert_stmt->value_from_select()) {
// insert into values xxxx
@ -66,24 +67,6 @@ int ObInsertLogPlan::generate_normal_raw_plan()
} else { /*do nothing*/}
}
// allocal optimizer stat gather operator.
bool need_osg = false;
OSGShareInfo osg_info;
if (OB_SUCC(ret)) {
if (OB_FAIL(compute_dml_parallel())) { // compute parallel before check allocate stats gather
LOG_WARN("failed to compute dml parallel", K(ret));
} else if (use_pdml() && OB_FAIL(set_is_direct_insert())) {
LOG_WARN("failed to set is direct insert", K(ret));
} else if (OB_FAIL(check_need_online_stats_gather(need_osg))) { // allocal optimizer stat gather operator.
LOG_WARN("fail to check wether we need optimizer stats gathering operator", K(ret));
} else if (need_osg) {
if (OB_FAIL(generate_osg_share_info(osg_info))) {
LOG_WARN("fail to generated share info", K(ret));
} else if (OB_FAIL(candi_allocate_optimizer_stats_gathering(osg_info))) {
LOG_WARN("failed to allocate optimizer stats gathering", K(ret));
}
}
}
// allocate subplan filter for "INSERT .. ON DUPLICATE KEY UPDATE c1 = (select...)"
if (OB_SUCC(ret) && insert_stmt->is_insert_up()) {
ObSEArray<ObRawExpr*, 4> subquery;
@ -96,17 +79,36 @@ int ObInsertLogPlan::generate_normal_raw_plan()
LOG_WARN("failed to allocate subplan", K(ret));
} else { /*do nothing*/ }
}
bool need_osg = false;
OSGShareInfo *osg_info = NULL;
if (OB_SUCC(ret)) {
// compute parallel before check allocate stats gather
if (OB_FAIL(compute_dml_parallel())) {
LOG_WARN("failed to compute dml parallel", K(ret));
} else if (use_pdml() && OB_FAIL(set_is_direct_insert())) {
LOG_WARN("failed to set is direct insert", K(ret));
} else if (OB_FAIL(check_need_online_stats_gather(need_osg))) {
LOG_WARN("fail to check wether we need optimizer stats gathering operator", K(ret));
} else if (need_osg && OB_FAIL(generate_osg_share_info(osg_info))) {
LOG_WARN("failed to generate osg share info");
} else if (need_osg && OB_ISNULL(osg_info)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get unexpected null");
}
}
if (OB_SUCC(ret)) {
if (OB_FAIL(prepare_dml_infos())) {
LOG_WARN("failed to prepare dml infos", K(ret));
} else if (use_pdml()) {
if (OB_FAIL(candi_allocate_pdml_insert())) {
if (OB_FAIL(candi_allocate_pdml_insert(osg_info))) {
LOG_WARN("failed to allocate pdml insert", K(ret));
} else {
LOG_TRACE("succeed to allocate pdml insert operator",
K(candidates_.candidate_plans_.count()));
}
} else if (OB_FAIL(candi_allocate_insert())) {
} else if (OB_FAIL(candi_allocate_insert(osg_info))) {
LOG_WARN("failed to allocate insert operator", K(ret));
} else {
LOG_TRACE("succeed to allocate insert operator", K(candidates_.candidate_plans_.count()));
@ -125,10 +127,10 @@ int ObInsertLogPlan::generate_normal_raw_plan()
* all information collection by other OSG.
*/
if (OB_SUCC(ret) && need_osg) {
if (OB_FAIL(candi_allocate_root_optimizer_stats_gathering(osg_info))) {
if (OB_FAIL(candi_allocate_optimizer_stats_merge(osg_info))) {
LOG_WARN("fail to allcate osg on top", K(ret));
} else {
LOG_TRACE("succeed to allocate optimizer stat gather",
LOG_TRACE("succeed to allocate optimizer stat merge",
K(candidates_.candidate_plans_.count()));
}
}
@ -153,6 +155,95 @@ int ObInsertLogPlan::generate_normal_raw_plan()
return ret;
}
int ObInsertLogPlan::generate_osg_share_info(OSGShareInfo *&info)
{
int ret = OB_SUCCESS;
const ObInsertStmt *stmt = NULL;
const ObTableSchema *tab_schema = NULL;
ObSqlSchemaGuard *schema_guard = NULL;
if (OB_ISNULL(schema_guard = get_optimizer_context().get_sql_schema_guard()) ||
OB_UNLIKELY(!get_stmt()->is_insert_stmt())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get unexpected null", K(ret));
} else if (OB_ISNULL(info = OB_NEWx(OSGShareInfo, (&get_allocator())))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("failed to allocate memory");
} else {
stmt = static_cast<const ObInsertStmt *>(get_stmt());
uint64_t table_id = stmt->get_insert_table_info().table_id_;
uint64_t ref_table_id = stmt->get_insert_table_info().ref_table_id_;
if (OB_FAIL(schema_guard->get_table_schema(table_id, ref_table_id, stmt, tab_schema))) {
LOG_WARN("fail to get table schema", K(ref_table_id), K(tab_schema), K(ret));
} else if (OB_ISNULL(tab_schema)) {
ret = OB_TABLE_NOT_EXIST;
LOG_WARN("get unexpected null pointer", K(ret));
} else {
const ObColumnSchemaV2 *col_schema = NULL;
const ObInsertTableInfo& table_info = stmt->get_insert_table_info();
ObSEArray<uint64_t, 4> generated_column_ids;
info->table_id_ = ref_table_id;
if (tab_schema->is_partitioned_table()) {
info->part_level_ = tab_schema->get_part_level();
if (OB_FAIL(gen_calc_part_id_expr(table_id,
ref_table_id,
CALC_PARTITION_TABLET_ID,
info->calc_part_id_expr_))) {
LOG_WARN("failed to init calc part id", K(ret));
} else if (OB_FAIL(ObOptimizerUtil::replace_column_with_select_for_partid(stmt,
get_optimizer_context(),
info->calc_part_id_expr_))) {
// using the select column/values expr to calc partid.
LOG_WARN("fail to replace column item with select item", K(ret));
} else {
LOG_TRACE("success to generate calc_part expr", K(ret), K(info->calc_part_id_expr_));
}
}
// column conv;
for (int64_t i = 0; OB_SUCC(ret) && i < table_info.column_exprs_.count(); i++) {
ObColumnRefRawExpr *tmp_col = table_info.column_exprs_.at(i);
ObRawExpr *col_conv_expr = table_info.column_conv_exprs_.at(i);
if (OB_ISNULL(tmp_col)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get unexpected null pointer", K(ret));
} else if (OB_FAIL(ObTransformUtils::get_base_column(stmt, tmp_col))) {
LOG_WARN("fail to get base column", K(ret));
} else {
// since the column_exprs may be a generated_table's column. e.g., insert into (select c2, c1 from t1) values (1,1);
// for t1, the column_ids of c1 and c2 are 16 and 17. However, in the insert stmt, the column c2 in subquery is 16.
// we need to get the real column id.
col_schema = tab_schema->get_column_schema(tmp_col->get_column_id());
if (OB_ISNULL(col_schema)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("can't get column schema", K(ret));
} else if (!pl::ObDbmsStats::check_column_validity(*tab_schema, *col_schema)) {
// do not gather stats for auto inc column.
// continue, shouldn't add these to osg's output.
} else if (!col_schema->is_generated_column()) {
if (OB_FAIL(info->col_conv_exprs_.push_back(col_conv_expr))) {
LOG_WARN("fail to push back column convert expr", K(ret));
} else if (OB_FAIL(info->column_ids_.push_back(tmp_col->get_column_id()))) {
LOG_WARN("fail to push back column ids", K(ret));
}
} else {
// generated column: no need to replace column expr with select_item in select clause. since this work has already done.
if (OB_FAIL(info->generated_column_exprs_.push_back(col_conv_expr))) {
LOG_WARN("fail to add generated column expr", K(ret));
} else if (OB_FAIL(generated_column_ids.push_back(tmp_col->get_column_id()))) {
LOG_WARN("fail to push back column ids", K(ret));
}
}
}
} // end for
if (OB_SUCC(ret)) {
// column ids of generated column should be add at the tail.
if (OB_FAIL(append(info->column_ids_, generated_column_ids))) {
LOG_WARN("fail to append column ids", K(ret));
}
}
}
}
return ret;
}
// Direct-insert is enabled only:
// 1. pdml insert
@ -188,10 +279,10 @@ int ObInsertLogPlan::check_need_online_stats_gather(bool &need_osg)
bool online_sys_var = false;
bool need_gathering = true;
ObObj online_sys_var_obj;
const ObInsertStmt *insert_stmt = get_stmt();
// check null pointer in the upper layer. If use this function in other place, need to add null check.
const ObInsertStmt *insert_stmt = NULL;
TableItem *ins_table = NULL;
if (OB_ISNULL(ins_table = insert_stmt->get_table_item_by_id(insert_stmt->get_insert_table_info().table_id_))) {
if (OB_ISNULL(insert_stmt = get_stmt()) ||
OB_ISNULL(ins_table = insert_stmt->get_table_item_by_id(insert_stmt->get_insert_table_info().table_id_))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get unexpected null pointer", K(ret), K(insert_stmt->get_insert_table_info()));
} else if (OB_UNLIKELY(ins_table->is_system_table_ || ins_table->is_index_table_)
@ -254,7 +345,7 @@ int ObInsertLogPlan::allocate_insert_values_as_top(ObLogicalOperator *&top)
return ret;
}
int ObInsertLogPlan::candi_allocate_insert()
int ObInsertLogPlan::candi_allocate_insert(OSGShareInfo *osg_info)
{
int ret = OB_SUCCESS;
ObConstRawExpr *lock_row_flag_expr = NULL;
@ -281,7 +372,8 @@ int ObInsertLogPlan::candi_allocate_insert()
lock_row_flag_expr,
force_no_multi_part,
force_multi_part,
insert_plans))) {
insert_plans,
osg_info))) {
LOG_WARN("failed to create insert plans", K(ret));
} else if (!insert_plans.empty()) {
LOG_TRACE("succeed to create insert plan using hint", K(insert_plans.count()));
@ -291,7 +383,8 @@ int ObInsertLogPlan::candi_allocate_insert()
insert_sharding,
lock_row_flag_expr,
false, false,
insert_plans))) {
insert_plans,
osg_info))) {
LOG_WARN("failed to create insert plans", K(ret));
} else {
LOG_TRACE("succeed to create insert plan ignore hint", K(insert_plans.count()));
@ -334,7 +427,8 @@ int ObInsertLogPlan::create_insert_plans(ObIArray<CandidatePlan> &candi_plans,
ObConstRawExpr *lock_row_flag_expr,
const bool force_no_multi_part,
const bool force_multi_part,
ObIArray<CandidatePlan> &insert_plans)
ObIArray<CandidatePlan> &insert_plans,
OSGShareInfo *osg_info)
{
int ret = OB_SUCCESS;
ObExchangeInfo exch_info;
@ -354,6 +448,10 @@ int ObInsertLogPlan::create_insert_plans(ObIArray<CandidatePlan> &candi_plans,
LOG_WARN("failed to check need multi-partition dml", K(ret));
} else if (is_multi_part_dml && force_no_multi_part) {
/*do nothing*/
} else if (osg_info != NULL &&
OB_FAIL(allocate_optimizer_stats_gathering_as_top(candi_plan.plan_tree_,
*osg_info))) {
LOG_WARN("failed to allocate sequence as top", K(ret));
} else if (candi_plan.plan_tree_->is_sharding() &&
(is_multi_part_dml || insert_sharding->is_local()) &&
OB_FAIL(allocate_exchange_as_top(candi_plan.plan_tree_, exch_info))) {
@ -429,7 +527,7 @@ int ObInsertLogPlan::allocate_insert_as_top(ObLogicalOperator *&top,
return ret;
}
int ObInsertLogPlan::candi_allocate_pdml_insert()
int ObInsertLogPlan::candi_allocate_pdml_insert(OSGShareInfo *osg_info)
{
int ret = OB_SUCCESS;
int64_t gidx_cnt = index_dml_infos_.count();
@ -439,7 +537,8 @@ int ObInsertLogPlan::candi_allocate_pdml_insert()
if (OB_FAIL(candi_allocate_one_pdml_insert(i > 0,
i == gidx_cnt - 1,
is_pdml_update_split,
index_dml_infos_.at(i)))) {
index_dml_infos_.at(i),
i == 0 ? osg_info : NULL))) {
LOG_WARN("failed to allocate one pdml insert", K(ret), K(i), K(index_dml_infos_));
} else {
LOG_TRACE("succeed to allocate one pdml insert");
@ -1340,47 +1439,17 @@ int ObInsertLogPlan::build_column_conv_for_shadow_pk(const ObInsertTableInfo& ta
return ret;
}
int ObInsertLogPlan::candi_allocate_optimizer_stats_gathering(const OSGShareInfo &osg_info)
{
int ret = OB_SUCCESS;
ObExchangeInfo exch_info;
CandidatePlan best_plan;
ObSEArray<CandidatePlan, 4> stats_gathering_plan;
ObSEArray<CandidatePlan, 8> best_candidates;
if (OB_FAIL(get_minimal_cost_candidates(candidates_.candidate_plans_, best_candidates))) {
LOG_WARN("failed to get minimal cost candidates", K(ret));
} else {
for (int64_t i = 0; OB_SUCC(ret) && i < best_candidates.count(); ++i) {
best_plan = best_candidates.at(i);
if (OB_ISNULL(best_plan.plan_tree_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get unexpected null", K(ret));
} else {
OSG_TYPE type = (!best_plan.plan_tree_->is_sharding()) ? OSG_TYPE::NORMAL_OSG : OSG_TYPE::GATHER_OSG;
if (OB_FAIL(allocate_optimizer_stats_gathering_as_top(best_plan.plan_tree_,
type,
osg_info))) {
LOG_WARN("failed to allocate sequence as top", K(ret));
} else if (OB_FAIL(stats_gathering_plan.push_back(best_plan))) {
LOG_WARN("failed to push back candidate plan", K(ret));
}
}
}
}
if (OB_SUCC(ret) && OB_FAIL(prune_and_keep_best_plans(stats_gathering_plan))) {
LOG_WARN("failed to prune and keep best plans", K(ret));
}
return ret;
}
int ObInsertLogPlan::candi_allocate_root_optimizer_stats_gathering(const OSGShareInfo &osg_info)
int ObInsertLogPlan::candi_allocate_optimizer_stats_merge(OSGShareInfo *osg_info)
{
int ret = OB_SUCCESS;
ObExchangeInfo exch_info;
CandidatePlan candidate_plan;
ObSEArray<CandidatePlan, 4> stats_gathering_plan;
ObSEArray<CandidatePlan, 8> best_candidates;
if (OB_FAIL(get_minimal_cost_candidates(candidates_.candidate_plans_, best_candidates))) {
if (OB_ISNULL(osg_info)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get unexpected null");
} else if (OB_FAIL(get_minimal_cost_candidates(candidates_.candidate_plans_, best_candidates))) {
LOG_WARN("failed to get minimal cost candidates", K(ret));
} else {
for (int64_t i = 0; OB_SUCC(ret) && i < best_candidates.count(); i++) {
@ -1388,15 +1457,13 @@ int ObInsertLogPlan::candi_allocate_root_optimizer_stats_gathering(const OSGShar
if (OB_ISNULL(best_candidates.at(i).plan_tree_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get unexpected null", K(ret));
} else if (best_candidates.at(i).plan_tree_->has_allocated_osg()) {
} else if (best_candidates.at(i).plan_tree_->need_osg_merge()) {
if (best_candidates.at(i).plan_tree_->is_sharding() &&
best_candidates.at(i).plan_tree_->get_phy_plan_type() != ObPhyPlanType::OB_PHY_PLAN_REMOTE &&
OB_FAIL(allocate_exchange_as_top(best_candidates.at(i).plan_tree_, exch_info))) {
LOG_WARN("failed to allocate exchange as top", K(ret));
} else if (OB_FAIL(allocate_optimizer_stats_gathering_as_top(
best_candidates.at(i).plan_tree_,
OSG_TYPE::MERGE_OSG,
osg_info))) {
*osg_info))) {
LOG_WARN("failed to allocate sequence as top", K(ret));
}
}
@ -1410,125 +1477,3 @@ int ObInsertLogPlan::candi_allocate_root_optimizer_stats_gathering(const OSGShar
}
return ret;
}
int ObInsertLogPlan::allocate_optimizer_stats_gathering_as_top(ObLogicalOperator *&old_top,
OSG_TYPE type,
const OSGShareInfo &info)
{
int ret = OB_SUCCESS;
ObLogOptimizerStatsGathering *osg = NULL;
if (OB_ISNULL(old_top)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("Get unexpected null", K(ret), K(old_top));
} else if (OB_ISNULL(osg = static_cast<ObLogOptimizerStatsGathering *>(get_log_op_factory().
allocate(*this, log_op_def::LOG_OPTIMIZER_STATS_GATHERING)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("failed to allocate sequence operator", K(ret));
} else {
osg->set_child(ObLogicalOperator::first_child, old_top);
osg->set_osg_type(type);
osg->set_table_id(info.table_id_);
osg->set_part_level(info.part_level_);
osg->set_generated_column_exprs(info.generated_column_exprs_);
osg->set_col_conv_exprs(info.col_conv_exprs_);
osg->set_column_ids(info.column_ids_);
if (type == OSG_TYPE::GATHER_OSG) {
osg->set_allocated_osg(true);
}
if (type != OSG_TYPE::MERGE_OSG) {
osg->set_calc_part_id_expr(info.calc_part_id_expr_);
}
if (OB_FAIL(osg->compute_property())) {
LOG_WARN("failed to compute property", K(ret));
} else {
old_top = osg;
}
}
return ret;
}
int ObInsertLogPlan::generate_osg_share_info(OSGShareInfo &info)
{
int ret = OB_SUCCESS;
const ObInsertStmt *stmt = NULL;
const ObTableSchema *tab_schema = NULL;
ObSqlSchemaGuard *schema_guard = NULL;
if (OB_ISNULL(schema_guard = get_optimizer_context().get_sql_schema_guard())
|| OB_ISNULL(stmt = get_stmt())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get unexpected null", K(ret));
} else {
uint64_t table_id = stmt->get_insert_table_info().table_id_;
uint64_t ref_table_id = stmt->get_insert_table_info().ref_table_id_;
if (OB_FAIL(schema_guard->get_table_schema(table_id, ref_table_id, stmt, tab_schema))) {
LOG_WARN("fail to get table schema", K(ref_table_id), K(tab_schema), K(ret));
} else if (OB_ISNULL(tab_schema)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get unexpected null pointer", K(ret));
} else {
const ObColumnSchemaV2 *col_schema = NULL;
const ObInsertTableInfo& table_info = stmt->get_insert_table_info();
ObSEArray<uint64_t, 4> generated_column_ids;
info.table_id_ = ref_table_id;
if (tab_schema->is_partitioned_table()) {
info.part_level_ = tab_schema->get_part_level();
if (OB_FAIL(gen_calc_part_id_expr(table_id,
ref_table_id,
CALC_PARTITION_TABLET_ID,
info.calc_part_id_expr_))) {
LOG_WARN("failed to gen calc part id expr");
} else if (OB_FAIL(ObOptimizerUtil::replace_column_with_select_for_partid(stmt,
get_optimizer_context(),
info.calc_part_id_expr_))) {
// using the select column/values expr to calc partid.
LOG_WARN("fail to replace column item with select item", K(ret));
} else {
LOG_TRACE("success to generate calc_part expr", K(ret), K(info.calc_part_id_expr_));
}
}
// column conv;
for (int64_t i = 0; OB_SUCC(ret) && i < table_info.column_exprs_.count(); i++) {
ObColumnRefRawExpr *tmp_col = table_info.column_exprs_.at(i);
ObRawExpr *col_conv_expr = table_info.column_conv_exprs_.at(i);
if (OB_ISNULL(tmp_col)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get unexpected null pointer", K(ret));
} else if (OB_FAIL(ObTransformUtils::get_base_column(stmt, tmp_col))) {
LOG_WARN("fail to get base column", K(ret));
} else {
// since the column_exprs may be a generated_table's column. e.g., insert into (select c2, c1 from t1) values (1,1);
// for t1, the column_ids of c1 and c2 are 16 and 17. However, in the insert stmt, the column c2 in subquery is 16.
// we need to get the real column id.
col_schema = tab_schema->get_column_schema(tmp_col->get_column_id());
if (OB_ISNULL(col_schema)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("can't get column schema", K(ret));
} else if (!pl::ObDbmsStats::check_column_validity(*tab_schema, *col_schema)) {
// do not gather stats for auto inc column.
// continue, shouldn't add these to osg's output.
} else if (!col_schema->is_generated_column()) {
if (OB_FAIL(info.col_conv_exprs_.push_back(col_conv_expr))) {
LOG_WARN("fail to push back column convert expr", K(ret));
} else if (OB_FAIL(info.column_ids_.push_back(tmp_col->get_column_id()))) {
LOG_WARN("fail to push back column ids", K(ret));
}
} else {
// generated column: no need to replace column expr with select_item in select clause. since this work has already done.
if (OB_FAIL(info.generated_column_exprs_.push_back(col_conv_expr))) {
LOG_WARN("fail to add generated column expr", K(ret));
} else if (OB_FAIL(generated_column_ids.push_back(tmp_col->get_column_id()))) {
LOG_WARN("fail to push back column ids", K(ret));
}
}
}
} // end for
if (OB_SUCC(ret)) {
// column ids of generated column should be add at the tail.
if (OB_FAIL(append(info.column_ids_, generated_column_ids))) {
LOG_WARN("fail to append column ids", K(ret));
}
}
}
}
return ret;
}