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

[FEAT MERGE] information_schema enhence

Browse Source
This commit is contained in:
obdev
2023-01-13 03:42:07 +00:00
committed by OB-robot
parent 9ce9e1a357
commit ec56715ea5
98 changed files with 4513 additions and 3156 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
src/sql/executor/ob_maintain_dependency_info_task.cpp
View File

@ -26,7 +26,9 @@ namespace sql
ObMaintainObjDepInfoTask::ObMaintainObjDepInfoTask (const uint64_t tenant_id)
: tenant_id_(tenant_id),
gctx_(GCTX),
rs_rpc_proxy_(*GCTX.rs_rpc_proxy_)
rs_rpc_proxy_(*GCTX.rs_rpc_proxy_),
view_schema_(&alloc_),
reset_view_column_infos_(false)
{
set_retry_times(0);
}
@ -39,7 +41,9 @@ ObMaintainObjDepInfoTask::ObMaintainObjDepInfoTask (
const DepObjKeyItemList &delete_dep_objs)
: tenant_id_(tenant_id),
gctx_(GCTX),
rs_rpc_proxy_(rs_rpc_proxy)
rs_rpc_proxy_(rs_rpc_proxy),
view_schema_(&alloc_),
reset_view_column_infos_(false)
{
set_retry_times(0);
insert_dep_objs_.assign(insert_dep_objs);
@ -88,6 +92,14 @@ int ObMaintainObjDepInfoTask::check_and_build_dep_info_arg(
int ret = OB_SUCCESS;
bool is_valid = false;
for (int64_t i = 0 ; OB_SUCC(ret) && i < dep_objs.count(); ++i) {
// multi views in one stmt share ref_obj_array, just update own dep
if (view_schema_.get_table_id() != dep_objs.at(i).dep_obj_key_.dep_obj_id_) {
/*
* create view will maintain -1 as dep obj id, but will not call this function
* if current view level is less than 1 will maintain -1 as dep obj id, try to remove task like this earlier
*/
continue;
}
if (OB_FAIL(check_cur_maintain_task_is_valid(
dep_objs.at(i).dep_obj_key_,
dep_objs.at(i).dep_obj_item_.dep_obj_schema_version_,
@ -132,6 +144,8 @@ share::ObAsyncTask *ObMaintainObjDepInfoTask::deep_copy(char *buf, const int64_t
insert_dep_objs_,
update_dep_objs_,
delete_dep_objs_);
OZ ((static_cast<ObMaintainObjDepInfoTask *> (task))->assign_view_schema(view_schema_));
OX ((static_cast<ObMaintainObjDepInfoTask *> (task))->reset_view_column_infos_ = reset_view_column_infos_);
}
return task;
}
@ -144,6 +158,7 @@ int ObMaintainObjDepInfoTask::process()
SMART_VAR(obrpc::ObDependencyObjDDLArg, dep_obj_info_arg) {
dep_obj_info_arg.tenant_id_ = tenant_id_;
dep_obj_info_arg.exec_tenant_id_ = tenant_id_;
dep_obj_info_arg.reset_view_column_infos_ = reset_view_column_infos_;
OZ (gctx_.schema_service_->async_refresh_schema(tenant_id_, last_version));
OZ (gctx_.schema_service_->get_tenant_schema_guard(tenant_id_, schema_guard));
OZ (check_and_build_dep_info_arg(schema_guard, dep_obj_info_arg,
@ -153,9 +168,12 @@ int ObMaintainObjDepInfoTask::process()
OZ (check_and_build_dep_info_arg(schema_guard, dep_obj_info_arg,
delete_dep_objs_, share::schema::ObReferenceObjTable::DELETE_OP));
if (OB_FAIL(ret)) {
} else if (view_schema_.is_valid() && OB_FAIL(dep_obj_info_arg.schema_.assign(view_schema_))) {
LOG_WARN("failed to assign view schema", K(ret));
} else if (dep_obj_info_arg.insert_dep_objs_.empty()
&& dep_obj_info_arg.update_dep_objs_.empty()
&& dep_obj_info_arg.delete_dep_objs_.empty()) {
&& dep_obj_info_arg.delete_dep_objs_.empty()
&& !dep_obj_info_arg.schema_.is_valid()) {
// do nothing
} else if (OB_FAIL(rs_rpc_proxy_.maintain_obj_dependency_info(dep_obj_info_arg))) {
LOG_WARN("failed to maintain_obj_dependency_info", K(ret), K(dep_obj_info_arg));
@ -164,6 +182,26 @@ int ObMaintainObjDepInfoTask::process()
return ret;
}
int ObMaintainObjDepInfoTask::assign_view_schema(const ObTableSchema &view_schema)
{
int ret = OB_SUCCESS;
if (OB_FAIL(view_schema_.assign(view_schema))) {
LOG_WARN("failed to copy view schema", K(ret));
}
return ret;
}
int ObMaintainDepInfoTaskQueue::init(const int64_t thread_cnt, const int64_t queue_size)
{
int ret = OB_SUCCESS;
if (OB_FAIL(ObAsyncTaskQueue::init(thread_cnt, queue_size))) {
LOG_WARN("failed to init base queue", K(ret));
} else if (OB_FAIL(view_info_set_.create(INIT_BKT_SIZE))) {
LOG_WARN("failed to init view set", K(ret));
}
return ret;
}
void ObMaintainDepInfoTaskQueue::run2()
{
int ret = OB_SUCCESS;
@ -174,7 +212,11 @@ void ObMaintainDepInfoTaskQueue::run2()
} else {
ObAddr zero_addr;
while (!stop_) {
if (REACH_TIME_INTERVAL(600 * 1000 * 1000)) {
if (REACH_TIME_INTERVAL(6 * 1000 * 1000)) {
if (0 == queue_.size() && 0 != view_info_set_.size()) {
LOG_WARN("queue size not match", K(queue_.size()), K(view_info_set_.size()));
view_info_set_.clear();
}
LOG_INFO("[ASYNC TASK QUEUE]", "queue_size", queue_.size());
}
if (last_execute_time_ > 0
@ -229,6 +271,15 @@ void ObMaintainDepInfoTaskQueue::run2()
rescheduled = true;
}
}
} else {
const ObTableSchema &view_schema = (static_cast<ObMaintainObjDepInfoTask *> (task))->get_view_schema();
if (view_schema.is_valid()) {
// ignore ret
int tmp_ret = OB_SUCCESS;
if (OB_SUCCESS != (tmp_ret = view_info_set_.erase_refactored(view_schema.get_table_id()))) {
LOG_WARN("failed to erase obj id", K(tmp_ret));
}
}
}
last_execute_time_ = ObTimeUtility::current_time();
if (!rescheduled) {

23
src/sql/executor/ob_maintain_dependency_info_task.h
View File

@ -16,6 +16,9 @@
#include "lib/container/ob_fixed_array.h"
#include "lib/thread/ob_async_task_queue.h"
#include "share/schema/ob_dependency_info.h"
#include "lib/hash/ob_hashset.h"
#include "share/schema/ob_multi_version_schema_service.h"
#include "share/schema/ob_schema_struct.h"
namespace oceanbase
{
@ -49,7 +52,7 @@ public:
DepObjKeyItemList& get_update_dep_objs() { return update_dep_objs_; }
DepObjKeyItemList& get_delete_dep_objs() { return delete_dep_objs_; }
bool is_empty_task() const
{ return insert_dep_objs_.empty() && update_dep_objs_.empty() && delete_dep_objs_.empty(); }
{ return !reset_view_column_infos_ && (insert_dep_objs_.empty() && update_dep_objs_.empty() && delete_dep_objs_.empty() && !view_schema_.is_valid()); }
// int check_and_refresh_schema(uint64_t effective_tenant_id);
int check_cur_maintain_task_is_valid(
const share::schema::ObReferenceObjTable::ObDependencyObjKey &dep_obj_key,
@ -65,6 +68,10 @@ public:
virtual int process() override;
virtual int64_t get_deep_copy_size() const override { return sizeof(ObMaintainObjDepInfoTask); }
ObAsyncTask *deep_copy(char *buf, const int64_t buf_size) const override;
int assign_view_schema(const share::schema::ObTableSchema &view_schema);
share::schema::ObTableSchema &get_view_schema() { return view_schema_; }
void set_reset_view_column_infos(bool flag) { reset_view_column_infos_ = flag; }
bool reset_view_column_infos() const { return reset_view_column_infos_; }
private:
uint64_t tenant_id_;
@ -73,21 +80,31 @@ private:
DepObjKeyItemList insert_dep_objs_;
DepObjKeyItemList update_dep_objs_;
DepObjKeyItemList delete_dep_objs_;
ObArenaAllocator alloc_;
share::schema::ObTableSchema view_schema_;
bool reset_view_column_infos_;
DISALLOW_COPY_AND_ASSIGN(ObMaintainObjDepInfoTask);
};
class ObMaintainDepInfoTaskQueue: public share::ObAsyncTaskQueue
{
public:
static const int64_t INIT_BKT_SIZE = 8192;
ObMaintainDepInfoTaskQueue() : last_execute_time_(0) {}
virtual ~ObMaintainDepInfoTaskQueue() {}
virtual ~ObMaintainDepInfoTaskQueue()
{
view_info_set_.destroy();
}
int init(const int64_t thread_cnt, const int64_t queue_size);
virtual void run2() override;
inline int64_t get_last_execute_time() const { return last_execute_time_; }
inline void set_last_execute_time(const int64_t execute_time)
{ last_execute_time_ = execute_time; }
int add_view_id_to_set(const uint64_t view_id) { return view_info_set_.set_refactored(view_id); }
int erase_view_id_from_set(const uint64_t view_id) { return view_info_set_.erase_refactored(view_id); }
private:
int64_t last_execute_time_;
common::hash::ObHashSet<uint64_t, common::hash::ReadWriteDefendMode> view_info_set_;
};
} // namespace sql