database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[fix](group commit) Fix compatibility issues on serializing and deserializing wal file (#32299)

Browse Source
This commit is contained in:
huanghaibin
2024-03-19 22:04:45 +08:00
committed by yiguolei
parent 2057886d30
commit 2196c534e8
16 changed files with 78 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
be/src/runtime/group_commit_mgr.cpp
View File

@ -314,8 +314,12 @@ Status GroupCommitTable::_create_group_commit_load(int be_exe_version) {
st = _exec_plan_fragment(_db_id, _table_id, label, txn_id, is_pipeline, params,
pipeline_params);
if (!st.ok()) {
static_cast<void>(_finish_group_commit_load(_db_id, _table_id, label, txn_id, instance_id,
st, nullptr));
auto finish_st = _finish_group_commit_load(_db_id, _table_id, label, txn_id, instance_id,
st, nullptr);
if (!finish_st.ok()) {
LOG(WARNING) << "finish group commit error, label=" << label
<< ", st=" << finish_st.to_string();
}
}
return st;
}
@ -432,8 +436,12 @@ Status GroupCommitTable::_exec_plan_fragment(int64_t db_id, int64_t table_id,
const TExecPlanFragmentParams& params,
const TPipelineFragmentParams& pipeline_params) {
auto finish_cb = [db_id, table_id, label, txn_id, this](RuntimeState* state, Status* status) {
static_cast<void>(_finish_group_commit_load(db_id, table_id, label, txn_id,
state->fragment_instance_id(), *status, state));
auto finish_st = _finish_group_commit_load(db_id, table_id, label, txn_id,
state->fragment_instance_id(), *status, state);
if (!finish_st.ok()) {
LOG(WARNING) << "finish group commit error, label=" << label
<< ", st=" << finish_st.to_string();
}
};
if (is_pipeline) {
return _exec_env->fragment_mgr()->exec_plan_fragment(pipeline_params, finish_cb);
@ -512,8 +520,8 @@ Status LoadBlockQueue::create_wal(int64_t db_id, int64_t tb_id, int64_t wal_id,
std::string real_label = config::group_commit_wait_replay_wal_finish
? import_label + "_test_wait"
: import_label;
RETURN_IF_ERROR(ExecEnv::GetInstance()->wal_mgr()->create_wal_path(db_id, tb_id, wal_id,
real_label, _wal_base_path));
RETURN_IF_ERROR(ExecEnv::GetInstance()->wal_mgr()->create_wal_path(
db_id, tb_id, wal_id, real_label, _wal_base_path, WAL_VERSION));
_v_wal_writer = std::make_shared<vectorized::VWalWriter>(
db_id, tb_id, wal_id, real_label, wal_manager, slot_desc, be_exe_version);
return _v_wal_writer->init();