chaoyli commented on a change in pull request #454: Refactor Tablet and
TabletMeta
URL: https://github.com/apache/incubator-doris/pull/454#discussion_r243519707
##########
File path: be/src/olap/tablet_meta.cpp
##########
@@ -17,968 +17,197 @@
#include "olap/tablet_meta.h"
-#include <algorithm>
-#include <fstream>
-#include <queue>
-#include <sstream>
-#include <string>
-#include <utility>
-#include <vector>
-
-#include "olap/field.h"
-#include "olap/wrapper_field.h"
-#include "olap/file_helper.h"
-#include "olap/utils.h"
-
-using google::protobuf::RepeatedPtrField;
-using std::ifstream;
-using std::ios;
-using std::list;
-using std::make_pair;
-using std::ofstream;
-using std::queue;
-using std::sort;
-using std::string;
-using std::stringstream;
-using std::unordered_map;
-using std::vector;
-
namespace doris {
-// related static functions of version graph
-
-// Construct version graph(using adjacency list) from header's information.
-static OLAPStatus construct_version_graph(
- const RepeatedPtrField<PDelta>& versions_in_header,
- vector<Vertex>* version_graph,
- unordered_map<int, int>* vertex_helper_map);
-
-// Clear version graph and vertex_helper_map, release memory hold by
version_graph.
-static OLAPStatus clear_version_graph(vector<Vertex>* version_graph,
- unordered_map<int, int>* vertex_helper_map);
-
-// Add version to graph, it is called near the end of add_version
-static OLAPStatus add_version_to_graph(const Version& version,
- vector<Vertex>* version_graph,
- unordered_map<int, int>* vertex_helper_map);
-
-// Delete version from graph, it is called near the end of delete_version
-static OLAPStatus delete_version_from_graph(
- const RepeatedPtrField<PDelta>& versions_in_header,
- const Version& version,
- vector<Vertex>* version_graph,
- unordered_map<int, int>* vertex_helper_map);
-// Add vertex to graph, if vertex already exists, still return SUCCESS.
-static OLAPStatus add_vertex_to_graph(int vertex_value,
- vector<Vertex>* version_graph,
- unordered_map<int, int>* vertex_helper_map);
-
-TabletMeta::~TabletMeta() {
- // Release memory of version graph.
- clear_version_graph(&_version_graph, &_vertex_helper_map);
- Clear();
+TabletMeta::TabletMeta(DataDir* data_dir) {
+ _data_dir = data_dir;
}
-OLAPStatus TabletMeta::init() {
- clear_version_graph(&_version_graph, &_vertex_helper_map);
- if (construct_version_graph(delta(),
- &_version_graph,
- &_vertex_helper_map) != OLAP_SUCCESS) {
- OLAP_LOG_WARNING("fail to construct version graph.");
- return OLAP_ERR_OTHER_ERROR;
- }
- if (_file_name == "") {
- stringstream file_stream;
- file_stream << tablet_id() << ".hdr";
- _file_name = file_stream.str();
- }
- return OLAP_SUCCESS;
-}
+NewStatus TabletMeta::serialize(string* meta_binary) {
+ std::lock_guard<std::mutex> lock(_mutex);
+ return serialize_unlock(meta_binary);
+};
-OLAPStatus TabletMeta::load_and_init() {
- // check the tablet_path is not empty
- if (_file_name == "") {
- LOG(WARNING) << "file_path is empty for header";
- return OLAP_ERR_DIR_NOT_EXIST;
- }
-
- FileHeader<OLAPHeaderMessage> file_header;
- FileHandler file_handler;
+NewStatus TabletMeta::serialize_unlock(string* meta_binary) {
+ _tablet_meta_pb.SerializeToString(meta_binary);
+ return NewStatus::OK();
+};
- if (file_handler.open(_file_name.c_str(), O_RDONLY) != OLAP_SUCCESS) {
- LOG(WARNING) << "fail to open index file. [file='" << _file_name <<
"']";
- return OLAP_ERR_IO_ERROR;
- }
-
- // In file_header.unserialize(), it validates file length, signature,
checksum of protobuf.
- if (file_header.unserialize(&file_handler) != OLAP_SUCCESS) {
- LOG(WARNING) << "fail to unserialize header. [path='" << _file_name <<
"']";
- return OLAP_ERR_PARSE_PROTOBUF_ERROR;
- }
-
- try {
- CopyFrom(file_header.message());
- } catch (...) {
- LOG(WARNING) << "fail to copy protocol buffer object. [path='" <<
_file_name << "']";
- return OLAP_ERR_PARSE_PROTOBUF_ERROR;
- }
-
- return init();
-}
-
-OLAPStatus TabletMeta::save() {
- return save(_file_name);
+NewStatus TabletMeta::deserialize(string* meta_binary) {
+ std::lock_guard<std::mutex> lock(_mutex);
+ return deserialize_unlock(meta_binary);
}
-OLAPStatus TabletMeta::save(const string& file_path) {
- // check the tablet_path is not empty
- if (file_path == "") {
- LOG(WARNING) << "file_path is empty for header";
- return OLAP_ERR_DIR_NOT_EXIST;
- }
-
- FileHeader<OLAPHeaderMessage> file_header;
- FileHandler file_handler;
-
- if (file_handler.open_with_mode(file_path.c_str(),
- O_CREAT | O_WRONLY | O_TRUNC, S_IRUSR | S_IWUSR) != OLAP_SUCCESS) {
- LOG(WARNING) << "fail to open header file. [file='" << file_path <<
"']";
- return OLAP_ERR_IO_ERROR;
- }
-
- try {
- file_header.mutable_message()->CopyFrom(*this);
- } catch (...) {
- LOG(WARNING) << "fail to copy protocol buffer object. [path='" <<
file_path << "']";
- return OLAP_ERR_OTHER_ERROR;
- }
-
- if (file_header.prepare(&file_handler) != OLAP_SUCCESS
- || file_header.serialize(&file_handler) != OLAP_SUCCESS) {
- LOG(WARNING) << "fail to serialize to file header. [path='" <<
file_path << "']";
- return OLAP_ERR_SERIALIZE_PROTOBUF_ERROR;
+NewStatus TabletMeta::deserialize_unlock(const string& meta_binary) {
+ _tablet_meta_pb.ParseFromString(meta_binary);
+ _table_id = _tablet_meta_pb.table_id();
+ _partition_id = _tablet_meta_pb.partition_id();
+ _tablet_id = _tablet_meta_pb.tablet_id();
+ _schema_hash = _tablet_meta_pb.schema_hash();
+ _shard_id = _tablet_meta_pb.shard_id();
+ RETURN_NOT_OK(_schema.deserialize_from_pb(_tablet_meta_pb.schema()));
+ for (auto& it : _tablet_meta_pb.rs_metas()) {
+ RowsetMeta rs_meta;
+ rs_meta.deserialize_from_pb(it);
+ _rs_metas.push_back(std::move(rs_meta));
}
-
- return OLAP_SUCCESS;
-}
-
-OLAPStatus TabletMeta::add_version(Version version, VersionHash version_hash,
- int32_t segment_group_id, int32_t num_segments,
- int64_t index_size, int64_t data_size, int64_t num_rows,
- bool empty, const std::vector<KeyRange>* column_statistics) {
- // Check whether version is valid.
- if (version.first > version.second) {
- LOG(WARNING) << "the version is not valid."
- << "version=" << version.first << "-" << version.second;
- return OLAP_ERR_HEADER_ADD_VERSION;
+ for (auto& it : _tablet_meta_pb.inc_rs_metas()) {
+ RowsetMeta rs_meta;
+ rs_meta.deserialize_from_pb(it);
+ _rs_metas.push_back(std::move(rs_meta));
}
- int delta_id = 0;
- for (int i = 0; i < delta_size(); ++i) {
- if (delta(i).start_version() == version.first
- && delta(i).end_version() == version.second) {
- for (const PSegmentGroup& segment_group :
delta(i).segment_group()) {
- if (segment_group.segment_group_id() == segment_group_id) {
- LOG(WARNING) << "the version is existed."
- << "version=" << version.first << "-"
- << version.second;
- return OLAP_ERR_HEADER_ADD_VERSION;
- }
- }
- delta_id = i;
+ // generate TabletState
+ switch (_tablet_meta_pb.tablet_state()) {
+ case NOT_READY:
+ _tablet_state = TabletState::kNotReady;
break;
- }
- }
-
- // if segment_group_id is greater or equal than zero, it is used
- // to streaming load
-
- // Try to add version to protobuf.
- PDelta* new_delta = nullptr;
- try {
- if (segment_group_id == -1 || segment_group_id == 0) {
- // snapshot will use segment_group_id which equals minus one
- new_delta = add_delta();
- new_delta->set_start_version(version.first);
- new_delta->set_end_version(version.second);
- new_delta->set_version_hash(version_hash);
- new_delta->set_creation_time(time(NULL));
- } else {
- new_delta = const_cast<PDelta*>(&delta(delta_id));
- }
- PSegmentGroup* new_segment_group = new_delta->add_segment_group();
- new_segment_group->set_segment_group_id(segment_group_id);
- new_segment_group->set_num_segments(num_segments);
- new_segment_group->set_index_size(index_size);
- new_segment_group->set_data_size(data_size);
- new_segment_group->set_num_rows(num_rows);
- new_segment_group->set_empty(empty);
- if (NULL != column_statistics) {
- for (size_t i = 0; i < column_statistics->size(); ++i) {
- ColumnPruning *column_pruning =
- new_segment_group->add_column_pruning();
-
column_pruning->set_min(column_statistics->at(i).first->to_string());
-
column_pruning->set_max(column_statistics->at(i).second->to_string());
-
column_pruning->set_null_flag(column_statistics->at(i).first->is_null());
- }
- }
- } catch (...) {
- OLAP_LOG_WARNING("add file version to protobf error");
- return OLAP_ERR_HEADER_ADD_VERSION;
- }
-
- if (add_version_to_graph(version, &_version_graph, &_vertex_helper_map) !=
OLAP_SUCCESS) {
- OLAP_LOG_WARNING("fail to add version to graph. [version='%d-%d']",
- version.first,
- version.second);
- return OLAP_ERR_HEADER_ADD_VERSION;
- }
-
- return OLAP_SUCCESS;
-}
-
-OLAPStatus TabletMeta::add_pending_version(
- int64_t partition_id, int64_t transaction_id,
- const std::vector<string>* delete_conditions) {
- for (int i = 0; i < pending_delta_size(); ++i) {
- if (pending_delta(i).transaction_id() == transaction_id) {
- LOG(WARNING) << "pending delta already exists in header."
- << "transaction_id: " << transaction_id;
- return OLAP_ERR_HEADER_ADD_PENDING_DELTA;
- }
- }
-
- try {
- PPendingDelta* new_pending_delta = add_pending_delta();
- new_pending_delta->set_partition_id(partition_id);
- new_pending_delta->set_transaction_id(transaction_id);
- new_pending_delta->set_creation_time(time(NULL));
-
- if (delete_conditions != nullptr) {
- DeleteConditionMessage* del_cond =
new_pending_delta->mutable_delete_condition();
- del_cond->set_version(0);
- for (const string& condition : *delete_conditions) {
- del_cond->add_sub_conditions(condition);
- LOG(INFO) << "store one sub-delete condition. condition=" <<
condition
- << ", transaction_id=" << transaction_id;
- }
- }
-
- } catch (...) {
- LOG(WARNING) << "fail to add pending segment_group to header protobf";
- return OLAP_ERR_HEADER_ADD_PENDING_DELTA;
- }
-
- return OLAP_SUCCESS;
-}
-
-OLAPStatus TabletMeta::add_pending_segment_group(
- int64_t transaction_id, int32_t num_segments,
- int32_t pending_segment_group_id, const PUniqueId& load_id,
- bool empty, const std::vector<KeyRange>* column_statistics) {
-
- int32_t delta_id = 0;
- for (int32_t i = 0; i < pending_delta_size(); ++i) {
- const PPendingDelta& delta = pending_delta(i);
- if (delta.transaction_id() == transaction_id) {
- delta_id = i;
- for (int j = 0; j < delta.pending_segment_group_size(); ++j) {
- const PPendingSegmentGroup& pending_segment_group =
delta.pending_segment_group(j);
- if (pending_segment_group.pending_segment_group_id() ==
pending_segment_group_id) {
- LOG(WARNING) << "pending segment_group already exists in
header."
- << "transaction_id:" << transaction_id
- << ", pending_segment_group_id: " <<
pending_segment_group_id;
- return OLAP_ERR_HEADER_ADD_PENDING_DELTA;
- }
- }
- }
- }
-
- try {
- PPendingSegmentGroup* new_pending_segment_group
- =
const_cast<PPendingDelta&>(pending_delta(delta_id)).add_pending_segment_group();
-
new_pending_segment_group->set_pending_segment_group_id(pending_segment_group_id);
- new_pending_segment_group->set_num_segments(num_segments);
- new_pending_segment_group->mutable_load_id()->set_hi(load_id.hi());
- new_pending_segment_group->mutable_load_id()->set_lo(load_id.lo());
- new_pending_segment_group->set_empty(empty);
- if (NULL != column_statistics) {
- for (size_t i = 0; i < column_statistics->size(); ++i) {
- ColumnPruning *column_pruning =
- new_pending_segment_group->add_column_pruning();
-
column_pruning->set_min(column_statistics->at(i).first->to_string());
-
column_pruning->set_max(column_statistics->at(i).second->to_string());
-
column_pruning->set_null_flag(column_statistics->at(i).first->is_null());
- }
- }
- } catch (...) {
- OLAP_LOG_WARNING("fail to add pending segment_group to protobf");
- return OLAP_ERR_HEADER_ADD_PENDING_DELTA;
- }
-
- return OLAP_SUCCESS;
-}
-
-OLAPStatus TabletMeta::add_incremental_version(Version version, VersionHash
version_hash,
- int32_t segment_group_id, int32_t num_segments,
- int64_t index_size, int64_t data_size, int64_t num_rows,
- bool empty, const std::vector<KeyRange>* column_statistics) {
- // Check whether version is valid.
- if (version.first != version.second) {
- OLAP_LOG_WARNING("the incremental version is not valid. [version=%d]",
version.first);
- return OLAP_ERR_HEADER_ADD_INCREMENTAL_VERSION;
- }
-
- // Check whether the version is existed.
- int32_t delta_id = 0;
- for (int i = 0; i < incremental_delta_size(); ++i) {
- const PDelta& incre_delta = incremental_delta(i);
- if (incre_delta.start_version() == version.first) {
- delta_id = i;
- for (int j = 0; j < incre_delta.segment_group_size(); ++j) {
- const PSegmentGroup& incremental_segment_group =
incre_delta.segment_group(j);
- if (incremental_segment_group.segment_group_id() ==
segment_group_id) {
- LOG(WARNING) << "segment_group already exists in header."
- << "version: " << version.first << "-" <<
version.second << ","
- << "segment_group_id: " << segment_group_id;
- return OLAP_ERR_HEADER_ADD_PENDING_DELTA;
- }
- }
- }
- }
-
- // Try to add version to protobuf.
- try {
- PDelta* new_incremental_delta = nullptr;
- if (segment_group_id == 0) {
- new_incremental_delta = add_incremental_delta();
- new_incremental_delta->set_start_version(version.first);
- new_incremental_delta->set_end_version(version.second);
- new_incremental_delta->set_version_hash(version_hash);
- new_incremental_delta->set_creation_time(time(NULL));
- } else {
- new_incremental_delta =
const_cast<PDelta*>(&incremental_delta(delta_id));
- }
- PSegmentGroup* new_incremental_segment_group =
new_incremental_delta->add_segment_group();
- new_incremental_segment_group->set_segment_group_id(segment_group_id);
- new_incremental_segment_group->set_num_segments(num_segments);
- new_incremental_segment_group->set_index_size(index_size);
- new_incremental_segment_group->set_data_size(data_size);
- new_incremental_segment_group->set_num_rows(num_rows);
- new_incremental_segment_group->set_empty(empty);
- if (NULL != column_statistics) {
- for (size_t i = 0; i < column_statistics->size(); ++i) {
- ColumnPruning *column_pruning =
- new_incremental_segment_group->add_column_pruning();
-
column_pruning->set_min(column_statistics->at(i).first->to_string());
-
column_pruning->set_max(column_statistics->at(i).second->to_string());
-
column_pruning->set_null_flag(column_statistics->at(i).first->is_null());
- }
- }
- } catch (...) {
- OLAP_LOG_WARNING("add incremental version to protobf error");
- return OLAP_ERR_HEADER_ADD_INCREMENTAL_VERSION;
- }
-
- return OLAP_SUCCESS;
-}
-
-void TabletMeta::add_delete_condition(const DeleteConditionMessage&
delete_condition,
- int64_t version) {
- // check whether condition exist
- DeleteConditionMessage* del_cond = NULL;
- int i = 0;
- for (; i < delete_data_conditions_size(); i++) {
- DeleteConditionMessage temp = delete_data_conditions().Get(i);
- if (temp.version() == version) {
+ case RUNNING:
+ _tablet_state = TabletState::kRunning;
break;
- }
- }
-
- // clear existed condition
- if (i < delete_data_conditions_size()) {
- del_cond = mutable_delete_data_conditions(i);
- del_cond->clear_sub_conditions();
- } else {
- del_cond = add_delete_data_conditions();
- del_cond->set_version(version);
- }
-
- for (const string& condition : delete_condition.sub_conditions()) {
- del_cond->add_sub_conditions(condition);
- }
- LOG(INFO) << "add delete condition. version=" << version;
-}
-
-const PPendingDelta* TabletMeta::get_pending_delta(int64_t transaction_id)
const {
- for (int i = 0; i < pending_delta_size(); i++) {
- if (pending_delta(i).transaction_id() == transaction_id) {
- return &pending_delta(i);
- }
- }
- return nullptr;
-}
-
-const PPendingSegmentGroup* TabletMeta::get_pending_segment_group(int64_t
transaction_id,
- int32_t pending_segment_group_id) const {
- for (int i = 0; i < pending_delta_size(); i++) {
- if (pending_delta(i).transaction_id() == transaction_id) {
- const PPendingDelta& delta = pending_delta(i);
- for (int j = 0; j < delta.pending_segment_group_size(); ++j) {
- const PPendingSegmentGroup& pending_segment_group =
delta.pending_segment_group(j);
- if (pending_segment_group.pending_segment_group_id() ==
pending_segment_group_id) {
- return &pending_segment_group;
- }
- }
- }
- }
- return nullptr;
-}
-
-const PDelta* TabletMeta::get_incremental_version(Version version) const {
- for (int i = 0; i < incremental_delta_size(); i++) {
- if (incremental_delta(i).start_version() == version.first
- && incremental_delta(i).end_version() == version.second) {
- return &incremental_delta(i);
- }
- }
- return nullptr;
-}
-
-OLAPStatus TabletMeta::delete_version(Version version) {
- // Find the version that need to be deleted.
- int index = -1;
- for (int i = 0; i < delta_size(); ++i) {
- if (delta(i).start_version() == version.first
- && delta(i).end_version() == version.second) {
- index = i;
+ case TOMBSTONED:
+ _tablet_state = TabletState::kTombstoned;
break;
- }
- }
-
- // Delete version from protobuf.
- if (index != -1) {
- RepeatedPtrField<PDelta>* version_ptr = mutable_delta();
- for (int i = index; i < delta_size() - 1; ++i) {
- version_ptr->SwapElements(i, i + 1);
- }
-
- version_ptr->RemoveLast();
- }
-
- // Atomic delete is not supported now.
- if (delete_version_from_graph(delta(), version,
- &_version_graph,
- &_vertex_helper_map) != OLAP_SUCCESS) {
- OLAP_LOG_WARNING("fail to delete version from graph.
[version='%d-%d']",
- version.first,
- version.second);
- return OLAP_ERR_HEADER_DELETE_VERSION;
- }
-
- return OLAP_SUCCESS;
-}
-
-OLAPStatus TabletMeta::delete_all_versions() {
- clear_file_version();
- clear_delta();
- clear_pending_delta();
- clear_incremental_delta();
- clear_version_graph(&_version_graph, &_vertex_helper_map);
-
- if (construct_version_graph(delta(),
- &_version_graph,
- &_vertex_helper_map) != OLAP_SUCCESS) {
- OLAP_LOG_WARNING("fail to construct version graph.");
- return OLAP_ERR_OTHER_ERROR;
- }
- return OLAP_SUCCESS;
-}
-
-void TabletMeta::delete_pending_delta(int64_t transaction_id) {
- int index = -1;
- for (int i = 0; i < pending_delta_size(); ++i) {
- if (pending_delta(i).transaction_id() == transaction_id) {
- index = i;
+ case STOPPED:
+ _tablet_state = TabletState::kStopped;
break;
- }
- }
-
- if (index != -1) {
- RepeatedPtrField<PPendingDelta>* pending_delta_ptr =
mutable_pending_delta();
- for (int i = index; i < pending_delta_size() - 1; ++i) {
- pending_delta_ptr->SwapElements(i, i + 1);
- }
-
- pending_delta_ptr->RemoveLast();
- }
-}
-
-void TabletMeta::delete_incremental_delta(Version version) {
- int index = -1;
- for (int i = 0; i < incremental_delta_size(); ++i) {
- if (incremental_delta(i).start_version() == version.first
- && incremental_delta(i).end_version() == version.second) {
- index = i;
+ case SHUTDOWN:
+ _tablet_state = TabletState::kShutdown;
break;
- }
+ default:
+ LOG(WARNNING) << "tablet has no state. tablet=" << _tablet_id
+ << ", schema_hash=" << _schema_hash;
}
- if (index != -1) {
- RepeatedPtrField<PDelta>* version_ptr = mutable_incremental_delta();
- for (int i = index; i < incremental_delta_size() - 1; ++i) {
- version_ptr->SwapElements(i, i + 1);
- }
-
- version_ptr->RemoveLast();
- }
+ // generate AlterTabletTask
+
RETURN(_alter_task.deserialize_from_pb(_tablet_meta_pb.alter_tablet_task()));
+ return NewStatus::OK();
}
-// This function is called when base-compaction, cumulative-compaction,
quering.
-// we use BFS algorithm to get the shortest version path.
-OLAPStatus TabletMeta::select_versions_to_span(const Version& target_version,
- vector<Version>* span_versions) {
- if (target_version.first > target_version.second) {
- OLAP_LOG_WARNING("invalid param target_version. [start_version_id=%d
end_version_id=%d]",
- target_version.first,
- target_version.second);
- return OLAP_ERR_INPUT_PARAMETER_ERROR;
- }
-
- if (span_versions == NULL) {
- OLAP_LOG_WARNING("param span_versions is NULL.");
- return OLAP_ERR_INPUT_PARAMETER_ERROR;
- }
-
- // bfs_queue's element is vertex_index.
- queue<int> bfs_queue;
- // predecessor[i] means the predecessor of vertex_index 'i'.
- vector<int> predecessor(_version_graph.size());
- // visited[int]==true means it had entered bfs_queue.
- vector<bool> visited(_version_graph.size());
- // [start_vertex_value, end_vertex_value)
- int start_vertex_value = target_version.first;
- int end_vertex_value = target_version.second + 1;
- // -1 is invalid vertex index.
- int start_vertex_index = -1;
- // -1 is valid vertex index.
- int end_vertex_index = -1;
-
- for (size_t i = 0; i < _version_graph.size(); ++i) {
- if (_version_graph[i].value == start_vertex_value) {
- start_vertex_index = i;
- }
- if (_version_graph[i].value == end_vertex_value) {
- end_vertex_index = i;
- }
- }
-
- if (start_vertex_index < 0 || end_vertex_index < 0) {
- OLAP_LOG_WARNING("fail to find version in version list. "
- "[start_version_id=%d end_version_id=%d tmp_start=%d
tmp_end=%d]",
- target_version.first,
- target_version.second,
- start_vertex_index,
- end_vertex_index);
- return OLAP_ERR_VERSION_NOT_EXIST;
- }
-
- for (int i = 0; i < static_cast<int>(_version_graph.size()); ++i) {
- visited[i] = false;
- }
-
- bfs_queue.push(start_vertex_index);
- visited[start_vertex_index] = true;
- // The predecessor of root is itself.
- predecessor[start_vertex_index] = start_vertex_index;
-
- while (bfs_queue.empty() == false && visited[end_vertex_index] == false) {
- int top_vertex_index = bfs_queue.front();
- bfs_queue.pop();
-
- for (list<int>::const_iterator it =
_version_graph[top_vertex_index].edges->begin();
- it != _version_graph[top_vertex_index].edges->end(); ++it) {
- if (visited[*it] == false) {
- // If we don't support reverse version in the path, and start
vertex
- // value is larger than the end vertex value, we skip this
edge.
- if (_version_graph[top_vertex_index].value >
_version_graph[*it].value) {
- continue;
- }
-
- visited[*it] = true;
- predecessor[*it] = top_vertex_index;
- bfs_queue.push(*it);
- }
- }
- }
-
- if (visited[end_vertex_index] == false) {
- OLAP_LOG_WARNING("fail to find path to end_version in version list. "
- "[start_version_id=%d end_version_id=%d]",
- target_version.first,
- target_version.second);
- return OLAP_ERR_VERSION_NOT_EXIST;
- }
-
- vector<int> reversed_path;
- int tmp_vertex_index = end_vertex_index;
- reversed_path.push_back(tmp_vertex_index);
-
- // For start_vertex_index, its predecessor must be itself.
- while (predecessor[tmp_vertex_index] != tmp_vertex_index) {
- tmp_vertex_index = predecessor[tmp_vertex_index];
- reversed_path.push_back(tmp_vertex_index);
- }
-
- // Make span_versions from reversed_path.
- stringstream shortest_path_for_debug;
- for (int path_id = reversed_path.size() - 1; path_id > 0; --path_id) {
- int tmp_start_vertex_value =
_version_graph[reversed_path[path_id]].value;
- int tmp_end_vertex_value = _version_graph[reversed_path[path_id -
1]].value;
-
- // tmp_start_vertex_value mustn't be equal to tmp_end_vertex_value
- if (tmp_start_vertex_value <= tmp_end_vertex_value) {
- span_versions->push_back(make_pair(tmp_start_vertex_value,
tmp_end_vertex_value - 1));
- } else {
- span_versions->push_back(make_pair(tmp_end_vertex_value,
tmp_start_vertex_value - 1));
- }
-
- shortest_path_for_debug << (*span_versions)[span_versions->size() -
1].first << '-'
- << (*span_versions)[span_versions->size() -
1].second << ' ';
- }
-
- VLOG(10) << "calculated shortest path. "
- << "version=" << target_version.first << "-" <<
target_version.second
- << "path=" << shortest_path_for_debug.str();
-
- return OLAP_SUCCESS;
+NewStatus TabletMeta::save_meta() {
+ std::lock_guard<std::mutex> lock(_mutex);
+ return save_meta_unlock();
}
-const PDelta* TabletMeta::get_lastest_delta_version() const {
- if (delta_size() == 0) {
- return nullptr;
- }
-
- const PDelta* max_delta = nullptr;
- for (int i = delta_size() - 1; i >= 0; --i) {
- if (delta(i).start_version() == delta(i).end_version()) {
- if (max_delta == nullptr) {
- max_delta = &delta(i);
- } else if (delta(i).start_version() > max_delta->start_version()) {
- max_delta = &delta(i);
- }
- }
+NewStatus TabletMeta::save_meta_unlock() {
+ string meta_binary;
+ serialize_unlock(meta_binary);
+ NewStatus status = TabletMetaManager::save(_data_dir, _tablet_id,
_schema_hash, meta_binary);
+ if (!status.ok()) {
+ LOG(WARNING) << "fail to save tablet_meta. status=" <<
status.to_string()
+ << ", tablet_id=" << _tablet_id
+ << ", schema_hash=" << _schema_hash;
}
- if (max_delta != nullptr) {
- LOG(INFO) << "max_delta:" << max_delta->start_version() << ","
- << max_delta->end_version();
- }
- return max_delta;
+ return status;
}
-const PDelta* TabletMeta::get_lastest_version() const {
- if (delta_size() == 0) {
- return nullptr;
- }
-
- const PDelta* max_delta = nullptr;
- for (int i = delta_size() - 1; i >= 0; --i) {
- if (max_delta == nullptr) {
- max_delta = &delta(i);
- } else if (delta(i).end_version() > max_delta->end_version()) {
- max_delta = &delta(i);
- } else if (delta(i).end_version() == max_delta->end_version()
- && delta(i).start_version() == delta(i).end_version()) {
- max_delta = &delta(i);
- }
- }
- return max_delta;
+NewStatus TabletMeta::to_meta_pb(TabletMetaPB* tablet_meta_pb) {
+ std::lock_guard<std::mutex> lock(_mutex);
+ return to_meta_pb_unlock(tablet_meta_pb);
}
-Version TabletMeta::get_latest_version() const {
- auto delta = get_lastest_version();
- return {delta->start_version(), delta->end_version()};
-}
+NewStatus TabletMeta::to_meta_pb_unlock(TabletMetaPB* tablet_meta_pb) {
+ tablet_meta_pb->set_table_id(_table_id);
+ tablet_meta_pb->set_partition_id(_partition_id);
+ tablet_meta_pb->set_table_id(_tablet_id);
+ tablet_meta_pb->set_schema_hash(_schema_hash);
+ tablet_meta_pb->set_shard_id(_shard_id);
-const PDelta* TabletMeta::get_delta(int index) const {
- if (delta_size() == 0) {
- return nullptr;
+ tablet_meta_pb->set_tablet_name(_tablet_name);
Review comment:
table_name may not stored in PB, otherwise cannot to be renamed.
----------------------------------------------------------------
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]
