morningman closed pull request #387: change OLAPTablePtr to TabletPtr
URL: https://github.com/apache/incubator-doris/pull/387
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/be/src/agent/pusher.cpp b/be/src/agent/pusher.cpp
index fc79bfb6..a26d0f11 100644
--- a/be/src/agent/pusher.cpp
+++ b/be/src/agent/pusher.cpp
@@ -57,7 +57,7 @@ AgentStatus Pusher::init() {
}
// Check replica exist
- OLAPTablePtr olap_table;
+ TabletSharedPtr olap_table;
olap_table = _engine->get_table(
_push_req.tablet_id,
_push_req.schema_hash);
diff --git a/be/src/agent/task_worker_pool.cpp
b/be/src/agent/task_worker_pool.cpp
index 9cb9df17..cb464efb 100644
--- a/be/src/agent/task_worker_pool.cpp
+++ b/be/src/agent/task_worker_pool.cpp
@@ -1042,7 +1042,7 @@ void* TaskWorkerPool::_clone_worker_thread_callback(void*
arg_this) {
string src_file_path;
TBackend src_host;
// Check local tablet exist or not
- OLAPTablePtr tablet =
+ TabletSharedPtr tablet =
worker_pool_this->_env->olap_engine()->get_table(
clone_req.tablet_id, clone_req.schema_hash);
if (tablet.get() != NULL) {
@@ -2293,7 +2293,7 @@ AgentStatus TaskWorkerPool::_move_dir(
bool overwrite,
std::vector<std::string>* error_msgs) {
- OLAPTablePtr tablet = _env->olap_engine()->get_table(
+ TabletSharedPtr tablet = _env->olap_engine()->get_table(
tablet_id, schema_hash);
if (tablet.get() == NULL) {
LOG(INFO) << "failed to get tablet. tablet_id:" << tablet_id
diff --git a/be/src/exec/olap_meta_reader.cpp b/be/src/exec/olap_meta_reader.cpp
index ceeda76d..023304a9 100644
--- a/be/src/exec/olap_meta_reader.cpp
+++ b/be/src/exec/olap_meta_reader.cpp
@@ -43,7 +43,7 @@ Status EngineMetaReader::get_hints(
RuntimeProfile* profile) {
auto tablet_id = scan_range->scan_range().tablet_id;
int32_t schema_hash =
strtoul(scan_range->scan_range().schema_hash.c_str(), NULL, 10);
- OLAPTablePtr table = OLAPEngine::get_instance()->get_table(
+ TabletSharedPtr table = OLAPEngine::get_instance()->get_table(
tablet_id, schema_hash);
if (table.get() == NULL) {
LOG(WARNING) << "tablet does not exist. tablet_id=" << tablet_id << ",
schema_hash="
diff --git a/be/src/exec/olap_scanner.h b/be/src/exec/olap_scanner.h
index 78be097b..054c4832 100644
--- a/be/src/exec/olap_scanner.h
+++ b/be/src/exec/olap_scanner.h
@@ -115,7 +115,7 @@ class OlapScanner {
ReaderParams _params;
std::unique_ptr<Reader> _reader;
- OLAPTablePtr _olap_table;
+ TabletSharedPtr _olap_table;
int64_t _version;
std::vector<uint32_t> _return_columns;
diff --git a/be/src/http/action/meta_action.cpp
b/be/src/http/action/meta_action.cpp
index c4d7ad7b..c349d4f6 100644
--- a/be/src/http/action/meta_action.cpp
+++ b/be/src/http/action/meta_action.cpp
@@ -49,7 +49,7 @@ Status MetaAction::_handle_header(HttpRequest *req,
std::string* json_header) {
}
uint64_t tablet_id = std::stoull(req_tablet_id);
uint32_t schema_hash = std::stoul(req_schema_hash);
- OLAPTablePtr olap_table = OLAPEngine::get_instance()->get_table(tablet_id,
schema_hash);
+ TabletSharedPtr olap_table =
OLAPEngine::get_instance()->get_table(tablet_id, schema_hash);
if (olap_table == nullptr) {
LOG(WARNING) << "no tablet for tablet_id:" << tablet_id << " schema
hash:" << schema_hash;
return Status("no tablet exist");
diff --git a/be/src/http/action/restore_tablet_action.cpp
b/be/src/http/action/restore_tablet_action.cpp
index e2fbb3f0..b2c816fe 100644
--- a/be/src/http/action/restore_tablet_action.cpp
+++ b/be/src/http/action/restore_tablet_action.cpp
@@ -84,7 +84,7 @@ Status RestoreTabletAction::_handle(HttpRequest *req) {
int32_t schema_hash = std::atoi(schema_hash_str.c_str());
LOG(INFO) << "get restore tablet action request: " << tablet_id << "-" <<
schema_hash;
- OLAPTablePtr tablet =
+ TabletSharedPtr tablet =
OLAPEngine::get_instance()->get_table(tablet_id, schema_hash);
if (tablet.get() != nullptr) {
LOG(WARNING) << "find tablet. tablet_id=" << tablet_id << "
schema_hash=" << schema_hash;
diff --git a/be/src/olap/base_compaction.cpp b/be/src/olap/base_compaction.cpp
index 3128cb6f..32f8b602 100644
--- a/be/src/olap/base_compaction.cpp
+++ b/be/src/olap/base_compaction.cpp
@@ -40,7 +40,7 @@ using std::vector;
namespace doris {
-OLAPStatus BaseCompaction::init(OLAPTablePtr table, bool is_manual_trigger) {
+OLAPStatus BaseCompaction::init(TabletSharedPtr table, bool is_manual_trigger)
{
// 表在首次查询或PUSH等操作时，会被加载到内存
// 如果表没有被加载，表明该表上目前没有任何操作，所以不进行BE操作
if (!table->is_loaded()) {
diff --git a/be/src/olap/base_compaction.h b/be/src/olap/base_compaction.h
index f0fc4510..3d2e82f8 100644
--- a/be/src/olap/base_compaction.h
+++ b/be/src/olap/base_compaction.h
@@ -55,7 +55,7 @@ class BaseCompaction {
// 返回值：
// - 如果init执行成功，即可以执行BE，则返回OLAP_SUCCESS；
// - 其它情况下，返回相应的错误码
- OLAPStatus init(OLAPTablePtr table, bool is_manual_trigger = false);
+ OLAPStatus init(TabletSharedPtr table, bool is_manual_trigger = false);
// 执行BaseCompaction, 可能会持续很长时间
//
@@ -164,7 +164,7 @@ class BaseCompaction {
}
// 需要进行操作的Table指针
- OLAPTablePtr _table;
+ TabletSharedPtr _table;
// 新base的version
Version _new_base_version;
// 现有base的version
diff --git a/be/src/olap/cumulative_compaction.cpp
b/be/src/olap/cumulative_compaction.cpp
index ba43b87d..f2a57075 100755
--- a/be/src/olap/cumulative_compaction.cpp
+++ b/be/src/olap/cumulative_compaction.cpp
@@ -32,7 +32,7 @@ using std::vector;
namespace doris {
-OLAPStatus CumulativeCompaction::init(OLAPTablePtr table) {
+OLAPStatus CumulativeCompaction::init(TabletSharedPtr table) {
LOG(INFO) << "init cumulative compaction handler. [table=" <<
table->full_name() << "]";
if (_is_init) {
diff --git a/be/src/olap/cumulative_compaction.h
b/be/src/olap/cumulative_compaction.h
index c9a923e0..47394dc4 100755
--- a/be/src/olap/cumulative_compaction.h
+++ b/be/src/olap/cumulative_compaction.h
@@ -54,7 +54,7 @@ class CumulativeCompaction {
// 返回值：
// - 如果触发cumulative compaction，返回OLAP_SUCCESS
// - 否则，返回对应错误码
- OLAPStatus init(OLAPTablePtr table);
+ OLAPStatus init(TabletSharedPtr table);
// 执行cumulative compaction
//
@@ -164,7 +164,7 @@ class CumulativeCompaction {
// 当delta文件的大小超过该值时，我们认为该delta文件是cumulative文件
size_t _max_delta_file_size;
// 待执行cumulative compaction的olap table
- OLAPTablePtr _table;
+ TabletSharedPtr _table;
// 新cumulative文件的版本
Version _cumulative_version;
// 新cumulative文件的version hash
diff --git a/be/src/olap/data_writer.cpp b/be/src/olap/data_writer.cpp
index 3cf80120..35fecd9f 100644
--- a/be/src/olap/data_writer.cpp
+++ b/be/src/olap/data_writer.cpp
@@ -26,7 +26,7 @@
namespace doris {
-ColumnDataWriter* ColumnDataWriter::create(OLAPTablePtr table, SegmentGroup*
segment_group, bool is_push_write) {
+ColumnDataWriter* ColumnDataWriter::create(TabletSharedPtr table,
SegmentGroup* segment_group, bool is_push_write) {
ColumnDataWriter* writer = NULL;
switch (table->data_file_type()) {
case COLUMN_ORIENTED_FILE:
@@ -40,7 +40,7 @@ ColumnDataWriter* ColumnDataWriter::create(OLAPTablePtr
table, SegmentGroup* seg
return writer;
}
-ColumnDataWriter::ColumnDataWriter(OLAPTablePtr table, SegmentGroup*
segment_group, bool is_push_write)
+ColumnDataWriter::ColumnDataWriter(TabletSharedPtr table, SegmentGroup*
segment_group, bool is_push_write)
: _is_push_write(is_push_write),
_table(table),
_column_statistics(_table->num_key_fields(),
diff --git a/be/src/olap/data_writer.h b/be/src/olap/data_writer.h
index d27c006d..1f4356f1 100644
--- a/be/src/olap/data_writer.h
+++ b/be/src/olap/data_writer.h
@@ -31,8 +31,8 @@ class ColumnDataWriter {
public:
// Factory function
// 调用者获得新建的对象, 并负责delete释放
- static ColumnDataWriter* create(OLAPTablePtr table, SegmentGroup*
segment_group, bool is_push_write);
- ColumnDataWriter(OLAPTablePtr table, SegmentGroup* segment_group, bool
is_push_write);
+ static ColumnDataWriter* create(TabletSharedPtr table, SegmentGroup*
segment_group, bool is_push_write);
+ ColumnDataWriter(TabletSharedPtr table, SegmentGroup* segment_group, bool
is_push_write);
~ColumnDataWriter();
OLAPStatus init();
OLAPStatus attached_by(RowCursor* row_cursor);
@@ -50,7 +50,7 @@ class ColumnDataWriter {
OLAPStatus _init_segment();
bool _is_push_write;
- OLAPTablePtr _table;
+ TabletSharedPtr _table;
// first is min, second is max
std::vector<std::pair<WrapperField*, WrapperField*>> _column_statistics;
uint32_t _row_index;
diff --git a/be/src/olap/delete_handler.cpp b/be/src/olap/delete_handler.cpp
index 93b34d1d..5aee5ffd 100644
--- a/be/src/olap/delete_handler.cpp
+++ b/be/src/olap/delete_handler.cpp
@@ -50,7 +50,7 @@ namespace doris {
// 1. 删除条件的版本要不是当前最大的delta版本号，要不是最大的delta版本号加1
// 2. 删除条件中指定的列在table中存在，必须是key列，且不能是double，float类型
OLAPStatus DeleteConditionHandler::store_cond(
- OLAPTablePtr table,
+ TabletSharedPtr table,
const int32_t version,
const vector<TCondition>& conditions) {
if (conditions.size() == 0 || _check_version_valid(table, version) !=
OLAP_SUCCESS) {
@@ -109,7 +109,7 @@ string
DeleteConditionHandler::construct_sub_conditions(const TCondition& condit
// 删除指定版本号的删除条件；需要注意的是，如果table上没有任何删除条件，或者
// 指定版本号的删除条件不存在，也会返回OLAP_SUCCESS。
-OLAPStatus DeleteConditionHandler::delete_cond(OLAPTablePtr table,
+OLAPStatus DeleteConditionHandler::delete_cond(TabletSharedPtr table,
const int32_t version,
bool delete_smaller_version_conditions) {
if (version < 0) {
@@ -157,7 +157,7 @@ OLAPStatus DeleteConditionHandler::delete_cond(OLAPTablePtr
table,
return OLAP_SUCCESS;
}
-OLAPStatus DeleteConditionHandler::log_conds(OLAPTablePtr table) {
+OLAPStatus DeleteConditionHandler::log_conds(TabletSharedPtr table) {
LOG(INFO) << "display all delete condition. tablet=" << table->full_name();
table->obtain_header_rdlock();
const del_cond_array& delete_conditions = table->delete_data_conditions();
@@ -181,7 +181,7 @@ OLAPStatus DeleteConditionHandler::log_conds(OLAPTablePtr
table) {
}
OLAPStatus DeleteConditionHandler::check_condition_valid(
- OLAPTablePtr table,
+ TabletSharedPtr table,
const TCondition& cond) {
// 检查指定列名的列是否存在
int field_index = table->get_field_index(cond.column_name);
@@ -257,7 +257,7 @@ OLAPStatus DeleteConditionHandler::check_condition_valid(
}
}
-OLAPStatus DeleteConditionHandler::_check_version_valid(OLAPTablePtr table,
+OLAPStatus DeleteConditionHandler::_check_version_valid(TabletSharedPtr table,
const int32_t filter_version) {
// 找到当前最大的delta文件版本号
vector<Version> all_file_versions;
@@ -280,7 +280,7 @@ OLAPStatus
DeleteConditionHandler::_check_version_valid(OLAPTablePtr table,
}
}
-int DeleteConditionHandler::_check_whether_condition_exist(OLAPTablePtr table,
int cond_version) {
+int DeleteConditionHandler::_check_whether_condition_exist(TabletSharedPtr
table, int cond_version) {
const del_cond_array& delete_conditions = table->delete_data_conditions();
if (delete_conditions.size() == 0) {
@@ -334,7 +334,7 @@ bool DeleteHandler::_parse_condition(const std::string&
condition_str, TConditio
return true;
}
-OLAPStatus DeleteHandler::init(OLAPTablePtr olap_table, int32_t version) {
+OLAPStatus DeleteHandler::init(TabletSharedPtr olap_table, int32_t version) {
if (_is_inited) {
OLAP_LOG_WARNING("reintialize delete handler.");
return OLAP_ERR_INIT_FAILED;
diff --git a/be/src/olap/delete_handler.h b/be/src/olap/delete_handler.h
index f78c16ac..7b69939e 100644
--- a/be/src/olap/delete_handler.h
+++ b/be/src/olap/delete_handler.h
@@ -55,7 +55,7 @@ class DeleteConditionHandler {
// 检查cond表示的删除条件是否符合要求；
// 如果不符合要求，返回OLAP_ERR_DELETE_INVALID_CONDITION；符合要求返回OLAP_SUCCESS
- OLAPStatus check_condition_valid(OLAPTablePtr table, const TCondition&
cond);
+ OLAPStatus check_condition_valid(TabletSharedPtr table, const TCondition&
cond);
// 存储指定版本号的删除条件到Header文件中。因此，调用之前需要对Header文件加写锁
//
@@ -68,7 +68,7 @@ class DeleteConditionHandler {
// * OLAP_ERR_DELETE_INVALID_PARAMETERS：函数参数不符合要求
// * OLAP_ERR_DELETE_INVALID_CONDITION：del_condition不符合要求
OLAPStatus store_cond(
- OLAPTablePtr table,
+ TabletSharedPtr table,
const int32_t version,
const std::vector<TCondition>& conditions);
@@ -90,7 +90,7 @@ class DeleteConditionHandler {
// * 这个表没有指定版本号的删除条件
// * OLAP_ERR_DELETE_INVALID_PARAMETERS：函数参数不符合要求
OLAPStatus delete_cond(
- OLAPTablePtr table, const int32_t version, bool
delete_smaller_version_conditions);
+ TabletSharedPtr table, const int32_t version, bool
delete_smaller_version_conditions);
// 将一个olap engine的表上存有的所有删除条件打印到log中。调用前只需要给Header文件加读锁
//
@@ -98,15 +98,15 @@ class DeleteConditionHandler {
// table: 要打印删除条件的olap engine表
// 返回值：
// OLAP_SUCCESS：调用成功
- OLAPStatus log_conds(OLAPTablePtr table);
+ OLAPStatus log_conds(TabletSharedPtr table);
private:
// 检查指定的删除条件版本是否符合要求；
// 如果不符合要求，返回OLAP_ERR_DELETE_INVALID_VERSION；符合要求返回OLAP_SUCCESS
- OLAPStatus _check_version_valid(OLAPTablePtr table, const int32_t
filter_version);
+ OLAPStatus _check_version_valid(TabletSharedPtr table, const int32_t
filter_version);
// 检查指定版本的删除条件是否已经存在。如果存在，返回指定版本删除条件的数组下标；不存在返回-1
- int _check_whether_condition_exist(OLAPTablePtr, int cond_version);
+ int _check_whether_condition_exist(TabletSharedPtr, int cond_version);
};
// 表示一个删除条件
@@ -154,7 +154,7 @@ class DeleteHandler {
// * OLAP_SUCCESS: 调用成功
// * OLAP_ERR_DELETE_INVALID_PARAMETERS: 参数不符合要求
// * OLAP_ERR_MALLOC_ERROR: 在填充_del_conds时，分配内存失败
- OLAPStatus init(OLAPTablePtr olap_table, int32_t version);
+ OLAPStatus init(TabletSharedPtr olap_table, int32_t version);
// 判定一条数据是否符合删除条件
//
diff --git a/be/src/olap/delta_writer.h b/be/src/olap/delta_writer.h
index 4778893f..539941ef 100644
--- a/be/src/olap/delta_writer.h
+++ b/be/src/olap/delta_writer.h
@@ -65,11 +65,11 @@ class DeltaWriter {
bool _is_init = false;
WriteRequest _req;
- OLAPTablePtr _table;
+ TabletSharedPtr _table;
SegmentGroup* _cur_segment_group;
std::vector<SegmentGroup*> _segment_group_vec;
std::vector<SegmentGroup*> _new_segment_group_vec;
- OLAPTablePtr _new_table;
+ TabletSharedPtr _new_table;
ColumnDataWriter* _writer;
MemTable* _mem_table;
Schema* _schema;
diff --git a/be/src/olap/merger.cpp b/be/src/olap/merger.cpp
index 3a84b0b8..3d218011 100644
--- a/be/src/olap/merger.cpp
+++ b/be/src/olap/merger.cpp
@@ -35,7 +35,7 @@ using std::vector;
namespace doris {
-Merger::Merger(OLAPTablePtr table, SegmentGroup* segment_group, ReaderType
type) :
+Merger::Merger(TabletSharedPtr table, SegmentGroup* segment_group, ReaderType
type) :
_table(table),
_segment_group(segment_group),
_reader_type(type),
diff --git a/be/src/olap/merger.h b/be/src/olap/merger.h
index e47469fd..38672351 100644
--- a/be/src/olap/merger.h
+++ b/be/src/olap/merger.h
@@ -29,7 +29,7 @@ class ColumnData;
class Merger {
public:
// parameter index is created by caller, and it is empty.
- Merger(OLAPTablePtr table, SegmentGroup* index, ReaderType type);
+ Merger(TabletSharedPtr table, SegmentGroup* index, ReaderType type);
virtual ~Merger() {};
@@ -44,7 +44,7 @@ class Merger {
return _row_count;
}
private:
- OLAPTablePtr _table;
+ TabletSharedPtr _table;
SegmentGroup* _segment_group;
ReaderType _reader_type;
uint64_t _row_count;
diff --git a/be/src/olap/olap_cond.h b/be/src/olap/olap_cond.h
index 8d923e9a..0ab3d45a 100644
--- a/be/src/olap/olap_cond.h
+++ b/be/src/olap/olap_cond.h
@@ -89,7 +89,7 @@ struct Cond {
// 所有归属于同一列上的条件二元组，聚合在一个CondColumn上
class CondColumn {
public:
- CondColumn(OLAPTablePtr table, int32_t index) : _col_index(index),
_table(table) {
+ CondColumn(TabletSharedPtr table, int32_t index) : _col_index(index),
_table(table) {
_conds.clear();
_is_key = _table->tablet_schema()[_col_index].is_key;
}
@@ -120,7 +120,7 @@ class CondColumn {
bool _is_key;
int32_t _col_index;
std::vector<Cond*> _conds;
- OLAPTablePtr _table;
+ TabletSharedPtr _table;
};
// 一次请求所关联的条件
@@ -139,7 +139,7 @@ class Conditions {
_columns.clear();
}
- void set_table(OLAPTablePtr table) {
+ void set_table(TabletSharedPtr table) {
long do_not_remove_me_until_you_want_a_heart_attacking =
table.use_count();
OLAP_UNUSED_ARG(do_not_remove_me_until_you_want_a_heart_attacking);
@@ -164,7 +164,7 @@ class Conditions {
}
private:
- OLAPTablePtr _table; // ref to OLAPTable to access schema
+ TabletSharedPtr _table; // ref to OLAPTable to access schema
CondColumns _columns; // list of condition column
};
diff --git a/be/src/olap/olap_engine.cpp b/be/src/olap/olap_engine.cpp
index b10e5d06..5051c90e 100644
--- a/be/src/olap/olap_engine.cpp
+++ b/be/src/olap/olap_engine.cpp
@@ -78,7 +78,7 @@ const std::string HTTP_REQUEST_FILE_PARAM = "&file=";
const uint32_t DOWNLOAD_FILE_MAX_RETRY = 3;
const uint32_t LIST_REMOTE_FILE_TIMEOUT = 15;
-bool _sort_table_by_create_time(const OLAPTablePtr& a, const OLAPTablePtr& b) {
+bool _sort_table_by_create_time(const TabletSharedPtr& a, const
TabletSharedPtr& b) {
return a->creation_time() < b->creation_time();
}
@@ -753,11 +753,11 @@ OLAPStatus OLAPEngine::clear() {
return OLAP_SUCCESS;
}
-OLAPTablePtr OLAPEngine::_get_table_with_no_lock(TTabletId tablet_id,
SchemaHash schema_hash) {
+TabletSharedPtr OLAPEngine::_get_table_with_no_lock(TTabletId tablet_id,
SchemaHash schema_hash) {
VLOG(3) << "begin to get olap table. tablet_id=" << tablet_id;
tablet_map_t::iterator it = _tablet_map.find(tablet_id);
if (it != _tablet_map.end()) {
- for (OLAPTablePtr table : it->second.table_arr) {
+ for (TabletSharedPtr table : it->second.table_arr) {
if (table->equal(tablet_id, schema_hash)) {
VLOG(3) << "get olap table success. tablet_id=" << tablet_id;
return table;
@@ -767,13 +767,13 @@ OLAPTablePtr
OLAPEngine::_get_table_with_no_lock(TTabletId tablet_id, SchemaHash
VLOG(3) << "fail to get olap table. tablet_id=" << tablet_id;
// Return empty olap_table if fail
- OLAPTablePtr olap_table;
+ TabletSharedPtr olap_table;
return olap_table;
}
-OLAPTablePtr OLAPEngine::get_table(TTabletId tablet_id, SchemaHash
schema_hash, bool load_table) {
+TabletSharedPtr OLAPEngine::get_table(TTabletId tablet_id, SchemaHash
schema_hash, bool load_table) {
_tablet_map_lock.rdlock();
- OLAPTablePtr olap_table;
+ TabletSharedPtr olap_table;
olap_table = _get_table_with_no_lock(tablet_id, schema_hash);
_tablet_map_lock.unlock();
@@ -794,14 +794,14 @@ OLAPTablePtr OLAPEngine::get_table(TTabletId tablet_id,
SchemaHash schema_hash,
OLAPStatus OLAPEngine::get_tables_by_id(
TTabletId tablet_id,
- list<OLAPTablePtr>* table_list) {
+ list<TabletSharedPtr>* table_list) {
OLAPStatus res = OLAP_SUCCESS;
VLOG(3) << "begin to get tables by id. tablet_id=" << tablet_id;
_tablet_map_lock.rdlock();
tablet_map_t::iterator it = _tablet_map.find(tablet_id);
if (it != _tablet_map.end()) {
- for (OLAPTablePtr olap_table : it->second.table_arr) {
+ for (TabletSharedPtr olap_table : it->second.table_arr) {
table_list->push_back(olap_table);
}
}
@@ -812,7 +812,7 @@ OLAPStatus OLAPEngine::get_tables_by_id(
return OLAP_ERR_TABLE_NOT_FOUND;
}
- for (std::list<OLAPTablePtr>::iterator it = table_list->begin();
+ for (std::list<TabletSharedPtr>::iterator it = table_list->begin();
it != table_list->end();) {
if (!(*it)->is_loaded()) {
if ((*it)->load() != OLAP_SUCCESS) {
@@ -843,7 +843,7 @@ bool OLAPEngine::check_tablet_id_exist(TTabletId tablet_id)
{
}
OLAPStatus OLAPEngine::add_table(TTabletId tablet_id, SchemaHash schema_hash,
- const OLAPTablePtr& table, bool force) {
+ const TabletSharedPtr& table, bool force) {
OLAPStatus res = OLAP_SUCCESS;
VLOG(3) << "begin to add olap table to OLAPEngine. "
<< "tablet_id=" << tablet_id << ", schema_hash=" << schema_hash
@@ -852,8 +852,8 @@ OLAPStatus OLAPEngine::add_table(TTabletId tablet_id,
SchemaHash schema_hash,
table->set_id(_global_table_id++);
- OLAPTablePtr table_item;
- for (OLAPTablePtr item : _tablet_map[tablet_id].table_arr) {
+ TabletSharedPtr table_item;
+ for (TabletSharedPtr item : _tablet_map[tablet_id].table_arr) {
if (item->equal(tablet_id, schema_hash)) {
table_item = item;
break;
@@ -967,7 +967,7 @@ void OLAPEngine::delete_transaction(
// delete transaction from tablet
if (delete_from_tablet) {
- OLAPTablePtr tablet = get_table(tablet_info.tablet_id,
tablet_info.schema_hash);
+ TabletSharedPtr tablet = get_table(tablet_info.tablet_id,
tablet_info.schema_hash);
if (tablet.get() != nullptr) {
tablet->delete_pending_data(transaction_id);
}
@@ -975,7 +975,7 @@ void OLAPEngine::delete_transaction(
}
}
-void OLAPEngine::get_transactions_by_tablet(OLAPTablePtr tablet, int64_t*
partition_id,
+void OLAPEngine::get_transactions_by_tablet(TabletSharedPtr tablet, int64_t*
partition_id,
set<int64_t>* transaction_ids) {
if (tablet.get() == nullptr || partition_id == nullptr || transaction_ids
== nullptr) {
OLAP_LOG_WARNING("parameter is null when get transactions by tablet");
@@ -1048,7 +1048,7 @@ OLAPStatus OLAPEngine::publish_version(const
TPublishVersionRequest& publish_ver
<< ", version=" << version.first
<< ", version_hash=" << version_hash
<< ", transaction_id=" << transaction_id;
- OLAPTablePtr tablet = get_table(tablet_info.tablet_id,
tablet_info.schema_hash);
+ TabletSharedPtr tablet = get_table(tablet_info.tablet_id,
tablet_info.schema_hash);
if (tablet.get() == NULL) {
OLAP_LOG_WARNING("can't get table when publish version.
[tablet_id=%ld schema_hash=%d]",
@@ -1136,7 +1136,7 @@ void OLAPEngine::clear_transaction_task(const
TTransactionId transaction_id,
LOG(INFO) << "finish to clear transaction task. transaction_id=" <<
transaction_id;
}
-OLAPStatus OLAPEngine::clone_incremental_data(OLAPTablePtr tablet, OLAPHeader&
clone_header,
+OLAPStatus OLAPEngine::clone_incremental_data(TabletSharedPtr tablet,
OLAPHeader& clone_header,
int64_t committed_version) {
LOG(INFO) << "begin to incremental clone. tablet=" << tablet->full_name()
<< ", committed_version=" << committed_version;
@@ -1207,7 +1207,7 @@ OLAPStatus
OLAPEngine::clone_incremental_data(OLAPTablePtr tablet, OLAPHeader& c
return clone_res;
}
-OLAPStatus OLAPEngine::clone_full_data(OLAPTablePtr tablet, OLAPHeader&
clone_header) {
+OLAPStatus OLAPEngine::clone_full_data(TabletSharedPtr tablet, OLAPHeader&
clone_header) {
Version clone_latest_version = clone_header.get_latest_version();
LOG(INFO) << "begin to full clone. table=" << tablet->full_name() << ","
<< "clone_latest_version=" << clone_latest_version.first << "-" <<
clone_latest_version.second;
@@ -1303,7 +1303,7 @@ OLAPStatus OLAPEngine::drop_table(
// Get table which need to be droped
_tablet_map_lock.rdlock();
- OLAPTablePtr dropped_table = _get_table_with_no_lock(tablet_id,
schema_hash);
+ TabletSharedPtr dropped_table = _get_table_with_no_lock(tablet_id,
schema_hash);
_tablet_map_lock.unlock();
if (dropped_table.get() == NULL) {
OLAP_LOG_WARNING("fail to drop not existed table. [tablet_id=%ld
schema_hash=%d]",
@@ -1334,7 +1334,7 @@ OLAPStatus OLAPEngine::drop_table(
bool is_drop_base_table = false;
_tablet_map_lock.rdlock();
- OLAPTablePtr related_table = _get_table_with_no_lock(
+ TabletSharedPtr related_table = _get_table_with_no_lock(
related_tablet_id, related_schema_hash);
_tablet_map_lock.unlock();
if (related_table.get() == NULL) {
@@ -1389,14 +1389,14 @@ OLAPStatus OLAPEngine::_drop_table_directly_unlocked(
TTabletId tablet_id, SchemaHash schema_hash, bool keep_files) {
OLAPStatus res = OLAP_SUCCESS;
- OLAPTablePtr dropped_table = _get_table_with_no_lock(tablet_id,
schema_hash);
+ TabletSharedPtr dropped_table = _get_table_with_no_lock(tablet_id,
schema_hash);
if (dropped_table.get() == NULL) {
OLAP_LOG_WARNING("fail to drop not existed table. [tablet_id=%ld
schema_hash=%d]",
tablet_id, schema_hash);
return OLAP_ERR_TABLE_NOT_FOUND;
}
- for (list<OLAPTablePtr>::iterator it =
_tablet_map[tablet_id].table_arr.begin();
+ for (list<TabletSharedPtr>::iterator it =
_tablet_map[tablet_id].table_arr.begin();
it != _tablet_map[tablet_id].table_arr.end();) {
if ((*it)->equal(tablet_id, schema_hash)) {
if (!keep_files) {
@@ -1432,13 +1432,13 @@ OLAPStatus OLAPEngine::drop_tables_on_error_root_path(
TSchemaHash schema_hash = tablet_info.schema_hash;
VLOG(3) << "drop_table begin. tablet_id=" << tablet_id
<< ", schema_hash=" << schema_hash;
- OLAPTablePtr dropped_table = _get_table_with_no_lock(tablet_id,
schema_hash);
+ TabletSharedPtr dropped_table = _get_table_with_no_lock(tablet_id,
schema_hash);
if (dropped_table.get() == NULL) {
OLAP_LOG_WARNING("dropping table not exist. [table=%ld
schema_hash=%d]",
tablet_id, schema_hash);
continue;
} else {
- for (list<OLAPTablePtr>::iterator it =
_tablet_map[tablet_id].table_arr.begin();
+ for (list<TabletSharedPtr>::iterator it =
_tablet_map[tablet_id].table_arr.begin();
it != _tablet_map[tablet_id].table_arr.end();) {
if ((*it)->equal(tablet_id, schema_hash)) {
it = _tablet_map[tablet_id].table_arr.erase(it);
@@ -1458,9 +1458,9 @@ OLAPStatus OLAPEngine::drop_tables_on_error_root_path(
return res;
}
-OLAPTablePtr OLAPEngine::create_table(
+TabletSharedPtr OLAPEngine::create_table(
const TCreateTabletReq& request, const string* ref_root_path,
- const bool is_schema_change_table, const OLAPTablePtr ref_olap_table) {
+ const bool is_schema_change_table, const TabletSharedPtr
ref_olap_table) {
// Get all available stores, use ref_root_path if the caller specified
std::vector<OlapStore*> stores;
if (ref_root_path == nullptr) {
@@ -1473,7 +1473,7 @@ OLAPTablePtr OLAPEngine::create_table(
stores.push_back(ref_olap_table->store());
}
- OLAPTablePtr olap_table;
+ TabletSharedPtr olap_table;
// Try to create table on each of all_available_root_path, util success
for (auto& store : stores) {
OLAPHeader* header = new OLAPHeader();
@@ -1505,7 +1505,7 @@ OLAPStatus OLAPEngine::create_init_version(TTabletId
tablet_id, SchemaHash schem
Version version, VersionHash
version_hash) {
VLOG(3) << "begin to create init version. "
<< "begin=" << version.first << ", end=" << version.second;
- OLAPTablePtr table;
+ TabletSharedPtr table;
ColumnDataWriter* writer = NULL;
SegmentGroup* new_segment_group = NULL;
OLAPStatus res = OLAP_SUCCESS;
@@ -1621,7 +1621,7 @@ void OLAPEngine::release_schema_change_lock(TTabletId
tablet_id) {
VLOG(3) << "release_schema_change_lock end. tablet_id=" << tablet_id;
}
-void OLAPEngine::_build_tablet_info(OLAPTablePtr olap_table, TTabletInfo*
tablet_info) {
+void OLAPEngine::_build_tablet_info(TabletSharedPtr olap_table, TTabletInfo*
tablet_info) {
tablet_info->tablet_id = olap_table->tablet_id();
tablet_info->schema_hash = olap_table->schema_hash();
@@ -1658,7 +1658,7 @@ OLAPStatus OLAPEngine::report_tablet_info(TTabletInfo*
tablet_info) {
OLAPStatus res = OLAP_SUCCESS;
- OLAPTablePtr olap_table = get_table(
+ TabletSharedPtr olap_table = get_table(
tablet_info->tablet_id, tablet_info->schema_hash);
if (olap_table.get() == NULL) {
OLAP_LOG_WARNING("can't find table. [table=%ld schema_hash=%d]",
@@ -1686,7 +1686,7 @@ OLAPStatus
OLAPEngine::report_all_tablets_info(std::map<TTabletId, TTablet>* tab
}
TTablet tablet;
- for (OLAPTablePtr olap_table : item.second.table_arr) {
+ for (TabletSharedPtr olap_table : item.second.table_arr) {
if (olap_table.get() == NULL) {
continue;
}
@@ -1747,7 +1747,7 @@ void OLAPEngine::_build_tablet_stat() {
TTabletStat stat;
stat.tablet_id = item.first;
- for (OLAPTablePtr olap_table : item.second.table_arr) {
+ for (TabletSharedPtr olap_table : item.second.table_arr) {
if (olap_table.get() == NULL) {
continue;
}
@@ -1767,7 +1767,7 @@ void OLAPEngine::_build_tablet_stat() {
_tablet_stat_cache_update_time_ms = UnixMillis();
}
-bool OLAPEngine::_can_do_compaction(OLAPTablePtr table) {
+bool OLAPEngine::_can_do_compaction(TabletSharedPtr table) {
// 如果table正在做schema change，则通过选路判断数据是否转换完成
// 如果选路成功，则转换完成，可以进行BE
// 如果选路失败，则转换未完成，不能进行BE
@@ -1798,7 +1798,7 @@ void OLAPEngine::start_clean_fd_cache() {
}
void OLAPEngine::perform_cumulative_compaction() {
- OLAPTablePtr best_table =
_find_best_tablet_to_compaction(CompactionType::CUMULATIVE_COMPACTION);
+ TabletSharedPtr best_table =
_find_best_tablet_to_compaction(CompactionType::CUMULATIVE_COMPACTION);
if (best_table == nullptr) { return; }
CumulativeCompaction cumulative_compaction;
@@ -1816,7 +1816,7 @@ void OLAPEngine::perform_cumulative_compaction() {
}
void OLAPEngine::perform_base_compaction() {
- OLAPTablePtr best_table =
_find_best_tablet_to_compaction(CompactionType::BASE_COMPACTION);
+ TabletSharedPtr best_table =
_find_best_tablet_to_compaction(CompactionType::BASE_COMPACTION);
if (best_table == nullptr) { return; }
BaseCompaction base_compaction;
@@ -1834,12 +1834,12 @@ void OLAPEngine::perform_base_compaction() {
}
}
-OLAPTablePtr OLAPEngine::_find_best_tablet_to_compaction(CompactionType
compaction_type) {
+TabletSharedPtr OLAPEngine::_find_best_tablet_to_compaction(CompactionType
compaction_type) {
ReadLock tablet_map_rdlock(&_tablet_map_lock);
uint32_t highest_score = 0;
- OLAPTablePtr best_table;
+ TabletSharedPtr best_table;
for (tablet_map_t::value_type& table_ins : _tablet_map){
- for (OLAPTablePtr& table_ptr : table_ins.second.table_arr) {
+ for (TabletSharedPtr& table_ptr : table_ins.second.table_arr) {
if (!table_ptr->is_loaded() || !_can_do_compaction(table_ptr)) {
continue;
}
@@ -1917,7 +1917,7 @@ OLAPStatus OLAPEngine::start_trash_sweep(double* usage) {
// clear expire incremental segment_group
_tablet_map_lock.rdlock();
for (const auto& item : _tablet_map) {
- for (OLAPTablePtr olap_table : item.second.table_arr) {
+ for (TabletSharedPtr olap_table : item.second.table_arr) {
if (olap_table.get() == NULL) {
continue;
}
@@ -1973,7 +1973,7 @@ OLAPStatus OLAPEngine::_create_new_table_header(
const TCreateTabletReq& request,
OlapStore* store,
const bool is_schema_change_table,
- const OLAPTablePtr ref_olap_table,
+ const TabletSharedPtr ref_olap_table,
OLAPHeader* header) {
uint64_t shard = 0;
OLAPStatus res = store->get_shard(&shard);
@@ -2157,9 +2157,9 @@ void OLAPEngine::_cancel_unfinished_schema_change() {
AlterTabletType type;
for (const auto& tablet_instance : _tablet_map) {
- for (OLAPTablePtr olap_table : tablet_instance.second.table_arr) {
+ for (TabletSharedPtr olap_table : tablet_instance.second.table_arr) {
if (olap_table.get() == NULL) {
- OLAP_LOG_WARNING("get empty OLAPTablePtr. [tablet_id=%ld]",
tablet_instance.first);
+ OLAP_LOG_WARNING("get empty TabletSharedPtr. [tablet_id=%ld]",
tablet_instance.first);
continue;
}
@@ -2169,7 +2169,7 @@ void OLAPEngine::_cancel_unfinished_schema_change() {
continue;
}
- OLAPTablePtr new_olap_table = get_table(tablet_id, schema_hash,
false);
+ TabletSharedPtr new_olap_table = get_table(tablet_id, schema_hash,
false);
if (new_olap_table.get() == NULL) {
OLAP_LOG_WARNING("the table referenced by schema change cannot
be found. "
"schema change cancelled. [tablet='%s']",
@@ -2228,7 +2228,7 @@ void OLAPEngine::add_unused_index(SegmentGroup*
segment_group) {
}
OLAPStatus OLAPEngine::_create_init_version(
- OLAPTablePtr olap_table, const TCreateTabletReq& request) {
+ TabletSharedPtr olap_table, const TCreateTabletReq& request) {
OLAPStatus res = OLAP_SUCCESS;
if (request.version < 1) {
@@ -2281,7 +2281,7 @@ OLAPStatus OLAPEngine::create_table(const
TCreateTabletReq& request) {
// return success if table with same tablet_id and schema_hash exist,
// false if table with same tablet_id but different schema_hash
exist
if (check_tablet_id_exist(request.tablet_id)) {
- OLAPTablePtr table = get_table(
+ TabletSharedPtr table = get_table(
request.tablet_id, request.tablet_schema.schema_hash);
if (table.get() != NULL) {
LOG(INFO) << "create table success for table already exist.";
@@ -2296,7 +2296,7 @@ OLAPStatus OLAPEngine::create_table(const
TCreateTabletReq& request) {
static Mutex create_table_lock;
MutexLock auto_lock(&create_table_lock);
- OLAPTablePtr olap_table;
+ TabletSharedPtr olap_table;
do {
// 3. Create table with only header, no deltas
olap_table = create_table(request, NULL, false, NULL);
@@ -2315,7 +2315,7 @@ OLAPStatus OLAPEngine::create_table(const
TCreateTabletReq& request) {
}
is_table_added = true;
- OLAPTablePtr olap_table_ptr = get_table(
+ TabletSharedPtr olap_table_ptr = get_table(
request.tablet_id, request.tablet_schema.schema_hash);
if (olap_table_ptr.get() == NULL) {
res = OLAP_ERR_TABLE_NOT_FOUND;
@@ -2421,7 +2421,7 @@ AlterTableStatus OLAPEngine::show_alter_table_status(
AlterTableStatus status = ALTER_TABLE_FINISHED;
- OLAPTablePtr table = OLAPEngine::get_instance()->get_table(tablet_id,
schema_hash);
+ TabletSharedPtr table = OLAPEngine::get_instance()->get_table(tablet_id,
schema_hash);
if (table.get() == NULL) {
OLAP_LOG_WARNING("fail to get table. [table=%ld schema_hash=%d]",
tablet_id, schema_hash);
@@ -2450,7 +2450,7 @@ OLAPStatus OLAPEngine::compute_checksum(
return OLAP_ERR_CE_CMD_PARAMS_ERROR;
}
- OLAPTablePtr tablet = get_table(tablet_id, schema_hash);
+ TabletSharedPtr tablet = get_table(tablet_id, schema_hash);
if (NULL == tablet.get()) {
OLAP_LOG_WARNING("can't find tablet. [tablet_id=%ld schema_hash=%d]",
tablet_id, schema_hash);
@@ -2534,7 +2534,7 @@ OLAPStatus OLAPEngine::cancel_delete(const
TCancelDeleteDataReq& request) {
OLAPStatus res = OLAP_SUCCESS;
// 1. Get all tablets with same tablet_id
- list<OLAPTablePtr> table_list;
+ list<TabletSharedPtr> table_list;
res = get_tables_by_id(request.tablet_id, &table_list);
if (res != OLAP_SUCCESS) {
OLAP_LOG_WARNING("can't find table. [table=%ld]", request.tablet_id);
@@ -2543,7 +2543,7 @@ OLAPStatus OLAPEngine::cancel_delete(const
TCancelDeleteDataReq& request) {
// 2. Remove delete conditions from each tablet.
DeleteConditionHandler cond_handler;
- for (OLAPTablePtr temp_table : table_list) {
+ for (TabletSharedPtr temp_table : table_list) {
temp_table->obtain_header_wrlock();
res = cond_handler.delete_cond(temp_table, request.version, false);
if (res != OLAP_SUCCESS) {
@@ -2564,7 +2564,7 @@ OLAPStatus OLAPEngine::cancel_delete(const
TCancelDeleteDataReq& request) {
}
// Show delete conditions in tablet header.
- for (OLAPTablePtr table : table_list) {
+ for (TabletSharedPtr table : table_list) {
cond_handler.log_conds(table);
}
@@ -2586,7 +2586,7 @@ OLAPStatus OLAPEngine::delete_data(
}
// 1. Get all tablets with same tablet_id
- OLAPTablePtr table = get_table(request.tablet_id, request.schema_hash);
+ TabletSharedPtr table = get_table(request.tablet_id, request.schema_hash);
if (table.get() == NULL) {
OLAP_LOG_WARNING("can't find table. [table=%ld schema_hash=%d]",
request.tablet_id, request.schema_hash);
@@ -2615,7 +2615,7 @@ OLAPStatus OLAPEngine::delete_data(
OLAPStatus OLAPEngine::recover_tablet_until_specfic_version(
const TRecoverTabletReq& recover_tablet_req) {
- OLAPTablePtr table = get_table(recover_tablet_req.tablet_id,
+ TabletSharedPtr table = get_table(recover_tablet_req.tablet_id,
recover_tablet_req.schema_hash);
if (table == nullptr) { return OLAP_ERR_TABLE_NOT_FOUND; }
RETURN_NOT_OK(table->recover_tablet_until_specfic_version(recover_tablet_req.version,
@@ -2623,7 +2623,7 @@ OLAPStatus
OLAPEngine::recover_tablet_until_specfic_version(
return OLAP_SUCCESS;
}
-string OLAPEngine::get_info_before_incremental_clone(OLAPTablePtr tablet,
+string OLAPEngine::get_info_before_incremental_clone(TabletSharedPtr tablet,
int64_t committed_version, vector<Version>* missing_versions) {
// get missing versions
@@ -2649,7 +2649,7 @@ string
OLAPEngine::get_info_before_incremental_clone(OLAPTablePtr tablet,
return tablet->tablet_path() + CLONE_PREFIX;
}
-OLAPStatus OLAPEngine::finish_clone(OLAPTablePtr tablet, const string&
clone_dir,
+OLAPStatus OLAPEngine::finish_clone(TabletSharedPtr tablet, const string&
clone_dir,
int64_t committed_version, bool
is_incremental_clone) {
OLAPStatus res = OLAP_SUCCESS;
vector<string> linked_success_files;
@@ -2854,7 +2854,7 @@ OLAPStatus OLAPEngine::clear_alter_task(const TTabletId
tablet_id,
const TSchemaHash schema_hash) {
LOG(INFO) << "begin to process clear alter task. tablet_id=" << tablet_id
<< ", schema_hash=" << schema_hash;
- OLAPTablePtr tablet = get_table(tablet_id, schema_hash);
+ TabletSharedPtr tablet = get_table(tablet_id, schema_hash);
if (tablet.get() == NULL) {
OLAP_LOG_WARNING("can't find tablet when process clear alter task. ",
"[tablet_id=%ld, schema_hash=%d]", tablet_id,
schema_hash);
@@ -2891,7 +2891,7 @@ OLAPStatus OLAPEngine::clear_alter_task(const TTabletId
tablet_id,
tablet->release_header_lock();
// clear related tablet's schema change info
- OLAPTablePtr related_table = get_table(related_tablet_id,
related_schema_hash);
+ TabletSharedPtr related_table = get_table(related_tablet_id,
related_schema_hash);
if (related_table.get() == NULL) {
OLAP_LOG_WARNING("related table not found when process clear alter
task. "
"[tablet_id=%ld schema_hash=%d "
@@ -2929,7 +2929,7 @@ OLAPStatus OLAPEngine::push(
return OLAP_ERR_CE_CMD_PARAMS_ERROR;
}
- OLAPTablePtr olap_table = OLAPEngine::get_instance()->get_table(
+ TabletSharedPtr olap_table = OLAPEngine::get_instance()->get_table(
request.tablet_id, request.schema_hash);
if (NULL == olap_table.get()) {
OLAP_LOG_WARNING("false to find table. [table=%ld schema_hash=%d]",
diff --git a/be/src/olap/olap_engine.h b/be/src/olap/olap_engine.h
index 9dc333c3..21dad561 100644
--- a/be/src/olap/olap_engine.h
+++ b/be/src/olap/olap_engine.h
@@ -86,9 +86,9 @@ class OLAPEngine {
}
// Get table pointer
- OLAPTablePtr get_table(TTabletId tablet_id, SchemaHash schema_hash, bool
load_table = true);
+ TabletSharedPtr get_table(TTabletId tablet_id, SchemaHash schema_hash,
bool load_table = true);
- OLAPStatus get_tables_by_id(TTabletId tablet_id, std::list<OLAPTablePtr>*
table_list);
+ OLAPStatus get_tables_by_id(TTabletId tablet_id,
std::list<TabletSharedPtr>* table_list);
bool check_tablet_id_exist(TTabletId tablet_id);
@@ -97,10 +97,10 @@ class OLAPEngine {
// Create new table for OLAPEngine
//
// Return OLAPTable * succeeded; Otherwise, return NULL if failed
- OLAPTablePtr create_table(const TCreateTabletReq& request,
+ TabletSharedPtr create_table(const TCreateTabletReq& request,
const std::string* ref_root_path,
const bool is_schema_change_table,
- const OLAPTablePtr ref_olap_table);
+ const TabletSharedPtr ref_olap_table);
// Add a table pointer to OLAPEngine
// If force, drop the existing table add this new one
@@ -109,7 +109,7 @@ class OLAPEngine {
// OLAP_ERR_TABLE_INSERT_DUPLICATION_ERROR, if find duplication
// OLAP_ERR_NOT_INITED, if not inited
OLAPStatus add_table(TTabletId tablet_id, SchemaHash schema_hash,
- const OLAPTablePtr& table, bool force = false);
+ const TabletSharedPtr& table, bool force = false);
OLAPStatus add_transaction(TPartitionId partition_id, TTransactionId
transaction_id,
TTabletId tablet_id, SchemaHash schema_hash,
@@ -119,7 +119,7 @@ class OLAPEngine {
TTabletId tablet_id, SchemaHash schema_hash,
bool delete_from_tablet = true);
- void get_transactions_by_tablet(OLAPTablePtr tablet, int64_t* partition_id,
+ void get_transactions_by_tablet(TabletSharedPtr tablet, int64_t*
partition_id,
std::set<int64_t>* transaction_ids);
bool has_transaction(TPartitionId partition_id, TTransactionId
transaction_id,
@@ -131,10 +131,10 @@ class OLAPEngine {
void clear_transaction_task(const TTransactionId transaction_id,
const std::vector<TPartitionId> partition_ids);
- OLAPStatus clone_incremental_data(OLAPTablePtr tablet, OLAPHeader&
clone_header,
+ OLAPStatus clone_incremental_data(TabletSharedPtr tablet, OLAPHeader&
clone_header,
int64_t committed_version);
- OLAPStatus clone_full_data(OLAPTablePtr tablet, OLAPHeader& clone_header);
+ OLAPStatus clone_full_data(TabletSharedPtr tablet, OLAPHeader&
clone_header);
// Add empty data for OLAPTable
//
@@ -320,10 +320,10 @@ class OLAPEngine {
// before doing incremental clone,
// need to calculate tablet's download dir and tablet's missing versions
- virtual std::string get_info_before_incremental_clone(OLAPTablePtr tablet,
+ virtual std::string get_info_before_incremental_clone(TabletSharedPtr
tablet,
int64_t committed_version, std::vector<Version>* missing_versions);
- virtual OLAPStatus finish_clone(OLAPTablePtr tablet, const std::string&
clone_dir,
+ virtual OLAPStatus finish_clone(TabletSharedPtr tablet, const std::string&
clone_dir,
int64_t committed_version, bool
is_incremental_clone);
// Obtain shard path for new tablet.
@@ -389,15 +389,15 @@ class OLAPEngine {
OLAPStatus _judge_and_update_effective_cluster_id(int32_t cluster_id);
OLAPStatus _calc_snapshot_id_path(
- const OLAPTablePtr& olap_table,
+ const TabletSharedPtr& olap_table,
std::string* out_path);
std::string _get_schema_hash_full_path(
- const OLAPTablePtr& ref_olap_table,
+ const TabletSharedPtr& ref_olap_table,
const std::string& location) const;
std::string _get_header_full_path(
- const OLAPTablePtr& ref_olap_table,
+ const TabletSharedPtr& ref_olap_table,
const std::string& schema_hash_path) const;
void _update_header_file_info(
@@ -406,25 +406,25 @@ class OLAPEngine {
OLAPStatus _link_index_and_data_files(
const std::string& header_path,
- const OLAPTablePtr& ref_olap_table,
+ const TabletSharedPtr& ref_olap_table,
const std::vector<VersionEntity>& version_entity_vec);
OLAPStatus _copy_index_and_data_files(
const std::string& header_path,
- const OLAPTablePtr& ref_olap_table,
+ const TabletSharedPtr& ref_olap_table,
std::vector<VersionEntity>& version_entity_vec);
OLAPStatus _create_snapshot_files(
- const OLAPTablePtr& ref_olap_table,
+ const TabletSharedPtr& ref_olap_table,
const TSnapshotRequest& request,
std::string* snapshot_path);
OLAPStatus _create_incremental_snapshot_files(
- const OLAPTablePtr& ref_olap_table,
+ const TabletSharedPtr& ref_olap_table,
const TSnapshotRequest& request,
std::string* snapshot_path);
- OLAPStatus _prepare_snapshot_dir(const OLAPTablePtr& ref_olap_table,
+ OLAPStatus _prepare_snapshot_dir(const TabletSharedPtr& ref_olap_table,
std::string* snapshot_id_path);
OLAPStatus _append_single_delta(
@@ -446,19 +446,19 @@ class OLAPEngine {
OLAPStatus _generate_new_header(
OlapStore* store,
const uint64_t new_shard,
- const OLAPTablePtr& tablet,
+ const TabletSharedPtr& tablet,
const std::vector<VersionEntity>& version_entity_vec, OLAPHeader*
new_olap_header);
OLAPStatus _create_hard_link(const std::string& from_path, const
std::string& to_path);
OLAPStatus _start_bg_worker();
- OLAPStatus _create_init_version(OLAPTablePtr olap_table, const
TCreateTabletReq& request);
+ OLAPStatus _create_init_version(TabletSharedPtr olap_table, const
TCreateTabletReq& request);
private:
struct TableInstances {
Mutex schema_change_lock;
- std::list<OLAPTablePtr> table_arr;
+ std::list<TabletSharedPtr> table_arr;
};
enum CompactionType {
@@ -497,7 +497,7 @@ class OLAPEngine {
typedef std::map<int64_t, TableInstances> tablet_map_t;
typedef std::map<std::string, uint32_t> file_system_task_count_t;
- OLAPTablePtr _get_table_with_no_lock(TTabletId tablet_id, SchemaHash
schema_hash);
+ TabletSharedPtr _get_table_with_no_lock(TTabletId tablet_id, SchemaHash
schema_hash);
// 遍历root所指定目录, 通过dirs返回此目录下所有有文件夹的名字, files返回所有文件的名字
OLAPStatus _dir_walk(const std::string& root,
@@ -510,20 +510,20 @@ class OLAPEngine {
OLAPStatus _create_new_table_header(const TCreateTabletReq& request,
OlapStore* store,
const bool is_schema_change_table,
- const OLAPTablePtr ref_olap_table,
+ const TabletSharedPtr
ref_olap_table,
OLAPHeader* header);
OLAPStatus _check_existed_or_else_create_dir(const std::string& path);
- OLAPTablePtr _find_best_tablet_to_compaction(CompactionType
compaction_type);
- bool _can_do_compaction(OLAPTablePtr table);
+ TabletSharedPtr _find_best_tablet_to_compaction(CompactionType
compaction_type);
+ bool _can_do_compaction(TabletSharedPtr table);
void _cancel_unfinished_schema_change();
OLAPStatus _do_sweep(
const std::string& scan_root, const time_t& local_tm_now, const
uint32_t expire);
- void _build_tablet_info(OLAPTablePtr olap_table, TTabletInfo* tablet_info);
+ void _build_tablet_info(TabletSharedPtr olap_table, TTabletInfo*
tablet_info);
void _build_tablet_stat();
EngineOptions _options;
diff --git a/be/src/olap/olap_snapshot.cpp b/be/src/olap/olap_snapshot.cpp
index 9d5e7e93..b63ed0f8 100644
--- a/be/src/olap/olap_snapshot.cpp
+++ b/be/src/olap/olap_snapshot.cpp
@@ -65,7 +65,7 @@ OLAPStatus OLAPEngine::make_snapshot(
return OLAP_ERR_INPUT_PARAMETER_ERROR;
}
- OLAPTablePtr ref_olap_table = get_table(request.tablet_id,
request.schema_hash);
+ TabletSharedPtr ref_olap_table = get_table(request.tablet_id,
request.schema_hash);
if (ref_olap_table.get() == NULL) {
OLAP_LOG_WARNING("failed to get olap table. [table=%ld
schema_hash=%d]",
request.tablet_id, request.schema_hash);
@@ -113,7 +113,7 @@ OLAPStatus OLAPEngine::release_snapshot(const string&
snapshot_path) {
}
OLAPStatus OLAPEngine::_calc_snapshot_id_path(
- const OLAPTablePtr& olap_table,
+ const TabletSharedPtr& olap_table,
string* out_path) {
OLAPStatus res = OLAP_SUCCESS;
if (out_path == nullptr) {
@@ -138,7 +138,7 @@ OLAPStatus OLAPEngine::_calc_snapshot_id_path(
}
string OLAPEngine::_get_schema_hash_full_path(
- const OLAPTablePtr& ref_olap_table,
+ const TabletSharedPtr& ref_olap_table,
const string& location) const {
stringstream schema_full_path_stream;
schema_full_path_stream << location
@@ -150,7 +150,7 @@ string OLAPEngine::_get_schema_hash_full_path(
}
string OLAPEngine::_get_header_full_path(
- const OLAPTablePtr& ref_olap_table,
+ const TabletSharedPtr& ref_olap_table,
const std::string& schema_hash_path) const {
stringstream header_name_stream;
header_name_stream << schema_hash_path << "/" <<
ref_olap_table->tablet_id() << ".hdr";
@@ -183,7 +183,7 @@ void OLAPEngine::_update_header_file_info(
OLAPStatus OLAPEngine::_link_index_and_data_files(
const string& schema_hash_path,
- const OLAPTablePtr& ref_olap_table,
+ const TabletSharedPtr& ref_olap_table,
const vector<VersionEntity>& version_entity_vec) {
OLAPStatus res = OLAP_SUCCESS;
@@ -229,7 +229,7 @@ OLAPStatus OLAPEngine::_link_index_and_data_files(
OLAPStatus OLAPEngine::_copy_index_and_data_files(
const string& schema_hash_path,
- const OLAPTablePtr& ref_olap_table,
+ const TabletSharedPtr& ref_olap_table,
vector<VersionEntity>& version_entity_vec) {
std::stringstream prefix_stream;
prefix_stream << schema_hash_path << "/" << ref_olap_table->tablet_id();
@@ -271,7 +271,7 @@ OLAPStatus OLAPEngine::_copy_index_and_data_files(
}
OLAPStatus OLAPEngine::_create_snapshot_files(
- const OLAPTablePtr& ref_olap_table,
+ const TabletSharedPtr& ref_olap_table,
const TSnapshotRequest& request,
string* snapshot_path) {
OLAPStatus res = OLAP_SUCCESS;
@@ -435,7 +435,7 @@ OLAPStatus OLAPEngine::_create_snapshot_files(
}
OLAPStatus OLAPEngine::_create_incremental_snapshot_files(
- const OLAPTablePtr& ref_olap_table,
+ const TabletSharedPtr& ref_olap_table,
const TSnapshotRequest& request,
string* snapshot_path) {
LOG(INFO) << "begin to create incremental snapshot files."
@@ -634,7 +634,7 @@ OLAPStatus OLAPEngine::storage_medium_migrate(
DorisMetrics::storage_migrate_requests_total.increment(1);
OLAPStatus res = OLAP_SUCCESS;
- OLAPTablePtr tablet = get_table(tablet_id, schema_hash);
+ TabletSharedPtr tablet = get_table(tablet_id, schema_hash);
if (tablet.get() == NULL) {
OLAP_LOG_WARNING("can't find olap table. [tablet_id=%ld
schema_hash=%d]",
tablet_id, schema_hash);
@@ -743,7 +743,7 @@ OLAPStatus OLAPEngine::storage_medium_migrate(
// if old table finished schema change, then the schema change status
of the new table is DONE
// else the schema change status of the new table is FAILED
- OLAPTablePtr new_tablet = get_table(tablet_id, schema_hash);
+ TabletSharedPtr new_tablet = get_table(tablet_id, schema_hash);
if (new_tablet.get() == NULL) {
OLAP_LOG_WARNING("get null olap table. [tablet_id=%ld
schema_hash=%d]",
tablet_id, schema_hash);
@@ -770,7 +770,7 @@ OLAPStatus OLAPEngine::storage_medium_migrate(
OLAPStatus OLAPEngine::_generate_new_header(
OlapStore* store,
const uint64_t new_shard,
- const OLAPTablePtr& tablet,
+ const TabletSharedPtr& tablet,
const vector<VersionEntity>& version_entity_vec, OLAPHeader*
new_olap_header) {
if (store == nullptr) {
LOG(WARNING) << "fail to generate new header for store is null";
diff --git a/be/src/olap/olap_table.cpp b/be/src/olap/olap_table.cpp
index 0ff94875..c1325dbd 100644
--- a/be/src/olap/olap_table.cpp
+++ b/be/src/olap/olap_table.cpp
@@ -54,7 +54,7 @@ using boost::filesystem::path;
namespace doris {
-OLAPTablePtr OLAPTable::create_from_header_file(
+TabletSharedPtr OLAPTable::create_from_header_file(
TTabletId tablet_id, TSchemaHash schema_hash,
const string& header_file, OlapStore* store) {
OLAPHeader* olap_header = NULL;
@@ -90,7 +90,7 @@ OLAPTablePtr OLAPTable::create_from_header_file(
return create_from_header(olap_header, store);
}
-OLAPTablePtr OLAPTable::create_from_header_file_for_check(
+TabletSharedPtr OLAPTable::create_from_header_file_for_check(
TTabletId tablet_id, TSchemaHash schema_hash, const string&
header_file) {
OLAPHeader* olap_header = NULL;
@@ -106,7 +106,7 @@ OLAPTablePtr OLAPTable::create_from_header_file_for_check(
return NULL;
}
- OLAPTablePtr olap_table = std::make_shared<OLAPTable>(olap_header);
+ TabletSharedPtr olap_table = std::make_shared<OLAPTable>(olap_header);
if (olap_table == NULL) {
OLAP_LOG_WARNING("fail to validate table. [header_file=%s]",
header_file.c_str());
delete olap_header;
@@ -128,7 +128,7 @@ OLAPTable::OLAPTable(OLAPHeader* header)
_table_for_check = true;
}
-OLAPTablePtr OLAPTable::create_from_header(
+TabletSharedPtr OLAPTable::create_from_header(
OLAPHeader* header,
OlapStore* store) {
auto olap_table = std::make_shared<OLAPTable>(header, store);
@@ -982,7 +982,7 @@ OLAPStatus OLAPTable::_handle_existed_version(int64_t
transaction_id, const Vers
return res;
} else if (!push_for_delete) {
DeleteConditionHandler del_cond_handler;
- OLAPTablePtr olap_table_ptr =
+ TabletSharedPtr olap_table_ptr =
OLAPEngine::get_instance()->get_table(_tablet_id,
_schema_hash);
if (olap_table_ptr.get() != nullptr) {
del_cond_handler.delete_cond(olap_table_ptr, version.first,
false);
@@ -2155,7 +2155,7 @@ void OLAPTable::set_schema_change_request(TTabletId
tablet_id,
schema_change_status->set_schema_change_type(alter_table_type);
}
-bool OLAPTable::remove_last_schema_change_version(OLAPTablePtr new_olap_table)
{
+bool OLAPTable::remove_last_schema_change_version(TabletSharedPtr
new_olap_table) {
if (!_header->has_schema_change_status()) {
return false;
}
diff --git a/be/src/olap/olap_table.h b/be/src/olap/olap_table.h
index ec2f4f52..1e7bc9df 100644
--- a/be/src/olap/olap_table.h
+++ b/be/src/olap/olap_table.h
@@ -44,7 +44,7 @@ class RowBlockPosition;
class OlapStore;
// Define OLAPTable's shared_ptr. It is used for
-typedef std::shared_ptr<OLAPTable> OLAPTablePtr;
+typedef std::shared_ptr<OLAPTable> TabletSharedPtr;
enum BaseCompactionStage {
BASE_COMPACTION_WAITING = 0,
@@ -94,17 +94,17 @@ struct SchemaChangeStatus {
class OLAPTable : public std::enable_shared_from_this<OLAPTable> {
public:
- static OLAPTablePtr create_from_header_file(
+ static TabletSharedPtr create_from_header_file(
TTabletId tablet_id,
TSchemaHash schema_hash,
const std::string& header_file,
OlapStore* store = nullptr);
- static OLAPTablePtr create_from_header_file_for_check(
+ static TabletSharedPtr create_from_header_file_for_check(
TTabletId tablet_id,
TSchemaHash schema_hash,
const std::string& header_file);
- static OLAPTablePtr create_from_header(
+ static TabletSharedPtr create_from_header(
OLAPHeader* header,
OlapStore* store = nullptr);
@@ -578,7 +578,7 @@ class OLAPTable : public
std::enable_shared_from_this<OLAPTable> {
const std::vector<Version>&
versions_to_changed,
const AlterTabletType alter_table_type);
- bool remove_last_schema_change_version(OLAPTablePtr new_olap_table);
+ bool remove_last_schema_change_version(TabletSharedPtr new_olap_table);
void clear_schema_change_request();
SchemaChangeStatus schema_change_status() {
diff --git a/be/src/olap/push_handler.cpp b/be/src/olap/push_handler.cpp
index a51c7950..f2fa3997 100644
--- a/be/src/olap/push_handler.cpp
+++ b/be/src/olap/push_handler.cpp
@@ -44,261 +44,8 @@ namespace doris {
// this usually means schema change is over,
// clear schema change info in both current table and related tables,
// finally we will only push for current tables
-OLAPStatus PushHandler::process(
- OLAPTablePtr olap_table,
- const TPushReq& request,
- PushType push_type,
- vector<TTabletInfo>* tablet_info_vec) {
- LOG(INFO) << "begin to push data. tablet=" << olap_table->full_name()
- << ", version=" << request.version;
-
- OLAPStatus res = OLAP_SUCCESS;
- _request = request;
- _olap_table_arr.clear();
- _olap_table_arr.push_back(olap_table);
- vector<TableVars> table_infoes(1);
- table_infoes[0].olap_table = olap_table;
-
- bool is_push_locked = false;
- bool is_new_tablet = false;
- bool is_new_tablet_effective = false;
-
- // 1. Get related tablets first if tablet in alter table status,
- TTabletId tablet_id;
- TSchemaHash schema_hash;
- AlterTabletType alter_table_type;
- OLAPTablePtr related_olap_table;
- _obtain_header_rdlock();
- bool is_schema_changing = olap_table->get_schema_change_request(
- &tablet_id, &schema_hash, NULL, &alter_table_type);
- _release_header_lock();
-
- if (is_schema_changing) {
- related_olap_table = OLAPEngine::get_instance()->get_table(tablet_id,
schema_hash);
- if (NULL == related_olap_table.get()) {
- OLAP_LOG_WARNING("can't find olap table, clear invalid schema
change info."
- "[table=%ld schema_hash=%d]", tablet_id,
schema_hash);
- _obtain_header_wrlock();
- olap_table->clear_schema_change_request();
- _release_header_lock();
- is_schema_changing = false;
- } else {
- // _olap_table_arr is used to obtain header lock,
- // to avoid deadlock, we must lock tablet header in time order.
- if (related_olap_table->creation_time() <
olap_table->creation_time()) {
- _olap_table_arr.push_front(related_olap_table);
- } else {
- _olap_table_arr.push_back(related_olap_table);
- }
- }
- }
-
- // Obtain push lock to avoid simultaneously PUSH and
- // conflict with alter table operations.
- for (OLAPTablePtr table : _olap_table_arr) {
- table->obtain_push_lock();
- }
- is_push_locked = true;
-
- if (is_schema_changing) {
- _obtain_header_rdlock();
- is_schema_changing = olap_table->get_schema_change_request(
- &tablet_id, &schema_hash, NULL, &alter_table_type);
- _release_header_lock();
-
- if (!is_schema_changing) {
- LOG(INFO) << "schema change info is cleared after base table get
related tablet, "
- << "maybe new tablet reach at the same time and load
firstly. "
- << ", old_tablet=" << olap_table->full_name()
- << ", new_tablet=" << related_olap_table->full_name()
- << ", version=" << _request.version;
- } else if (related_olap_table->creation_time() >
olap_table->creation_time()) {
- // If current table is old table, append it to table_infoes
- table_infoes.push_back(TableVars());
- TableVars& new_item = table_infoes.back();
- new_item.olap_table = related_olap_table;
- } else {
- // if current table is new table, clear schema change info
- res = _clear_alter_table_info(olap_table, related_olap_table);
- if (res != OLAP_SUCCESS) {
- OLAP_LOG_WARNING("fail to clear schema change info. [res=%d]",
res);
- goto EXIT;
- }
-
- LOG(INFO) << "data of new table is generated, stop convert from
base table. "
- << "old_tablet=" << olap_table->full_name()
- << ", new_tablet=" << related_olap_table->full_name()
- << ", version=" << _request.version;
- is_new_tablet_effective = true;
- }
- }
-
- // To keep logic of alter_table/rollup_table consistent
- if (table_infoes.size() == 1) {
- table_infoes.resize(2);
- }
-
- // 2. validate request: version and version_hash chek
- _obtain_header_rdlock();
- res = _validate_request(table_infoes[0].olap_table,
- table_infoes[1].olap_table,
- is_new_tablet_effective,
- push_type);
- _release_header_lock();
- if (res != OLAP_SUCCESS) {
- OLAP_LOG_WARNING("fail to validate request. [res=%d table='%s'
version=%ld]",
- res, olap_table->full_name().c_str(),
_request.version);
- goto EXIT;
- }
-
- // 3. Remove reverted version including delta and cumulative,
- // which will be deleted by background thread
- _obtain_header_wrlock();
- for (TableVars& table_var : table_infoes) {
- if (NULL == table_var.olap_table.get()) {
- continue;
- }
-
- res = _get_versions_reverted(table_var.olap_table,
- is_new_tablet,
- push_type,
- &(table_var.unused_versions));
- if (res != OLAP_SUCCESS) {
- OLAP_LOG_WARNING("failed to get reverted versions. "
- "[res=%d table='%s' version=%ld]",
- res, table_var.olap_table->full_name().c_str(),
_request.version);
- goto EXIT;
- }
-
- if (table_var.unused_versions.size() != 0) {
- res = _update_header(table_var.olap_table,
- &(table_var.unused_versions),
- &(table_var.added_indices),
- &(table_var.unused_indices));
- if (res != OLAP_SUCCESS) {
- OLAP_LOG_WARNING("fail to update header for revert. "
- "[res=%d table='%s' version=%ld]",
- res,
table_var.olap_table->full_name().c_str(), _request.version);
- goto EXIT;
- }
-
- _delete_old_indices(&(table_var.unused_indices));
- }
-
- // If there are more than one table, others is doing alter table
- is_new_tablet = true;
- }
- _release_header_lock();
-
- // 4. Save delete condition when push for delete
- if (push_type == PUSH_FOR_DELETE) {
- _obtain_header_wrlock();
- DeleteConditionHandler del_cond_handler;
-
- for (TableVars& table_var : table_infoes) {
- if (table_var.olap_table.get() == NULL) {
- continue;
- }
-
- res = del_cond_handler.store_cond(
- table_var.olap_table, request.version,
request.delete_conditions);
- if (res != OLAP_SUCCESS) {
- OLAP_LOG_WARNING("fail to store delete condition. [res=%d
table='%s']",
- res,
table_var.olap_table->full_name().c_str());
- goto EXIT;
- }
-
- res = table_var.olap_table->save_header();
- if (res != OLAP_SUCCESS) {
- LOG(FATAL) << "fail to save header. res=" << res
- << ", table=" << table_var.olap_table->full_name();
- goto EXIT;
- }
- }
-
- _release_header_lock();
- }
-
- // 5. Convert local data file into delta_file and build index,
- // which may take a long time
- res = _convert(table_infoes[0].olap_table,
- table_infoes[1].olap_table,
- &(table_infoes[0].added_indices),
- &(table_infoes[1].added_indices),
- alter_table_type);
- if (res != OLAP_SUCCESS) {
- OLAP_LOG_WARNING("fail to convert data. [res=%d]", res);
- goto EXIT;
- }
-
- // Update table header: add new version and remove reverted version
- _obtain_header_wrlock();
- for (TableVars& table_var : table_infoes) {
- if (NULL == table_var.olap_table.get()) {
- continue;
- }
-
- res = _update_header(table_var.olap_table,
- &(table_var.unused_versions),
- &(table_var.added_indices),
- &(table_var.unused_indices));
- if (res != OLAP_SUCCESS) {
- OLAP_LOG_WARNING("fail to update header of new delta."
- "[res=%d table='%s' version=%ld]",
- res, table_var.olap_table->full_name().c_str(),
_request.version);
- goto EXIT;
- }
- }
- _release_header_lock();
-
- // 6. Delete unused versions which include delta and commulative,
- // which, in fact, is added to list and deleted by background thread
- for (TableVars& table_var : table_infoes) {
- if (NULL == table_var.olap_table.get()) {
- continue;
- }
-
- _delete_old_indices(&(table_var.unused_indices));
- }
-
-EXIT:
- _release_header_lock();
-
- // Get tablet infos for output
- if (res == OLAP_SUCCESS || res == OLAP_ERR_PUSH_VERSION_ALREADY_EXIST) {
- if (tablet_info_vec != NULL) {
- _get_tablet_infos(table_infoes, tablet_info_vec);
- }
- res = OLAP_SUCCESS;
- }
-
- // Clear added_indices when error happens
- for (TableVars& table_var : table_infoes) {
- if (table_var.olap_table.get() == NULL) {
- continue;
- }
-
- for (SegmentGroup* segment_group : table_var.added_indices) {
- segment_group->delete_all_files();
- SAFE_DELETE(segment_group);
- }
- }
-
- // Release push lock
- if (is_push_locked) {
- for (OLAPTablePtr table : _olap_table_arr) {
- table->release_push_lock();
- }
- }
- _olap_table_arr.clear();
-
- LOG(INFO) << "finish to process push. res=" << res;
-
- return res;
-}
-
OLAPStatus PushHandler::process_realtime_push(
- OLAPTablePtr olap_table,
+ TabletSharedPtr olap_table,
const TPushReq& request,
PushType push_type,
vector<TTabletInfo>* tablet_info_vec) {
@@ -354,7 +101,7 @@ OLAPStatus PushHandler::process_realtime_push(
<< ", related_tablet_id=" << related_tablet_id
<< ", related_schema_hash=" << related_schema_hash
<< ", transaction_id=" << request.transaction_id;
- OLAPTablePtr related_olap_table =
OLAPEngine::get_instance()->get_table(
+ TabletSharedPtr related_olap_table =
OLAPEngine::get_instance()->get_table(
related_tablet_id, related_schema_hash);
// if related tablet not exists, only push current tablet
@@ -526,8 +273,8 @@ void PushHandler::_get_tablet_infos(
}
OLAPStatus PushHandler::_convert(
- OLAPTablePtr curr_olap_table,
- OLAPTablePtr new_olap_table,
+ TabletSharedPtr curr_olap_table,
+ TabletSharedPtr new_olap_table,
Indices* curr_olap_indices,
Indices* new_olap_indices,
AlterTabletType alter_table_type) {
@@ -715,8 +462,8 @@ OLAPStatus PushHandler::_convert(
}
OLAPStatus PushHandler::_validate_request(
- OLAPTablePtr olap_table_for_raw,
- OLAPTablePtr olap_table_for_schema_change,
+ TabletSharedPtr olap_table_for_raw,
+ TabletSharedPtr olap_table_for_schema_change,
bool is_new_tablet_effective,
PushType push_type) {
const PDelta* latest_delta = olap_table_for_raw->lastest_delta();
@@ -782,7 +529,7 @@ OLAPStatus PushHandler::_validate_request(
// user submit a push job and cancel it soon, but some
// tablets already push success.
OLAPStatus PushHandler::_get_versions_reverted(
- OLAPTablePtr olap_table,
+ TabletSharedPtr olap_table,
bool is_new_tablet,
PushType push_type,
Versions* unused_versions) {
@@ -832,7 +579,7 @@ OLAPStatus PushHandler::_get_versions_reverted(
}
OLAPStatus PushHandler::_update_header(
- OLAPTablePtr olap_table,
+ TabletSharedPtr olap_table,
Versions* unused_versions,
Indices* new_indices,
Indices* unused_indices) {
@@ -876,8 +623,8 @@ void PushHandler::_delete_old_indices(Indices*
unused_indices) {
}
OLAPStatus PushHandler::_clear_alter_table_info(
- OLAPTablePtr tablet,
- OLAPTablePtr related_tablet) {
+ TabletSharedPtr tablet,
+ TabletSharedPtr related_tablet) {
OLAPStatus res = OLAP_SUCCESS;
_obtain_header_wrlock();
@@ -957,7 +704,7 @@ BinaryReader::BinaryReader()
}
OLAPStatus BinaryReader::init(
- OLAPTablePtr table,
+ TabletSharedPtr table,
BinaryFile* file) {
OLAPStatus res = OLAP_SUCCESS;
@@ -1087,7 +834,7 @@ LzoBinaryReader::LzoBinaryReader()
}
OLAPStatus LzoBinaryReader::init(
- OLAPTablePtr table,
+ TabletSharedPtr table,
BinaryFile* file) {
OLAPStatus res = OLAP_SUCCESS;
diff --git a/be/src/olap/push_handler.h b/be/src/olap/push_handler.h
index c84def76..f4f69352 100644
--- a/be/src/olap/push_handler.h
+++ b/be/src/olap/push_handler.h
@@ -42,7 +42,7 @@ class ColumnMapping;
class RowCursor;
struct TableVars {
- OLAPTablePtr olap_table;
+ TabletSharedPtr olap_table;
Versions unused_versions;
Indices unused_indices;
Indices added_indices;
@@ -56,14 +56,8 @@ class PushHandler {
~PushHandler() {}
// Load local data file into specified tablet.
- OLAPStatus process(
- OLAPTablePtr olap_table,
- const TPushReq& request,
- PushType push_type,
- std::vector<TTabletInfo>* tablet_info_vec);
-
OLAPStatus process_realtime_push(
- OLAPTablePtr olap_table,
+ TabletSharedPtr olap_table,
const TPushReq& request,
PushType push_type,
std::vector<TTabletInfo>* tablet_info_vec);
@@ -73,8 +67,8 @@ class PushHandler {
private:
// Validate request, mainly data version check.
OLAPStatus _validate_request(
- OLAPTablePtr olap_table_for_raw,
- OLAPTablePtr olap_table_for_schema_change,
+ TabletSharedPtr olap_table_for_raw,
+ TabletSharedPtr olap_table_for_schema_change,
bool is_rollup_new_table,
PushType push_type);
@@ -82,7 +76,7 @@ class PushHandler {
// user submit a push job and cancel it soon, but some
// tablets already push success.
OLAPStatus _get_versions_reverted(
- OLAPTablePtr olap_table,
+ TabletSharedPtr olap_table,
bool is_schema_change_tablet,
PushType push_type,
Versions* unused_versions);
@@ -90,15 +84,15 @@ class PushHandler {
// Convert local data file to internal formatted delta,
// return new delta's SegmentGroup
OLAPStatus _convert(
- OLAPTablePtr curr_olap_table,
- OLAPTablePtr new_olap_table_vec,
+ TabletSharedPtr curr_olap_table,
+ TabletSharedPtr new_olap_table_vec,
Indices* curr_olap_indices,
Indices* new_olap_indices,
AlterTabletType alter_table_type);
// Update header info when new version add or dirty version removed.
OLAPStatus _update_header(
- OLAPTablePtr olap_table,
+ TabletSharedPtr olap_table,
Versions* unused_versions,
Indices* new_indices,
Indices* unused_indices);
@@ -108,15 +102,15 @@ class PushHandler {
// Clear schema change information.
OLAPStatus _clear_alter_table_info(
- OLAPTablePtr olap_table,
- OLAPTablePtr related_olap_table);
+ TabletSharedPtr olap_table,
+ TabletSharedPtr related_olap_table);
// Only for debug
std::string _debug_version_list(const Versions& versions) const;
// Lock tablet header before read header info.
void _obtain_header_rdlock() {
- for (std::list<OLAPTablePtr>::iterator it = _olap_table_arr.begin();
+ for (std::list<TabletSharedPtr>::iterator it = _olap_table_arr.begin();
it != _olap_table_arr.end(); ++it) {
VLOG(3) << "obtain all header locks rd. tablet=" <<
(*it)->full_name();
(*it)->obtain_header_rdlock();
@@ -127,7 +121,7 @@ class PushHandler {
// Locak tablet header before write header info.
void _obtain_header_wrlock() {
- for (std::list<OLAPTablePtr>::iterator it = _olap_table_arr.begin();
+ for (std::list<TabletSharedPtr>::iterator it = _olap_table_arr.begin();
it != _olap_table_arr.end(); ++it) {
VLOG(3) << "obtain all header locks wr. tablet=" <<
(*it)->full_name();
(*it)->obtain_header_wrlock();
@@ -139,7 +133,7 @@ class PushHandler {
// Release tablet header lock.
void _release_header_lock() {
if (_header_locked) {
- for (std::list<OLAPTablePtr>::reverse_iterator it =
_olap_table_arr.rbegin();
+ for (std::list<TabletSharedPtr>::reverse_iterator it =
_olap_table_arr.rbegin();
it != _olap_table_arr.rend(); ++it) {
VLOG(3) << "release all header locks. tablet=" <<
(*it)->full_name();
(*it)->release_header_lock();
@@ -158,7 +152,7 @@ class PushHandler {
// maily contains specified tablet object
// contains related tables also if in schema change, tablet split or rollup
- std::list<OLAPTablePtr> _olap_table_arr;
+ std::list<TabletSharedPtr> _olap_table_arr;
// lock tablet header before modify tabelt header
bool _header_locked;
@@ -202,7 +196,7 @@ class IBinaryReader {
static IBinaryReader* create(bool need_decompress);
virtual ~IBinaryReader() {}
- virtual OLAPStatus init(OLAPTablePtr table, BinaryFile* file) = 0;
+ virtual OLAPStatus init(TabletSharedPtr table, BinaryFile* file) = 0;
virtual OLAPStatus finalize() = 0;
virtual OLAPStatus next(RowCursor* row, MemPool* mem_pool) = 0;
@@ -224,7 +218,7 @@ class IBinaryReader {
}
BinaryFile* _file;
- OLAPTablePtr _table;
+ TabletSharedPtr _table;
size_t _content_len;
size_t _curr;
uint32_t _adler_checksum;
@@ -239,7 +233,7 @@ class BinaryReader: public IBinaryReader {
finalize();
}
- virtual OLAPStatus init(OLAPTablePtr table, BinaryFile* file);
+ virtual OLAPStatus init(TabletSharedPtr table, BinaryFile* file);
virtual OLAPStatus finalize();
virtual OLAPStatus next(RowCursor* row, MemPool* mem_pool);
@@ -260,7 +254,7 @@ class LzoBinaryReader: public IBinaryReader {
finalize();
}
- virtual OLAPStatus init(OLAPTablePtr table, BinaryFile* file);
+ virtual OLAPStatus init(TabletSharedPtr table, BinaryFile* file);
virtual OLAPStatus finalize();
virtual OLAPStatus next(RowCursor* row, MemPool* mem_pool);
diff --git a/be/src/olap/reader.h b/be/src/olap/reader.h
index 0eb6598b..46919dbf 100644
--- a/be/src/olap/reader.h
+++ b/be/src/olap/reader.h
@@ -48,7 +48,7 @@ class RuntimeState;
// Params for Reader,
// mainly include tablet, data version and fetch range.
struct ReaderParams {
- OLAPTablePtr olap_table;
+ TabletSharedPtr olap_table;
ReaderType reader_type;
bool aggregation;
Version version;
@@ -198,7 +198,7 @@ class Reader {
Version _version;
- OLAPTablePtr _olap_table;
+ TabletSharedPtr _olap_table;
// _own_data_sources is data source that reader aquire from olap_table, so
we need to
// release these when reader closing
diff --git a/be/src/olap/schema_change.cpp b/be/src/olap/schema_change.cpp
index 89ebc6e4..ff56ea38 100644
--- a/be/src/olap/schema_change.cpp
+++ b/be/src/olap/schema_change.cpp
@@ -55,12 +55,12 @@ bool version_entity_sorter(const VersionEntity& a, const
VersionEntity& b) {
}
RowBlockChanger::RowBlockChanger(const std::vector<FieldInfo> &tablet_schema,
- const OLAPTablePtr &ref_olap_table) {
+ const TabletSharedPtr &ref_olap_table) {
_schema_mapping.resize(tablet_schema.size());
}
RowBlockChanger::RowBlockChanger(const vector<FieldInfo>& tablet_schema,
- const OLAPTablePtr& ref_olap_table,
+ const TabletSharedPtr& ref_olap_table,
const DeleteHandler& delete_handler) {
_schema_mapping.resize(tablet_schema.size());
_delete_handler = delete_handler;
@@ -533,7 +533,7 @@ void RowBlockAllocator::release(RowBlock* row_block) {
delete row_block;
}
-RowBlockMerger::RowBlockMerger(OLAPTablePtr olap_table) :
_olap_table(olap_table) {}
+RowBlockMerger::RowBlockMerger(TabletSharedPtr olap_table) :
_olap_table(olap_table) {}
RowBlockMerger::~RowBlockMerger() {}
@@ -644,12 +644,12 @@ bool RowBlockMerger::_pop_heap() {
}
LinkedSchemaChange::LinkedSchemaChange(
- OLAPTablePtr base_olap_table, OLAPTablePtr new_olap_table) :
+ TabletSharedPtr base_olap_table, TabletSharedPtr new_olap_table) :
_base_olap_table(base_olap_table),
_new_olap_table(new_olap_table) {}
SchemaChangeDirectly::SchemaChangeDirectly(
- OLAPTablePtr olap_table,
+ TabletSharedPtr olap_table,
const RowBlockChanger& row_block_changer) :
_olap_table(olap_table),
_row_block_changer(row_block_changer),
@@ -890,7 +890,7 @@ bool SchemaChangeDirectly::process(ColumnData* olap_data,
SegmentGroup* new_segm
return result;
}
-SchemaChangeWithSorting::SchemaChangeWithSorting(OLAPTablePtr olap_table,
+SchemaChangeWithSorting::SchemaChangeWithSorting(TabletSharedPtr olap_table,
const RowBlockChanger&
row_block_changer,
size_t memory_limitation) :
_olap_table(olap_table),
@@ -1226,12 +1226,12 @@ OLAPStatus
SchemaChangeHandler::clear_schema_change_single_info(
AlterTabletType* alter_table_type,
bool only_one,
bool check_only) {
- OLAPTablePtr olap_table = OLAPEngine::get_instance()->get_table(tablet_id,
schema_hash);
+ TabletSharedPtr olap_table =
OLAPEngine::get_instance()->get_table(tablet_id, schema_hash);
return clear_schema_change_single_info(olap_table, alter_table_type,
only_one, check_only);
}
OLAPStatus SchemaChangeHandler::clear_schema_change_single_info(
- OLAPTablePtr olap_table,
+ TabletSharedPtr olap_table,
AlterTabletType* type,
bool only_one,
bool check_only) {
@@ -1263,7 +1263,7 @@ OLAPStatus
SchemaChangeHandler::clear_schema_change_single_info(
}
OLAPStatus SchemaChangeHandler::_check_and_clear_schema_change_info(
- OLAPTablePtr olap_table,
+ TabletSharedPtr olap_table,
const TAlterTabletReq& request) {
// check for schema change chain ( A->B)
// broken old relation if a chain was found and there is no version to be
changed
@@ -1317,7 +1317,7 @@ OLAPStatus
SchemaChangeHandler::_check_and_clear_schema_change_info(
}
// clear schema change info of related tablet
- OLAPTablePtr tablet = OLAPEngine::get_instance()->get_table(
+ TabletSharedPtr tablet = OLAPEngine::get_instance()->get_table(
tablet_id, schema_hash);
if (tablet.get() == NULL) {
OLAP_LOG_WARNING("get null tablet! [tablet_id=%ld schema_hash=%d]",
@@ -1360,7 +1360,7 @@ OLAPStatus SchemaChangeHandler::process_alter_table(
}
// 2. Get base table
- OLAPTablePtr ref_olap_table = OLAPEngine::get_instance()->get_table(
+ TabletSharedPtr ref_olap_table = OLAPEngine::get_instance()->get_table(
request.base_tablet_id, request.base_schema_hash);
if (ref_olap_table.get() == NULL) {
OLAP_LOG_WARNING("fail to find base table. [base_table=%ld
base_schema_hash=%d]",
@@ -1380,7 +1380,7 @@ OLAPStatus SchemaChangeHandler::process_alter_table(
}
// 4. return failed if new table already exist in OLAPEngine.
- OLAPTablePtr new_tablet = OLAPEngine::get_instance()->get_table(
+ TabletSharedPtr new_tablet = OLAPEngine::get_instance()->get_table(
request.new_tablet_req.tablet_id,
request.new_tablet_req.tablet_schema.schema_hash);
if (new_tablet.get() != NULL) {
res = OLAP_SUCCESS;
@@ -1395,10 +1395,10 @@ OLAPStatus SchemaChangeHandler::process_alter_table(
OLAPStatus SchemaChangeHandler::_do_alter_table(
AlterTabletType type,
- OLAPTablePtr ref_olap_table,
+ TabletSharedPtr ref_olap_table,
const TAlterTabletReq& request) {
OLAPStatus res = OLAP_SUCCESS;
- OLAPTablePtr new_olap_table;
+ TabletSharedPtr new_olap_table;
string base_root_path = ref_olap_table->storage_root_path_name();
LOG(INFO) << "begin to do alter tablet job. new_table_id=" <<
request.new_tablet_req.tablet_id;
@@ -1632,10 +1632,10 @@ OLAPStatus SchemaChangeHandler::_do_alter_table(
}
OLAPStatus SchemaChangeHandler::_create_new_olap_table(
- const OLAPTablePtr ref_olap_table,
+ const TabletSharedPtr ref_olap_table,
const TCreateTabletReq& request,
const string* ref_root_path,
- OLAPTablePtr* out_new_olap_table) {
+ TabletSharedPtr* out_new_olap_table) {
OLAPStatus res = OLAP_SUCCESS;
OLAPTable* new_olap_table = NULL;
bool is_table_added = false;
@@ -1646,7 +1646,7 @@ OLAPStatus SchemaChangeHandler::_create_new_olap_table(
do {
// 2. Create table with only header, no deltas
- OLAPTablePtr new_olap_table = OLAPEngine::get_instance()->create_table(
+ TabletSharedPtr new_olap_table =
OLAPEngine::get_instance()->create_table(
request, ref_root_path, true, ref_olap_table);
if (new_olap_table == NULL) {
OLAP_LOG_WARNING("failed to create table. [table=%ld xml_path=%d]",
@@ -1697,7 +1697,7 @@ OLAPStatus SchemaChangeHandler::_create_new_olap_table(
break;
}
- OLAPTablePtr olap_table;
+ TabletSharedPtr olap_table;
olap_table = OLAPEngine::get_instance()->get_table(
request.tablet_id, request.tablet_schema.schema_hash);
if (olap_table.get() == NULL) {
@@ -1732,8 +1732,8 @@ OLAPStatus SchemaChangeHandler::_create_new_olap_table(
}
OLAPStatus SchemaChangeHandler::schema_version_convert(
- OLAPTablePtr src_olap_table,
- OLAPTablePtr dest_olap_table,
+ TabletSharedPtr src_olap_table,
+ TabletSharedPtr dest_olap_table,
vector<SegmentGroup*>* ref_segment_groups,
vector<SegmentGroup*>* new_segment_groups) {
if (NULL == new_segment_groups) {
@@ -1862,7 +1862,7 @@ OLAPStatus SchemaChangeHandler::schema_version_convert(
}
OLAPStatus SchemaChangeHandler::_get_versions_to_be_changed(
- OLAPTablePtr ref_olap_table,
+ TabletSharedPtr ref_olap_table,
vector<Version>& versions_to_be_changed) {
int32_t request_version = 0;
const PDelta* lastest_version = ref_olap_table->lastest_version();
@@ -1902,8 +1902,8 @@ OLAPStatus
SchemaChangeHandler::_get_versions_to_be_changed(
// 增加A->(B|C|...) 的schema_change信息
OLAPStatus SchemaChangeHandler::_save_schema_change_info(
AlterTabletType alter_table_type,
- OLAPTablePtr ref_olap_table,
- OLAPTablePtr new_olap_table,
+ TabletSharedPtr ref_olap_table,
+ TabletSharedPtr new_olap_table,
const vector<Version>& versions_to_be_changed) {
// check new table exists,
@@ -2203,8 +2203,8 @@ OLAPStatus
SchemaChangeHandler::_alter_table(SchemaChangeParams* sc_params) {
// @static
// 分析column的mapping以及filter key的mapping
-OLAPStatus SchemaChangeHandler::_parse_request(OLAPTablePtr ref_olap_table,
- OLAPTablePtr new_olap_table,
+OLAPStatus SchemaChangeHandler::_parse_request(TabletSharedPtr ref_olap_table,
+ TabletSharedPtr new_olap_table,
RowBlockChanger* rb_changer,
bool* sc_sorting,
bool* sc_directly) {
@@ -2369,7 +2369,7 @@ OLAPStatus SchemaChange::create_init_version(
VLOG(3) << "begin to create init version. "
<< "begin=" << version.first << ", end=" << version.second;
- OLAPTablePtr table;
+ TabletSharedPtr table;
ColumnDataWriter* writer = NULL;
OLAPStatus res = OLAP_SUCCESS;
diff --git a/be/src/olap/schema_change.h b/be/src/olap/schema_change.h
index 84cafa6f..163a7d70 100644
--- a/be/src/olap/schema_change.h
+++ b/be/src/olap/schema_change.h
@@ -57,11 +57,11 @@ class RowBlockChanger {
typedef std::vector<ColumnMapping> SchemaMapping;
RowBlockChanger(const std::vector<FieldInfo>& tablet_schema,
- const OLAPTablePtr& ref_olap_table,
+ const TabletSharedPtr& ref_olap_table,
const DeleteHandler& delete_handler);
RowBlockChanger(const std::vector<FieldInfo>& tablet_schema,
- const OLAPTablePtr& ref_olap_table);
+ const TabletSharedPtr& ref_olap_table);
virtual ~RowBlockChanger();
@@ -119,7 +119,7 @@ class RowBlockAllocator {
class RowBlockMerger {
public:
- explicit RowBlockMerger(OLAPTablePtr olap_table);
+ explicit RowBlockMerger(TabletSharedPtr olap_table);
virtual ~RowBlockMerger();
bool merge(
@@ -141,7 +141,7 @@ class RowBlockMerger {
bool _make_heap(const std::vector<RowBlock*>& row_block_arr);
bool _pop_heap();
- OLAPTablePtr _olap_table;
+ TabletSharedPtr _olap_table;
std::priority_queue<MergeElement> _heap;
};
@@ -191,14 +191,14 @@ class SchemaChange {
class LinkedSchemaChange : public SchemaChange {
public:
explicit LinkedSchemaChange(
- OLAPTablePtr base_olap_table,
- OLAPTablePtr new_olap_table);
+ TabletSharedPtr base_olap_table,
+ TabletSharedPtr new_olap_table);
~LinkedSchemaChange() {}
bool process(ColumnData* olap_data, SegmentGroup* new_segment_group);
private:
- OLAPTablePtr _base_olap_table;
- OLAPTablePtr _new_olap_table;
+ TabletSharedPtr _base_olap_table;
+ TabletSharedPtr _new_olap_table;
DISALLOW_COPY_AND_ASSIGN(LinkedSchemaChange);
};
@@ -208,14 +208,14 @@ class SchemaChangeDirectly : public SchemaChange {
// @params olap_table the instance of table which has new schema.
// @params row_block_changer changer to modifiy the data of RowBlock
explicit SchemaChangeDirectly(
- OLAPTablePtr olap_table,
+ TabletSharedPtr olap_table,
const RowBlockChanger& row_block_changer);
virtual ~SchemaChangeDirectly();
virtual bool process(ColumnData* olap_data, SegmentGroup*
new_segment_group);
private:
- OLAPTablePtr _olap_table;
+ TabletSharedPtr _olap_table;
const RowBlockChanger& _row_block_changer;
RowBlockAllocator* _row_block_allocator;
RowCursor* _src_cursor;
@@ -230,7 +230,7 @@ class SchemaChangeDirectly : public SchemaChange {
class SchemaChangeWithSorting : public SchemaChange {
public:
explicit SchemaChangeWithSorting(
- OLAPTablePtr olap_table,
+ TabletSharedPtr olap_table,
const RowBlockChanger& row_block_changer,
size_t memory_limitation);
virtual ~SchemaChangeWithSorting();
@@ -247,7 +247,7 @@ class SchemaChangeWithSorting : public SchemaChange {
std::vector<SegmentGroup*>& src_segment_group_arr,
SegmentGroup* segment_group);
- OLAPTablePtr _olap_table;
+ TabletSharedPtr _olap_table;
const RowBlockChanger& _row_block_changer;
size_t _memory_limitation;
Version _temp_delta_versions;
@@ -264,8 +264,8 @@ class SchemaChangeHandler {
OLAPStatus process_alter_table(AlterTabletType alter_table_type,
const TAlterTabletReq& request);
- OLAPStatus schema_version_convert(OLAPTablePtr ref_olap_table,
- OLAPTablePtr new_olap_table,
+ OLAPStatus schema_version_convert(TabletSharedPtr ref_olap_table,
+ TabletSharedPtr new_olap_table,
std::vector<SegmentGroup*>*
ref_segment_groups,
std::vector<SegmentGroup*>*
new_segment_groups);
@@ -290,7 +290,7 @@ class SchemaChangeHandler {
bool only_one,
bool check_only);
- static OLAPStatus clear_schema_change_single_info(OLAPTablePtr olap_table,
+ static OLAPStatus clear_schema_change_single_info(TabletSharedPtr
olap_table,
AlterTabletType*
alter_table_type,
bool only_one,
bool check_only);
@@ -302,21 +302,21 @@ class SchemaChangeHandler {
// Returns:
// 成功：如果存在历史信息，没有问题的就清空；或者没有历史信息
// 失败：否则如果有历史信息且无法清空的（有version还没有完成）
- OLAPStatus _check_and_clear_schema_change_info(OLAPTablePtr olap_table,
+ OLAPStatus _check_and_clear_schema_change_info(TabletSharedPtr olap_table,
const TAlterTabletReq&
request);
- OLAPStatus _get_versions_to_be_changed(OLAPTablePtr ref_olap_table,
+ OLAPStatus _get_versions_to_be_changed(TabletSharedPtr ref_olap_table,
std::vector<Version>&
versions_to_be_changed);
OLAPStatus _do_alter_table(AlterTabletType type,
- OLAPTablePtr ref_olap_table,
+ TabletSharedPtr ref_olap_table,
const TAlterTabletReq& request);
struct SchemaChangeParams {
// 为了让calc_split_key也可使用普通schema_change的线程，才设置了此type
AlterTabletType alter_table_type;
- OLAPTablePtr ref_olap_table;
- OLAPTablePtr new_olap_table;
+ TabletSharedPtr ref_olap_table;
+ TabletSharedPtr new_olap_table;
std::vector<ColumnData*> ref_olap_data_arr;
std::string debug_message;
DeleteHandler delete_handler;
@@ -326,23 +326,23 @@ class SchemaChangeHandler {
};
// 根据给定的table_desc，创建OLAPTable，并挂接到OLAPEngine中
- OLAPStatus _create_new_olap_table(const OLAPTablePtr ref_olap_table,
+ OLAPStatus _create_new_olap_table(const TabletSharedPtr ref_olap_table,
const TCreateTabletReq&
create_tablet_req,
const std::string* ref_root_path,
- OLAPTablePtr* out_new_olap_table);
+ TabletSharedPtr* out_new_olap_table);
// 增加A->(B|C|...) 的schema_change信息
// 在split table时，增加split-table status相关的信息
// 其他的都增加在schema-change status中
OLAPStatus _save_schema_change_info(AlterTabletType alter_table_type,
- OLAPTablePtr ref_olap_table,
- OLAPTablePtr new_olap_table,
+ TabletSharedPtr ref_olap_table,
+ TabletSharedPtr new_olap_table,
const std::vector<Version>&
versions_to_be_changed);
static OLAPStatus _alter_table(SchemaChangeParams* sc_params);
- static OLAPStatus _parse_request(OLAPTablePtr ref_olap_table,
- OLAPTablePtr new_olap_table,
+ static OLAPStatus _parse_request(TabletSharedPtr ref_olap_table,
+ TabletSharedPtr new_olap_table,
RowBlockChanger* rb_changer,
bool* sc_sorting,
bool* sc_directly);
diff --git a/be/src/olap/segment_writer.cpp b/be/src/olap/segment_writer.cpp
index d801d637..82752753 100644
--- a/be/src/olap/segment_writer.cpp
+++ b/be/src/olap/segment_writer.cpp
@@ -27,7 +27,7 @@ namespace doris {
SegmentWriter::SegmentWriter(
const std::string& file_name,
- OLAPTablePtr table,
+ TabletSharedPtr table,
uint32_t stream_buffer_size) :
_file_name(file_name),
_table(table),
diff --git a/be/src/olap/segment_writer.h b/be/src/olap/segment_writer.h
index c6303888..7fabb72c 100644
--- a/be/src/olap/segment_writer.h
+++ b/be/src/olap/segment_writer.h
@@ -30,7 +30,7 @@ class ColumnDataHeaderMessage;
class SegmentWriter {
public:
explicit SegmentWriter(const std::string& file_name,
- OLAPTablePtr table,
+ TabletSharedPtr table,
uint32_t stream_buffer_size);
~SegmentWriter();
OLAPStatus init(uint32_t write_mbytes_per_sec);
@@ -45,7 +45,7 @@ class SegmentWriter {
OLAPStatus _make_file_header(ColumnDataHeaderMessage* file_header);
std::string _file_name;
- OLAPTablePtr _table;
+ TabletSharedPtr _table;
uint32_t _stream_buffer_size; // 输出缓冲区大小
std::vector<ColumnWriter*> _root_writers;
OutStreamFactory* _stream_factory;
diff --git a/be/src/olap/store.cpp b/be/src/olap/store.cpp
index ccc4a4f1..a154c061 100755
--- a/be/src/olap/store.cpp
+++ b/be/src/olap/store.cpp
@@ -506,7 +506,7 @@ OLAPStatus OlapStore::_load_table_from_header(OLAPEngine*
engine, TTabletId tabl
}
return OLAP_ERR_HEADER_INIT_FAILED;
}
- OLAPTablePtr olap_table =
+ TabletSharedPtr olap_table =
OLAPTable::create_from_header(olap_header.release(), this);
if (olap_table == nullptr) {
LOG(WARNING) << "fail to new table. tablet_id=" << tablet_id << ",
schema_hash:" << schema_hash;
diff --git a/be/src/runtime/snapshot_loader.cpp
b/be/src/runtime/snapshot_loader.cpp
index 4b51babd..1e5ce262 100644
--- a/be/src/runtime/snapshot_loader.cpp
+++ b/be/src/runtime/snapshot_loader.cpp
@@ -556,7 +556,7 @@ Status SnapshotLoader::move(
// than we merge the 2 .hdr file before reloading it.
// load header in tablet dir to get the base vesion
- OLAPTablePtr tablet = OLAPEngine::get_instance()->get_table(
+ TabletSharedPtr tablet = OLAPEngine::get_instance()->get_table(
tablet_id, schema_hash);
if (tablet.get() == NULL) {
std::stringstream ss;
diff --git a/be/test/olap/delete_handler_test.cpp
b/be/test/olap/delete_handler_test.cpp
index 87437fc4..4c2d3171 100644
--- a/be/test/olap/delete_handler_test.cpp
+++ b/be/test/olap/delete_handler_test.cpp
@@ -206,7 +206,7 @@ class TestDeleteConditionHandler : public testing::Test {
typedef RepeatedPtrField<DeleteConditionMessage> del_cond_array;
std::string _tablet_path;
- OLAPTablePtr _olap_table;
+ TabletSharedPtr _olap_table;
TCreateTabletReq _create_tablet;
DeleteConditionHandler _delete_condition_handler;
};
@@ -507,7 +507,7 @@ class TestDeleteConditionHandler2 : public testing::Test {
typedef RepeatedPtrField<DeleteConditionMessage> del_cond_array;
std::string _tablet_path;
- OLAPTablePtr _olap_table;
+ TabletSharedPtr _olap_table;
TCreateTabletReq _create_tablet;
};
@@ -831,7 +831,7 @@ class TestDeleteHandler : public testing::Test {
std::string _tablet_path;
RowCursor _data_row_cursor;
- OLAPTablePtr _olap_table;
+ TabletSharedPtr _olap_table;
TCreateTabletReq _create_tablet;
DeleteHandler _delete_handler;
};
diff --git a/be/test/olap/delta_writer_test.cpp
b/be/test/olap/delta_writer_test.cpp
index 94a35277..c8cb18a8 100644
--- a/be/test/olap/delta_writer_test.cpp
+++ b/be/test/olap/delta_writer_test.cpp
@@ -397,7 +397,7 @@ TEST_F(TestDeltaWriter, write) {
ASSERT_EQ(res, OLAP_SUCCESS);
// publish version success
- OLAPTablePtr table =
OLAPEngine::get_instance()->get_table(write_req.tablet_id,
write_req.schema_hash);
+ TabletSharedPtr table =
OLAPEngine::get_instance()->get_table(write_req.tablet_id,
write_req.schema_hash);
TPublishVersionRequest publish_req;
publish_req.transaction_id = write_req.transaction_id;
TPartitionVersionInfo info;
@@ -591,7 +591,7 @@ TEST_F(TestSchemaChange, schema_change) {
ASSERT_EQ(res, OLAP_SUCCESS);
// publish version success
- OLAPTablePtr table =
OLAPEngine::get_instance()->get_table(write_req.tablet_id,
write_req.schema_hash);
+ TabletSharedPtr table =
OLAPEngine::get_instance()->get_table(write_req.tablet_id,
write_req.schema_hash);
TPublishVersionRequest publish_req;
publish_req.transaction_id = write_req.transaction_id;
TPartitionVersionInfo info;
diff --git a/be/test/olap/mock_command_executor.h
b/be/test/olap/mock_command_executor.h
index 75b1eed6..d293e24d 100644
--- a/be/test/olap/mock_command_executor.h
+++ b/be/test/olap/mock_command_executor.h
@@ -26,7 +26,7 @@ namespace doris {
class MockCommandExecutor : public OLAPEngine {
public:
MOCK_METHOD1(create_table, OLAPStatus(const TCreateTabletReq& request));
- MOCK_METHOD2(get_table, OLAPTablePtr(TTabletId tablet_id, TSchemaHash
schema_hash));
+ MOCK_METHOD2(get_table, TabletSharedPtr(TTabletId tablet_id, TSchemaHash
schema_hash));
MOCK_METHOD1(drop_table, OLAPStatus(const TDropTabletReq& request));
MOCK_METHOD2(
push,
@@ -95,13 +95,13 @@ class MockCommandExecutor : public OLAPEngine {
MOCK_METHOD3(
get_info_before_incremental_clone,
std::string(
- OLAPTablePtr tablet,
+ TabletSharedPtr tablet,
int64_t committed_version,
std::vector<Version>* missing_versions));
MOCK_METHOD4(
finish_clone,
OLAPStatus(
- OLAPTablePtr tablet,
+ TabletSharedPtr tablet,
const std::string& clone_dir,
int64_t committed_version,
bool is_incremental_clone));
diff --git a/be/test/olap/olap_reader_test.cpp
b/be/test/olap/olap_reader_test.cpp
index 8a731b6f..692e96e2 100755
--- a/be/test/olap/olap_reader_test.cpp
+++ b/be/test/olap/olap_reader_test.cpp
@@ -170,7 +170,7 @@ class TestOLAPReaderRow : public testing::Test {
CommandExecutor command_executor = CommandExecutor();
res = command_executor.create_table(_create_tablet);
ASSERT_EQ(OLAP_SUCCESS, res);
- OLAPTablePtr tablet = command_executor.get_table(
+ TabletSharedPtr tablet = command_executor.get_table(
_create_tablet.tablet_id,
_create_tablet.tablet_schema.schema_hash);
ASSERT_TRUE(tablet.get() != NULL);
_tablet_path = tablet->tablet_path();
@@ -706,7 +706,7 @@ class TestOLAPReaderColumn : public testing::Test {
CommandExecutor command_executor = CommandExecutor();
res = command_executor.create_table(_create_tablet);
ASSERT_EQ(OLAP_SUCCESS, res);
- OLAPTablePtr tablet = command_executor.get_table(
+ TabletSharedPtr tablet = command_executor.get_table(
_create_tablet.tablet_id,
_create_tablet.tablet_schema.schema_hash);
ASSERT_TRUE(tablet.get() != NULL);
_tablet_path = tablet->tablet_path();
@@ -1189,7 +1189,7 @@ class TestOLAPReaderColumnDeleteCondition : public
testing::Test {
CommandExecutor command_executor = CommandExecutor();
res = command_executor.create_table(_create_tablet);
ASSERT_EQ(OLAP_SUCCESS, res);
- OLAPTablePtr tablet = command_executor.get_table(
+ TabletSharedPtr tablet = command_executor.get_table(
_create_tablet.tablet_id,
_create_tablet.tablet_schema.schema_hash);
ASSERT_TRUE(tablet.get() != NULL);
_tablet_path = tablet->tablet_path();
diff --git a/be/test/olap/vectorized_olap_reader_test.cpp
b/be/test/olap/vectorized_olap_reader_test.cpp
index 99816e6b..5fe41382 100644
--- a/be/test/olap/vectorized_olap_reader_test.cpp
+++ b/be/test/olap/vectorized_olap_reader_test.cpp
@@ -172,7 +172,7 @@ class TestVectorizedOLAPReader : public testing::Test {
CommandExecutor command_executor = CommandExecutor();
res = command_executor.create_table(_create_tablet);
ASSERT_EQ(OLAP_SUCCESS, res);
- OLAPTablePtr tablet = command_executor.get_table(
+ TabletSharedPtr tablet = command_executor.get_table(
_create_tablet.tablet_id,
_create_tablet.tablet_schema.schema_hash);
ASSERT_TRUE(tablet.get() != NULL);
_tablet_name = tablet->tablet_name();
@@ -202,7 +202,7 @@ class TestVectorizedOLAPReader : public testing::Test {
CommandExecutor command_executor = CommandExecutor();
res = command_executor.create_table(_create_tablet);
ASSERT_EQ(OLAP_SUCCESS, res);
- OLAPTablePtr tablet = command_executor.get_table(
+ TabletSharedPtr tablet = command_executor.get_table(
_create_tablet.tablet_id,
_create_tablet.tablet_schema.schema_hash);
ASSERT_TRUE(tablet.get() != NULL);
_tablet_name = tablet->tablet_name();
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
