hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
d9ca3d13de39ff4e8c123149f1b1dfe4f37dafee
oceanbase/src/storage/blocksstable/ob_micro_block_cache.cpp

1495 lines
55 KiB
C++
Raw 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.
*/
#define USING_LOG_PREFIX STORAGE
#include "storage/blocksstable/ob_micro_block_cache.h"
#include "storage/blocksstable/ob_block_manager.h"
#include "storage/blocksstable/ob_macro_block_handle.h"
#include "storage/blocksstable/ob_shared_macro_block_manager.h"
#include "storage/blocksstable/ob_macro_block_handle.h"
#include "storage/blocksstable/ob_shared_macro_block_manager.h"
#include "storage/blocksstable/cs_encoding/ob_cs_micro_block_transformer.h"
namespace oceanbase
{
using namespace common;
using namespace storage;
namespace blocksstable
{
/**
* -----------------------------------------------------ObMicroBlockCacheKey--------------------------------------------------
*/
ObMicroBlockCacheKey::ObMicroBlockCacheKey(
const uint64_t tenant_id,
const MacroBlockId &macro_id,
const int64_t offset,
const int64_t size)
: tenant_id_(tenant_id),
block_id_(macro_id, offset, size)
{
}
ObMicroBlockCacheKey::ObMicroBlockCacheKey(
const uint64_t tenant_id,
const ObMicroBlockId &block_id)
: tenant_id_(tenant_id), block_id_(block_id)
{
}
ObMicroBlockCacheKey::ObMicroBlockCacheKey()
: tenant_id_(common::OB_INVALID_TENANT_ID), block_id_()
{
}
ObMicroBlockCacheKey::ObMicroBlockCacheKey(const ObMicroBlockCacheKey &other)
{
tenant_id_ = other.tenant_id_;
block_id_ = other.block_id_;
}
ObMicroBlockCacheKey::~ObMicroBlockCacheKey()
{
}
void ObMicroBlockCacheKey::set(const uint64_t tenant_id,
const MacroBlockId &block_id,
const int64_t offset,
const int64_t size)
{
tenant_id_ = tenant_id;
block_id_.macro_id_ = block_id;
block_id_.offset_ = offset;
block_id_.size_ = size;
}
bool ObMicroBlockCacheKey::operator ==(const ObIKVCacheKey &other) const
{
const ObMicroBlockCacheKey &other_key = reinterpret_cast<const ObMicroBlockCacheKey &> (other);
return tenant_id_ == other_key.tenant_id_ && block_id_ == other_key.block_id_;
}
uint64_t ObMicroBlockCacheKey::get_tenant_id() const
{
return tenant_id_;
}
uint64_t ObMicroBlockCacheKey::hash() const
{
return murmurhash(this, sizeof(ObMicroBlockCacheKey), 0);
}
int64_t ObMicroBlockCacheKey::size() const
{
return sizeof(*this);
}
int ObMicroBlockCacheKey::deep_copy(char *buf, const int64_t buf_len, ObIKVCacheKey *&key) const
{
int ret = OB_SUCCESS;
if (OB_UNLIKELY(NULL == buf || buf_len < size())) {
ret = OB_INVALID_ARGUMENT;
STORAGE_LOG(WARN, "Invalid argument, ", K(ret));
} else if (OB_UNLIKELY(
0 == tenant_id_ || OB_INVALID_TENANT_ID == tenant_id_ || !block_id_.is_valid())) {
ret = OB_INVALID_DATA;
STORAGE_LOG(WARN, "The micro block cache key is invalid, ", K(*this), K(ret));
} else {
key = new (buf) ObMicroBlockCacheKey(tenant_id_, block_id_);
}
return ret;
}
/**
* -----------------------------------------------------ObMicroBlockCacheValue--------------------------------------------------
*/
ObMicroBlockCacheValue::ObMicroBlockCacheValue() : block_data_(), alloc_by_block_io_(false)
{
}
ObMicroBlockCacheValue::ObMicroBlockCacheValue(
const char *buf,
const int64_t size,
const char *extra_buf /* = NULL */,
const int64_t extra_size /* = 0 */,
const ObMicroBlockData::Type block_type /* = DATA_BLOCK */)
: block_data_(buf, size, extra_buf, extra_size, block_type),
alloc_by_block_io_(false)
{
}
ObMicroBlockCacheValue::~ObMicroBlockCacheValue()
{
}
int64_t ObMicroBlockCacheValue::size() const
{
return sizeof(blocksstable::ObMicroBlockCacheValue) + block_data_.total_size();
}
int ObMicroBlockCacheValue::deep_copy(char *buf, const int64_t buf_len, ObIKVCacheValue *&value) const
{
int ret = OB_SUCCESS;
ObMicroBlockCacheValue *pvalue = NULL;
if (OB_UNLIKELY(NULL == buf || buf_len < size())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("Invalid argument, ", K(ret));
} else if (OB_UNLIKELY(!block_data_.is_valid())) {
//buffer_ is allowed to be NULL
ret = OB_INVALID_DATA;
LOG_WARN("The micro block cache value is not valid, ", K(*this), K(ret));
} else {
char *new_buf = buf + sizeof(blocksstable::ObMicroBlockCacheValue);
MEMCPY(new_buf, block_data_.get_buf(), block_data_.get_buf_size());
if (NULL != block_data_.get_extra_buf() && block_data_.get_extra_size() > 0) {
switch (block_data_.type_) {
case ObMicroBlockData::Type::DATA_BLOCK: {
MEMCPY(new_buf + block_data_.get_buf_size(), block_data_.get_extra_buf(), block_data_.get_extra_size());
if (block_data_.get_store_type() == ObRowStoreType::CS_ENCODING_ROW_STORE) {
// no need update cached coder for CS_ENCODING_ROW_STORE
} else if (OB_FAIL(ObMicroBlockDecoder::update_cached_decoders(
new_buf + block_data_.get_buf_size(),
block_data_.get_extra_size(),
block_data_.get_buf(),
new_buf,
block_data_.get_buf_size()))) {
LOG_WARN(" Update cached pointer failed", K(ret), K_(block_data), KP(new_buf));
}
break;
}
case ObMicroBlockData::Type::INDEX_BLOCK: {
const ObIndexBlockDataHeader *src_idx_header
= reinterpret_cast<const ObIndexBlockDataHeader *>(block_data_.get_extra_buf());
char *new_extra_buf = new_buf + block_data_.get_buf_size();
const int64_t new_extra_buf_size = buf_len - block_data_.get_buf_size();
ObIndexBlockDataHeader *dst_idx_header = reinterpret_cast<ObIndexBlockDataHeader *>(new_extra_buf);
int64_t pos = sizeof(ObIndexBlockDataHeader);
if (OB_FAIL(dst_idx_header->deep_copy_transformed_index_block(
*src_idx_header, new_extra_buf_size, new_extra_buf, pos))) {
LOG_WARN("Fail to deep copy transformed index block", K(ret));
}
break;
}
default:
ret = OB_NOT_SUPPORTED;
LOG_WARN("Not Supported block data type", K(ret), K_(block_data));
}
if (OB_SUCC(ret)) {
pvalue = new (buf) ObMicroBlockCacheValue(
new_buf, block_data_.get_buf_size(),
new_buf + block_data_.get_buf_size(), block_data_.get_extra_size(), block_data_.type_);
pvalue->alloc_by_block_io_ = alloc_by_block_io_;
}
} else {
pvalue = new (buf) ObMicroBlockCacheValue(new_buf, block_data_.get_buf_size(),
nullptr, 0, block_data_.type_);
pvalue->alloc_by_block_io_ = alloc_by_block_io_;
}
value = pvalue;
}
return ret;
}
/*---------------------------------Multi Block IO parameters--------------------------------------*/
ObMultiBlockIOResult::ObMultiBlockIOResult()
{
reset();
}
ObMultiBlockIOResult::~ObMultiBlockIOResult()
{
}
void ObMultiBlockIOResult::reset()
{
micro_blocks_ = NULL;
handles_ = NULL;
block_count_ = 0;
ret_code_ = OB_SUCCESS;
}
int ObMultiBlockIOResult::get_block_data(
const int64_t index, ObMicroBlockData &block_data) const
{
int ret = OB_SUCCESS;
if (OB_SUCCESS != ret_code_) {
ret = ret_code_;
STORAGE_LOG(WARN, "async process block failed", K(ret));
} else if (NULL == micro_blocks_ || NULL == handles_ || block_count_ <= 0) {
ret = OB_INNER_STAT_ERROR;
STORAGE_LOG(WARN, "inner stat error", K(ret), KP(micro_blocks_), KP(handles_), K_(block_count));
} else if (index >= block_count_ || index < 0) {
ret = OB_INVALID_ARGUMENT;
STORAGE_LOG(ERROR, "invalid index", K(ret), K(index), K_(block_count));
} else if (NULL == micro_blocks_[index]) {
ret = OB_INNER_STAT_ERROR;
STORAGE_LOG(WARN, "micro_block is null", K(ret),
"handle validity", handles_[index].is_valid(), K(index));
} else {
block_data = micro_blocks_[index]->get_block_data();
}
return ret;
}
void ObMultiBlockIOParam::reset()
{
micro_index_infos_ = nullptr;
start_index_ = -1;
block_count_ = -1;
}
bool ObMultiBlockIOParam::is_valid() const
{
bool is_same_block = false;
const bool basic_valid = nullptr != micro_index_infos_
&& start_index_ >= 0
&& block_count_ > 0
&& micro_index_infos_->count() >= start_index_ + block_count_;
if (basic_valid) {
const ObMicroIndexInfo &first_micro = micro_index_infos_->at(start_index_);
const ObMicroIndexInfo &last_micro = micro_index_infos_->at(start_index_ + block_count_ - 1);
is_same_block = first_micro.get_macro_id() == last_micro.get_macro_id();
}
return basic_valid && is_same_block;
}
void ObMultiBlockIOParam::get_io_range(int64_t &offset, int64_t &size) const
{
offset = 0;
size = 0;
if (block_count_ > 0) {
const int64_t end_index = start_index_ + block_count_ - 1;
offset = micro_index_infos_->at(start_index_).get_block_offset();
ObMicroIndexInfo &end_micro_index = micro_index_infos_->at(end_index);
size = end_micro_index.get_block_offset() - offset + end_micro_index.get_block_size();
}
}
int ObMultiBlockIOParam::get_block_des_info(ObIMicroBlockIOCallback &io_callback) const
{
int ret = OB_SUCCESS;
if (OB_UNLIKELY(!is_valid())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("Invalid multi block io parameter", K(ret), K(*this));
} else {
ObMicroIndexInfo &start_info = micro_index_infos_->at(start_index_);
io_callback.set_micro_des_meta(start_info.row_header_);
}
return ret;
}
void ObMultiBlockIOCtx::reset()
{
micro_index_infos_ = nullptr;
block_count_ = 0;
}
bool ObMultiBlockIOCtx::is_valid() const
{
return OB_NOT_NULL(micro_index_infos_) && block_count_ > 0;
}
/*---------------------------------------ObIMicroBlockIOCallback-------------------------------------*/
ObIMicroBlockIOCallback::ObIMicroBlockIOCallback()
: cache_(nullptr),
put_size_stat_(nullptr),
allocator_(nullptr),
data_buffer_(nullptr),
tenant_id_(OB_INVALID_TENANT_ID),
block_id_(),
offset_(0),
block_des_meta_(),
use_block_cache_(true)
{
MEMSET(encrypt_key_, 0, sizeof(encrypt_key_));
block_des_meta_.encrypt_key_ = encrypt_key_;
static_assert(sizeof(*this) <= CALLBACK_BUF_SIZE, "IOCallback buf size not enough");
}
ObIMicroBlockIOCallback::~ObIMicroBlockIOCallback()
{
if (OB_NOT_NULL(allocator_) && OB_NOT_NULL(data_buffer_)) {
allocator_->free(data_buffer_);
data_buffer_ = nullptr;
}
allocator_ = nullptr;
}
void ObIMicroBlockIOCallback::set_micro_des_meta(const ObIndexBlockRowHeader *idx_row_header)
{
OB_ASSERT(nullptr != idx_row_header);
block_des_meta_.compressor_type_ = idx_row_header->get_compressor_type();
block_des_meta_.encrypt_id_ = idx_row_header->get_encrypt_id();
block_des_meta_.master_key_id_ = idx_row_header->get_master_key_id();
block_des_meta_.row_store_type_ = idx_row_header->get_row_store_type();
MEMCPY(encrypt_key_, idx_row_header->get_encrypt_key(), share::OB_MAX_TABLESPACE_ENCRYPT_KEY_LENGTH);
}
int ObIMicroBlockIOCallback::alloc_data_buf(const char *io_data_buffer, const int64_t data_size)
{
//UNUSED NOW
int ret = OB_SUCCESS;
if (OB_ISNULL(allocator_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("Unexpected error, the allocator is NULL, ", KP_(allocator), K(ret));
} else if (OB_UNLIKELY(data_size <= 0)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid data buffer size", K(ret), K(data_size));
} else {
data_buffer_ = static_cast<char *>(allocator_->alloc(data_size));
for (int64_t i = 1; OB_ISNULL(data_buffer_) && i <= ALLOC_BUF_RETRY_TIMES; i++) {
ob_usleep(ALLOC_BUF_RETRY_INTERVAL * i);
data_buffer_ = static_cast<char *>(allocator_->alloc(data_size));
}
if (OB_ISNULL(data_buffer_)) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("Fail to allocate memory", K(ret), K_(offset), K(data_size), KP(data_buffer_));
} else {
MEMCPY(data_buffer_, io_data_buffer, data_size);
}
}
return ret;
}
int ObIMicroBlockIOCallback::process_block(
ObMacroBlockReader *reader,
const char *buffer,
const int64_t offset,
const int64_t size,
const ObMicroBlockCacheValue *&micro_block,
common::ObKVCacheHandle &cache_handle)
{
int ret = OB_SUCCESS;
ObMicroBlockData block_data;
ObMicroBlockHeader header;
int64_t pos = 0;
int64_t payload_size = 0;
const char *payload_buf = nullptr;
if (OB_UNLIKELY(NULL == reader || NULL == buffer || offset < 0 || size < 0)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid arguments", K(ret), KP(reader), KP(buffer), K(offset), K(size));
} else if (OB_FAIL(header.deserialize(buffer, size, pos))) {
LOG_ERROR("Fail to deserialize record header", K(ret), K_(block_id), K(offset));
} else if (OB_FAIL(header.check_and_get_record(
buffer, size, MICRO_BLOCK_HEADER_MAGIC, payload_buf, payload_size))) {
LOG_ERROR("Micro block data is corrupted", K(ret), K_(block_id), K(offset),
K(size), K_(tenant_id), KP(buffer), KP(this));
} else {
if (OB_UNLIKELY(!use_block_cache_)) {
// Won't put in cache
if (OB_FAIL(read_block_and_copy(header, *reader, buffer, size, block_data, micro_block, cache_handle))) {
LOG_WARN("Fail to read micro block and copy to cache value", K(ret));
}
} else {
ObIMicroBlockCache::BaseBlockCache *kvcache = nullptr;
ObMicroBlockCacheKey key(tenant_id_, block_id_, offset, size);
if (OB_FAIL(cache_->get_cache(kvcache))) {
LOG_WARN("Fail to get kvcache", K(ret));
} else if (OB_UNLIKELY(OB_SUCCESS == (ret = kvcache->get(key, micro_block, cache_handle)))) {
// entry exist, no need to put
} else if (OB_FAIL(cache_->put_cache_block(
block_des_meta_, buffer, key, *reader, *allocator_, micro_block, cache_handle))) {
LOG_WARN("Failed to put block to cache", K(ret));
}
}
// NOTE: if block has column level compress and don't use kvcache,
// the block not be full transformed here and left to be part transformed in
// cs micro block decoder.
}
return ret;
}
int ObIMicroBlockIOCallback::read_block_and_copy(
const ObMicroBlockHeader &header,
ObMacroBlockReader &reader,
const char *buffer,
const int64_t size,
ObMicroBlockData &block_data,
const ObMicroBlockCacheValue *&micro_block,
ObKVCacheHandle &handle)
{
int ret = OB_SUCCESS;
block_data.type_ = cache_->get_type();
// only full_transform for index_block with cs_encoding when don't use block cache
if (ObMicroBlockData::INDEX_BLOCK == block_data.type_ &&
ObStoreFormat::is_row_store_type_with_cs_encoding(static_cast<ObRowStoreType>(header.row_store_type_))) {
if (OB_FAIL(reader.decrypt_and_full_transform_data(header, block_des_meta_,
buffer, size, block_data.get_buf(), block_data.get_buf_size(), nullptr))) {
LOG_WARN("fail to decrypt_and_full_transform_data", K(ret), K(header), K_(block_des_meta));
}
} else {
bool is_compressed = false;
if (OB_FAIL(reader.do_decrypt_and_decompress_data(
header, block_des_meta_, buffer, size,
block_data.get_buf(), block_data.get_buf_size(),
is_compressed, false/*need deep copy*/, nullptr))) {
LOG_WARN("fail to do decrypt and decompress data", K(ret));
}
}
if (OB_SUCC(ret)) {
ObMicroBlockCacheValue value(
block_data.get_buf(), block_data.get_buf_size(), nullptr, 0, block_data.type_);
value.set_alloc_by_block_io();
char *buf = nullptr;
const int64_t buf_len = value.size();
handle.reset();
micro_block = nullptr;
ObIKVCacheValue *value_copy = nullptr;
if (OB_ISNULL(buf = static_cast<char *>(allocator_->alloc(buf_len)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("Failed to allocate value", K(ret), K(buf_len));
} else if (OB_FAIL(value.deep_copy(buf, buf_len, value_copy))) {
LOG_WARN("Failed to deep copy value", K(ret));
allocator_->free(buf);
} else {
micro_block = static_cast<const ObMicroBlockCacheValue *>(value_copy);
}
}
return ret;
}
/*-----------------------------------ObAsyncSingleMicroBlockIOCallback-----------------------------------*/
ObAsyncSingleMicroBlockIOCallback::ObAsyncSingleMicroBlockIOCallback()
: ObIMicroBlockIOCallback(),
micro_block_(nullptr),
cache_handle_()
{
STATIC_ASSERT(sizeof(*this) <= CALLBACK_BUF_SIZE, "IOCallback buf size not enough");
}
ObAsyncSingleMicroBlockIOCallback::~ObAsyncSingleMicroBlockIOCallback()
{
// release micro_block_ outside
if (OB_NOT_NULL(allocator_) && OB_NOT_NULL(micro_block_) && !cache_handle_.is_valid()) {
allocator_->free(const_cast<ObMicroBlockCacheValue *>(micro_block_));
micro_block_ = nullptr;
}
}
int64_t ObAsyncSingleMicroBlockIOCallback::size() const
{
return sizeof(*this);
}
int ObAsyncSingleMicroBlockIOCallback::inner_process(const char *data_buffer, const int64_t size)
{
int ret = OB_SUCCESS;
ObTimeGuard time_guard("AsyncSingle_Callback_Process", 100000); //100ms
if (OB_ISNULL(cache_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("Invalid micro block cache callback, ", KP_(cache), K(ret));
} else if (OB_UNLIKELY(size <= 0 || data_buffer == nullptr)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid data buffer size", K(ret), K(size), KP(data_buffer));
} else {
ObMacroBlockReader *reader = nullptr;
if (OB_ISNULL(reader = GET_TSI_MULT(ObMacroBlockReader, 1))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("Fail to allocate ObMacroBlockReader, ", K(ret));
} else if (OB_FAIL(process_block(reader, data_buffer, offset_, size, micro_block_, cache_handle_))) {
LOG_WARN("process_block failed", K(ret));
}
}
return ret;
}
const char *ObAsyncSingleMicroBlockIOCallback::get_data()
{
return reinterpret_cast<const char *>(micro_block_);
}
/*-----------------------------------ObMultiDataBlockIOCallback-----------------------------------*/
ObMultiDataBlockIOCallback::ObMultiDataBlockIOCallback()
: ObIMicroBlockIOCallback(),
io_ctx_(),
io_result_()
{
STATIC_ASSERT(sizeof(*this) <= CALLBACK_BUF_SIZE, "IOCallback buf size not enough");
}
ObMultiDataBlockIOCallback::~ObMultiDataBlockIOCallback()
{
free_result();
}
int64_t ObMultiDataBlockIOCallback::size() const
{
return sizeof(*this);
}
int ObMultiDataBlockIOCallback::inner_process(const char *data_buffer, const int64_t size)
{
int ret = OB_SUCCESS;
ObTimeGuard time_guard("MultiData_Callback_Process", 100000); //100ms
if (OB_ISNULL(cache_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("Invalid micro block cache callback, ", KP_(cache), K(ret));
} else if (OB_UNLIKELY(size <= 0 || data_buffer == nullptr)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid data buffer size", K(ret), K(size), KP(data_buffer));
} else {
ObMacroBlockReader *reader = nullptr;
if (OB_ISNULL(reader = GET_TSI_MULT(ObMacroBlockReader, 1))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("Fail to allocate ObMacroBlockReader, ", K(ret));
} else if (OB_FAIL(alloc_result())) {
LOG_WARN("alloc_result failed", K(ret));
}
const int64_t block_count = io_ctx_.block_count_;
for (int64_t i = 0; OB_SUCC(ret) && i < block_count; ++i) {
const int64_t data_size = io_ctx_.micro_index_infos_[i].get_block_size();
const int64_t data_offset = io_ctx_.micro_index_infos_[i].get_block_offset() - offset_;
if (OB_FAIL(process_block(
reader,
data_buffer + data_offset,
offset_ + data_offset,
data_size,
io_result_.micro_blocks_[i],
io_result_.handles_[i]))) {
LOG_WARN("process_block failed", K(ret));
}
}
}
if (OB_FAIL(ret)) {
io_result_.ret_code_ = ret;
}
return ret;
}
const char *ObMultiDataBlockIOCallback::get_data()
{
const char *data = nullptr;
data = reinterpret_cast<const char*>(&io_result_);
return data;
}
int ObMultiDataBlockIOCallback::set_io_ctx(
const ObMultiBlockIOParam &io_param)
{
int ret = OB_SUCCESS;
if (OB_UNLIKELY(!io_param.is_valid())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid io_param", K(ret), K(io_param));
} else {
io_ctx_.micro_index_infos_ = &io_param.micro_index_infos_->at(io_param.start_index_);
io_ctx_.block_count_ = io_param.block_count_;
}
return ret;
}
int ObMultiDataBlockIOCallback::deep_copy_ctx(
const ObMultiBlockIOCtx &io_ctx)
{
int ret = OB_SUCCESS;
if (OB_UNLIKELY(!io_ctx.is_valid())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid io_ctx", K(ret), K(io_ctx));
} else if (OB_ISNULL(allocator_)) {
ret = OB_INNER_STAT_ERROR;
LOG_WARN("allocator_ is null", K(ret), KP(allocator_));
} else {
void *ptr = nullptr;
int64_t alloc_size = sizeof(ObMicroIndexInfo) * io_ctx.block_count_;
if (OB_ISNULL(ptr = allocator_->alloc(alloc_size))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("alloc memory failed", K(ret), K(alloc_size));
} else {
io_ctx_.micro_index_infos_ = reinterpret_cast<ObMicroIndexInfo *>(ptr);
MEMCPY(io_ctx_.micro_index_infos_, io_ctx.micro_index_infos_, alloc_size);
}
if (OB_SUCC(ret)) {
io_ctx_.block_count_ = io_ctx.block_count_;
}
}
return ret;
}
int ObMultiDataBlockIOCallback::alloc_result()
{
int ret = OB_SUCCESS;
void *ptr = nullptr;
const int64_t block_count = io_ctx_.block_count_;
if (OB_ISNULL(ptr = allocator_->alloc(sizeof(ObMicroBlockCacheValue *) * block_count))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("alloc failed", K(ret));
} else {
io_result_.micro_blocks_ = reinterpret_cast<const ObMicroBlockCacheValue **>(ptr);
MEMSET(io_result_.micro_blocks_, 0, sizeof(ObMicroBlockCacheValue *) * block_count);
}
if (OB_SUCC(ret)) {
if (OB_ISNULL(ptr = allocator_->alloc(sizeof(ObKVCacheHandle) * block_count))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("alloc failed", K(ret));
} else {
io_result_.handles_ = new (ptr) ObKVCacheHandle[block_count];
io_result_.block_count_ = block_count;
}
}
return ret;
}
void ObMultiDataBlockIOCallback::free_result()
{
if (OB_NOT_NULL(allocator_)) {
if (OB_NOT_NULL(io_result_.micro_blocks_)) {
if (OB_NOT_NULL(io_result_.handles_)) {
for (int64_t i = 0; i < io_result_.block_count_; ++i) {
if (!io_result_.handles_[i].is_valid()
&& OB_NOT_NULL(io_result_.micro_blocks_[i])) {
allocator_->free(const_cast<ObMicroBlockCacheValue *>(io_result_.micro_blocks_[i]));
}
}
}
allocator_->free(io_result_.micro_blocks_);
io_result_.micro_blocks_ = nullptr;
}
if (OB_NOT_NULL(io_result_.handles_)) {
for (int64_t i = 0; i < io_result_.block_count_; ++i) {
io_result_.handles_[i].~ObKVCacheHandle();
}
allocator_->free(io_result_.handles_);
io_result_.handles_ = nullptr;
io_result_.block_count_ = 0;
}
}
}
/* ---------------------------------------- ObSyncSingleMicroBLockIOCallback ---------------------------------------- */
ObSyncSingleMicroBLockIOCallback::ObSyncSingleMicroBLockIOCallback()
: ObIMicroBlockIOCallback(),
macro_reader_(nullptr),
block_data_(nullptr),
is_data_block_(true)
{
}
ObSyncSingleMicroBLockIOCallback::~ObSyncSingleMicroBLockIOCallback()
{
}
int64_t ObSyncSingleMicroBLockIOCallback::size() const
{
return sizeof(*this);
}
int ObSyncSingleMicroBLockIOCallback::inner_process(const char *data_buffer, const int64_t size)
{
int ret = OB_SUCCESS;
bool is_compressed = false;
ObTimeGuard time_guard("SyncSingle_Callback_Process", 100000); //100ms
if (OB_UNLIKELY(size <= 0 || data_buffer == nullptr)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid data buffer size", K(ret), K(size), KP(data_buffer));
} else {
if (OB_UNLIKELY(nullptr == macro_reader_ || nullptr == block_data_ || nullptr == allocator_)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("Unexpected reader or block data", K(ret), KP(macro_reader_), KP(block_data_), KP_(allocator));
} else {
const char *src_block_buf = data_buffer;
const int64_t src_buf_size = size;
ObMicroBlockHeader header;
if (OB_FAIL(header.deserialize_and_check_header(src_block_buf, src_buf_size))) {
LOG_WARN("fail to deserialize_and_check_header", K(ret), KP(src_block_buf), K(src_buf_size));
} else if (ObStoreFormat::is_row_store_type_with_cs_encoding(static_cast<ObRowStoreType>(header.row_store_type_))) {
if (OB_FAIL(macro_reader_->decrypt_and_full_transform_data(
header, block_des_meta_, src_block_buf, src_buf_size,
block_data_->get_buf(), block_data_->get_buf_size(), allocator_))) {
LOG_WARN("fail to decrypt_and_full_transform_data", K(ret), K(header), K(block_des_meta_), K(is_data_block_));
}
} else { // not cs_encoding
if (OB_FAIL(macro_reader_->do_decrypt_and_decompress_data(
header, block_des_meta_, src_block_buf, src_buf_size, block_data_->get_buf(),
block_data_->get_buf_size(), is_compressed, true /* need_deep_copy */, allocator_))) {
LOG_WARN("Fail to decrypt and decompress micro block data buf", K(ret));
}
}
}
}
return ret;
}
const char *ObSyncSingleMicroBLockIOCallback::get_data()
{
return reinterpret_cast<char *>(block_data_);
}
/*----------------------------------------ObIMicroBlockCache--------------------------------------*/
int ObIMicroBlockCache::get_cache_block(
const uint64_t tenant_id,
const MacroBlockId block_id,
const int64_t offset,
const int64_t size,
ObMicroBlockBufferHandle &handle)
{
int ret = OB_SUCCESS;
BaseBlockCache *cache = NULL;
if (OB_UNLIKELY(0 == tenant_id || OB_INVALID_TENANT_ID == tenant_id || !block_id.is_valid()
|| offset < 0 || size <= 0)) {
ret = OB_INVALID_ARGUMENT;
STORAGE_LOG(WARN, "invalid arguments", K(ret),
KP(cache), K(tenant_id), K(block_id), K(offset), K(size));
} else if (OB_FAIL(get_cache(cache))) {
STORAGE_LOG(WARN, "get_cache failed", K(ret));
} else {
ObMicroBlockCacheKey key(tenant_id, block_id, offset, size);
if (OB_FAIL(cache->get(key, handle.micro_block_, handle.handle_))) {
if (OB_ENTRY_NOT_EXIST != ret) {
STORAGE_LOG(WARN, "Fail to get micro block from block cache, ", K(ret));
}
} else {
EVENT_INC(ObStatEventIds::BLOCK_CACHE_HIT);
}
}
return ret;
}
int ObIMicroBlockCache::prefetch(
const uint64_t tenant_id,
const MacroBlockId &macro_id,
const ObMicroIndexInfo& idx_row,
const bool use_cache,
ObMacroBlockHandle &macro_handle,
ObIAllocator *allocator)
{
int ret = OB_SUCCESS;
const ObIndexBlockRowHeader *idx_header = idx_row.row_header_;
if (OB_ISNULL(idx_header)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("Invalid null index block row header", K(ret), K(idx_row));
} else if (OB_UNLIKELY(!idx_header->is_valid() || 0 >= idx_header->get_block_size() || nullptr == allocator)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("Invalid data index block row header ", K(ret), K(idx_row), KP(allocator));
} else {
void *buf = nullptr;
ObAsyncSingleMicroBlockIOCallback *callback = nullptr;
if (OB_ISNULL(buf = allocator->alloc(sizeof(ObAsyncSingleMicroBlockIOCallback)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("allocate callback memory failed", K(ret));
} else {
callback = new (buf) ObAsyncSingleMicroBlockIOCallback;
callback->allocator_ = allocator;
callback->use_block_cache_ = use_cache;
if (OB_FAIL(prefetch(tenant_id, macro_id, idx_row, macro_handle, *callback))) {
LOG_WARN("Fail to prefetch data micro block", K(ret));
}
if (OB_FAIL(ret) && OB_NOT_NULL(callback->get_allocator())) { //Avoid double_free with io_handle
callback->~ObAsyncSingleMicroBlockIOCallback();
allocator->free(callback);
}
}
}
return ret;
}
int ObIMicroBlockCache::prefetch(
const uint64_t tenant_id,
const MacroBlockId &macro_id,
const ObMicroIndexInfo& idx_row,
ObMacroBlockHandle &macro_handle,
ObIMicroBlockIOCallback &callback)
{
int ret = OB_SUCCESS;
const ObIndexBlockRowHeader *idx_row_header = idx_row.row_header_;
if (OB_ISNULL(idx_row_header)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret));
} else {
// fill callback
callback.cache_ = this;
callback.put_size_stat_ = this;
callback.tenant_id_ = tenant_id;
callback.block_id_ = macro_id;
callback.offset_ = idx_row.get_block_offset();
callback.set_micro_des_meta(idx_row_header);
// fill read info
ObMacroBlockReadInfo read_info;
read_info.macro_block_id_ = macro_id;
read_info.io_desc_.set_wait_event(ObWaitEventIds::DB_FILE_DATA_READ);
read_info.io_desc_.set_group_id(ObIOModule::MICRO_BLOCK_CACHE_IO);
read_info.io_callback_ = &callback;
read_info.offset_ = idx_row.get_block_offset();
read_info.size_ = idx_row.get_block_size();
read_info.io_timeout_ms_ = max(THIS_WORKER.get_timeout_remain() / 1000, 0);
if (OB_FAIL(ObBlockManager::async_read_block(read_info, macro_handle))) {
STORAGE_LOG(WARN, "Fail to async read block, ", K(ret), K(read_info));
} else {
EVENT_INC(ObStatEventIds::IO_READ_PREFETCH_MICRO_COUNT);
EVENT_ADD(ObStatEventIds::IO_READ_PREFETCH_MICRO_BYTES, idx_row.get_block_size());
}
}
return ret;
}
int ObIMicroBlockCache::prefetch(
const uint64_t tenant_id,
const MacroBlockId &macro_id,
const ObMultiBlockIOParam &io_param,
const bool use_cache,
ObMacroBlockHandle &macro_handle,
ObIMicroBlockIOCallback &callback)
{
int ret = OB_SUCCESS;
int64_t offset = 0;
int64_t size = 0;
if (OB_FAIL(io_param.get_block_des_info(callback))) {
LOG_WARN("Fail to get meta data for deserializing block data", K(ret), K(io_param));
} else {
// fill callback
io_param.get_io_range(offset, size);
callback.cache_ = this;
callback.put_size_stat_ = this;
callback.tenant_id_ = tenant_id;
callback.block_id_ = macro_id;
callback.offset_ = offset;
callback.use_block_cache_ = use_cache;
// fill read info
ObMacroBlockReadInfo read_info;
read_info.macro_block_id_ = macro_id;
read_info.io_desc_.set_wait_event(ObWaitEventIds::DB_FILE_DATA_READ);
read_info.io_desc_.set_group_id(ObIOModule::MICRO_BLOCK_CACHE_IO);
read_info.io_callback_ = &callback;
read_info.offset_ = offset;
read_info.size_ = size;
read_info.io_timeout_ms_ = max(THIS_WORKER.get_timeout_remain() / 1000, 0);
if (OB_FAIL(ObBlockManager::async_read_block(read_info, macro_handle))) {
STORAGE_LOG(WARN, "Fail to async read block, ", K(ret), K(read_info));
} else {
EVENT_ADD(ObStatEventIds::IO_READ_PREFETCH_MICRO_COUNT, io_param.block_count_);
EVENT_ADD(ObStatEventIds::IO_READ_PREFETCH_MICRO_BYTES, size);
}
}
return ret;
}
int ObIMicroBlockCache::add_put_size(const int64_t put_size)
{
UNUSED(put_size);
return OB_SUCCESS;
}
/*-----------------------------------ObMicroBlockBufTranformer-----------------------------------*/
ObMicroBlockBufTransformer::ObMicroBlockBufTransformer(
const ObMicroBlockDesMeta &block_des_meta,
ObMacroBlockReader *reader,
ObMicroBlockHeader &header,
const char *buf,
const int64_t buf_size)
: is_inited_(false),
is_cs_full_transfrom_(ObStoreFormat::is_row_store_type_with_cs_encoding(
static_cast<ObRowStoreType>(header.row_store_type_))),
block_des_meta_(block_des_meta),
reader_(reader), header_(header),
payload_buf_(buf + header.header_size_),
payload_size_(buf_size - header.header_size_)
{
}
int ObMicroBlockBufTransformer::init()
{
int ret = OB_SUCCESS;
if (is_cs_full_transfrom_) {
bool is_compressed = false;
#ifdef OB_BUILD_TDE_SECURITY
if (share::ObEncryptionUtil::need_encrypt(static_cast<ObCipherOpMode>(block_des_meta_.encrypt_id_))) {
const char *decrypt_buf = NULL;
int64_t decrypt_len = 0;
if (OB_FAIL(reader_->decrypt_buf(
block_des_meta_, payload_buf_, payload_size_, decrypt_buf, decrypt_len))) {
LOG_WARN("Fail to decrypt data", K(ret));
} else {
payload_buf_ = decrypt_buf;
payload_size_ = decrypt_len;
is_compressed = (header_.data_length_ != decrypt_len);
}
}
#else
is_compressed = header_.is_compressed_data();
#endif
if (OB_SUCC(ret)) {
if (OB_UNLIKELY(is_compressed)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("cs encoding must has no block-level compression", K(ret), K(header_));
} else if (OB_FAIL(transformer_.init(&header_, payload_buf_, payload_size_))) {
LOG_WARN("fail to init cs micro block transformer", K(ret), K_(header));
}
}
}
if (OB_SUCC(ret)) {
is_inited_ = true;
}
return ret;
}
int ObMicroBlockBufTransformer::get_buf_size(int64_t &buf_size) const
{
int ret = OB_SUCCESS;
if (IS_NOT_INIT) {
ret = OB_NOT_INIT;
LOG_WARN("not inited", K(ret));
} else if (!is_cs_full_transfrom_) {
buf_size = header_.header_size_ + header_.data_length_;
} else if (OB_FAIL(transformer_.calc_full_transform_size(buf_size))) {
LOG_WARN("fail to calc transformed size", K(ret), K_(header));
}
return ret;
}
int ObMicroBlockBufTransformer::transfrom(char *block_buf, const int64_t buf_size)
{
int ret = OB_SUCCESS;
if (IS_NOT_INIT) {
ret = OB_NOT_INIT;
LOG_WARN("not inited", K(ret));
} else if (OB_UNLIKELY(nullptr == block_buf || buf_size <= 0)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), KP(block_buf), K(buf_size));
} else if (!is_cs_full_transfrom_) {
ObMicroBlockHeader *micro_header = nullptr;
int64_t pos = 0;
if (OB_FAIL(header_.deep_copy(block_buf, buf_size, pos, micro_header))) {
LOG_WARN("Fail to deep copy header", K(ret), K_(header));
} else {
#ifdef OB_BUILD_TDE_SECURITY
const char *decrypt_buf = NULL;
int64_t decrypt_len = 0;
if (OB_FAIL(reader_->decrypt_buf(
block_des_meta_, payload_buf_, payload_size_, decrypt_buf, decrypt_len))) {
LOG_WARN("fail to decrypt data", K(ret), K_(header));
} else {
payload_buf_ = decrypt_buf;
payload_size_ = decrypt_len;
}
#endif
if (OB_SUCC(ret)) {
if (OB_FAIL(reader_->decompress_data_with_prealloc_buf(
block_des_meta_.compressor_type_, payload_buf_, payload_size_,
block_buf + pos, buf_size - pos))) {
LOG_WARN("Fail to decompress data with preallocated buffer", K(ret), K_(header));
}
}
}
} else { // is_cs_full_transfrom_
int64_t pos = 0;
if (OB_FAIL(transformer_.full_transform(block_buf, buf_size, pos))) {
LOG_WARN("fail to transfrom cs encoding mirco blcok", K(ret));
} else if (OB_UNLIKELY(pos != buf_size)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("pos should equal to buf_size", K(ret), K(pos), K(buf_size));
}
}
return ret;
}
/*-------------------------------------ObDataMicroBlockCache--------------------------------------*/
int ObDataMicroBlockCache::init(const char *cache_name, const int64_t priority)
{
int ret = OB_SUCCESS;
const int64_t mem_limit = 4 * 1024 * 1024 * 1024LL;
if (OB_SUCCESS != (ret = common::ObKVCache<ObMicroBlockCacheKey, ObMicroBlockCacheValue>::init(
cache_name, priority))) {
STORAGE_LOG(WARN, "Fail to init kv cache, ", K(ret));
} else if (OB_FAIL(allocator_.init(mem_limit, OB_MALLOC_MIDDLE_BLOCK_SIZE, OB_MALLOC_MIDDLE_BLOCK_SIZE))) {
STORAGE_LOG(WARN, "Fail to init io allocator, ", K(ret));
} else {
allocator_.set_attr(SET_USE_500(ObMemAttr(OB_SERVER_TENANT_ID, ObModIds::OB_SSTABLE_MICRO_BLOCK_ALLOCATOR)));
}
return ret;
}
void ObDataMicroBlockCache::destroy()
{
common::ObKVCache<ObMicroBlockCacheKey, ObMicroBlockCacheValue>::destroy();
allocator_.destroy();
}
int ObDataMicroBlockCache::prefetch(
const uint64_t tenant_id,
const MacroBlockId &macro_id,
const ObMultiBlockIOParam &io_param,
const bool use_cache,
ObMacroBlockHandle &macro_handle)
{
int ret = OB_SUCCESS;
ObMultiDataBlockIOCallback *callback = nullptr;
ObIAllocator *allocator = nullptr;
if (OB_FAIL(get_allocator(allocator))) {
LOG_WARN("Fail to get allocator", K(ret));
} else {
void *buf = nullptr;
if (OB_ISNULL(buf = allocator->alloc(sizeof(ObMultiDataBlockIOCallback)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("allocate callback memory failed", K(ret));
} else {
callback = new (buf) ObMultiDataBlockIOCallback;
callback->allocator_ = allocator;
if (OB_UNLIKELY(!io_param.is_valid() || 0 == tenant_id || OB_INVALID_TENANT_ID == tenant_id)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("Invalid input parameters", K(ret), K(tenant_id));
} else if (OB_FAIL(callback->set_io_ctx(io_param))) {
LOG_WARN("Set io context failed", K(ret), K(io_param));
} else if (OB_FAIL(ObIMicroBlockCache::prefetch(
tenant_id, macro_id, io_param, use_cache, macro_handle, *callback))) {
LOG_WARN("Fail to prefetch multi data blocks", K(ret));
}
if (OB_FAIL(ret) && OB_NOT_NULL(callback->get_allocator())) { //Avoid double_free with io_handle
callback->~ObMultiDataBlockIOCallback();
allocator->free(callback);
}
}
}
return ret;
}
int ObDataMicroBlockCache::load_block(
const ObMicroBlockId &micro_block_id,
const ObMicroBlockDesMeta &des_meta,
ObMacroBlockReader *macro_reader,
ObMicroBlockData &block_data,
ObIAllocator *allocator)
{
int ret = OB_SUCCESS;
ObMacroBlockReadInfo macro_read_info;
ObMacroBlockHandle macro_handle;
bool is_compressed = false;
if (OB_UNLIKELY(!micro_block_id.is_valid() || nullptr == macro_reader || nullptr == allocator)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("Invalid argument", K(ret), K(micro_block_id), KP(macro_reader), KP(allocator));
} else {
void *buf = nullptr;
ObSyncSingleMicroBLockIOCallback *callback = nullptr;
if (OB_ISNULL(buf = allocator->alloc(sizeof(ObSyncSingleMicroBLockIOCallback)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("allocate callback memory failed", K(ret));
} else {
callback = new (buf) ObSyncSingleMicroBLockIOCallback;
callback->allocator_ = allocator;
callback->offset_ = micro_block_id.offset_;
callback->block_des_meta_ = des_meta;
callback->block_data_ = &block_data;
callback->macro_reader_ = macro_reader;
callback->is_data_block_ = true;
macro_read_info.macro_block_id_ = micro_block_id.macro_id_;
macro_read_info.io_desc_.set_wait_event(ObWaitEventIds::DB_FILE_DATA_READ);
macro_read_info.io_desc_.set_group_id(ObIOModule::MICRO_BLOCK_CACHE_IO);
macro_read_info.io_callback_ = callback;
macro_read_info.offset_ = micro_block_id.offset_;
macro_read_info.size_ = micro_block_id.size_;
macro_read_info.io_timeout_ms_ = GCONF._data_storage_io_timeout / 1000L;
if (OB_FAIL(ObBlockManager::read_block(macro_read_info, macro_handle))) {
LOG_WARN("Fail to sync read block", K(ret), K(macro_read_info));
} else {
block_data.type_ = ObMicroBlockData::DATA_BLOCK;
EVENT_INC(ObStatEventIds::IO_READ_PREFETCH_MICRO_COUNT);
EVENT_ADD(ObStatEventIds::IO_READ_PREFETCH_MICRO_BYTES, micro_block_id.size_);
}
if (OB_FAIL(ret) && OB_NOT_NULL(callback->get_allocator())) { //Avoid double_free with io_handle
callback->~ObSyncSingleMicroBLockIOCallback();
allocator->free(callback);
}
}
}
return ret;
}
int ObDataMicroBlockCache::get_cache(BaseBlockCache *&cache)
{
int ret = OB_SUCCESS;
cache = this;
return ret;
}
int ObDataMicroBlockCache::get_allocator(common::ObIAllocator *&allocator)
{
int ret = OB_SUCCESS;
allocator = &allocator_;
return ret;
}
int64_t ObDataMicroBlockCache::calc_value_size(const int64_t data_length,
const ObRowStoreType &type,
bool &need_decoder)
{
need_decoder = false;
int64_t value_size = sizeof(ObMicroBlockCacheValue) + data_length;
if (ObStoreFormat::is_row_store_type_with_encoding(type)) {
need_decoder = true;
value_size += ObMicroBlockDecoder::MAX_CACHED_DECODER_BUF_SIZE;
}
return value_size;
}
int ObDataMicroBlockCache::write_extra_buf(const ObRowStoreType row_store_type,
const char *block_buf,
const int64_t block_size,
char *extra_buf,
ObMicroBlockData &micro_data)
{
int ret = OB_SUCCESS;
int64_t decoder_size = 0;
if (ObStoreFormat::is_row_store_type_with_cs_encoding(row_store_type)) {
if (OB_FAIL(ObMicroBlockCSDecoder::get_decoder_cache_size(block_buf, block_size, decoder_size))) {
LOG_WARN("Fail to get decoder cache size", K(ret));
} else if (OB_FAIL(ObMicroBlockCSDecoder::cache_decoders(extra_buf, decoder_size, block_buf, block_size))) {
LOG_WARN("Fail to set cache decoder", K(ret));
}
} else if (OB_FAIL(ObMicroBlockDecoder::get_decoder_cache_size(block_buf, block_size, decoder_size))) {
LOG_WARN("Fail to get decoder cache size", K(ret));
} else if (OB_FAIL(ObMicroBlockDecoder::cache_decoders(extra_buf, decoder_size, block_buf, block_size))) {
LOG_WARN("Fail to set cache decoder", K(ret));
}
if (OB_FAIL(ret)) {
} else {
micro_data.get_extra_buf() = extra_buf;
micro_data.get_extra_size() = decoder_size;
}
return ret;
}
int ObDataMicroBlockCache::put_cache_block(
const ObMicroBlockDesMeta &des_meta,
const char *raw_block_buf,
const ObMicroBlockCacheKey &key,
ObMacroBlockReader &reader,
ObIAllocator &allocator,
const ObMicroBlockCacheValue *&micro_block,
common::ObKVCacheHandle &cache_handle)
{
UNUSED(allocator);
int ret = OB_SUCCESS;
ObMicroBlockHeader header;
int64_t pos = 0;
int64_t payload_size = 0;
const char *payload_buf = nullptr;
const ObMicroBlockId &micro_id = key.get_micro_block_id();
ObIMicroBlockCache::BaseBlockCache *kvcache = nullptr;
if (OB_UNLIKELY(!des_meta.is_valid() || !micro_id.is_valid()) || OB_ISNULL(raw_block_buf)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("Invalid argument", K(ret), K(key), K(des_meta));
} else if (OB_FAIL(header.deserialize(raw_block_buf, micro_id.size_, pos))) {
LOG_ERROR("Fail to deserialize record header", K(ret), K(micro_id));
} else if (OB_FAIL(header.check_and_get_record(
raw_block_buf, micro_id.size_, MICRO_BLOCK_HEADER_MAGIC, payload_buf, payload_size))) {
LOG_ERROR("Micro block data is corrupted", K(ret), K(key), KP(raw_block_buf), KP(this));
} else if (OB_FAIL(get_cache(kvcache))) {
LOG_WARN("Fail to get kvcache", K(ret));
} else {
ObKVCacheInstHandle inst_handle;
ObKVCachePair *kvpair = nullptr;
int64_t block_size = 0;
int64_t value_size = 0;
int64_t extra_size = 0;
bool need_decoder = false;
ObMicroBlockBufTransformer buf_transformer(des_meta, &reader, header, payload_buf, payload_size);
if (OB_FAIL(buf_transformer.init())) {
LOG_WARN("Fail to init buf transformer", K(ret));
} else if (OB_FAIL(buf_transformer.get_buf_size(block_size))) {
LOG_WARN("Fail to get block buf size", K(ret));
} else if (FALSE_IT(value_size = calc_value_size(block_size, des_meta.row_store_type_, need_decoder))) {
} else if (OB_FAIL(alloc(
key.get_tenant_id(), sizeof(ObMicroBlockCacheKey), value_size, kvpair, cache_handle, inst_handle))) {
LOG_WARN("Fail to allocate kvpair from kvcache", K(ret), K(value_size), K(key));
} else {
char *block_buf = reinterpret_cast<char *>(kvpair->value_) + sizeof(ObMicroBlockCacheValue);
kvpair->key_ = new (kvpair->key_) ObMicroBlockCacheKey(key);
ObMicroBlockCacheValue *cache_value = new (kvpair->value_) ObMicroBlockCacheValue(block_buf, block_size);
ObMicroBlockData &micro_data = cache_value->get_block_data();
micro_data.type_ = get_type();
if (OB_FAIL(buf_transformer.transfrom(block_buf, block_size))) {
LOG_WARN("fail to transfrom", K(ret));
} else if (need_decoder && OB_FAIL(write_extra_buf(
des_meta.row_store_type_, block_buf, block_size, block_buf + block_size, micro_data))) {
LOG_WARN("Fail to cache decoder on extra buffer for data block", K(ret), K(header), KPC(cache_value));
} else if (FALSE_IT(micro_block = cache_value)) {
} else if (OB_FAIL(put_kvpair(inst_handle, kvpair, cache_handle, false /* overwrite */))) {
if (OB_ENTRY_EXIST != ret) {
LOG_WARN("Fail to put micro block cache", K(ret));
} else {
ret = OB_SUCCESS;
}
} else {
const int64_t put_size = ObKVStoreMemBlock::get_align_size(key, *cache_value);
if (OB_FAIL(add_put_size(put_size))) {
LOG_WARN("add_put_size failed", K(ret), K(put_size));
}
}
if (OB_FAIL(ret)) {
cache_handle.reset();
micro_block = nullptr;
}
}
}
return ret;
}
int ObDataMicroBlockCache::reserve_kvpair(
const ObMicroBlockDesc &micro_block_desc,
ObKVCacheInstHandle &inst_handle,
ObKVCacheHandle &cache_handle,
ObKVCachePair *&kvpair,
int64_t &kvpair_size)
{
int ret = OB_SUCCESS;
kvpair_size = 0;
int64_t block_size = 0;
int64_t extra_size = 0;
int64_t value_size = 0;
bool need_decoder = false;
const int64_t key_size = sizeof(ObMicroBlockCacheKey);
const ObRowStoreType row_store_type = static_cast<ObRowStoreType>(micro_block_desc.header_->row_store_type_);
ObCSMicroBlockTransformer transformer;
if (OB_UNLIKELY(!micro_block_desc.is_valid() || inst_handle.is_valid()
|| cache_handle.is_valid() || nullptr != kvpair)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("Invalid argument", K(ret), K(micro_block_desc), K(inst_handle), K(cache_handle), KP(kvpair));
} else if (ObStoreFormat::is_row_store_type_with_cs_encoding(row_store_type)) {
if (OB_FAIL(transformer.init(micro_block_desc.header_, micro_block_desc.buf_, micro_block_desc.buf_size_))) {
LOG_WARN("fail to init transformer", K(ret), K(micro_block_desc));
} else if (OB_FAIL(transformer.calc_full_transform_size(block_size))) {
LOG_WARN("fail to calc_full_transform_size", K(ret), K(micro_block_desc));
}
} else {
block_size = micro_block_desc.header_->header_size_ + micro_block_desc.data_size_;
}
if (OB_SUCC(ret)) {
value_size = calc_value_size(block_size, row_store_type, need_decoder);
BaseBlockCache *kvcache = nullptr;
if (OB_FAIL(get_cache(kvcache))) {
LOG_WARN("Fail to get cache", K(ret));
} else if (OB_FAIL(kvcache->alloc(MTL_ID(), key_size, value_size, kvpair, cache_handle, inst_handle))) {
LOG_WARN("Fail to alloc cache buf", K(ret), K(key_size), K(value_size));
} else {
char *block_buf = reinterpret_cast<char *>(kvpair->value_) + sizeof(ObMicroBlockCacheValue);
kvpair->key_ = new (kvpair->key_) ObMicroBlockCacheKey();
ObMicroBlockCacheValue *cache_value = new (kvpair->value_) ObMicroBlockCacheValue(block_buf, block_size);
int64_t pos = 0;
if (ObStoreFormat::is_row_store_type_with_cs_encoding(row_store_type)) {
if (OB_FAIL(transformer.full_transform(block_buf, block_size, pos))) {
LOG_WARN("fail to full transform", K(ret), KP(block_buf), K(block_size));
}
} else {
MEMCPY(block_buf, micro_block_desc.header_, micro_block_desc.header_->header_size_);
MEMCPY(block_buf + micro_block_desc.header_->header_size_, micro_block_desc.buf_, micro_block_desc.buf_size_);
}
}
}
if (OB_FAIL(ret)) {
} else if (need_decoder && OB_FAIL(write_extra_buf(
static_cast<ObRowStoreType>(micro_block_desc.header_->row_store_type_),
reinterpret_cast<char *>(kvpair->value_) + sizeof(ObMicroBlockCacheValue),
block_size,
reinterpret_cast<char *>(kvpair->value_) + sizeof(ObMicroBlockCacheValue) + block_size,
static_cast<ObMicroBlockCacheValue *>(kvpair->value_)->get_block_data()))) {
LOG_WARN("Fail to write decoder in extra buf", K(ret));
} else {
kvpair_size = sizeof(ObMicroBlockCacheKey) + value_size;
}
return ret;
}
ObMicroBlockData::Type ObDataMicroBlockCache::get_type()
{
return ObMicroBlockData::DATA_BLOCK;
}
/*-------------------------------------ObIndexMicroBlockCache-------------------------------------*/
ObIndexMicroBlockCache::ObIndexMicroBlockCache()
: ObDataMicroBlockCache()
{
}
ObIndexMicroBlockCache::~ObIndexMicroBlockCache()
{
}
int ObIndexMicroBlockCache::init(const char *cache_name, const int64_t priority)
{
return ObDataMicroBlockCache::init(cache_name, priority);
}
int ObIndexMicroBlockCache::load_block(
const ObMicroBlockId &micro_block_id,
const ObMicroBlockDesMeta &des_meta,
ObMacroBlockReader *macro_reader,
ObMicroBlockData &block_data,
ObIAllocator *allocator)
{
UNUSED(macro_reader);
int ret = OB_SUCCESS;
ObMacroBlockReadInfo macro_read_info;
ObMacroBlockHandle macro_handle;
// TODO: @chengji make deserialize micro block with allocator static and remove tmp inner_macro_reader
ObMacroBlockReader inner_macro_reader;
bool is_compressed = false;
if (OB_UNLIKELY(!micro_block_id.is_valid()) || OB_ISNULL(allocator)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("Invalid argument", K(ret), K(micro_block_id), KP(allocator));
} else {
void *buf = nullptr;
ObSyncSingleMicroBLockIOCallback *callback = nullptr;
if (OB_ISNULL(buf = allocator->alloc(sizeof(ObSyncSingleMicroBLockIOCallback)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("allocate callback memory failed", K(ret));
} else {
callback = new (buf) ObSyncSingleMicroBLockIOCallback;
callback->allocator_ = allocator;
callback->offset_ = micro_block_id.offset_;
callback->block_des_meta_ = des_meta;
callback->block_data_ = &block_data;
callback->macro_reader_ = &inner_macro_reader;
callback->is_data_block_ = false;
macro_read_info.macro_block_id_ = micro_block_id.macro_id_;
macro_read_info.io_desc_.set_wait_event(ObWaitEventIds::DB_FILE_DATA_READ);
macro_read_info.io_desc_.set_group_id(ObIOModule::MICRO_BLOCK_CACHE_IO);
macro_read_info.io_callback_ = callback;
macro_read_info.offset_ = micro_block_id.offset_;
macro_read_info.size_ = micro_block_id.size_;
macro_read_info.io_timeout_ms_ = GCONF._data_storage_io_timeout / 1000L;
ObIndexBlockDataTransformer idx_transformer;
char *transform_buf = nullptr;
char *raw_idx_block_buf = nullptr;
if (OB_FAIL(ObBlockManager::read_block(macro_read_info, macro_handle))) {
LOG_WARN("Fail to sync read block", K(ret), K(macro_read_info));
} else if (FALSE_IT(raw_idx_block_buf = const_cast<char *>(block_data.get_buf()))) {
} else if (OB_FAIL(idx_transformer.transform(block_data, block_data, *allocator, transform_buf))) {
LOG_WARN("Fail to transform index block to memory format", K(ret));
} else {
EVENT_INC(ObStatEventIds::IO_READ_PREFETCH_MICRO_COUNT);
EVENT_ADD(ObStatEventIds::IO_READ_PREFETCH_MICRO_BYTES, micro_block_id.size_);
}
if (nullptr != raw_idx_block_buf) {
allocator->free(raw_idx_block_buf);
}
if (OB_FAIL(ret)) {
if (nullptr != callback->get_allocator()) { //Avoid double_free with io_handle
callback->~ObSyncSingleMicroBLockIOCallback();
allocator->free(callback);
}
if (nullptr != transform_buf) {
allocator->free(transform_buf);
}
block_data.reset();
}
}
}
return ret;
}
int ObIndexMicroBlockCache::put_cache_block(
const ObMicroBlockDesMeta &des_meta,
const char *raw_block_buf,
const ObMicroBlockCacheKey &key,
ObMacroBlockReader &reader,
ObIAllocator &allocator,
const ObMicroBlockCacheValue *&micro_block,
common::ObKVCacheHandle &cache_handle)
{
int ret = OB_SUCCESS;
ObMicroBlockHeader header;
int64_t pos = 0;
int64_t payload_size = 0;
const char *payload_buf = nullptr;
const ObMicroBlockId &micro_id = key.get_micro_block_id();
ObIMicroBlockCache::BaseBlockCache *kvcache = nullptr;
if (OB_UNLIKELY(!des_meta.is_valid() || !micro_id.is_valid()) || OB_ISNULL(raw_block_buf)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("Invalid argument", K(ret), K(key), K(des_meta));
} else if (OB_FAIL(header.deserialize(raw_block_buf, micro_id.size_, pos))) {
LOG_ERROR("Fail to deserialize record header", K(ret), K(micro_id));
} else if (OB_FAIL(header.check_and_get_record(
raw_block_buf, micro_id.size_, MICRO_BLOCK_HEADER_MAGIC, payload_buf, payload_size))) {
LOG_ERROR("Micro block data is corrupted", K(ret), K(key), KP(raw_block_buf), KP(this));
} else if (OB_FAIL(get_cache(kvcache))) {
LOG_WARN("Fail to get kvcache", K(ret));
} else {
int64_t block_size = 0;
int64_t value_size = 0;
char *block_buf = nullptr;
ObMicroBlockBufTransformer buf_transformer(des_meta, &reader, header, payload_buf, payload_size);
ObIndexBlockDataTransformer idx_transformer;
if (OB_FAIL(buf_transformer.init())) {
LOG_WARN("Fail to init block buf transformer", K(ret));
} else if (OB_FAIL(buf_transformer.get_buf_size(block_size))) {
LOG_WARN("Fail to get transformed buf size", K(ret));
} else if (OB_ISNULL(block_buf = static_cast<char *>(allocator.alloc(block_size)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("Fail to allocate memory", K(ret), K(block_size));
} else if (OB_FAIL(buf_transformer.transfrom(block_buf, block_size))) {
LOG_WARN("fail to transfrom block buf", K(ret));
} else {
ObMicroBlockCacheValue cache_value(block_buf, block_size);
ObMicroBlockData &block_data = cache_value.get_block_data();
block_data.type_ = get_type();
char *allocated_buf = nullptr;
if (OB_FAIL(idx_transformer.transform(block_data, block_data, allocator, allocated_buf))) {
LOG_WARN("Fail to transform index block to memory format", K(ret));
} else if (OB_FAIL(put_and_fetch(key, cache_value, micro_block, cache_handle, false /* overwrite */))) {
if (OB_ENTRY_EXIST != ret) {
LOG_WARN("Fail to put micro block cache", K(ret));
} else {
ret = OB_SUCCESS;
}
} else {
const int64_t put_size = ObKVStoreMemBlock::get_align_size(key, cache_value);
if (OB_FAIL(add_put_size(put_size))) {
LOG_WARN("add_put_size failed", K(ret), K(put_size));
}
}
if (nullptr != allocated_buf) {
allocator.free(allocated_buf);
}
}
if (nullptr != block_buf) {
allocator.free(block_buf);
}
if (OB_FAIL(ret)) {
cache_handle.reset();
micro_block = nullptr;
}
}
return ret;
}
ObMicroBlockData::Type ObIndexMicroBlockCache::get_type()
{
return ObMicroBlockData::INDEX_BLOCK;
}
}//end namespace blocksstable
}//end namespace oceanbase
Reference in New Issue View Git Blame Copy Permalink