hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity
Files
generic
oceanbase/tools/ob_admin/clog_tool/ob_log_entry_parser.cpp

2305 lines
87 KiB
C++
Raw Permalink Blame History

/**
* Copyright (c) 2021 OceanBase
* OceanBase CE is licensed under Mulan PubL v2.
* You can use this software according to the terms and conditions of the Mulan PubL v2.
* You may obtain a copy of Mulan PubL v2 at:
* http://license.coscl.org.cn/MulanPubL-2.0
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
* See the Mulan PubL v2 for more details.
*/
#include <string.h>
#include "ob_log_entry_parser.h"
#include "clog/ob_info_block_handler.h"
#include "storage/memtable/ob_memtable_context.h"
#include "storage/ob_pg_log.h"
#include "share/ob_rpc_struct.h"
#include "clog/ob_log_compress.h"
#include "share/ob_encrypt_kms.h"
#include "share/ob_encryption_util.h"
namespace oceanbase {
using namespace common;
using namespace share;
using namespace memtable;
using namespace storage;
namespace clog {
int ObLogStat::init()
{
int ret = OB_SUCCESS;
const int64_t HASH_BUCKET_NUM = 241;
if (OB_FAIL(compressed_tenant_ids_.create(HASH_BUCKET_NUM))) {
CLOG_LOG(ERROR, "failed to init");
}
return ret;
}
bool ObInfoEntryDumpFunctor::operator()(const ObPartitionKey& partition_key, const uint64_t min_log_id)
{
int bool_ret = true;
int ret = OB_SUCCESS;
if ((!partition_key.is_valid()) || OB_INVALID_FILE_ID == file_id_) {
ret = OB_INVALID_ARGUMENT;
bool_ret = false;
CLOG_LOG(ERROR, "ObInfoEntryDumpFunctor get invalid argument", K(ret), K(partition_key), K(file_id_));
} else {
fprintf(stdout,
"||IndexInfoBlock|file_id:%lu|partition_key:%s|min_log_id:%ld|\n",
file_id_,
to_cstring(partition_key),
min_log_id);
}
return bool_ret;
}
int ObLogEntryParser::init(uint64_t file_id, char* buf, int64_t buf_len, const ObLogEntryFilter& filter,
const common::ObString& host, const int32_t port, const char* config_file, const bool is_ofs)
{
int ret = OB_SUCCESS;
if (OB_UNLIKELY(is_inited_)) {
ret = OB_INIT_TWICE;
} else if (OB_ISNULL(buf) || OB_UNLIKELY(buf_len <= 0) || OB_UNLIKELY(OB_INVALID_FILE_ID == file_id)) {
ret = OB_INVALID_ARGUMENT;
CLOG_LOG(WARN, "invalid buf or buf_len", KP(buf), K(buf_len), K(file_id), K(ret));
} else if (OB_ISNULL(compress_rbuf_.buf_ = static_cast<char*>(
allocator_.alloc_aligned(OB_MALLOC_BIG_BLOCK_SIZE, CLOG_DIO_ALIGN_SIZE)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
CLOG_LOG(WARN, "failed to allocate memory", K(ret));
} else if (OB_FAIL(log_stat_.init())) {
CLOG_LOG(WARN, "failed to init log_stat", K(file_id));
} else if (OB_FAIL(ObLogEntryParserImpl::init(file_id, filter, host, port, config_file))) {
CLOG_LOG(WARN, "failed to init ObLogEntryParserImpl");
} else {
is_inited_ = true;
log_file_type_ = OB_UNKNOWN_FILE_TYPE;
buf_cur_ = buf;
buf_end_ = buf + buf_len;
is_ofs_ = is_ofs;
compress_rbuf_.buf_len_ = OB_MALLOC_BIG_BLOCK_SIZE;
}
return ret;
}
void ObLogEntryParser::advance_(const int64_t step)
{
cur_offset_ += static_cast<offset_t>(step);
buf_cur_ += step;
CLOG_LOG(DEBUG, "advance", K(step));
}
int ObLogEntryParser::advance_to_next_align_()
{
int ret = OB_SUCCESS;
bool break_on_fail = true;
CLOG_LOG(INFO, "DEBUG for break", K(cur_offset_));
const char* env_val = getenv("CLOG_TOOL_BREAK_ON_FAIL");
if (NULL != env_val) {
break_on_fail = strcasecmp(env_val, "false") ? true : false;
}
if (break_on_fail) {
ret = OB_ITER_END;
} else {
ObLogBlockMetaV2 block_meta;
const int64_t step = SKIP_STEP - cur_offset_ % SKIP_STEP;
if ((buf_end_ - buf_cur_) < step + block_meta.get_serialize_size()) {
ret = OB_ITER_END;
} else {
advance_(step);
}
}
return ret;
}
int ObLogEntryParser::dump_block_(const ObLogBlockMetaV2& meta)
{
int ret = OB_SUCCESS;
if (!meta.check_meta_checksum()) {
// fatal error, data corrupt, but go on
CLOG_LOG(ERROR, "check block meta checksum error", K(ret), K(meta));
advance_(meta.get_serialize_size());
} else {
if (!filter_.is_valid()) {
fprintf(stdout,
"$$$file_id:%lu offset:%ld len:%ld BlockMeta: %s|\n",
file_id_,
cur_offset_,
meta.get_serialize_size(),
to_cstring(meta));
}
ObBlockType type = meta.get_block_type();
switch (type) {
case OB_DATA_BLOCK:
advance_(meta.get_serialize_size());
break;
case OB_HEADER_BLOCK: // not used any more
// file header is CLOG_DIO_ALIGN_SIZE bytes,
// skip it
advance_(meta.get_total_len());
break;
case OB_INFO_BLOCK: {
if (OB_UNKNOWN_FILE_TYPE == log_file_type_) {
log_file_type_ = OB_CLOG_FILE_TYPE;
}
if (0 == cur_offset_) {
is_ofs_ = true;
}
advance_(meta.get_serialize_size());
int64_t pos = 0;
if (OB_CLOG_FILE_TYPE == log_file_type_) {
ObCommitInfoBlockHandler commit_info_block_handler;
if (OB_FAIL(commit_info_block_handler.init())) {
CLOG_LOG(WARN, "failed to init clog handler", K(ret));
} else if (OB_FAIL(commit_info_block_handler.resolve_info_block(buf_cur_, buf_end_ - buf_cur_, pos))) {
CLOG_LOG(WARN, "failed to resolve_info_block in clog file", K(ret));
} else {
CLOG_LOG(INFO, "dump info block success", K(meta), K(is_ofs_));
advance_(meta.get_data_len());
}
} else if (OB_ILOG_FILE_TYPE == log_file_type_) {
ObIndexInfoBlockHandler index_info_block_handler;
if (OB_FAIL(index_info_block_handler.init())) {
CLOG_LOG(WARN, "failed to init clog handler", K(ret));
} else if (OB_FAIL(index_info_block_handler.resolve_info_block(buf_cur_, buf_end_ - buf_cur_, pos))) {
CLOG_LOG(WARN, "failed to resolve_info_block in ilog file", K(ret));
} else {
ObInfoEntryDumpFunctor fn(file_id_);
ObIndexInfoBlockHandler::MinLogIdInfo min_log_id_info;
if (OB_FAIL(index_info_block_handler.get_all_min_log_id_info(min_log_id_info))) {
CLOG_LOG(WARN, "get_all_min_log_id_info failed", K(ret));
} else if (OB_FAIL(min_log_id_info.for_each(fn))) {
CLOG_LOG(ERROR, "hash_map parse fail: for_each", K(ret));
} else {
CLOG_LOG(INFO, "index log info block print complete", K(meta));
}
}
} else {
CLOG_LOG(ERROR, "invalid clog type while dump info block", K(log_file_type_));
}
}
if (!is_ofs_) {
ret = OB_ITER_END; // info block is the end for local file
}
break;
case OB_TRAILER_BLOCK:
ret = OB_ITER_END;
// iter will never encounter a trailer for InfoBlock lies ahead
break;
default:
ret = OB_INVALID_DATA;
CLOG_LOG(ERROR, "unknown block meta type", K(ret), K(meta), K(file_id_), K(cur_offset_));
}
}
return ret;
}
// if return ILOG_ENTRY, that means ilog_entry prepared well
int ObLogEntryParser::get_type_(ObCLogItemType& type)
{
int ret = OB_SUCCESS;
type = UNKNOWN_TYPE;
if (IS_NOT_INIT) {
ret = OB_NOT_INIT;
CLOG_LOG(WARN, "log entry parser is not inited", K(ret));
} else if (OB_ISNULL(buf_cur_)) {
ret = OB_ERR_UNEXPECTED;
CLOG_LOG(ERROR, "buf cur is null", KP(buf_cur_), K(ret));
} else if (0 == buf_end_ - buf_cur_) {
CLOG_LOG(INFO, "reach end", K(cur_offset_));
ret = OB_ITER_END;
} else if (OB_UNLIKELY((buf_end_ - buf_cur_) < 2)) {
ret = OB_ERR_UNEXPECTED;
CLOG_LOG(ERROR, "buf not enough", K_(cur_offset), K(ret));
} else if (OB_FAIL(clog::parse_log_item_type(buf_cur_, buf_end_ - buf_cur_, type))) {
CLOG_LOG(ERROR, "parse_log_item_type fail", K(ret), KP_(buf_cur), K_(buf_end), K(type));
}
return ret;
}
int ObLogEntryParser::dump_all_entry(bool is_hex, bool without_data)
{
int ret = OB_SUCCESS;
if (OB_UNLIKELY(!is_inited_)) {
ret = OB_NOT_INIT;
CLOG_LOG(ERROR, "log entry parser is not inited", K(is_inited_), K(ret));
} else {
dump_hex_ = is_hex;
without_data_ = without_data;
while (OB_SUCC(ret)) {
if (OB_FAIL(parse_next_entry())) {
if (OB_ITER_END != ret) {
CLOG_LOG(WARN, "failed to parse_next_entry", K(ret));
}
}
}
}
return ret;
}
int ObLogEntryParser::format_dump_entry()
{
int ret = OB_SUCCESS;
if (OB_UNLIKELY(!is_inited_)) {
ret = OB_NOT_INIT;
CLOG_LOG(ERROR, "log entry parser is not inited", K(is_inited_), K(ret));
} else {
while (OB_SUCC(ret)) {
if (OB_FAIL(format_dump_next_entry())) {
if (OB_ITER_END != ret)
CLOG_LOG(WARN, "failed to format_dump_entry", K(ret));
}
}
}
return ret;
}
int ObLogEntryParser::stat_log()
{
int ret = OB_SUCCESS;
if (OB_UNLIKELY(!is_inited_)) {
ret = OB_NOT_INIT;
CLOG_LOG(ERROR, "log entry parser is not inited", K(is_inited_), K(ret));
} else {
while (OB_SUCC(ret)) {
if (OB_FAIL(stat_next_entry())) {
if (OB_ITER_END != ret)
CLOG_LOG(WARN, "failed to stat_log", K(ret));
}
}
}
return ret;
}
// return OB_EAGAIN: to prepare buffer and do get_next_entry_ again
int ObLogEntryParser::parse_next_entry()
{
int ret = OB_SUCCESS;
ObCLogItemType item_type = UNKNOWN_TYPE;
if (OB_FAIL(get_type_(item_type)) && OB_ITER_END != ret) {
CLOG_LOG(ERROR, "failed to get item type", K(ret));
} else if (OB_ITER_END == ret) {
// reach end
} else if (ILOG_ENTRY == item_type) {
if (OB_ILOG_FILE_TYPE != log_file_type_) {
log_file_type_ = OB_ILOG_FILE_TYPE;
}
CLOG_LOG(DEBUG, "ilog entry magic", K(cur_offset_), K(item_type));
int64_t pos = 0;
ObIndexEntry entry;
if (OB_FAIL(entry.deserialize(buf_cur_, buf_end_ - buf_cur_, pos))) {
CLOG_LOG(ERROR,
"ilog entry deserialize error, maybe corrupted",
K(ret),
K(item_type),
KP(buf_cur_),
KP(buf_end_),
K(pos),
K(cur_offset_),
K(file_id_));
// just go to next item
if (OB_FAIL(advance_to_next_align_())) {
if (OB_ITER_END != ret) {
CLOG_LOG(ERROR, "unexpected error", K(ret));
}
}
} else {
advance_(pos);
if (OB_FAIL(dump_ilog_entry(entry))) {
CLOG_LOG(WARN, "failed to dump_ilog_entry", K(ret), K(cur_offset_), KP(buf_cur_), KP(buf_end_), K(entry));
ret = OB_SUCCESS; // skip this entry, go on to handle next entry
}
}
} else if (CLOG_ENTRY == item_type) {
if (OB_CLOG_FILE_TYPE != log_file_type_) {
log_file_type_ = OB_CLOG_FILE_TYPE;
}
CLOG_LOG(DEBUG, "clog entry magic", K(cur_offset_), K(item_type));
int64_t pos = 0;
ObLogEntry entry;
if (OB_FAIL(entry.deserialize(buf_cur_, buf_end_ - buf_cur_, pos))) {
CLOG_LOG(ERROR,
"clog entry deserialize error, maybe corrupted",
K(ret),
K(item_type),
KP(buf_cur_),
KP(buf_end_),
K(pos),
K(cur_offset_),
K(file_id_));
// just go to next item
if (OB_FAIL(advance_to_next_align_())) {
if (OB_ITER_END != ret) {
CLOG_LOG(ERROR, "unexpected error", K(ret));
}
}
} else {
if (OB_FAIL(dump_clog_entry(entry, pos))) {
// fatal error, data corrupt
CLOG_LOG(WARN, "failed to dump_clog_entry", K(ret), K(cur_offset_), KP(buf_cur_), KP(buf_end_), K(entry));
ret = OB_SUCCESS; // skip this entry, go on to handle next entry
}
advance_(pos);
}
} else if (EOF_BUF == item_type) {
CLOG_LOG(DEBUG, "eof magic", K(cur_offset_));
// pointing to start of EOF
ret = OB_ITER_END;
CLOG_LOG(INFO, "reach eof", K(file_id_), K(cur_offset_));
} else if (BLOCK_META == item_type) {
CLOG_LOG(DEBUG, "block magic", K(cur_offset_));
ObLogBlockMetaV2 meta;
int64_t pos = 0;
if (OB_FAIL(meta.deserialize(buf_cur_, buf_end_ - buf_cur_, pos))) {
CLOG_LOG(ERROR,
"log block meta deserialize error, maybe corrupted",
K(ret),
K(item_type),
KP(buf_cur_),
KP(buf_end_),
K(pos),
K(cur_offset_),
K(file_id_));
if (OB_FAIL(advance_to_next_align_())) {
if (OB_ITER_END != ret) {
CLOG_LOG(ERROR, "unexpected error", K(ret));
}
}
} else if (OB_FAIL(dump_block_(meta))) {
if (OB_ITER_END != ret) {
CLOG_LOG(WARN, "dump block error", K(ret), K(meta), K(file_id_), K(cur_offset_), KP(buf_cur_), KP(buf_end_));
}
}
} else if (PADDING_ENTRY == item_type) {
CLOG_LOG(DEBUG, "padding entry magic", K(cur_offset_));
// data written by new writer
ObPaddingEntry pe;
int64_t pos = 0;
if (OB_FAIL(pe.deserialize(buf_cur_, buf_end_ - buf_cur_, pos))) {
CLOG_LOG(ERROR, "padding entry deserialize error");
if (OB_FAIL(advance_to_next_align_())) {
if (OB_ITER_END != ret) {
CLOG_LOG(ERROR, "unexpected error", K(ret));
}
}
} else {
if (!filter_.is_valid()) {
fprintf(stdout, "Padding: %s|\n", to_cstring(pe));
}
// skip padding entry
advance_(pe.get_entry_size());
}
} else if (CLOG_ENTRY_COMPRESSED_ZSTD == item_type || CLOG_ENTRY_COMPRESSED_LZ4 == item_type ||
CLOG_ENTRY_COMPRESSED_ZSTD_138 == item_type) {
int64_t local_pos = 0;
int64_t uncompress_len = 0;
int64_t pos = 0;
ObLogEntry entry;
if (OB_FAIL(clog::uncompress(
buf_cur_, buf_end_ - buf_cur_, compress_rbuf_.buf_, compress_rbuf_.buf_len_, uncompress_len, pos))) {
CLOG_LOG(ERROR, "failed to uncompress clog entry", K(ret), K(item_type));
} else if (OB_FAIL(entry.deserialize(compress_rbuf_.buf_, uncompress_len, local_pos))) {
CLOG_LOG(ERROR, "failed to deserialize clog entry", K(ret));
}
if (OB_FAIL(ret)) {
if (OB_FAIL(advance_to_next_align_())) {
if (OB_ITER_END != ret) {
CLOG_LOG(ERROR, "unexpected error", K(ret));
}
}
} else {
if (OB_FAIL(dump_clog_entry(entry, pos))) {
CLOG_LOG(WARN, "failed to dump_clog_entry", K(ret), K(entry));
ret = OB_SUCCESS;
}
advance_(pos);
}
} else {
ret = OB_ERR_UNEXPECTED;
CLOG_LOG(ERROR, "error unexpected", K(ret), K(item_type));
}
return ret;
}
int ObLogEntryParserImpl::init(const int64_t file_id, const ObLogEntryFilter& filter, const common::ObString& host,
const int32_t port, const char* config_file)
{
int ret = OB_SUCCESS;
UNUSED(config_file);
if (OB_UNLIKELY(file_id <= 0)) {
ret = OB_INVALID_ARGUMENT;
} else if (OB_UNLIKELY(NULL == (print_buf_ = static_cast<char*>(allocator_.alloc(PRINT_BUF_SIZE))))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
CLOG_LOG(WARN, "failed to allocate memory", K(ret));
} else {
file_id_ = file_id;
cur_offset_ = 0;
filter_ = filter;
is_inited_ = true;
}
if (!host.empty() && port > 0) {
if (OB_SUCCESS != client_.init()) {
CLOG_LOG(WARN, "failed to init net client", K(ret));
} else if (OB_SUCCESS != client_.get_proxy(rpc_proxy_)) {
CLOG_LOG(WARN, "failed to get_proxy", K(ret));
} else {
host_addr_.set_ip_addr(host, port);
rpc_proxy_.set_server(host_addr_);
}
}
return ret;
}
int ObLogEntryParser::format_dump_next_entry()
{
int ret = OB_SUCCESS;
ObCLogItemType item_type = UNKNOWN_TYPE;
if (OB_UNLIKELY(!is_inited_)) {
ret = OB_NOT_INIT;
CLOG_LOG(ERROR, "log entry parser is not inited", K(is_inited_), K(ret));
} else if (OB_FAIL(get_type_(item_type)) && OB_ITER_END != ret) {
CLOG_LOG(ERROR, "failed to get item type", K(ret));
} else if (OB_ITER_END == ret) {
// reach end
} else if (ILOG_ENTRY == item_type) {
ret = OB_ITER_END;
CLOG_LOG(ERROR, "should be clog file rather than ilog file", K(ret));
} else if (CLOG_ENTRY == item_type) {
if (OB_CLOG_FILE_TYPE != log_file_type_) {
log_file_type_ = OB_CLOG_FILE_TYPE;
}
CLOG_LOG(DEBUG, "clog entry magic", K(cur_offset_), K(item_type));
int64_t pos = 0;
ObLogEntry entry;
if (OB_FAIL(entry.deserialize(buf_cur_, buf_end_ - buf_cur_, pos))) {
CLOG_LOG(ERROR,
"entry deserialize error, maybe corrupted",
K(ret),
K(item_type),
KP(buf_cur_),
KP(buf_end_),
K(pos),
K(cur_offset_),
K(file_id_));
if (OB_FAIL(advance_to_next_align_())) {
if (OB_ITER_END != ret) {
CLOG_LOG(ERROR, "unexpected error", K(ret));
}
}
} else {
if (OB_FAIL(format_dump_clog_entry(entry))) {
// fatal error, data corrupt
CLOG_LOG(ERROR, "failed to dump_clog_entry", K(ret), K(cur_offset_), KP(buf_cur_), KP(buf_end_), K(entry));
ret = OB_SUCCESS; // skip this entry, go on to handle next entry
} else { /*do nothing*/
}
advance_(pos);
}
} else if (EOF_BUF == item_type) {
CLOG_LOG(DEBUG, "eof magic", K(cur_offset_));
// pointing to start of EOF
ret = OB_ITER_END;
CLOG_LOG(INFO, "reach eof", K(file_id_), K(cur_offset_)); // FIXME
} else if (BLOCK_META == item_type) {
CLOG_LOG(DEBUG, "block magic", K(cur_offset_));
ObLogBlockMetaV2 meta;
int64_t pos = 0;
if (OB_FAIL(meta.deserialize(buf_cur_, buf_end_ - buf_cur_, pos))) {
CLOG_LOG(ERROR,
"log block meta deserialize error, maybe corrupted",
K(ret),
K(item_type),
KP(buf_cur_),
KP(buf_end_),
K(pos),
K(cur_offset_),
K(file_id_));
if (OB_FAIL(advance_to_next_align_())) {
if (OB_ITER_END != ret) {
CLOG_LOG(ERROR, "unexpected error", K(ret));
}
}
} else if (OB_FAIL(skip_block_(meta))) {
if (OB_ITER_END != ret) {
CLOG_LOG(WARN, "dump block error", K(ret), K(meta), K(file_id_), K(cur_offset_), KP(buf_cur_), KP(buf_end_));
}
}
} else if (PADDING_ENTRY == item_type) {
CLOG_LOG(DEBUG, "padding entry magic", K(cur_offset_));
// data written by new writer
ObPaddingEntry pe;
int64_t pos = 0;
if (OB_FAIL(pe.deserialize(buf_cur_, buf_end_ - buf_cur_, pos))) {
CLOG_LOG(ERROR, "padding entry deserialize error");
if (OB_FAIL(advance_to_next_align_())) {
if (OB_ITER_END != ret) {
CLOG_LOG(ERROR, "unexpected error", K(ret));
}
}
} else {
// skip padding entry
advance_(pe.get_entry_size());
}
} else {
ret = OB_ERR_UNEXPECTED;
CLOG_LOG(ERROR, "error unexpected", K(ret), K(item_type));
}
return ret;
}
int ObLogEntryParser::stat_next_entry()
{
int ret = OB_SUCCESS;
ObCLogItemType item_type = UNKNOWN_TYPE;
if (OB_FAIL(get_type_(item_type)) && OB_ITER_END != ret) {
CLOG_LOG(ERROR, "failed to get item type", K(ret));
} else if (OB_ITER_END == ret) {
// reach end
} else if (ILOG_ENTRY == item_type) {
ret = OB_ITER_END;
// fprintf(stdout, "invalid argument is ilog file, stat only support clog file");
} else if (CLOG_ENTRY == item_type) {
if (OB_CLOG_FILE_TYPE != log_file_type_) {
log_file_type_ = OB_CLOG_FILE_TYPE;
}
CLOG_LOG(DEBUG, "clog entry magic", K(cur_offset_), K(item_type));
int64_t pos = 0;
ObLogEntry entry;
if (OB_FAIL(entry.deserialize(buf_cur_, buf_end_ - buf_cur_, pos))) {
CLOG_LOG(ERROR,
"entry deserialize error, maybe corrupted",
K(ret),
K(item_type),
KP(buf_cur_),
KP(buf_end_),
K(pos),
K(cur_offset_),
K(file_id_));
if (OB_FAIL(advance_to_next_align_())) {
if (OB_ITER_END != ret) {
CLOG_LOG(ERROR, "unexpected error", K(ret));
}
}
} else {
if (OB_FAIL(stat_clog_entry(entry, pos, false /*is_compressed=false*/))) {
CLOG_LOG(ERROR, "fail to stat_clog_entry", K(ret));
}
advance_(pos);
}
} else if (EOF_BUF == item_type) {
CLOG_LOG(DEBUG, "eof magic", K(cur_offset_));
// pointing to start of EOF
ret = OB_ITER_END;
CLOG_LOG(INFO, "reach eof", K(file_id_), K(cur_offset_), K(ret));
} else if (BLOCK_META == item_type) {
CLOG_LOG(DEBUG, "block magic", K(cur_offset_));
ObLogBlockMetaV2 meta;
int64_t pos = 0;
if (OB_FAIL(meta.deserialize(buf_cur_, buf_end_ - buf_cur_, pos))) {
CLOG_LOG(ERROR,
"log block meta deserialize error, maybe corrupted",
K(ret),
K(item_type),
KP(buf_cur_),
KP(buf_end_),
K(pos),
K(cur_offset_),
K(file_id_));
if (OB_FAIL(advance_to_next_align_())) {
if (OB_ITER_END != ret) {
CLOG_LOG(ERROR, "unexpected error", K(ret));
}
}
} else if (OB_FAIL(stat_block_(meta))) {
if (OB_ITER_END != ret) {
CLOG_LOG(WARN, "dump block error", K(ret), K(meta), K(file_id_), K(cur_offset_), KP(buf_cur_), KP(buf_end_));
}
}
} else if (PADDING_ENTRY == item_type) {
CLOG_LOG(DEBUG, "padding entry magic", K(cur_offset_));
// data written by new writer
ObPaddingEntry pe;
int64_t pos = 0;
if (OB_FAIL(pe.deserialize(buf_cur_, buf_end_ - buf_cur_, pos))) {
CLOG_LOG(ERROR, "padding entry deserialize error");
if (OB_FAIL(advance_to_next_align_())) {
if (OB_ITER_END != ret) {
CLOG_LOG(ERROR, "unexpected error", K(ret));
}
}
} else {
// skip padding entry
advance_(pe.get_entry_size());
log_stat_.padding_size_ += pe.get_entry_size();
}
} else if (CLOG_ENTRY_COMPRESSED_ZSTD == item_type || CLOG_ENTRY_COMPRESSED_LZ4 == item_type ||
CLOG_ENTRY_COMPRESSED_ZSTD_138 == item_type) {
int64_t local_pos = 0;
int64_t uncompress_len = 0;
int64_t pos = 0;
ObLogEntry entry;
if (OB_FAIL(clog::uncompress(
buf_cur_, buf_end_ - buf_cur_, compress_rbuf_.buf_, compress_rbuf_.buf_len_, uncompress_len, pos))) {
CLOG_LOG(ERROR, "failed to uncompress clog entry", K(ret), K(item_type));
} else if (OB_FAIL(entry.deserialize(compress_rbuf_.buf_, uncompress_len, local_pos))) {
CLOG_LOG(ERROR, "failed to deserialize clog entry", K(ret));
}
if (OB_FAIL(ret)) {
if (OB_FAIL(advance_to_next_align_())) {
if (OB_ITER_END != ret) {
CLOG_LOG(ERROR, "unexpected error", K(ret));
}
}
} else {
if (OB_FAIL(stat_clog_entry(entry, pos, true /*is_compressed=true*/))) {
CLOG_LOG(WARN, "failed to stat_clog_entry", K(ret), K(entry));
ret = OB_SUCCESS;
}
advance_(pos);
}
} else {
ret = OB_ERR_UNEXPECTED;
CLOG_LOG(ERROR, "error unexpected", K(ret), K(item_type));
}
return ret;
}
int ObLogEntryParserImpl::dump_memtable_mutator(const char* buf, int64_t len)
{
int ret = OB_SUCCESS;
int64_t pos = 0;
ObMemtableMutatorIterator mmi;
ObMemtableMutatorMeta meta;
int64_t meta_pos = 0;
if (OB_FAIL(meta.deserialize(buf, len, meta_pos))) {
CLOG_LOG(ERROR, "meta.deserialize fail", K(ret));
} else if (ObTransRowFlag::is_normal_row(meta.get_flags())) {
if (OB_FAIL(mmi.deserialize(buf, len, pos))) {
CLOG_LOG(ERROR, "mmi.deserialize fail", K(ret));
} else {
fprintf(stdout, " %s", to_cstring(mmi.get_meta()));
}
while (OB_SUCCESS == ret) {
ObMemtableMutatorRow row;
if (OB_FAIL(mmi.get_next_row(row)) && OB_ITER_END != ret) {
CLOG_LOG(ERROR, "mmi.get_next_row fail", K(ret));
} else if (OB_ITER_END == ret) {
ret = OB_SUCCESS;
break;
} else {
if (without_data_) {
fprintf(stdout, " MutatorRow={*} | OLD_ROW={");
} else if (dump_hex_) {
int64_t pos = row.rowkey_.to_smart_string(print_buf_, PRINT_BUF_SIZE - 1);
print_buf_[pos] = '\0';
fprintf(stdout, " MutatorRow={%s} HexedRowkey={%s} | OLD_ROW={", to_cstring(row), print_buf_);
} else {
fprintf(stdout, " MutatorRow={%s} | OLD_ROW={", to_cstring(row));
}
ObCellReader new_cci;
ObCellReader old_cci;
if (NULL != row.old_row_.data_ && row.old_row_.size_ > 0) {
if (OB_FAIL(old_cci.init(row.old_row_.data_, row.old_row_.size_, SPARSE))) {
CLOG_LOG(ERROR, "old cci.init fail", K(ret));
}
while (OB_SUCC(ret) && OB_SUCC(old_cci.next_cell())) {
uint64_t column_id = OB_INVALID_ID;
const ObObj* value = NULL;
if (OB_FAIL(old_cci.get_cell(column_id, value))) {
CLOG_LOG(ERROR, "failed to get cell", K(ret));
} else if (NULL == value) {
ret = OB_ERR_UNEXPECTED;
CLOG_LOG(ERROR, "got value is NULL", K(ret));
} else if (ObExtendType == value->get_type() && ObActionFlag::OP_END_FLAG == value->get_ext()) {
ret = OB_ITER_END;
} else if (OB_FAIL(dump_obj(*value, column_id))) {
CLOG_LOG(ERROR, "failed to dump obj", K(*value), K(ret));
} else { /*do nothing*/
}
}
if (OB_ITER_END == ret) {
ret = OB_SUCCESS;
}
} else if (NULL == row.old_row_.data_ && 0 == row.old_row_.size_) {
// no data of old row
} else {
CLOG_LOG(ERROR, "old cci.init fail", K(row), K(ret));
}
fprintf(stdout, "} | NEW_ROW={");
if (OB_SUCC(ret) && OB_FAIL(new_cci.init(row.new_row_.data_, row.new_row_.size_, SPARSE))) {
CLOG_LOG(ERROR, "new cci.init fail", K(ret));
} else {
while (OB_SUCC(ret) && OB_SUCC(new_cci.next_cell())) {
uint64_t column_id = OB_INVALID_ID;
const ObObj* value = NULL;
if (OB_FAIL(new_cci.get_cell(column_id, value))) {
} else if (NULL == value) {
ret = OB_ERR_UNEXPECTED;
CLOG_LOG(ERROR, "value is NULL", K(ret));
} else if (ObExtendType == value->get_type() && ObActionFlag::OP_END_FLAG == value->get_ext()) {
ret = OB_ITER_END;
} else if (OB_FAIL(dump_obj(*value, column_id))) {
CLOG_LOG(ERROR, "failed to dump obj", K(*value), K(ret));
} else { /*do nothing*/
}
}
if (OB_ITER_END == ret) {
ret = OB_SUCCESS;
}
}
fprintf(stdout, "}");
}
}
} else if (ObTransRowFlag::is_big_row_start(meta.get_flags())) {
fprintf(stdout, "BIG_ROW_START: %s", to_cstring(meta));
} else if (ObTransRowFlag::is_big_row_mid(meta.get_flags())) {
fprintf(stdout, "BIG_ROW_MID: %s", to_cstring(meta));
} else if (ObTransRowFlag::is_big_row_end(meta.get_flags())) {
fprintf(stdout, "BIG_ROW_END: %s", to_cstring(meta));
}
if (OB_ITER_END == ret) {
ret = OB_SUCCESS;
}
return ret;
}
int ObLogEntryParserImpl::dump_sp_trans_redo_log(const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
ObSpTransRedoLogHelper helper;
ObSpTransRedoLog log(helper);
int64_t pos = 0;
if (OB_FAIL(log.init_for_deserialize())) {
CLOG_LOG(WARN, "redo log init for deserialize fail", K(ret));
} else if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialize log", K(ret));
} else if (OB_FAIL(log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(log.decrypt_table_key())) {
fprintf(stdout, " %s|||", to_cstring(log));
dump_memtable_mutator(log.get_mutator().get_data(), log.get_mutator().get_position());
} else if (ret == OB_EAGAIN) {
// impossible branch
} else {
CLOG_LOG(WARN, "log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParserImpl::dump_trans_redo_log(const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
ObTransRedoLogHelper helper;
ObTransRedoLog log(helper);
int64_t pos = 0;
if (OB_FAIL(log.init_for_deserialize())) {
CLOG_LOG(WARN, "redo log init for deserialize fail", K(ret));
} else if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialize log", K(ret));
} else if (OB_FAIL(log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(log.decrypt_table_key())) {
fprintf(stdout, " %s|||", to_cstring(log));
dump_memtable_mutator(log.get_mutator().get_data(), log.get_mutator().get_position());
} else if (ret == OB_EAGAIN) {
// impossible branch
} else {
CLOG_LOG(WARN, "log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParserImpl::dump_trans_prepare_log(const char* data, int64_t len)
{
int ret = OB_SUCCESS;
ObTransPrepareLogHelper helper;
ObTransPrepareLog log(helper);
int64_t pos = 0;
if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "prepare log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " %s|||", to_cstring(log));
}
return ret;
}
int ObLogEntryParserImpl::dump_sp_trans_commit_log(const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
ObSpTransCommitLogHelper helper;
ObSpTransCommitLog log(helper);
int64_t pos = 0;
// an sp commit log may contain redo data
if (OB_FAIL(log.init_for_deserialize())) {
CLOG_LOG(WARN, "sp redo log init for deserialize fail", K(ret));
} else if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialize log", K(ret));
} else if (OB_FAIL(log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(log.decrypt_table_key())) {
fprintf(stdout, " %s|||", to_cstring(log));
dump_memtable_mutator(log.get_mutator().get_data(), log.get_mutator().get_position());
} else if (ret == OB_EAGAIN) {
// impossible branch
} else {
CLOG_LOG(WARN, "commit log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParserImpl::dump_trans_commit_log(const char* data, int64_t len)
{
int ret = OB_SUCCESS;
PartitionLogInfoArray partition_log_info_array;
ObTransCommitLog log(partition_log_info_array);
int64_t pos = 0;
if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "commit log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " %s|||", to_cstring(log));
}
return ret;
}
int ObLogEntryParserImpl::dump_sp_trans_abort_log(const char* data, int64_t len)
{
int ret = OB_SUCCESS;
ObSpTransAbortLog log;
int64_t pos = 0;
if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "sp abort log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " %s|||", to_cstring(log));
}
return ret;
}
int ObLogEntryParserImpl::dump_trans_abort_log(const char* data, int64_t len)
{
int ret = OB_SUCCESS;
ObTransAbortLog log;
int64_t pos = 0;
if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "commit log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " %s|||", to_cstring(log));
}
return ret;
}
int ObLogEntryParserImpl::dump_trans_clear_log(const char* data, int64_t len)
{
int ret = OB_SUCCESS;
ObTransClearLog log;
int64_t pos = 0;
if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "clear log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " %s|||", to_cstring(log));
}
return ret;
}
int ObLogEntryParserImpl::dump_trans_prepare_commit_log(const char* data, int64_t len)
{
int ret = OB_SUCCESS;
int64_t pos = 0;
ObTransPrepareLogHelper helper;
ObTransPrepareLog p_log(helper);
PartitionLogInfoArray partition_log_info_array;
ObTransCommitLog c_log(partition_log_info_array);
if (OB_FAIL(p_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "prepare log deserialize error", K(ret), KP(data), K(len), K(pos));
} else if (OB_FAIL(c_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "commit log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " %s|||", to_cstring(p_log));
fprintf(stdout, " %s|||", to_cstring(c_log));
}
return ret;
}
int ObLogEntryParserImpl::dump_trans_redo_prepare_log(const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
ObTransRedoLogHelper helper;
ObTransRedoLog r_log(helper);
int64_t pos = 0;
if (OB_FAIL(r_log.init_for_deserialize())) {
CLOG_LOG(WARN, "redo log init for deserialize fail", K(ret));
} else if (OB_FAIL(r_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialize log", K(ret));
} else if (OB_FAIL(r_log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(r_log.decrypt_table_key()) || ret == OB_EAGAIN) {
if (OB_SUCC(ret)) {
fprintf(stdout, " %s|||", to_cstring(r_log));
dump_memtable_mutator(r_log.get_mutator().get_data(), r_log.get_mutator().get_position());
} else {
// impossible branch
}
ObTransPrepareLogHelper helper;
ObTransPrepareLog p_log(helper);
if (OB_FAIL(p_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "prepare log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " %s|||", to_cstring(p_log));
}
} else {
CLOG_LOG(WARN, "redo log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParserImpl::dump_trans_redo_prepare_commit_log(
const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
int64_t pos = 0;
ObTransRedoLogHelper helper;
ObTransRedoLog r_log(helper);
if (OB_FAIL(r_log.init_for_deserialize())) {
CLOG_LOG(WARN, "redo log init for deserialize fail", K(ret));
} else if (OB_FAIL(r_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserializde log", K(ret));
} else if (OB_FAIL(r_log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(r_log.decrypt_table_key()) || ret == OB_EAGAIN) {
if (OB_SUCC(ret)) {
fprintf(stdout, " %s|||", to_cstring(r_log));
dump_memtable_mutator(r_log.get_mutator().get_data(), r_log.get_mutator().get_position());
} else {
// impossible branch
}
ObTransPrepareLogHelper helper;
ObTransPrepareLog p_log(helper);
PartitionLogInfoArray partition_log_info_array;
ObTransCommitLog c_log(partition_log_info_array);
if (OB_FAIL(p_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "prepare log deserialize error", K(ret), KP(data), K(len), K(pos));
} else if (OB_FAIL(c_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "commit log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " %s|||", to_cstring(p_log));
fprintf(stdout, " %s|||", to_cstring(c_log));
}
} else {
CLOG_LOG(WARN, "redo log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParserImpl::dump_trans_prepare_commit_clear_log(const char* data, int64_t len)
{
int ret = OB_SUCCESS;
int64_t pos = 0;
ObTransPrepareLogHelper helper;
ObTransPrepareLog p_log(helper);
PartitionLogInfoArray partition_log_info_array;
ObTransCommitLog c_log(partition_log_info_array);
ObTransClearLog cl_log;
if (OB_FAIL(p_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "prepare log deserialize error", K(ret), KP(data), K(len), K(pos));
} else if (OB_FAIL(c_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "commit log deserialize error", K(ret), KP(data), K(len), K(pos));
} else if (OB_FAIL(cl_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "clear log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " %s|||", to_cstring(p_log));
fprintf(stdout, " %s|||", to_cstring(c_log));
fprintf(stdout, " %s|||", to_cstring(cl_log));
}
return ret;
}
int ObLogEntryParserImpl::dump_trans_redo_prepare_commit_clear_log(
const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
int64_t pos = 0;
ObTransRedoLogHelper helper;
ObTransRedoLog r_log(helper);
if (OB_FAIL(r_log.init_for_deserialize())) {
CLOG_LOG(WARN, "redo log init for deserialize fail", K(ret));
} else if (OB_FAIL(r_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialize log", K(ret));
} else if (OB_FAIL(r_log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(r_log.decrypt_table_key()) || ret == OB_EAGAIN) {
if (OB_SUCC(ret)) {
fprintf(stdout, " %s|||", to_cstring(r_log));
dump_memtable_mutator(r_log.get_mutator().get_data(), r_log.get_mutator().get_position());
} else {
// impossible branch
}
ObTransPrepareLogHelper helper;
ObTransPrepareLog p_log(helper);
PartitionLogInfoArray partition_log_info_array;
ObTransCommitLog c_log(partition_log_info_array);
ObTransClearLog cl_log;
if (OB_FAIL(p_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "prepare log deserialize error", K(ret), KP(data), K(len), K(pos));
} else if (OB_FAIL(c_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "commit log deserialize error", K(ret), KP(data), K(len), K(pos));
} else if (OB_FAIL(cl_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "clear log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " %s|||", to_cstring(p_log));
fprintf(stdout, " %s|||", to_cstring(c_log));
fprintf(stdout, " %s|||", to_cstring(cl_log));
}
} else {
CLOG_LOG(WARN, "redo log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParserImpl::dump_trans_mutator_log(const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
ObTransMutatorLogHelper helper;
ObTransMutatorLog log(helper);
int64_t pos = 0;
if (OB_FAIL(log.init_for_deserialize())) {
CLOG_LOG(WARN, "trans mutator log init for deserialize fail", K(ret));
} else if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialize log", K(ret));
} else if (OB_FAIL(log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(log.decrypt_table_key())) {
fprintf(stdout, " %s|||", to_cstring(log));
dump_memtable_mutator(log.get_mutator().get_data(), log.get_mutator().get_position());
} else {
CLOG_LOG(WARN, "log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParserImpl::dump_trans_mutator_abort_log(const char* data, int64_t len)
{
int ret = OB_SUCCESS;
ObTransMutatorAbortLog log;
int64_t pos = 0;
if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " %s|||", to_cstring(log));
}
return ret;
}
int ObLogEntryParserImpl::dump_trans_state_log(const char* data, int64_t len)
{
int ret = OB_SUCCESS;
ObTransStateLog log;
int64_t pos = 0;
if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " %s|||", to_cstring(log));
}
return ret;
}
int ObLogEntryParserImpl::dump_trans_mutator_state_log(const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
ObTransMutatorLogHelper helper;
ObTransMutatorLog m_log(helper);
int64_t pos = 0;
if (OB_FAIL(m_log.init_for_deserialize())) {
CLOG_LOG(WARN, "trans mutator log init for deserialize fail", K(ret));
} else if (OB_FAIL(m_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialize log", K(ret));
} else if (OB_FAIL(m_log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(m_log.decrypt_table_key()) || ret == OB_EAGAIN) {
if (OB_SUCC(ret)) {
fprintf(stdout, " %s|||", to_cstring(m_log));
dump_memtable_mutator(m_log.get_mutator().get_data(), m_log.get_mutator().get_position());
} else {
// impossible branch
}
ObTransStateLog s_log;
if (OB_FAIL(s_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " %s|||", to_cstring(s_log));
}
} else {
CLOG_LOG(WARN, "log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParserImpl::dump_part_split_src_log(const char* data, int64_t len)
{
int ret = OB_SUCCESS;
ObPartitionSplitSourceLog log;
int64_t pos = 0;
if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " %s|||", to_cstring(log));
}
return ret;
}
int ObLogEntryParserImpl::dump_part_split_dest_log(const char* data, int64_t len)
{
int ret = OB_SUCCESS;
ObPartitionSplitDestLog log;
int64_t pos = 0;
if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " %s|||", to_cstring(log));
}
return ret;
}
int ObLogEntryParserImpl::dump_trans_checkpoint_log(const char* data, int64_t len)
{
int ret = OB_SUCCESS;
ObCheckpointLog log;
int64_t pos = 0;
if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " %s|||", to_cstring(log));
}
return ret;
}
int ObLogEntryParserImpl::dump_partition_schema_version_change_log(const char* data, int64_t len)
{
int ret = OB_SUCCESS;
ObPGSchemaChangeLog log;
int64_t pos = 0;
if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "ObPGSchemaChangeLog deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " OB_LOG_PARTITION_SCHEMA_VERSION_CHANGE %s|||", to_cstring(log));
}
return ret;
}
int ObLogEntryParserImpl::dump_new_offline_partition_log(const char* data, int64_t len)
{
int ret = OB_SUCCESS;
ObOfflinePartitionLog log;
int64_t pos = 0;
if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "ObOfflinePartitionLog log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " OB_LOG_OFFLINE_PARTITION_LOG_V2 %s|||", to_cstring(log));
}
return ret;
}
int ObLogEntryParserImpl::dump_add_partition_to_pg_log(const char* data, int64_t len)
{
int ret = OB_SUCCESS;
ObAddPartitionToPGLog log;
int64_t pos = 0;
if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "ObAddPartitionToPGLog log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " OB_LOG_ADD_PARTITION_TO_PG %s|||", to_cstring(log));
}
return ret;
}
int ObLogEntryParserImpl::dump_remove_partition_from_pg_log(const char* data, int64_t len)
{
int ret = OB_SUCCESS;
ObRemovePartitionFromPGLog log;
int64_t pos = 0;
if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "ObRemovePartitionFroPGLog log deserialize error", K(ret), KP(data), K(len), K(pos));
} else {
fprintf(stdout, " OB_LOG_REMOVE_PARTITION_FROM_PG %s|||", to_cstring(log));
}
return ret;
}
int ObLogEntryParserImpl::dump_clog_entry(ObLogEntry& entry, int64_t pos)
{
int ret = OB_SUCCESS;
bool need_print = false;
ObLogEntry* entry_ptr = &entry;
ObLogEntry decrypted_entry;
if (!entry.check_integrity(true /*ignore_batch_commited*/)) {
ret = OB_INVALID_DATA;
CLOG_LOG(ERROR, "entry.check_integrity() fail", K(entry), K(ret));
} else if (OB_FAIL(check_filter(entry, need_print))) {
CLOG_LOG(ERROR, "check filter failed", K(ret), K(entry));
} else if (!need_print) {
CLOG_LOG(DEBUG, "ignore log", K(entry));
} else {
if (OB_SUCC(ret)) {
ret = dump_clog_entry_(*entry_ptr, pos);
}
}
return ret;
}
int ObLogEntryParserImpl::dump_clog_entry_(ObLogEntry& entry, int64_t pos)
{
int ret = OB_SUCCESS;
clog::ObLogType log_type = entry.get_header().get_log_type();
fprintf(stdout,
"$$$file_id:%lu offset:%ld len:%ld %s|log_id:%lu|%s |||",
file_id_,
cur_offset_,
pos,
get_log_type(log_type),
entry.get_header().get_log_id(),
to_cstring(entry));
if (OB_LOG_MEMBERSHIP == log_type) {
int64_t pos = 0;
ObMembershipLog member_log;
if (OB_FAIL(member_log.deserialize(entry.get_buf(), entry.get_header().get_data_len(), pos))) {
CLOG_LOG(ERROR, "failed to deserialize member_log", K(entry), K(ret));
} else {
fprintf(stdout, "||ObMembershipLog:%s", to_cstring(member_log));
}
} else if (OB_LOG_RENEW_MEMBERSHIP == log_type) {
int64_t pos = 0;
ObRenewMembershipLog renew_ms_log;
if (OB_FAIL(renew_ms_log.deserialize(entry.get_buf(), entry.get_header().get_data_len(), pos))) {
CLOG_LOG(ERROR, "failed to deserialize ObRenewMembershipLog", K(entry), K(ret));
} else {
fprintf(stdout, "||ObRenewMembershipLog:%s", to_cstring(renew_ms_log));
}
} else if (OB_LOG_SUBMIT == log_type) {
int64_t trans_inc = -1;
int64_t submit_log_type_tmp = -1;
const char* buf = entry.get_buf();
int64_t buf_len = entry.get_header().get_data_len();
int64_t pos = 0;
if (OB_FAIL(serialization::decode_i64(buf, buf_len, pos, &submit_log_type_tmp))) {
CLOG_LOG(ERROR, "failed to decode submit_log_type", K(entry), K(ret));
} else {
ObStorageLogType submit_log_type = static_cast<ObStorageLogType>(submit_log_type_tmp);
const uint64_t real_tenant_id = entry.get_header().get_partition_key().get_tenant_id();
if (ObStorageLogTypeChecker::is_trans_log(submit_log_type)) {
if (OB_FAIL(serialization::decode_i64(buf, buf_len, pos, &trans_inc))) {
CLOG_LOG(ERROR, "failed to decode trans_inc", K(ret), KP(buf), K(pos), K(buf_len));
} else if (OB_FAIL(dump_trans_log(submit_log_type, trans_inc, real_tenant_id, buf, buf_len, pos))) {
CLOG_LOG(ERROR, "failed to dump_trans_log", K(ret), KP(buf), K(pos), K(buf_len));
}
} else if (submit_log_type == OB_LOG_MAJOR_FREEZE) {
fprintf(stdout, "\t|||freeze log type:%s ", get_submit_log_type(submit_log_type));
ObStorageLogType log_type = storage::OB_LOG_UNKNOWN;
ObFreezeType freeze_type = INVALID_FREEZE;
ObPartitionKey partition_key;
ObSavedStorageInfo info;
int64_t frozen_version = -1;
if (OB_FAIL(dump_freeze_log(buf, buf_len, log_type, freeze_type, partition_key, frozen_version, info))) {
CLOG_LOG(ERROR, "failed to dump freeze log", K(ret));
} else {
fprintf(stdout,
"|FreezeType:%s|frozen_version:%ld|frozen_timestamp:%s|schema_version:%ld|saved_storage_info:%s |||",
get_freeze_type(freeze_type),
frozen_version,
time2str(info.get_frozen_timestamp()),
info.get_schema_version(),
to_cstring(info));
}
} else if (OB_LOG_OFFLINE_PARTITION == submit_log_type) {
fprintf(stdout, "|| OB_LOG_OFFLINE_PARTITION");
} else if (OB_LOG_OFFLINE_PARTITION_V2 == submit_log_type) {
if (OB_FAIL(dump_new_offline_partition_log(buf + pos, buf_len - pos))) {
CLOG_LOG(WARN, "dump_new_offline_partition_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
} else if (OB_PARTITION_SCHEMA_VERSION_CHANGE_LOG == submit_log_type) {
if (OB_FAIL(dump_partition_schema_version_change_log(buf + pos, buf_len - pos))) {
CLOG_LOG(WARN, "dump partition schema change log fail", K(ret), K(buf), K(pos), K(buf_len));
}
} else if (OB_LOG_ADD_PARTITION_TO_PG == submit_log_type) {
if (OB_FAIL(dump_add_partition_to_pg_log(buf + pos, buf_len - pos))) {
CLOG_LOG(WARN, "dump_add_partition_to_pg_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
} else if (OB_LOG_REMOVE_PARTITION_FROM_PG == submit_log_type) {
if (OB_FAIL(dump_remove_partition_from_pg_log(buf + pos, buf_len - pos))) {
CLOG_LOG(WARN, "dump_remove_partition_from_pg_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
} else if (ObStorageLogTypeChecker::is_split_log(submit_log_type)) {
switch (submit_log_type) {
case OB_LOG_SPLIT_SOURCE_PARTITION: {
if (OB_FAIL(dump_part_split_src_log(buf + pos, buf_len - pos))) {
CLOG_LOG(ERROR, "dump_part_split_src_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_SPLIT_DEST_PARTITION: {
if (OB_FAIL(dump_part_split_dest_log(buf + pos, buf_len - pos))) {
CLOG_LOG(ERROR, "dump_part_split_dest_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
default: {
CLOG_LOG(ERROR, "unknown log type", K(submit_log_type));
break;
}
}
} else if (submit_log_type == OB_LOG_TRANS_CHECKPOINT) {
if (OB_FAIL(dump_trans_checkpoint_log(buf + pos, buf_len - pos))) {
CLOG_LOG(ERROR, "dump_trans_checkpoint_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
} else { /*todo may be other logs need be printed?*/
}
}
} else if (OB_LOG_AGGRE == log_type) {
const char* log_buf = entry.get_buf();
int64_t buf_len = entry.get_header().get_data_len();
int32_t next_log_offset = 0;
const uint64_t real_tenant_id = entry.get_header().get_partition_key().get_tenant_id();
while (OB_SUCC(ret) && next_log_offset < buf_len) {
int64_t saved_pos = next_log_offset;
int64_t pos = next_log_offset;
int64_t log_type_in_buf = storage::OB_LOG_UNKNOWN;
ObStorageLogType log_type = storage::OB_LOG_UNKNOWN;
int64_t trans_inc = 0;
int64_t submit_timestamp = 0;
if (OB_FAIL(serialization::decode_i32(log_buf, buf_len, pos, &next_log_offset))) {
REPLAY_LOG(ERROR, "serialization decode_i32 failed", K(ret));
} else if (OB_FAIL(serialization::decode_i64(log_buf, buf_len, pos, &submit_timestamp))) {
REPLAY_LOG(ERROR, "serialization::decode_i64 failed", K(ret));
} else if (OB_FAIL(serialization::decode_i64(log_buf, buf_len, pos, &log_type_in_buf))) {
REPLAY_LOG(ERROR, "serialization::decode_i64 failed", K(ret));
} else if (OB_FAIL(serialization::decode_i64(log_buf, buf_len, pos, &trans_inc))) {
REPLAY_LOG(ERROR, "serialization::decode_i64 failed", K(ret));
} else {
const int64_t buf_len = next_log_offset - saved_pos - AGGRE_LOG_RESERVED_SIZE;
log_type = static_cast<ObStorageLogType>(log_type_in_buf);
const char* buf = log_buf + saved_pos + AGGRE_LOG_RESERVED_SIZE;
int64_t pos = 16;
if (ObStorageLogTypeChecker::is_trans_log(log_type)) {
if (OB_FAIL(dump_trans_log(log_type, trans_inc, real_tenant_id, buf, buf_len, pos))) {
CLOG_LOG(ERROR, "failed to dump_trans_log", K(entry), K(ret), KP(buf), K(pos), K(buf_len));
}
}
}
}
} else if (OB_LOG_ARCHIVE_CHECKPOINT == log_type || OB_LOG_ARCHIVE_KICKOFF == log_type) {
int64_t pos = 0;
ObLogArchiveInnerLog log;
if (OB_FAIL(log.deserialize(entry.get_buf(), entry.get_header().get_data_len(), pos))) {
CLOG_LOG(ERROR, "failed to deserialize archive_inner_log", K(entry), K(ret));
} else {
fprintf(stdout, "|| ObArchiveInnerLog:%s", to_cstring(log));
}
} else { /*todo may be other logs need be printed?*/
}
fprintf(stdout, "\n");
return ret;
}
int ObLogEntryParser::dump_ilog_entry(ObIndexEntry& entry)
{
int ret = OB_SUCCESS;
if (!entry.check_integrity()) {
ret = OB_INVALID_DATA;
CLOG_LOG(ERROR, "entry.check_integrity() fail", K(entry), K(ret));
} else {
fprintf(stdout, "INDEX_LOG: %s||\n", to_cstring(entry));
}
return ret;
}
int ObLogEntryParserImpl::dump_freeze_log(const char* buf, const int64_t buf_len, ObStorageLogType& log_type,
ObFreezeType& freeze_type, ObPartitionKey& pkey, int64_t& frozen_version, ObSavedStorageInfo& info)
{
int ret = OB_SUCCESS;
int64_t pos = 0;
log_type = storage::OB_LOG_UNKNOWN;
freeze_type = INVALID_FREEZE;
int64_t local_log_type = storage::OB_LOG_UNKNOWN;
int64_t local_freeze_type = INVALID_FREEZE;
if (OB_ISNULL(buf) || buf_len <= 0) {
ret = OB_INVALID_ARGUMENT;
CLOG_LOG(ERROR, "invalid argument", KP(buf), K(buf_len), K(ret));
} else if (OB_FAIL(serialization::decode_i64(buf, buf_len, pos, &local_log_type))) {
CLOG_LOG(ERROR, "deserialize log_type error", K(buf_len), K(pos), K(ret));
} else if (OB_FAIL(serialization::decode_i64(buf, buf_len, pos, &local_freeze_type))) {
CLOG_LOG(ERROR, "deserialize freeze_type error", K(buf_len), K(pos), K(ret));
} else if (OB_FAIL(pkey.deserialize(buf, buf_len, pos))) {
CLOG_LOG(ERROR, "deserialize partition key error", K(buf_len), K(pos), K(ret));
} else if (OB_FAIL(serialization::decode_i64(buf, buf_len, pos, &frozen_version))) {
CLOG_LOG(ERROR, "deserialize frozen version error", K(buf_len), K(pos), K(ret));
} else if (OB_FAIL(info.deserialize(buf, buf_len, pos))) {
CLOG_LOG(ERROR, "deserialize saved storage info error", K(buf_len), K(pos), K(ret));
} else {
log_type = static_cast<ObStorageLogType>(local_log_type);
freeze_type = static_cast<ObFreezeType>(local_freeze_type);
}
return ret;
}
int ObLogEntryParserImpl::dump_obj(const common::ObObj& obj, uint64_t column_id)
{
int ret = OB_SUCCESS;
if (without_data_) {
fprintf(stdout, "*");
} else if (dump_hex_) {
int64_t buf_len = PRINT_BUF_SIZE - 1; // reserver one character for '\0'
int64_t pos = 0;
if (OB_UNLIKELY(!is_inited_)) {
ret = OB_NOT_INIT;
CLOG_LOG(ERROR, "log entry parser is not inited", K(is_inited_), K(ret));
} else if (OB_FAIL(obj.print_smart(print_buf_, buf_len, pos))) {
CLOG_LOG(ERROR, "failed to print smart obj", K(column_id), K(obj), K(ret));
} else if (pos >= PRINT_BUF_SIZE - 1) {
ret = OB_ERR_UNEXPECTED;
CLOG_LOG(ERROR, "UNEXPECTED pos", K(pos), K(buf_len), K(ret));
} else {
print_buf_[pos] = '\0';
fprintf(stdout, " %ld:%s", column_id, print_buf_);
}
} else {
fprintf(stdout, " %ld:%s", column_id, to_cstring(obj));
}
return ret;
}
int ObLogEntryParser::skip_block_(const ObLogBlockMetaV2& meta)
{
int ret = OB_SUCCESS;
if (!meta.check_meta_checksum()) {
// fatal error, data corrupt
ret = OB_INVALID_DATA;
CLOG_LOG(ERROR, "check block meta checksum error", K(ret), K(meta));
} else {
ObBlockType type = meta.get_block_type();
switch (type) {
case OB_DATA_BLOCK:
advance_(meta.get_serialize_size());
break;
case OB_HEADER_BLOCK: // not used any more
// file header is CLOG_DIO_ALIGN_SIZE bytes,
// skip it
advance_(meta.get_total_len());
break;
case OB_INFO_BLOCK:
if (is_ofs_) {
advance_(meta.get_serialize_size() + meta.get_data_len());
} else {
ret = OB_ITER_END; // info block is the end for local file
}
break;
case OB_TRAILER_BLOCK:
ret = OB_ITER_END; // trailer block is the end
break;
default:
ret = OB_INVALID_DATA;
CLOG_LOG(ERROR, "unknown block meta type", K(ret), K(meta), K(file_id_), K(cur_offset_));
}
}
return ret;
}
int ObLogEntryParserImpl::format_dump_clog_entry(ObLogEntry& entry)
{
int ret = OB_SUCCESS;
if (!entry.check_integrity()) {
ret = OB_INVALID_DATA;
CLOG_LOG(ERROR, "entry.check_integrity() fail", K(entry), K(ret));
} else {
// only handle redo log
if (OB_LOG_SUBMIT == entry.get_header().get_log_type()) {
int64_t trans_inc = -1;
int64_t submit_log_type_tmp = -1;
const char* buf = entry.get_buf();
int64_t buf_len = entry.get_header().get_data_len();
int64_t pos = 0;
if (OB_FAIL(serialization::decode_i64(buf, buf_len, pos, &submit_log_type_tmp))) {
CLOG_LOG(ERROR, "failed to decode submit_log_type", K(entry), K(ret));
} else {
int submit_log_type = static_cast<int>(submit_log_type_tmp);
const uint64_t real_tenant_id = entry.get_header().get_partition_key().get_tenant_id();
if (((OB_LOG_SP_TRANS_REDO & submit_log_type) || (OB_LOG_TRANS_REDO & submit_log_type) ||
OB_LOG_SP_ELR_TRANS_COMMIT == submit_log_type || OB_LOG_SP_TRANS_COMMIT == submit_log_type) &&
submit_log_type < OB_LOG_TRANS_MAX) {
if (OB_FAIL(serialization::decode_i64(buf, buf_len, pos, &trans_inc))) {
CLOG_LOG(ERROR, "failed to decode trans_inc", K(ret), KP(buf), K(pos), K(buf_len));
} else {
switch (submit_log_type) {
case OB_LOG_SP_TRANS_REDO: {
if (OB_FAIL(format_dump_sp_trans_redo_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(ERROR, "format_dump_sp_trans_redo_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_SP_TRANS_COMMIT:
case OB_LOG_SP_ELR_TRANS_COMMIT: {
if (OB_FAIL(format_dump_sp_trans_commit_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(ERROR, "format_dump_sp_trans_redo_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_TRANS_REDO: {
if (OB_FAIL(format_dump_trans_redo_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(ERROR, "format_dump_trans_redo_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_TRANS_REDO_WITH_PREPARE: {
if (OB_FAIL(format_dump_trans_redo_prepare_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(ERROR, "format_dump_trans_redo_prepare_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_TRANS_REDO_WITH_PREPARE_WITH_COMMIT: {
if (OB_FAIL(format_dump_trans_redo_prepare_commit_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(WARN, "format_dump_trans_redo_prepare_commit_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_TRANS_REDO_WITH_PREPARE_WITH_COMMIT_WITH_CLEAR: {
if (OB_FAIL(
format_dump_trans_redo_prepare_commit_clear_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(
WARN, "format_dump_trans_redo_prepare_commit_clear_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
default: {
CLOG_LOG(ERROR, "unknown log type", K(submit_log_type));
break;
}
}
}
}
}
}
}
return ret;
}
int ObLogEntryParserImpl::format_dump_sp_trans_redo_log(const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
ObSpTransRedoLogHelper helper;
ObSpTransRedoLog log(helper);
int64_t pos = 0;
if (OB_FAIL(log.init_for_deserialize())) {
CLOG_LOG(WARN, "redo log init for deserialize fail", K(ret));
} else if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialize log", K(ret));
} else if (OB_FAIL(log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(log.decrypt_table_key())) {
cur_trans_id_ = log.get_trans_id();
format_dump_memtable_mutator(log.get_mutator().get_data(), log.get_mutator().get_position());
} else if (ret == OB_EAGAIN) {
// impossible branch
} else {
CLOG_LOG(WARN, "log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParserImpl::format_dump_sp_trans_commit_log(const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
ObSpTransCommitLogHelper helper;
ObSpTransCommitLog log(helper);
int64_t pos = 0;
if (OB_FAIL(log.init_for_deserialize())) {
CLOG_LOG(WARN, "redo log init for deserialize fail", K(ret));
} else if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialize log", K(ret));
} else if (OB_FAIL(log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(log.decrypt_table_key())) {
cur_trans_id_ = log.get_trans_id();
format_dump_memtable_mutator(log.get_mutator().get_data(), log.get_mutator().get_position());
} else if (ret == OB_EAGAIN) {
// impossible branch
} else {
CLOG_LOG(WARN, "log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParserImpl::format_dump_trans_redo_log(const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
ObTransRedoLogHelper helper;
ObTransRedoLog log(helper);
int64_t pos = 0;
if (OB_FAIL(log.init_for_deserialize())) {
CLOG_LOG(WARN, "redo log init for deserialize fail", K(ret));
} else if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialize log", K(ret));
} else if (OB_FAIL(log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(log.decrypt_table_key())) {
cur_trans_id_ = log.get_trans_id();
format_dump_memtable_mutator(log.get_mutator().get_data(), log.get_mutator().get_position());
} else if (ret == OB_EAGAIN) {
// impossible branch
} else {
CLOG_LOG(WARN, "log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParserImpl::format_dump_trans_redo_prepare_log(
const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
ObTransRedoLogHelper helper;
ObTransRedoLog r_log(helper);
int64_t pos = 0;
if (OB_FAIL(r_log.init_for_deserialize())) {
CLOG_LOG(WARN, "redo log init for deserialize fail", K(ret));
} else if (OB_FAIL(r_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialize log", K(ret));
} else if (OB_FAIL(r_log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(r_log.decrypt_table_key())) {
cur_trans_id_ = r_log.get_trans_id();
format_dump_memtable_mutator(r_log.get_mutator().get_data(), r_log.get_mutator().get_position());
} else if (ret == OB_EAGAIN) {
// impossible branch
} else {
CLOG_LOG(WARN, "redo log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParserImpl::format_dump_trans_redo_prepare_commit_log(
const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
int64_t pos = 0;
ObTransRedoLogHelper helper;
ObTransRedoLog r_log(helper);
if (OB_FAIL(r_log.init_for_deserialize())) {
CLOG_LOG(WARN, "redo log init for deserialize fail", K(ret));
} else if (OB_FAIL(r_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialize log", K(ret));
} else if (OB_FAIL(r_log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(r_log.decrypt_table_key())) {
cur_trans_id_ = r_log.get_trans_id();
format_dump_memtable_mutator(r_log.get_mutator().get_data(), r_log.get_mutator().get_position());
} else if (ret == OB_EAGAIN) {
// impossible branch
} else {
CLOG_LOG(WARN, "redo log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParserImpl::format_dump_trans_redo_prepare_commit_clear_log(
const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
int64_t pos = 0;
ObTransRedoLogHelper helper;
ObTransRedoLog r_log(helper);
if (OB_FAIL(r_log.init_for_deserialize())) {
CLOG_LOG(WARN, "redo log init for deserialize fail", K(ret));
} else if (OB_FAIL(r_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialize log", K(ret));
} else if (OB_FAIL(r_log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(r_log.decrypt_table_key())) {
cur_trans_id_ = r_log.get_trans_id();
format_dump_memtable_mutator(r_log.get_mutator().get_data(), r_log.get_mutator().get_position());
} else if (ret == OB_EAGAIN) {
// impossible branch
} else {
CLOG_LOG(WARN, "redo log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParserImpl::format_dump_memtable_mutator(const char* buf, int64_t len)
{
int ret = OB_SUCCESS;
int64_t pos = 0;
ObMemtableMutatorIterator mmi;
ObMemtableMutatorMeta meta;
int64_t meta_pos = 0;
if (OB_FAIL(meta.deserialize(buf, len, meta_pos))) {
CLOG_LOG(ERROR, "meta.deserialize fail", K(ret));
} else if (ObTransRowFlag::is_normal_row(meta.get_flags())) {
if (OB_FAIL(mmi.deserialize(buf, len, pos))) {
CLOG_LOG(ERROR, "mmi.deserialize fail", K(ret));
}
while (OB_SUCCESS == ret) {
ObMemtableMutatorRow row;
if (OB_FAIL(mmi.get_next_row(row)) && OB_ITER_END != ret) {
CLOG_LOG(ERROR, "mmi.get_next_row fail", K(ret));
} else if (OB_ITER_END == ret) {
ret = OB_SUCCESS;
break;
} else {
if (T_DML_LOCK == row.dml_type_) {
// do nothing
} else {
fprintf(stdout, "%s\t", to_cstring(cur_trans_id_));
fprintf(stdout, "%s\t", get_row_dml_type_str(row.dml_type_));
fprintf(stdout, "%lu\t", row.table_id_);
int64_t pos = row.rowkey_.to_format_string(print_buf_, PRINT_BUF_SIZE - 1);
print_buf_[pos] = '\0';
fprintf(stdout, "%s\t", print_buf_);
ObCellReader old_cci;
if (NULL != row.old_row_.data_ && row.old_row_.size_ > 0) {
if (OB_FAIL(old_cci.init(row.old_row_.data_, row.old_row_.size_, SPARSE))) {
CLOG_LOG(ERROR, "old cci.init fail", K(ret));
}
int64_t index = 0;
while (OB_SUCC(ret) && OB_SUCC(old_cci.next_cell())) {
uint64_t column_id = OB_INVALID_ID;
const ObObj* value = NULL;
if (OB_FAIL(old_cci.get_cell(column_id, value))) {
CLOG_LOG(ERROR, "failed to get cell", K(ret));
} else if (NULL == value) {
ret = OB_ERR_UNEXPECTED;
CLOG_LOG(ERROR, "got value is NULL", K(ret));
} else if (ObExtendType == value->get_type() && ObActionFlag::OP_END_FLAG == value->get_ext()) {
ret = OB_ITER_END;
} else {
if (index > 0) {
fprintf(stdout, " "); // separate columns
}
if (OB_SUCC(ret) && OB_FAIL(format_dump_obj(*value, column_id))) {
CLOG_LOG(ERROR, "failed to dump obj", K(*value), K(ret));
} else {
++index;
}
}
}
if (OB_ITER_END == ret) {
ret = OB_SUCCESS;
}
} else if (NULL == row.old_row_.data_ && 0 == row.old_row_.size_) {
// no data of old row
} else {
CLOG_LOG(ERROR, "old cci.init fail", K(row), K(ret));
}
if (OB_SUCC(ret) && T_DML_DELETE != row.dml_type_) {
ObCellReader new_cci;
if (OB_FAIL(new_cci.init(row.new_row_.data_, row.new_row_.size_, SPARSE))) {
CLOG_LOG(ERROR, "new cci.init fail", K(ret));
} else {
int64_t index = 0;
while (OB_SUCC(ret) && OB_SUCC(new_cci.next_cell())) {
uint64_t column_id = OB_INVALID_ID;
const ObObj* value = NULL;
if (OB_FAIL(new_cci.get_cell(column_id, value))) {
} else if (NULL == value) {
ret = OB_ERR_UNEXPECTED;
} else if (ObExtendType == value->get_type() && ObActionFlag::OP_END_FLAG == value->get_ext()) {
ret = OB_ITER_END;
} else {
if (OB_SUCC(ret) && OB_FAIL(format_dump_obj(*value, column_id))) {
CLOG_LOG(ERROR, "failed to dump obj", K(*value), K(ret));
} else {
++index;
}
}
}
if (OB_ITER_END == ret) {
ret = OB_SUCCESS;
}
}
}
fprintf(stdout, "\n");
}
}
}
} else {
fprintf(stdout, "BIG_ROW: %s\n", to_cstring(mmi.get_meta()));
}
if (OB_ITER_END == ret) {
ret = OB_SUCCESS;
}
return ret;
}
int ObLogEntryParserImpl::format_dump_obj(const common::ObObj& obj, uint64_t column_id)
{
int ret = OB_SUCCESS;
int64_t buf_len = PRINT_BUF_SIZE - 1; // reserver one character for '\0'
int64_t pos = 0;
if (OB_UNLIKELY(!is_inited_)) {
ret = OB_NOT_INIT;
CLOG_LOG(ERROR, "log entry parser is not inited", K(is_inited_), K(ret));
} else if (OB_FAIL(obj.print_format(print_buf_, buf_len, pos))) {
CLOG_LOG(ERROR, "failed to print smart obj", K(column_id), K(obj), K(ret));
} else if (pos >= PRINT_BUF_SIZE - 1) {
ret = OB_ERR_UNEXPECTED;
CLOG_LOG(ERROR, "UNEXPECTED pos", K(pos), K(buf_len), K(ret));
} else {
print_buf_[pos] = '\0';
fprintf(stdout, "%lu:%s", column_id, print_buf_);
}
return ret;
}
int ObLogEntryParser::stat_block_(const ObLogBlockMetaV2& meta)
{
int ret = OB_SUCCESS;
if (!meta.check_meta_checksum()) {
// fatal error, data corrupt
ret = OB_INVALID_DATA;
CLOG_LOG(ERROR, "check block meta checksum error", K(ret), K(meta));
} else {
ObBlockType type = meta.get_block_type();
switch (type) {
case OB_DATA_BLOCK: {
int64_t size = meta.get_serialize_size();
advance_(size);
log_stat_.data_block_header_size_ += size;
break;
}
case OB_HEADER_BLOCK:
// not used any more
// file header is CLOG_DIO_ALIGN_SIZE bytes,
// skip it
advance_(meta.get_total_len());
break;
case OB_INFO_BLOCK: {
ret = OB_ITER_END; // info block is the end
break;
}
case OB_TRAILER_BLOCK:
ret = OB_ITER_END;
break;
// iter will never encounter a trailer for InfoBlock lies ahead
default:
ret = OB_INVALID_DATA;
CLOG_LOG(ERROR, "unknown block meta type", K(ret), K(meta), K(file_id_), K(cur_offset_));
}
}
return ret;
}
int ObLogEntryParser::stat_clog_entry(const ObLogEntry& entry, const int64_t pos, const bool is_compressed)
{
int ret = OB_SUCCESS;
if (!entry.check_integrity()) {
ret = OB_INVALID_DATA;
CLOG_LOG(ERROR, "entry.check_integrity() fail", K(entry), K(ret));
} else if (is_compressed && OB_FAIL(log_stat_.compressed_tenant_ids_.set_refactored(
entry.get_header().get_partition_key().get_tenant_id(), 1 /* overwrite*/))) {
CLOG_LOG(ERROR, "failed to set_refactored", K(entry), K(ret));
} else {
log_stat_.log_header_size_ += entry.get_header().get_serialize_size();
log_stat_.log_size_ += entry.get_header().get_data_len();
log_stat_.primary_table_id_ = entry.get_header().get_partition_key().get_table_id();
if (OB_LOG_SUBMIT == entry.get_header().get_log_type()) {
if (is_compressed) {
log_stat_.compressed_log_cnt_++;
log_stat_.compressed_log_size_ += pos;
log_stat_.original_log_size_ += entry.get_total_len();
} else {
log_stat_.non_compressed_log_cnt_++;
}
++log_stat_.total_log_count_;
int64_t trans_inc = -1;
int64_t submit_log_type_tmp = -1;
const char* buf = entry.get_buf();
int64_t buf_len = entry.get_header().get_data_len();
int64_t pos = 0;
if (OB_FAIL(serialization::decode_i64(buf, buf_len, pos, &submit_log_type_tmp))) {
CLOG_LOG(ERROR, "failed to decode submit_log_type", K(entry), K(ret));
} else {
int submit_log_type = static_cast<int>(submit_log_type_tmp);
const uint64_t real_tenant_id = entry.get_header().get_partition_key().get_tenant_id();
if ((OB_LOG_SP_TRANS_COMMIT & submit_log_type) || (OB_LOG_SP_ELR_TRANS_COMMIT & submit_log_type)) {
++log_stat_.sp_trans_count_;
} else if (OB_LOG_TRANS_PREPARE & submit_log_type) {
++log_stat_.dist_trans_count_;
}
if (((OB_LOG_SP_TRANS_REDO & submit_log_type) || (OB_LOG_TRANS_REDO & submit_log_type) ||
OB_LOG_SP_ELR_TRANS_COMMIT == submit_log_type || OB_LOG_SP_TRANS_COMMIT == submit_log_type) &&
submit_log_type < OB_LOG_TRANS_MAX) {
if (OB_FAIL(serialization::decode_i64(buf, buf_len, pos, &trans_inc))) {
CLOG_LOG(ERROR, "failed to decode trans_inc", K(ret), KP(buf), K(pos), K(buf_len));
} else {
log_stat_.trans_log_size_ += buf_len - pos;
switch (submit_log_type) {
case OB_LOG_TRANS_REDO: {
if (OB_FAIL(stat_trans_redo_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(ERROR, "dump_trans_redo_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_SP_TRANS_REDO: {
if (OB_FAIL(stat_sp_trans_redo_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(ERROR, "dump_sp_trans_redo_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_TRANS_REDO_WITH_PREPARE: {
if (OB_FAIL(stat_trans_redo_prepare_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(ERROR, "dump_trans_redo_prepare_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_SP_TRANS_COMMIT:
case OB_LOG_SP_ELR_TRANS_COMMIT: {
if (OB_FAIL(stat_sp_trans_commit_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(WARN, "dump_sp_trans_commit_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_TRANS_REDO_WITH_PREPARE_WITH_COMMIT: {
if (OB_FAIL(stat_trans_redo_prepare_commit_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(WARN, "dump_trans_redo_prepare_commit_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_TRANS_REDO_WITH_PREPARE_WITH_COMMIT_WITH_CLEAR: {
if (OB_FAIL(stat_trans_redo_prepare_commit_clear_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(WARN, "dump_trans_redo_prepare_commit_clear_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
default: {
CLOG_LOG(ERROR, "unknown log type", K(submit_log_type));
break;
}
}
}
}
}
}
}
return ret;
}
int ObLogEntryParser::stat_sp_trans_redo_log(const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
ObSpTransRedoLogHelper helper;
ObSpTransRedoLog log(helper);
int64_t pos = 0;
if (OB_FAIL(log.init_for_deserialize())) {
CLOG_LOG(WARN, "redo log init for deserialize fail", K(ret));
} else if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialize log", K(ret));
} else if (OB_FAIL(log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(log.decrypt_table_key())) {
log_stat_.mutator_size_ += log.get_mutator().get_position();
stat_memtable_mutator(log.get_mutator().get_data(), log.get_mutator().get_position());
} else if (ret == OB_EAGAIN) {
// impossible branch
} else {
CLOG_LOG(WARN, "log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParser::stat_sp_trans_commit_log(const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
ObSpTransCommitLogHelper helper;
ObSpTransCommitLog log(helper);
int64_t pos = 0;
if (OB_FAIL(log.init_for_deserialize())) {
CLOG_LOG(WARN, "redo log init for deserialize fail", K(ret));
} else if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialize log", K(ret));
} else if (OB_FAIL(log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(log.decrypt_table_key())) {
log_stat_.mutator_size_ += log.get_mutator().get_position();
stat_memtable_mutator(log.get_mutator().get_data(), log.get_mutator().get_position());
} else if (ret == OB_EAGAIN) {
// impossible branch
} else {
CLOG_LOG(WARN, "log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParser::stat_trans_redo_log(const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
ObTransRedoLogHelper helper;
ObTransRedoLog log(helper);
int64_t pos = 0;
if (OB_FAIL(log.init_for_deserialize())) {
CLOG_LOG(WARN, "redo log init for deserialize fail", K(ret));
} else if (OB_FAIL(log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialize log", K(ret));
} else if (OB_FAIL(log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(log.decrypt_table_key())) {
log_stat_.mutator_size_ += log.get_mutator().get_position();
stat_memtable_mutator(log.get_mutator().get_data(), log.get_mutator().get_position());
} else if (ret == OB_EAGAIN) {
// impossible branch
} else {
CLOG_LOG(WARN, "log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParser::stat_trans_redo_prepare_log(const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
ObTransRedoLogHelper helper;
ObTransRedoLog r_log(helper);
int64_t pos = 0;
if (OB_FAIL(r_log.init_for_deserialize())) {
CLOG_LOG(WARN, "redo log init for deserialize fail", K(ret));
} else if (OB_FAIL(r_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialzie log", K(ret));
} else if (OB_FAIL(r_log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(r_log.decrypt_table_key())) {
log_stat_.mutator_size_ += r_log.get_mutator().get_position();
stat_memtable_mutator(r_log.get_mutator().get_data(), r_log.get_mutator().get_position());
} else if (ret == OB_EAGAIN) {
// impossible branch
} else {
CLOG_LOG(WARN, "redo log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParser::stat_trans_redo_prepare_commit_log(const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
int64_t pos = 0;
ObTransRedoLogHelper helper;
ObTransRedoLog r_log(helper);
if (OB_FAIL(r_log.init_for_deserialize())) {
CLOG_LOG(WARN, "redo log init for deserialize fail", K(ret));
} else if (OB_FAIL(r_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialize log", K(ret));
} else if (OB_FAIL(r_log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(r_log.decrypt_table_key())) {
log_stat_.mutator_size_ += r_log.get_mutator().get_position();
stat_memtable_mutator(r_log.get_mutator().get_data(), r_log.get_mutator().get_position());
} else if (ret == OB_EAGAIN) {
// impossible branch
} else {
CLOG_LOG(WARN, "redo log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParser::stat_trans_redo_prepare_commit_clear_log(
const char* data, int64_t len, const uint64_t real_tenant_id)
{
int ret = OB_SUCCESS;
int64_t pos = 0;
ObTransRedoLogHelper helper;
ObTransRedoLog r_log(helper);
if (OB_FAIL(r_log.init_for_deserialize())) {
CLOG_LOG(WARN, "redo log init for deserialize fail", K(ret));
} else if (OB_FAIL(r_log.deserialize(data, len, pos))) {
CLOG_LOG(WARN, "failed to deserialize log", K(ret));
} else if (OB_FAIL(r_log.replace_encrypt_info_tenant_id(real_tenant_id))) {
CLOG_LOG(WARN, "failed to replace encrypt info tenant id", K(ret));
} else if (OB_SUCC(r_log.decrypt_table_key())) {
log_stat_.mutator_size_ += r_log.get_mutator().get_position();
stat_memtable_mutator(r_log.get_mutator().get_data(), r_log.get_mutator().get_position());
} else if (ret == OB_EAGAIN) {
// impossible branch
} else {
CLOG_LOG(WARN, "redo log deserialize error", K(ret), KP(data), K(len), K(pos));
}
return ret;
}
int ObLogEntryParser::stat_memtable_mutator(const char* buf, int64_t len)
{
int ret = OB_SUCCESS;
int64_t pos = 0;
ObMemtableMutatorIterator mmi;
ObMemtableMutatorMeta meta;
int64_t meta_pos = 0;
if (OB_FAIL(meta.deserialize(buf, len, meta_pos))) {
CLOG_LOG(ERROR, "meta.deserialize fail", K(ret));
} else if (ObTransRowFlag::is_normal_row(meta.get_flags())) {
if (OB_FAIL(mmi.deserialize(buf, len, pos))) {
CLOG_LOG(ERROR, "mmi.deserialize fail", K(ret));
}
while (OB_SUCC(ret)) {
ObMemtableMutatorRow row;
if (OB_FAIL(mmi.get_next_row(row)) && OB_ITER_END != ret) {
CLOG_LOG(ERROR, "mmi.get_next_row fail", K(ret));
} else if (OB_ITER_END == ret) {
ret = OB_SUCCESS;
break;
} else {
++log_stat_.total_row_count_;
if (row.table_id_ == log_stat_.primary_table_id_) {
++log_stat_.primary_row_count_;
log_stat_.new_primary_row_size_ += row.new_row_.size_;
}
log_stat_.old_row_size_ += row.old_row_.size_;
log_stat_.new_row_size_ += row.new_row_.size_;
}
}
} else { /*do nothing*/
}
return ret;
}
int ObLogEntryParserImpl::check_filter(const ObLogEntry& entry, bool& need_print)
{
int ret = OB_SUCCESS;
need_print = true;
const ObLogEntryHeader& header = entry.get_header();
const ObPartitionKey& pkey = header.get_partition_key();
const uint64_t log_id = header.get_log_id();
if ((filter_.is_table_id_valid() && filter_.get_table_id() != pkey.get_table_id()) ||
(filter_.is_partition_id_valid() && filter_.get_partition_id() != pkey.get_partition_id()) ||
(filter_.is_log_id_valid() && filter_.get_log_id() != log_id)) {
need_print = false;
}
if (need_print) {
CLOG_LOG(INFO, "find target log entry", K(header));
}
return ret;
}
int ObLogEntryParserImpl::dump_trans_log(const ObStorageLogType log_type, const int64_t trans_inc,
const uint64_t real_tenant_id, const char* buf, const int64_t buf_len, int64_t& pos)
{
int ret = OB_SUCCESS;
fprintf(stdout, "\t|||Trans: log_type:%s, trans_inc:%ld", get_submit_log_type(log_type), trans_inc);
switch (log_type) {
case OB_LOG_TRANS_REDO: {
if (OB_FAIL(dump_trans_redo_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(ERROR, "dump_trans_redo_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_SP_TRANS_REDO: {
if (OB_FAIL(dump_sp_trans_redo_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(ERROR, "dump_sp_trans_redo_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_TRANS_PREPARE: {
if (OB_FAIL(dump_trans_prepare_log(buf + pos, buf_len - pos))) {
CLOG_LOG(ERROR, "dump_trans_prepare_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_TRANS_REDO_WITH_PREPARE: {
if (OB_FAIL(dump_trans_redo_prepare_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(ERROR, "dump_trans_redo_prepare_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_TRANS_COMMIT: {
if (OB_FAIL(dump_trans_commit_log(buf + pos, buf_len - pos))) {
CLOG_LOG(WARN, "dump_trans_commit_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_SP_TRANS_COMMIT:
case OB_LOG_SP_ELR_TRANS_COMMIT: {
if (OB_FAIL(dump_sp_trans_commit_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(WARN, "dump_sp_trans_commit_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_TRANS_CLEAR: {
if (OB_FAIL(dump_trans_clear_log(buf + pos, buf_len - pos))) {
CLOG_LOG(WARN, "dump_trans_clear_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_TRANS_ABORT: {
if (OB_FAIL(dump_trans_abort_log(buf + pos, buf_len - pos))) {
CLOG_LOG(WARN, "dump_sp_trans_abort_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_SP_TRANS_ABORT: {
if (OB_FAIL(dump_sp_trans_abort_log(buf + pos, buf_len - pos))) {
CLOG_LOG(WARN, "dump_sp_trans_abort_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_TRANS_PREPARE_WITH_COMMIT: {
if (OB_FAIL(dump_trans_prepare_commit_log(buf + pos, buf_len - pos))) {
CLOG_LOG(WARN, "dump_trans_prepare_commit_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_TRANS_REDO_WITH_PREPARE_WITH_COMMIT: {
if (OB_FAIL(dump_trans_redo_prepare_commit_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(WARN, "dump_trans_redo_prepare_commit_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_TRANS_PREPARE_WITH_COMMIT_WITH_CLEAR: {
if (OB_FAIL(dump_trans_prepare_commit_clear_log(buf + pos, buf_len - pos))) {
CLOG_LOG(WARN, "dump_trans_prepare_commit_clear_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_TRANS_REDO_WITH_PREPARE_WITH_COMMIT_WITH_CLEAR: {
if (OB_FAIL(dump_trans_redo_prepare_commit_clear_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(WARN, "dump_trans_redo_prepare_commit_clear_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_MUTATOR: {
if (OB_FAIL(dump_trans_mutator_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(WARN, "dump_trans_mutator_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_TRANS_STATE: {
if (OB_FAIL(dump_trans_state_log(buf + pos, buf_len - pos))) {
CLOG_LOG(WARN, "dump_trans_state_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_MUTATOR_WITH_STATE: {
if (OB_FAIL(dump_trans_mutator_state_log(buf + pos, buf_len - pos, real_tenant_id))) {
CLOG_LOG(WARN, "dump_trans_mutator_state_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
case OB_LOG_MUTATOR_ABORT: {
if (OB_FAIL(dump_trans_mutator_abort_log(buf + pos, buf_len - pos))) {
CLOG_LOG(WARN, "dump_trans_mutator_abort_log fail", K(ret), K(buf), K(pos), K(buf_len));
}
break;
}
default: {
CLOG_LOG(ERROR, "unknown log type", K(log_type));
break;
}
}
return ret;
}
} // end of namespace clog
} // end of namespace oceanbase
Reference in New Issue View Git Blame Copy Permalink