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

fix sql_statistics

Browse Source
This commit is contained in:
yongshige
2023-07-13 08:47:59 +00:00
committed by ob-robot
parent 775b4c0b62
commit bac3530b36
13 changed files with 354 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/observer/table_load/control/ob_table_load_control_rpc_executor.cpp
View File

@ -259,7 +259,7 @@ int ObDirectLoadControlCommitExecutor::process()
ObTableLoadStore store(table_ctx);
if (OB_FAIL(store.init())) {
LOG_WARN("fail to init store", KR(ret));
} else if (OB_FAIL(store.commit(res_.result_info_, res_.sql_statistics_))) {
} else if (OB_FAIL(store.commit(res_.result_info_))) {
LOG_WARN("fail to store commit", KR(ret));
} else if (OB_FAIL(ObTableLoadService::remove_ctx(table_ctx))) {
LOG_WARN("fail to remove table ctx", KR(ret), K(key));

107
src/observer/table_load/ob_table_load_commit_processor.cpp Normal file
View File

@ -0,0 +1,107 @@
/**
* 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 SERVER
#include "observer/table_load/ob_table_load_commit_processor.h"
#include "observer/table_load/ob_table_load_coordinator.h"
#include "observer/table_load/ob_table_load_service.h"
#include "observer/table_load/ob_table_load_store.h"
#include "sql/engine/ob_exec_context.h"
namespace oceanbase
{
namespace observer
{
using namespace table;
using namespace sql;
/**
* ObTableLoadCommitP
*/
int ObTableLoadCommitP::process()
{
int ret = OB_SUCCESS;
if (OB_FAIL(check_user_access(arg_.credential_))) {
LOG_WARN("fail to check_user_access", KR(ret));
} else if (OB_FAIL(ObTableLoadService::check_tenant())) {
LOG_WARN("fail to check tenant", KR(ret));
} else {
ObTableLoadTableCtx *table_ctx = nullptr;
ObTableLoadUniqueKey key(arg_.table_id_, arg_.task_id_);
if (OB_FAIL(ObTableLoadService::get_ctx(key, table_ctx))) {
LOG_WARN("fail to get table ctx", KR(ret), K(key));
} else {
ObTableLoadCoordinator coordinator(table_ctx);
if (OB_FAIL(coordinator.init())) {
LOG_WARN("fail to init coordinator", KR(ret));
} else if (OB_FAIL(coordinator.commit(result_.result_info_))) {
LOG_WARN("fail to coordinator commit", KR(ret));
} else if (OB_FAIL(ObTableLoadService::remove_ctx(table_ctx))) {
LOG_WARN("fail to remove table ctx", KR(ret), K(key));
}
}
if (OB_NOT_NULL(table_ctx)) {
ObTableLoadService::put_ctx(table_ctx);
table_ctx = nullptr;
}
}
return ret;
}
int ObTableLoadCommitP::check_user_access(const ObString &credential_str)
{
return ObTableLoadUtils::check_user_access(credential_str, gctx_, credential_);
}
/**
* ObTableLoadCommitPeerP
*/
int ObTableLoadCommitPeerP::process()
{
int ret = OB_SUCCESS;
if (OB_FAIL(check_user_access(arg_.credential_))) {
LOG_WARN("fail to check_user_access", KR(ret));
} else if (OB_FAIL(ObTableLoadService::check_tenant())) {
LOG_WARN("fail to check tenant", KR(ret));
} else {
ObTableLoadTableCtx *table_ctx = nullptr;
ObTableLoadUniqueKey key(arg_.table_id_, arg_.task_id_);
if (OB_FAIL(ObTableLoadService::get_ctx(key, table_ctx))) {
LOG_WARN("fail to get table ctx", KR(ret), K(key));
} else {
ObTableLoadStore store(table_ctx);
if (OB_FAIL(store.init())) {
LOG_WARN("fail to init store", KR(ret));
} else if (OB_FAIL(store.commit(result_.result_info_))) {
LOG_WARN("fail to store commit", KR(ret));
} else if (OB_FAIL(ObTableLoadService::remove_ctx(table_ctx))) {
LOG_WARN("fail to remove table ctx", KR(ret), K(key));
}
}
if (OB_NOT_NULL(table_ctx)) {
ObTableLoadService::put_ctx(table_ctx);
table_ctx = nullptr;
}
}
return ret;
}
int ObTableLoadCommitPeerP::check_user_access(const ObString &credential_str)
{
return ObTableLoadUtils::check_user_access(credential_str, gctx_, credential_);
}
} // namespace observer
} // namespace oceanbase

38
src/observer/table_load/ob_table_load_coordinator.cpp
View File

@ -580,7 +580,7 @@ int ObTableLoadCoordinator::add_check_merge_result_task()
* commit
*/
int ObTableLoadCoordinator::commit_peers(ObTableLoadSqlStatistics &sql_statistics)
int ObTableLoadCoordinator::commit_peers()
{
int ret = OB_SUCCESS;
ObTableLoadArray<ObAddr> all_addr_array;
@ -598,7 +598,7 @@ int ObTableLoadCoordinator::commit_peers(ObTableLoadSqlStatistics &sql_statistic
ObTableLoadStore store(ctx_);
if (OB_FAIL(store.init())) {
LOG_WARN("fail to init store", KR(ret));
} else if (OB_FAIL(store.commit(res.result_info_, res.sql_statistics_))) {
} else if (OB_FAIL(store.commit(res.result_info_))) {
LOG_WARN("fail to commit store", KR(ret));
}
} else { // 远端, 发送rpc
@ -609,9 +609,6 @@ int ObTableLoadCoordinator::commit_peers(ObTableLoadSqlStatistics &sql_statistic
ATOMIC_AAF(&coordinator_ctx_->result_info_.deleted_, res.result_info_.deleted_);
ATOMIC_AAF(&coordinator_ctx_->result_info_.skipped_, res.result_info_.skipped_);
ATOMIC_AAF(&coordinator_ctx_->result_info_.warnings_, res.result_info_.warnings_);
if (OB_FAIL(sql_statistics.add(res.sql_statistics_))) {
LOG_WARN("fail to add result sql stats", KR(ret), K(addr), K(res));
}
}
}
}
@ -648,11 +645,11 @@ int ObTableLoadCoordinator::commit(ObTableLoadResultInfo &result_info)
ObTableLoadSqlStatistics sql_statistics;
if (OB_FAIL(coordinator_ctx_->check_status(ObTableLoadStatusType::MERGED))) {
LOG_WARN("fail to check coordinator status", KR(ret));
} else if (OB_FAIL(commit_peers(sql_statistics))) {
} else if (OB_FAIL(commit_peers())) {
LOG_WARN("fail to commit peers", KR(ret));
} else if (param_.online_opt_stat_gather_ &&
OB_FAIL(
drive_sql_stat(coordinator_ctx_->exec_ctx_->get_exec_ctx(), sql_statistics))) {
drive_sql_stat(coordinator_ctx_->exec_ctx_->get_exec_ctx()))) {
LOG_WARN("fail to drive sql stat", KR(ret));
} else if (OB_FAIL(commit_redef_table())) {
LOG_WARN("fail to commit redef table", KR(ret));
@ -679,11 +676,11 @@ int ObTableLoadCoordinator::px_commit_data()
ObTableLoadSqlStatistics sql_statistics;
if (OB_FAIL(coordinator_ctx_->check_status(ObTableLoadStatusType::MERGED))) {
LOG_WARN("fail to check coordinator status", KR(ret));
} else if (OB_FAIL(commit_peers(sql_statistics))) {
} else if (OB_FAIL(commit_peers())) {
LOG_WARN("fail to commit peers", KR(ret));
} else if (param_.online_opt_stat_gather_ &&
OB_FAIL(
drive_sql_stat(coordinator_ctx_->exec_ctx_->get_exec_ctx(), sql_statistics))) {
drive_sql_stat(coordinator_ctx_->exec_ctx_->get_exec_ctx()))) {
LOG_WARN("fail to drive sql stat", KR(ret));
}
}
@ -711,8 +708,7 @@ int ObTableLoadCoordinator::px_commit_ddl()
return ret;
}
int ObTableLoadCoordinator::drive_sql_stat(ObExecContext *ctx,
ObTableLoadSqlStatistics &sql_statistics)
int ObTableLoadCoordinator::drive_sql_stat(ObExecContext *ctx)
{
int ret = OB_SUCCESS;
const uint64_t tenant_id = MTL_ID();
@ -720,9 +716,9 @@ int ObTableLoadCoordinator::drive_sql_stat(ObExecContext *ctx,
ObSchemaGetterGuard schema_guard;
ObSchemaGetterGuard *tmp_schema_guard = nullptr;
const ObTableSchema *table_schema = nullptr;
if (OB_UNLIKELY(nullptr == ctx || sql_statistics.is_empty())) {
if (OB_UNLIKELY(nullptr == ctx)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid args", KR(ret), KPC(ctx), K(sql_statistics));
LOG_WARN("invalid args", KR(ret), KPC(ctx));
} else if (OB_FAIL(ObTableLoadSchema::get_table_schema(tenant_id, table_id, schema_guard,
table_schema))) {
LOG_WARN("fail to get table schema", KR(ret), K(tenant_id), K(table_id));
@ -731,16 +727,14 @@ int ObTableLoadCoordinator::drive_sql_stat(ObExecContext *ctx,
ctx->get_sql_ctx()->schema_guard_ = &schema_guard;
ctx->get_das_ctx().set_sql_ctx(ctx->get_sql_ctx());
}
ObSEArray<ObOptColumnStat *, 64> part_column_stats;
if (OB_FAIL(ret)) {
} else if (OB_FAIL(sql_statistics.get_col_stat_array(part_column_stats))) {
LOG_WARN("failed to get column stat array");
} else if (OB_FAIL(ObIncrementalStatEstimator::derive_global_stat_by_direct_load(
*ctx, sql_statistics.table_stat_array_, part_column_stats))) {
LOG_WARN("fail to drive global stat by direct load", KR(ret));
if (OB_SUCC(ret)) {
if (OB_FAIL(ObIncrementalStatEstimator::derive_global_stat_by_direct_load(
*ctx, table_id))) {
LOG_WARN("fail to drive global stat by direct load", KR(ret));
}
ctx->get_sql_ctx()->schema_guard_ = tmp_schema_guard;
ctx->get_das_ctx().set_sql_ctx(ctx->get_sql_ctx());
}
ctx->get_sql_ctx()->schema_guard_ = tmp_schema_guard;
ctx->get_das_ctx().set_sql_ctx(ctx->get_sql_ctx());
return ret;
}

4
src/observer/table_load/ob_table_load_coordinator.h
View File

@ -50,9 +50,9 @@ private:
int confirm_begin_peers();
int pre_merge_peers();
int start_merge_peers();
int commit_peers(table::ObTableLoadSqlStatistics &sql_statistics);
int commit_peers();
int commit_redef_table();
int drive_sql_stat(sql::ObExecContext *ctx, table::ObTableLoadSqlStatistics &sql_statistics);
int drive_sql_stat(sql::ObExecContext *ctx);
private:
int add_check_merge_result_task();
int check_peers_merge_result(bool &is_finish);

35
src/observer/table_load/ob_table_load_store.cpp
View File

@ -17,6 +17,7 @@
#include "observer/table_load/ob_table_load_utils.h"
#include "storage/direct_load/ob_direct_load_insert_table_ctx.h"
#include "share/stat/ob_opt_stat_monitor_manager.h"
#include "share/stat/ob_dbms_stats_utils.h"
namespace oceanbase
{
@ -276,7 +277,7 @@ int ObTableLoadStore::start_merge()
return ret;
}
int ObTableLoadStore::commit(ObTableLoadResultInfo &result_info, ObTableLoadSqlStatistics &sql_statistics)
int ObTableLoadStore::commit(ObTableLoadResultInfo &result_info)
{
int ret = OB_SUCCESS;
if (IS_NOT_INIT) {
@ -286,6 +287,7 @@ int ObTableLoadStore::commit(ObTableLoadResultInfo &result_info, ObTableLoadSqlS
LOG_INFO("store commit");
ObMutexGuard guard(store_ctx_->get_op_lock());
ObTableLoadDmlStat dml_stats;
ObTableLoadSqlStatistics sql_statistics;
if (OB_FAIL(store_ctx_->check_status(ObTableLoadStatusType::MERGED))) {
LOG_WARN("fail to check store status", KR(ret));
} else if (OB_FAIL(store_ctx_->insert_table_ctx_->commit())) {
@ -295,6 +297,9 @@ int ObTableLoadStore::commit(ObTableLoadResultInfo &result_info, ObTableLoadSqlS
} else if (param_.online_opt_stat_gather_ &&
OB_FAIL(store_ctx_->merger_->collect_sql_statistics(sql_statistics))) {
LOG_WARN("fail to collect sql stats", KR(ret));
} else if (param_.online_opt_stat_gather_ &&
OB_FAIL(commit_sql_statistics(sql_statistics))) {
LOG_WARN("fail to commit sql stats", KR(ret));
} else if (OB_FAIL(store_ctx_->merger_->collect_dml_stat(dml_stats))) {
LOG_WARN("fail to build dml stat", KR(ret));
} else if (OB_FAIL(ObOptStatMonitorManager::get_instance().update_dml_stat_info_from_direct_load(dml_stats.dml_stat_array_))) {
@ -308,6 +313,34 @@ int ObTableLoadStore::commit(ObTableLoadResultInfo &result_info, ObTableLoadSqlS
return ret;
}
int ObTableLoadStore::commit_sql_statistics(const ObTableLoadSqlStatistics &sql_statistics)
{
int ret = OB_SUCCESS;
const uint64_t tenant_id = MTL_ID();
const uint64_t table_id = param_.table_id_;
ObSchemaGetterGuard schema_guard;
const ObTableSchema *table_schema = nullptr;
ObSEArray<ObOptColumnStat *, 64> part_column_stats;
ObSEArray<ObOptTableStat *, 64> part_table_stats;
if (sql_statistics.is_empty()) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("sql statistics is empty", K(ret));
} else if (OB_FAIL(ObTableLoadSchema::get_table_schema(tenant_id, table_id, schema_guard,
table_schema))) {
LOG_WARN("fail to get table schema", KR(ret), K(tenant_id), K(table_id));
} else if (OB_FAIL(sql_statistics.get_col_stat_array(part_column_stats))) {
LOG_WARN("failed to get column stat array");
} else if (OB_FAIL(sql_statistics.get_table_stat_array(part_table_stats))) {
LOG_WARN("failed to get table stat array");
} else if (OB_FAIL(ObDbmsStatsUtils::split_batch_write(
&schema_guard, store_ctx_->ctx_->session_info_, GCTX.sql_proxy_, part_table_stats,
part_column_stats))) {
LOG_WARN("failed to batch write stats", K(ret), K(sql_statistics.table_stat_array_),
K(sql_statistics.col_stat_array_));
}
return ret;
}
int ObTableLoadStore::get_status(ObTableLoadStatusType &status, int &error_code)
{
int ret = OB_SUCCESS;

4
src/observer/table_load/ob_table_load_store.h
View File

@ -38,8 +38,10 @@ public:
int confirm_begin();
int pre_merge(const table::ObTableLoadArray<table::ObTableLoadTransId> &committed_trans_id_array);
int start_merge();
int commit(table::ObTableLoadResultInfo &result_info, table::ObTableLoadSqlStatistics &sql_statistics);
int commit(table::ObTableLoadResultInfo &result_info);
int get_status(table::ObTableLoadStatusType &status, int &error_code);
private:
int commit_sql_statistics(const table::ObTableLoadSqlStatistics &sql_statistics);
private:
class MergeTaskProcessor;
class MergeTaskCallback;

35
src/share/stat/ob_dbms_stats_utils.cpp
View File

@ -355,6 +355,29 @@ int ObDbmsStatsUtils::split_batch_write(sql::ObExecContext &ctx,
const bool is_index_stat/*default false*/,
const bool is_history_stat/*default false*/,
const bool is_online_stat /*default false*/)
{
int ret = OB_SUCCESS;
if (OB_FAIL(split_batch_write(ctx.get_virtual_table_ctx().schema_guard_,
ctx.get_my_session(),
ctx.get_sql_proxy(),
table_stats,
column_stats,
is_index_stat,
is_history_stat,
is_online_stat))) {
LOG_WARN("failed to split batch write", K(ret));
}
return ret;
}
int ObDbmsStatsUtils::split_batch_write(share::schema::ObSchemaGetterGuard *schema_guard,
sql::ObSQLSessionInfo *session_info,
common::ObMySQLProxy *sql_proxy,
ObIArray<ObOptTableStat*> &table_stats,
ObIArray<ObOptColumnStat*> &column_stats,
const bool is_index_stat/*default false*/,
const bool is_history_stat/*default false*/,
const bool is_online_stat /*default false*/)
{
int ret = OB_SUCCESS;
int64_t idx_tab_stat = 0;
@ -366,10 +389,10 @@ int ObDbmsStatsUtils::split_batch_write(sql::ObExecContext &ctx,
int64_t current_time = ObTimeUtility::current_time();
ObMySQLTransaction trans;
//begin trans before writing stats
if (OB_ISNULL(ctx.get_my_session())) {
if (OB_ISNULL(session_info) || OB_ISNULL(sql_proxy)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get unexpected null", K(ret), K(ctx.get_my_session()));
} else if (OB_FAIL(trans.start(ctx.get_sql_proxy(), ctx.get_my_session()->get_effective_tenant_id()))) {
LOG_WARN("get unexpected null", K(ret), K(session_info), K(sql_proxy));
} else if (OB_FAIL(trans.start(sql_proxy, session_info->get_effective_tenant_id()))) {
LOG_WARN("fail to start transaction", K(ret));
}
while (OB_SUCC(ret) &&
@ -408,8 +431,8 @@ int ObDbmsStatsUtils::split_batch_write(sql::ObExecContext &ctx,
}
}
if (OB_SUCC(ret)) {
if (OB_FAIL(ObDbmsStatsUtils::batch_write(ctx.get_virtual_table_ctx().schema_guard_,
ctx.get_my_session()->get_effective_tenant_id(),
if (OB_FAIL(ObDbmsStatsUtils::batch_write(schema_guard,
session_info->get_effective_tenant_id(),
trans,
write_table_stats,
write_column_stats,
@ -417,7 +440,7 @@ int ObDbmsStatsUtils::split_batch_write(sql::ObExecContext &ctx,
is_index_stat,
is_history_stat,
is_online_stat,
CREATE_OBJ_PRINT_PARAM(ctx.get_my_session())))) {
CREATE_OBJ_PRINT_PARAM(session_info)))) {
LOG_WARN("failed to batch write stats", K(ret), K(idx_tab_stat), K(idx_col_stat));
} else {/*do nothing*/}
}

9
src/share/stat/ob_dbms_stats_utils.h
View File

@ -46,6 +46,15 @@ public:
const bool is_history_stat = false,
const bool is_online_stat = false);
static int split_batch_write(share::schema::ObSchemaGetterGuard *schema_guard,
sql::ObSQLSessionInfo *session_info,
common::ObMySQLProxy *sql_proxy,
ObIArray<ObOptTableStat*> &table_stats,
ObIArray<ObOptColumnStat*> &column_stats,
const bool is_index_stat = false,
const bool is_history_stat = false,
const bool is_online_stat = false);
static int batch_write_history_stats(sql::ObExecContext &ctx,
ObIArray<ObOptTableStatHandle> &history_tab_handles,
ObIArray<ObOptColumnStatHandle> &history_col_handles);

232
src/share/stat/ob_incremental_stat_estimator.cpp
View File

@ -21,6 +21,7 @@
#include "share/stat/ob_topk_hist_estimator.h"
#include "pl/sys_package/ob_dbms_stats.h"
#include "share/stat/ob_dbms_stats_history_manager.h"
#include "share/rc/ob_tenant_base.h"
namespace oceanbase {
using namespace pl;
@ -59,110 +60,81 @@ int ObIncrementalStatEstimator::try_derive_global_stat(ObExecContext &ctx,
}
int ObIncrementalStatEstimator::derive_global_stat_by_direct_load(ObExecContext &ctx,
ObIArray<ObOptTableStat*> &part_tab_stats,
const ObIArray<ObOptColumnStat*> &part_column_stats)
const uint64_t table_id)
{
int ret = OB_SUCCESS;
const share::schema::ObTableSchema *table_schema = NULL;
share::schema::ObSchemaGetterGuard *schema_guard = ctx.get_virtual_table_ctx().schema_guard_;
ObSEArray<ObOptStat, 4> part_opt_stats;
ObSEArray<ObOptStat, 4> all_derive_opt_stats;
ObArenaAllocator alloc("ObIncrStats");
ObTableStatParam param;
param.allocator_ = &alloc;
if (part_tab_stats.empty()) {
//do nothing
} else if (OB_ISNULL(part_tab_stats.at(0)) || OB_UNLIKELY(!part_tab_stats.at(0)->is_valid())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get unexpected error", K(ret), KPC(part_tab_stats.at(0)));
} else if (OB_FAIL(gen_opt_stat_param_by_direct_load(ctx, alloc,
part_tab_stats.at(0)->get_table_id(),
param))) {
LOG_WARN("failed to gen opt stat param by direct load", K(ret));
} else if (param.part_level_ != share::schema::PARTITION_LEVEL_ONE &&
param.part_level_ != share::schema::PARTITION_LEVEL_TWO) {
/*do nothing*/
} else if (OB_FAIL(generate_all_opt_stat(part_tab_stats, part_column_stats,
param.column_params_.count(),
part_opt_stats))) {
LOG_WARN("failed to generate all opt stat", K(ret));
} else if (OB_FAIL(param.all_part_infos_.assign(param.part_infos_))||
OB_FAIL(param.all_subpart_infos_.assign(param.subpart_infos_))) {
LOG_WARN("failed to assign", K(ret));
} else {
bool nee_derive_part = param.part_level_ == share::schema::PARTITION_LEVEL_TWO;
//derive part stat first
if (nee_derive_part) {
ObSEArray<ObOptStat, 1> empty_opt_stats;
if (OB_FAIL(do_derive_part_stats_from_subpart_stats(ctx, alloc, param, empty_opt_stats,
part_opt_stats, all_derive_opt_stats))) {
LOG_WARN("failed to derive part stat by direct load", K(ret));
}
}
//derive global stat
if (OB_SUCC(ret)) {
ObOptStat global_opt_stat;
bool need_derive_hist = false;
bool need_gather_hybrid_hist = false;
if (OB_FAIL(do_derive_global_stat(ctx, alloc, param,
nee_derive_part ? all_derive_opt_stats : part_opt_stats,
need_derive_hist,
TABLE_LEVEL, param.global_part_id_, need_gather_hybrid_hist,
global_opt_stat))) {
LOG_WARN("Failed to derive global stat from part stat", K(ret));
} else if (OB_FAIL(all_derive_opt_stats.push_back(global_opt_stat))) {
LOG_WARN("faield to push back", K(ret));
ObSEArray<ObOptTableStat, 4> part_tab_stats;
ObSEArray<ObOptColumnStatHandle, 4> part_col_handles;
ObArenaAllocator alloc("ObIncrStats", OB_MALLOC_NORMAL_BLOCK_SIZE, MTL_ID());
SMART_VAR(ObTableStatParam, param) {
param.allocator_ = &alloc;
if (OB_FAIL(gen_opt_stat_param_by_direct_load(ctx, alloc, table_id, param))) {
LOG_WARN("failed to gen opt stat param by direct load", K(ret));
} else if (param.part_level_ != share::schema::PARTITION_LEVEL_ONE &&
param.part_level_ != share::schema::PARTITION_LEVEL_TWO) {
//if we don't derive, we need update some cache info.
if (OB_FAIL(ObBasicStatsEstimator::update_last_modified_count(ctx, param))) {
LOG_WARN("failed to update last modified count", K(ret));
} else if (OB_FAIL(pl::ObDbmsStats::update_stat_cache(ctx.get_my_session()->get_rpc_tenant_id(), param))) {
LOG_WARN("fail to update stat cache", K(ret));
}
} else if (OB_FAIL(get_all_part_opt_stat_by_direct_load(param,
part_tab_stats,
part_col_handles,
part_opt_stats))) {
LOG_WARN("failed to get all part opt stat by direct load", K(ret));
} else if (OB_FAIL(param.all_part_infos_.assign(param.part_infos_))||
OB_FAIL(param.all_subpart_infos_.assign(param.subpart_infos_))) {
LOG_WARN("failed to assign", K(ret));
} else {
bool nee_derive_part = param.part_level_ == share::schema::PARTITION_LEVEL_TWO;
//derive part stat first
if (nee_derive_part) {
ObSEArray<ObOptStat, 1> empty_opt_stats;
if (OB_FAIL(do_derive_part_stats_from_subpart_stats(ctx, alloc, param, empty_opt_stats,
part_opt_stats, all_derive_opt_stats))) {
LOG_WARN("failed to derive part stat by direct load", K(ret));
}
}
//derive global stat
if (OB_SUCC(ret)) {
ObOptStat global_opt_stat;
bool need_derive_hist = false;
bool need_gather_hybrid_hist = false;
if (OB_FAIL(do_derive_global_stat(ctx, alloc, param,
nee_derive_part ? all_derive_opt_stats : part_opt_stats,
need_derive_hist,
TABLE_LEVEL, param.global_part_id_, need_gather_hybrid_hist,
global_opt_stat))) {
LOG_WARN("Failed to derive global stat from part stat", K(ret));
} else if (OB_FAIL(all_derive_opt_stats.push_back(global_opt_stat))) {
LOG_WARN("faield to push back", K(ret));
}
}
//write all stat
if (OB_SUCC(ret)) {
ObSEArray<ObOptTableStat *, 4> all_tstats;
ObSEArray<ObOptColumnStat *, 4> all_cstats;
//direct load does't process history stats.
if (OB_FAIL(ObDbmsStatsUtils::calssify_opt_stat(all_derive_opt_stats,
all_tstats,
all_cstats))) {
LOG_WARN("failed to calssify opt stat", K(ret));
} else if (OB_FAIL(ObDbmsStatsUtils::split_batch_write(ctx, all_tstats, all_cstats))) {
LOG_WARN("failed to split batch write", K(ret));
} else if (OB_FAIL(ObBasicStatsEstimator::update_last_modified_count(ctx, param))) {
LOG_WARN("failed to update last modified count", K(ret));
} else if (OB_FAIL(pl::ObDbmsStats::update_stat_cache(ctx.get_my_session()->get_rpc_tenant_id(), param))) {
LOG_WARN("fail to update stat cache", K(ret));
}
}
}
LOG_TRACE("succeed to derive global stat by direct load", K(param), K(part_tab_stats));
}
//write all stat
if (OB_SUCC(ret)) {
ObSEArray<ObOptTableStat *, 4> all_tstats;
ObSEArray<ObOptColumnStat *, 4> all_cstats;
ObSEArray<ObOptTableStatHandle, 4> history_tab_handles;
ObSEArray<ObOptColumnStatHandle, 4> history_col_handles;
if (OB_FAIL(append(all_tstats, part_tab_stats))) {
LOG_WARN("failed to append", K(ret));
} else if (OB_FAIL(append(all_cstats, part_column_stats))) {
LOG_WARN("failed to append", K(ret));
} else if (OB_FAIL(ObDbmsStatsUtils::calssify_opt_stat(all_derive_opt_stats,
all_tstats,
all_cstats))) {
LOG_WARN("failed to calssify opt stat", K(ret));
} else if (OB_FAIL(write_all_opt_stats_by_dircet_load(ctx, param, all_tstats, all_cstats))) {
LOG_WARN("failed to write all opt stats by dircet load", K(ret));
} else {/*do nothing*/}
}
LOG_TRACE("succeed to derive global stat by direct load", K(param), K(part_tab_stats),
K(part_column_stats));
return ret;
}
int ObIncrementalStatEstimator::write_all_opt_stats_by_dircet_load(
ObExecContext &ctx,
const ObTableStatParam &param,
ObIArray<ObOptTableStat *> &all_tstats,
ObIArray<ObOptColumnStat *> &all_cstats)
{
int ret = OB_SUCCESS;
ObSEArray<ObOptTableStatHandle, 4> history_tab_handles;
ObSEArray<ObOptColumnStatHandle, 4> history_col_handles;
//before write, we need record history stats.
if (OB_FAIL(ObDbmsStatsHistoryManager::get_history_stat_handles(ctx, param,
history_tab_handles,
history_col_handles))) {
LOG_WARN("failed to get history stat handles", K(ret));
} else if (OB_FAIL(ObDbmsStatsUtils::split_batch_write(ctx, all_tstats, all_cstats))) {
LOG_WARN("failed to split batch write", K(ret));
} else if (OB_FAIL(ObDbmsStatsUtils::batch_write_history_stats(ctx,
history_tab_handles,
history_col_handles))) {
LOG_WARN("failed to batch write history stats", K(ret));
} else if (OB_FAIL(ObBasicStatsEstimator::update_last_modified_count(ctx, param))) {
LOG_WARN("failed to update last modified count", K(ret));
} else if (OB_FAIL(pl::ObDbmsStats::update_stat_cache(ctx.get_my_session()->get_rpc_tenant_id(), param))) {
LOG_WARN("fail to update stat cache", K(ret));
} else {/*do nothing*/}
return ret;
}
@ -398,38 +370,6 @@ int ObIncrementalStatEstimator::generate_all_opt_stat(ObIArray<ObOptTableStat> &
return ret;
}
int ObIncrementalStatEstimator::generate_all_opt_stat(ObIArray<ObOptTableStat*> &table_stats,
const ObIArray<ObOptColumnStat*> &column_stats,
int64_t col_cnt,
ObIArray<ObOptStat> &all_opt_stats)
{
int ret = OB_SUCCESS;
for (int64_t i = 0; OB_SUCC(ret) && i < table_stats.count(); ++i) {
ObOptStat opt_stat;
opt_stat.table_stat_ = table_stats.at(i);
for (int64_t j = 0; OB_SUCC(ret) && opt_stat.column_stats_.count() < col_cnt && j < column_stats.count(); ++j) {
if (OB_ISNULL(column_stats.at(j)) || OB_ISNULL(opt_stat.table_stat_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get unexpected error", K(ret), K(column_stats.at(j)));
} else if (opt_stat.table_stat_->get_partition_id() == column_stats.at(j)->get_partition_id()) {
if (OB_FAIL(opt_stat.column_stats_.push_back(
const_cast<ObOptColumnStat*>(column_stats.at(j))))) {
LOG_WARN("failed to push back col stat", K(ret));
} else {/*do nothing*/}
}
}
if (OB_SUCC(ret)) {
if (OB_UNLIKELY(opt_stat.column_stats_.count() != col_cnt)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get unexpected error", K(ret), K(opt_stat.column_stats_), K(col_cnt));
} else if (OB_FAIL(all_opt_stats.push_back(opt_stat))) {
LOG_WARN("failed to push back opt stat", K(ret));
} else {/*do nothing*/}
}
}
return ret;
}
int ObIncrementalStatEstimator::get_table_and_column_stats(
ObOptStat &src_opt_stat,
const ObTableStatParam &param,
@ -1001,5 +941,43 @@ int ObIncrementalStatEstimator::gen_opt_stat_param_by_direct_load(ObExecContext
return ret;
}
int ObIncrementalStatEstimator::get_all_part_opt_stat_by_direct_load(
const ObTableStatParam param,
ObIArray<ObOptTableStat> &part_tab_stats,
ObIArray<ObOptColumnStatHandle> &part_col_handles,
ObIArray<ObOptStat> &part_opt_stats)
{
int ret = OB_SUCCESS;
ObSEArray<uint64_t, 4> column_ids;
const ObIArray<int64_t> &partition_ids = param.part_level_ == share::schema::PARTITION_LEVEL_ONE ?
param.part_ids_ : param.subpart_ids_;
if (OB_UNLIKELY(param.part_level_ != share::schema::PARTITION_LEVEL_ONE &&
param.part_level_ != share::schema::PARTITION_LEVEL_TWO)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get unexpected error", K(ret), K(param));
} else if (OB_FAIL(get_column_ids(param.column_params_, column_ids))) {
LOG_WARN("failed to get column ids", K(ret));
} else if (OB_FAIL(ObOptStatManager::get_instance().get_table_stat(param.tenant_id_,
param.table_id_,
partition_ids,
part_tab_stats))) {
LOG_WARN("failed to get table stat", K(ret));
} else if (OB_FAIL(ObOptStatManager::get_instance().get_column_stat(param.tenant_id_,
param.table_id_,
partition_ids,
column_ids,
part_col_handles))) {
LOG_WARN("failed to get column stat", K(ret));
} else if (OB_FAIL(generate_all_opt_stat(part_tab_stats,
part_col_handles,
column_ids.count(),
part_opt_stats))) {
LOG_WARN("failed to generate all opt stat", K(ret));
} else {
LOG_TRACE("Succeed get all part opt stat by direct load", K(param), K(part_tab_stats));
}
return ret;
}
} // namespace common
} // namespace oceanbase

17
src/share/stat/ob_incremental_stat_estimator.h
View File

@ -37,9 +37,7 @@ public:
int64_t subpart_cnt,
bool is_gather_part);
static int derive_global_stat_by_direct_load(ObExecContext &ctx,
ObIArray<ObOptTableStat*> &part_tab_stats,
const ObIArray<ObOptColumnStat*> &part_column_stats);
static int derive_global_stat_by_direct_load(ObExecContext &ctx, const uint64_t table_id);
private:
static int derive_global_stat_from_part_stats(ObExecContext &ctx,
@ -75,11 +73,6 @@ private:
int64_t col_cnt,
ObIArray<ObOptStat> &all_opt_stats);
static int generate_all_opt_stat(ObIArray<ObOptTableStat*> &table_stats,
const ObIArray<ObOptColumnStat*> &column_stats,
int64_t col_cnt,
ObIArray<ObOptStat> &all_opt_stats);
static int do_derive_global_stat(ObExecContext &ctx,
ObIAllocator &alloc,
const ObTableStatParam &param,
@ -142,10 +135,10 @@ private:
const uint64_t table_id,
ObTableStatParam &param);
static int write_all_opt_stats_by_dircet_load(ObExecContext &ctx,
const ObTableStatParam &param,
ObIArray<ObOptTableStat *> &all_tstats,
ObIArray<ObOptColumnStat *> &all_cstats);
static int get_all_part_opt_stat_by_direct_load(const ObTableStatParam param,
ObIArray<ObOptTableStat> &part_tab_stats,
ObIArray<ObOptColumnStatHandle> &part_col_handles,
ObIArray<ObOptStat> &part_opt_stats);
};

16
src/share/table/ob_table_load_sql_statistics.cpp
View File

@ -135,7 +135,21 @@ int ObTableLoadSqlStatistics::add(const ObTableLoadSqlStatistics& other)
return ret;
}
int ObTableLoadSqlStatistics::get_col_stat_array(ObIArray<ObOptColumnStat*> &col_stat_array)
int ObTableLoadSqlStatistics::get_table_stat_array(ObIArray<ObOptTableStat*> &table_stat_array) const
{
int ret = OB_SUCCESS;
for (int64_t i = 0; OB_SUCC(ret) && i < table_stat_array_.count(); ++i) {
if (OB_ISNULL(table_stat_array_.at(i))) {
ret = OB_ERR_UNEXPECTED;
OB_LOG(WARN, "get unexpected null");
} else if (OB_FAIL(table_stat_array.push_back(table_stat_array_.at(i)))) {
OB_LOG(WARN, "failed to push back col stat");
}
}
return ret;
}
int ObTableLoadSqlStatistics::get_col_stat_array(ObIArray<ObOptColumnStat*> &col_stat_array) const
{
int ret = OB_SUCCESS;
for (int64_t i = 0; OB_SUCC(ret) && i < col_stat_array_.count(); ++i) {

3
src/share/table/ob_table_load_sql_statistics.h
View File

@ -35,7 +35,8 @@ public:
int allocate_table_stat(ObOptTableStat *&table_stat);
int allocate_col_stat(ObOptOSGColumnStat *&col_stat);
int add(const ObTableLoadSqlStatistics& other);
int get_col_stat_array(ObIArray<ObOptColumnStat*> &col_stat_array);
int get_table_stat_array(ObIArray<ObOptTableStat*> &table_stat_array) const;
int get_col_stat_array(ObIArray<ObOptColumnStat*> &col_stat_array) const;
int persistence_col_stats();
TO_STRING_KV(K_(col_stat_array), K_(table_stat_array));
public:

61
src/storage/direct_load/ob_direct_load_merge_ctx.cpp
View File

@ -1,6 +1,14 @@
// Copyright (c) 2022-present Oceanbase Inc. All Rights Reserved.
// Author:
// suzhi.yt <>
/**
* 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 STORAGE
@ -236,39 +244,23 @@ int ObDirectLoadTabletMergeCtx::collect_sql_statistics(
// scan task_array
for (int64_t j = 0; OB_SUCC(ret) && j < task_array_.count(); ++j) {
ObOptOSGColumnStat *tmp_col_stat = task_array_.at(j)->get_column_stat_array().at(i);
ObOptOSGColumnStat *copied_col_stat = NULL;
if (OB_ISNULL(tmp_col_stat)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get unexpected null");
} else if (OB_ISNULL(copied_col_stat =
ObOptOSGColumnStat::create_new_osg_col_stat(sql_statistics.allocator_))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("failed to allocate memory");
} else if (OB_FAIL(copied_col_stat->deep_copy(*tmp_col_stat))) {
LOG_WARN("fail to copy colstat", KR(ret));
} else if (OB_FAIL(osg_col_stat->merge_column_stat(*copied_col_stat))) {
LOG_WARN("fail to merge column stat", KR(ret));
} else {
row_count += task_array_.at(j)->get_row_count();
if (task_array_.at(j)->get_row_count() != 0) {
if (OB_FAIL(osg_col_stat->merge_column_stat(*tmp_col_stat))) {
LOG_WARN("fail to merge column stat", KR(ret));
} else {
row_count += task_array_.at(j)->get_row_count();
}
}
}
// scan fast heap table
for (int64_t j = 0; OB_SUCC(ret) && j < fast_heap_table_array.count(); ++j) {
ObOptOSGColumnStat *tmp_col_stat = fast_heap_table_array.at(j)->get_column_stat_array().at(i);
ObOptOSGColumnStat *copied_col_stat = NULL;
if (OB_ISNULL(tmp_col_stat)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get unexpected null");
} else if (OB_ISNULL(copied_col_stat =
ObOptOSGColumnStat::create_new_osg_col_stat(sql_statistics.allocator_))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("failed to allocate memory");
} else if (OB_FAIL(copied_col_stat->deep_copy(*tmp_col_stat))) {
LOG_WARN("fail to copy colstat", KR(ret));
} else if (OB_FAIL(osg_col_stat->merge_column_stat(*copied_col_stat))) {
LOG_WARN("fail to merge column stat", KR(ret));
} else {
row_count += fast_heap_table_array.at(j)->get_row_count();
if (fast_heap_table_array.at(j)->get_row_count() != 0) {
if (OB_FAIL(osg_col_stat->merge_column_stat(*tmp_col_stat))) {
LOG_WARN("fail to merge column stat", KR(ret));
} else {
row_count += fast_heap_table_array.at(j)->get_row_count();
}
}
}
if (OB_SUCC(ret)) {
@ -280,6 +272,9 @@ int ObDirectLoadTabletMergeCtx::collect_sql_statistics(
osg_col_stat->col_stat_->set_stat_level(stat_level);
osg_col_stat->col_stat_->set_column_id(param_.col_descs_->at(col_id).col_id_);
osg_col_stat->col_stat_->set_num_distinct(ObGlobalNdvEval::get_ndv_from_llc(osg_col_stat->col_stat_->get_llc_bitmap()));
if (OB_FAIL(osg_col_stat->set_min_max_datum_to_obj())) {
LOG_WARN("failed to set min max datum to obj", K(ret));
}
}
}
if (OB_SUCC(ret)) {
@ -289,10 +284,6 @@ int ObDirectLoadTabletMergeCtx::collect_sql_statistics(
table_stat->set_row_count(table_row_cnt);
table_stat->set_avg_row_size(table_avg_len);
}
// persistence col stat once a merge task finished
if (OB_SUCC(ret) && OB_FAIL(sql_statistics.persistence_col_stats())) {
LOG_WARN("failed to persistence col stats");
}
}
}
return ret;