/**

* 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 SQL_ENG

#include "ob_ra_row_store.h"

#include "lib/container/ob_se_array_iterator.h"

#include "storage/blocksstable/ob_tmp_file.h"

#include "lib/utility/ob_tracepoint.h"

#include "share/config/ob_server_config.h"

namespace oceanbase

{

using namespace common;

namespace sql

{

int ObRARowStore::ShrinkBuffer::init(char *buf, const int64_t buf_size)

{

int ret = OB_SUCCESS;

if (NULL == buf || buf_size <= 0) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(ret));

} else {

data_ = buf;

head_ = 0;

tail_ = buf_size;

cap_ = buf_size;

}

return ret;

}

namespace ra_row_store {

template <typename T, typename B>

void pointer2off(T *&pointer, B *base)

{

pointer = reinterpret_cast<T *>(

reinterpret_cast<const char *>(pointer) - reinterpret_cast<const char *>(base));

}

template <typename T, typename B>

void off2pointer(T *&pointer, B *base)

{

pointer = reinterpret_cast<T *>(

reinterpret_cast<intptr_t>(pointer) + reinterpret_cast<char *>(base));

}

}

int64_t ObRARowStore::StoreRow::row_copy_size(const ObNewRow &row)

{

int64_t size = sizeof(common::ObObj) * row.get_count();

for (int64_t i = 0; i < row.get_count(); ++i) {

size += row.get_cell(i).get_deep_copy_size();

}

return size;

}

int ObRARowStore::StoreRow::copy_row(const ObNewRow &r, char *buf, const int64_t size)

{

int ret = OB_SUCCESS;

if (payload_ != buf || size < 0) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(ret), KP(payload_), KP(buf), K(size));

} else {

readable_ = true;

cnt_ = ~(1U << 31) & static_cast<int32>(r.get_count());

int64_t pos = sizeof(ObObj) * cnt_;

for (int64_t i = 0; OB_SUCC(ret) && i < cnt_; ++i) {

new (&cells()[i])ObObj();

if (OB_FAIL(cells()[i].deep_copy(r.get_cell(i), buf, size, pos))) {

LOG_WARN("copy cell failed", K(ret));

}

}

}

return ret;

}

int ObRARowStore::StoreRow::to_copyable()

{

int ret = OB_SUCCESS;

if (0 != readable_) {

for (int64_t i = 0; i < cnt_; ++i) {

if (cells()[i].need_deep_copy()) {

ra_row_store::pointer2off(cells()[i].v_.string_, this);

}

}

readable_ = false;

}

return ret;

}

int ObRARowStore::StoreRow::to_readable()

{

int ret = OB_SUCCESS;

if (0 == readable_) {

for (int64_t i = 0; i < cnt_; ++i) {

if (cells()[i].need_deep_copy()) {

ra_row_store::off2pointer(cells()[i].v_.string_, this);

}

}

readable_ = true;

}

return ret;

}

int ObRARowStore::Block::add_row(

ShrinkBuffer &buf, const ObNewRow &row, const int64_t row_size)

{

int ret = OB_SUCCESS;

if (!buf.is_inited() || row_size <= ROW_INDEX_SIZE) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(ret), K(buf), K(row_size));

} else if (row_size > buf.remain()) {

ret = OB_BUF_NOT_ENOUGH;

LOG_WARN("buffer not enough", K(row_size), "remain", buf.remain());

} else {

StoreRow *sr = new (buf.head())StoreRow;

if (OB_FAIL(sr->copy_row(

row, buf.head() + sizeof(*sr), row_size - sizeof(*sr) - ROW_INDEX_SIZE))) {

LOG_WARN("copy row failed", K(ret), K(row), K(row_size));

} else if (OB_FAIL(buf.fill_tail(ROW_INDEX_SIZE))) {

LOG_WARN("fill buffer tail failed", K(ret), K(buf), LITERAL_K(ROW_INDEX_SIZE));

} else {

*reinterpret_cast<row_idx_t *>(buf.tail()) = static_cast<row_idx_t>(

buf.head() - payload_);

idx_off_ -= static_cast<int32_t>(ROW_INDEX_SIZE);

if (OB_FAIL(buf.fill_head(row_size - ROW_INDEX_SIZE))) {

LOG_WARN("fill buffer head failed", K(ret), K(buf), K(row_size - ROW_INDEX_SIZE));

} else {

rows_++;

}

}

}

return ret;

}

int ObRARowStore::Block::get_store_row(const int64_t row_id, const StoreRow *&sr)

{

int ret = OB_SUCCESS;

if (!contain(row_id)) {

ret = OB_INDEX_OUT_OF_RANGE;

LOG_WARN("invalid index", K(ret), K(row_id), K(*this));

} else {

StoreRow *row = reinterpret_cast<StoreRow *>(

&payload_[indexes()[rows_ - (row_id - row_id_) - 1]]);

if (0 == row->readable_) {

if (OB_FAIL(row->to_readable())) {

LOG_WARN("store row to readable failed", K(ret));

}

}

if (OB_SUCC(ret)) {

sr = row;

}

}

return ret;

}

int ObRARowStore::Block::compact(ShrinkBuffer &buf)

{

int ret = OB_SUCCESS;

if (!buf.is_inited()) {

LOG_WARN("invalid argument", K(ret), K(buf));

} else if (OB_FAIL(buf.compact())) {

LOG_WARN("block buffer compact failed", K(ret));

} else {

idx_off_ = static_cast<int32_t>(buf.head() - rows_ * ROW_INDEX_SIZE - payload_);

}

return ret;

}

int ObRARowStore::Block::to_copyable()

{

int ret = OB_SUCCESS;

for (int64_t i = 0; OB_SUCC(ret) && i < rows_; ++i) {

StoreRow *sr = reinterpret_cast<StoreRow *>(&payload_[indexes()[i]]);

if (OB_FAIL(sr->to_copyable())) {

LOG_WARN("convert store row to copyable row failed", K(ret));

}

}

return ret;

}

ObRARowStore::ObRARowStore(common::ObIAllocator *alloc /* = NULL */,

const bool keep_projector /* = false */ )

: inited_(false), tenant_id_(0), label_(0), ctx_id_(0), mem_limit_(0),

idx_blk_(NULL), save_row_cnt_(0), row_cnt_(0), fd_(-1), dir_id_(-1), file_size_(0),

inner_reader_(*this), mem_hold_(0), allocator_(NULL == alloc ? inner_allocator_ : *alloc),

keep_projector_(keep_projector), projector_(NULL), projector_size_(0)

{

}

int ObRARowStore::init(int64_t mem_limit,

uint64_t tenant_id /* = common::OB_SERVER_TENANT_ID */,

int64_t mem_ctx_id /* = common::ObCtxIds::DEFAULT_CTX_ID */,

const char *label /* = common::ObNewModIds::OB_SQL_ROW_STORE) */)

{

int ret = OB_SUCCESS;

if (inited_) {

ret = OB_INIT_TWICE;

LOG_WARN("init twice", K(ret));

} else {

tenant_id_ = tenant_id;

ctx_id_ = mem_ctx_id;

label_ = label;

mem_limit_ = mem_limit;

inited_ = true;

}

return ret;

}

void ObRARowStore::reset()

{

int ret = OB_SUCCESS;

tenant_id_ = common::OB_SERVER_TENANT_ID;

label_ = common::ObModIds::OB_SQL_ROW_STORE;

ctx_id_ = common::ObCtxIds::DEFAULT_CTX_ID;

mem_limit_ = 0;

blkbuf_.reset();

idx_blk_ = NULL;

save_row_cnt_ = 0;

row_cnt_ = 0;

inner_reader_.reset();

if (is_file_open()) {

if (OB_FAIL(FILE_MANAGER_INSTANCE_V2.remove(fd_))) {

LOG_WARN("remove file failed", K(ret), K_(fd));

} else {

LOG_INFO("close file success", K(ret), K_(fd));

}

fd_ = -1;

dir_id_ = -1;

file_size_ = 0;

}

while (!blk_mem_list_.is_empty()) {

LinkNode *node = blk_mem_list_.remove_first();

if (NULL != node) {

node->~LinkNode();

allocator_.free(node);

}

}

blocks_.reset();

mem_hold_ = 0;

if (NULL != projector_) {

allocator_.free(projector_);

}

projector_ = NULL;

projector_size_ = 0;

inited_ = false;

}

void ObRARowStore::reuse()

{

int ret = OB_SUCCESS;

save_row_cnt_ = 0;

row_cnt_ = 0;

inner_reader_.reuse();

if (is_file_open()) {

if (OB_FAIL(FILE_MANAGER_INSTANCE_V2.remove(fd_))) {

LOG_WARN("remove file failed", K(ret), K_(fd));

} else {

LOG_INFO("close file success", K(ret), K_(fd));

}

fd_ = -1;

dir_id_ = -1;

file_size_ = 0;

}

idx_blk_ = NULL;

DLIST_FOREACH_REMOVESAFE_NORET(node, blk_mem_list_) {

if (&(*node) + 1 != static_cast<LinkNode *>(static_cast<void *>(blkbuf_.buf_.data()))) {

node->unlink();

node->~LinkNode();

allocator_.free(node);

}

}

if (NULL != blkbuf_.buf_.data()) {

if (OB_FAIL(setup_block(blkbuf_))) {

LOG_WARN("setup block failed", K(ret));

}

mem_hold_ = blkbuf_.buf_.capacity() + sizeof(LinkNode);

}

blocks_.reset();

}

int ObRARowStore::setup_block(BlockBuf &blkbuf) const

{

int ret = OB_SUCCESS;

if (!is_inited()) {

ret = OB_NOT_INIT;

LOG_WARN("not init", K(ret));

} else if (!blkbuf.buf_.is_inited()) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("block buffer not inited", K(ret));

} else {

blkbuf.buf_.reuse();

blkbuf.blk_ = new (blkbuf.buf_.head()) Block;

blkbuf.blk_->row_id_ = row_cnt_;

blkbuf.blk_->idx_off_ = static_cast<int32_t>(

blkbuf.buf_.tail() - blkbuf.blk_->payload_);

if (OB_FAIL(blkbuf.buf_.fill_head(sizeof(Block)))) {

LOG_WARN("fill buffer head failed", K(ret), K(blkbuf.buf_), K(sizeof(Block)));

}

}

return ret;

}

void *ObRARowStore::alloc_blk_mem(const int64_t size)

{

void *blk = NULL;

int ret = OB_SUCCESS;

if (size < 0) {

LOG_WARN("invalid argument", K(size));

} else {

ObMemAttr attr(tenant_id_, label_, ctx_id_);

void *mem = allocator_.alloc(size + sizeof(LinkNode), attr);

if (NULL == mem) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("alloc memory failed", K(ret), KP(mem));

} else {

LinkNode *node = new (mem) LinkNode;

if (!blk_mem_list_.add_last(node)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("add node to list failed", K(ret));

node->~LinkNode();

allocator_.free(mem);

} else {

blk = static_cast<char *>(mem) + sizeof(LinkNode);

mem_hold_ += size + sizeof(LinkNode);

}

}

}

return blk;

}

void ObRARowStore::free_blk_mem(void *mem, const int64_t size /* = 0 */)

{

if (NULL != mem) {

LinkNode *node = static_cast<LinkNode *>(mem) - 1;

if (NULL != node->get_next()) {

node->unlink();

}

node->~LinkNode();

allocator_.free(node);

mem_hold_ -= (size + sizeof(LinkNode));

}

}

int ObRARowStore::alloc_block(BlockBuf &blkbuf, const int64_t min_size)

{

int ret = OB_SUCCESS;

int64_t size = std::max(static_cast<int64_t>(BLOCK_SIZE), min_size);

if (row_cnt_ > 0 || need_dump()) {

size = std::max(size, static_cast<int64_t>(BIG_BLOCK_SIZE));

}

size += sizeof(LinkNode);

size = next_pow2(size);

size -= sizeof(LinkNode);

if (!is_inited()) {

ret = OB_NOT_INIT;

LOG_WARN("not init", K(ret));

} else {

void *mem = alloc_blk_mem(size);

if (OB_ISNULL(mem)) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("alloc memory failed", K(ret), K(size));

} else if (OB_FAIL(blkbuf.buf_.init(static_cast<char *>(mem), size))) {

LOG_WARN("init shrink buffer failed", K(ret));

} else if (OB_FAIL(setup_block(blkbuf))) {

LOG_WARN("setup block buffer fail", K(ret));

}

if (OB_FAIL(ret) && !OB_ISNULL(mem)) {

blkbuf.reset();

free_blk_mem(mem, size);

}

}

return ret;

}

int ObRARowStore::switch_block(const int64_t min_size)

{

int ret = OB_SUCCESS;

if (!is_inited()) {

ret = OB_NOT_INIT;

LOG_WARN("not init", K(ret));

} else if (min_size < 0 || OB_ISNULL(blkbuf_.blk_)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(ret), K(min_size));

} else if (OB_FAIL(blkbuf_.blk_->compact(blkbuf_.buf_))) {

LOG_WARN("block compact failed", K(ret));

} else {

const bool finish_add = (0 == min_size);

BlockBuf new_blkbuf;

const bool force_new_block = (min_size > blkbuf_.buf_.capacity())

|| !GCONF.is_sql_operator_dump_enabled();

BlockIndex bi;

bi.is_idx_block_ = false;

bi.on_disk_ = false;

bi.row_id_ = ~(0b11UL << 62) & save_row_cnt_;

bi.blk_ = blkbuf_.blk_;

bi.length_ = static_cast<int32_t>(blkbuf_.buf_.head_size());

bool dump = need_dump();

if (!finish_add && (force_new_block || !dump)) { // need alloc new block

if (OB_FAIL(alloc_block(new_blkbuf, min_size))) {

LOG_WARN("alloc block failed", K(ret), K(min_size));

if (!force_new_block) {

dump = true;

ret = OB_SUCCESS;

}

}

}

if (OB_SUCC(ret) && dump) {

if (OB_FAIL(blkbuf_.blk_->to_copyable())) {

LOG_WARN("convert block to copyable failed", K(ret));

} else {

if (OB_FAIL(write_file(bi, blkbuf_.buf_.data(), blkbuf_.buf_.head_size()))) {

LOG_WARN("write block to file failed");

}

}

}

if (OB_FAIL(ret)) {

} else if (OB_FAIL(add_block_idx(bi))) {

LOG_WARN("add block index failed", K(ret));

} else {

save_row_cnt_ = row_cnt_;

if (!dump) {

blkbuf_.reset();

}

if (new_blkbuf.buf_.is_inited()) {

if (blkbuf_.buf_.is_inited()) {

// free block, if block dumped and need bigger block

free_blk_mem(blkbuf_.buf_.data(), blkbuf_.buf_.capacity());

blkbuf_.reset();

}

blkbuf_ = new_blkbuf;

new_blkbuf.reset();

} else if (blkbuf_.buf_.is_inited()) {

if (OB_FAIL(setup_block(blkbuf_))) {

LOG_WARN("setup block failed", K(ret));

}

}

}

if (OB_FAIL(ret) && new_blkbuf.buf_.is_inited()) {

free_blk_mem(new_blkbuf.blk_, new_blkbuf.buf_.capacity());

new_blkbuf.reset();

}

}

return ret;

}

int ObRARowStore::add_block_idx(const BlockIndex &bi)

{

int ret = OB_SUCCESS;

if (!is_inited()) {

ret = OB_NOT_INIT;

LOG_WARN("not init", K(ret));

} else {

if (NULL == idx_blk_) {

if (OB_FAIL(blocks_.push_back(bi))) {

LOG_WARN("add block index to array failed", K(ret));

} else {

if (blocks_.count() >= DEFAULT_BLOCK_CNT) {

if (OB_FAIL(build_idx_block())) {

LOG_WARN("build index block failed", K(ret));

}

}

}

} else {

if (idx_blk_->is_full()) {

if (OB_FAIL(switch_idx_block())) {

LOG_WARN("switch index block failed", K(ret));

}

}

if (OB_SUCC(ret)) {

idx_blk_->block_indexes_[idx_blk_->cnt_++] = bi;

}

}

}

return ret;

}

int ObRARowStore::alloc_idx_block(IndexBlock *&ib)

{

int ret = OB_SUCCESS;

if (!is_inited()) {

ret = OB_NOT_INIT;

LOG_WARN("not init", K(ret));

} else {

void *mem = alloc_blk_mem(IndexBlock::INDEX_BLOCK_SIZE);

if (OB_ISNULL(mem)) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("alloc memory failed", K(ret));

} else {

ib = new (mem) IndexBlock;

}

}

return ret;

}

int ObRARowStore::build_idx_block()

{

STATIC_ASSERT(IndexBlock::capacity() > DEFAULT_BLOCK_CNT,

"DEFAULT_BLOCK_CNT block indexes must fit in one index block");

int ret = OB_SUCCESS;

if (!is_inited()) {

ret = OB_NOT_INIT;

LOG_WARN("not init", K(ret));

} else if (OB_FAIL(alloc_idx_block(idx_blk_))) {

LOG_WARN("alloc idx block failed", K(ret));

} else if (NULL == idx_blk_) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("alloc null index block", K(ret));

} else {

for (int64_t i = 0; OB_SUCC(ret) && i < blocks_.count(); ++i) {

if (OB_FAIL(add_block_idx(blocks_.at(i)))) {

LOG_WARN("add block idx failed", K(ret));

}

}

if (OB_SUCC(ret)) {

blocks_.reset();

}

}

return ret;

}

int ObRARowStore::switch_idx_block(bool finish_add /* = false */)

{

int ret = OB_SUCCESS;

if (!is_inited()) {

ret = OB_NOT_INIT;

LOG_WARN("not init", K(ret));

} else if (NULL == idx_blk_) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("index block should not be null");

} else {

IndexBlock *ib = NULL;

BlockIndex bi;

bi.is_idx_block_ = true;

bi.on_disk_ = false;

bi.row_id_ = idx_blk_->block_indexes_[0].row_id_;

bi.idx_blk_ = idx_blk_;

bi.length_ = static_cast<int32_t>(idx_blk_->buffer_size());

bool dump = need_dump();

if (!finish_add && !dump) {

if (OB_FAIL(alloc_idx_block(ib))) {

LOG_WARN("alloc index block failed", K(ret));

if (GCONF.is_sql_operator_dump_enabled()) {

ret = OB_SUCCESS;

dump = true;

}

} else if (OB_ISNULL(ib)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("alloc null block", K(ret));

}

}

if (OB_SUCC(ret) && dump) {

if (OB_FAIL(write_file(bi, idx_blk_, idx_blk_->buffer_size()))) {

LOG_WARN("write index block to file failed");

}

}

if (OB_FAIL(ret)) {

} else if (OB_FAIL(blocks_.push_back(bi))) {

LOG_WARN("add block index to array failed", K(ret));

} else {

if (!dump) {

idx_blk_ = NULL;

}

if (NULL != ib) {

idx_blk_ = ib;

ib = NULL;

} else if (NULL != idx_blk_) {

idx_blk_->reset();

}

}

if (!OB_SUCC(ret) && NULL != ib) {

ib->~IndexBlock();

free_blk_mem(ib, IndexBlock::INDEX_BLOCK_SIZE);

ib = NULL;

}

}

return ret;

}

int ObRARowStore::add_row(const common::ObNewRow &row)

{

int ret = OB_SUCCESS;

if (!is_inited()) {

ret = OB_NOT_INIT;

LOG_WARN("not init", K(ret));

} else {

int64_t bak_projector_size = 0;

int32_t *bak_projector_ = NULL;

ObNewRow &r = const_cast<ObNewRow &>(row);

if (keep_projector_ && NULL != row.projector_ && row.projector_size_ > 0) {

if (OB_ISNULL(projector_)) {

ObMemAttr attr(tenant_id_, label_, ctx_id_);

projector_size_ = row.projector_size_;

const int64_t size = projector_size_ * sizeof(*projector_);

projector_ = static_cast<int32_t *>(allocator_.alloc(size));

if (OB_ISNULL(projector_)) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("allocate memory failed", K(ret));

} else {

MEMCPY(projector_, row.projector_, size);

}

} else if (projector_size_ != row.projector_size_) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("projector size mismatch", K(ret), K(projector_size_), K(row.projector_size_));

}

if (OB_SUCC(ret)) {

bak_projector_ = r.projector_;

bak_projector_size = r.projector_size_;

r.projector_size_ = 0;

r.projector_ = NULL;

}

}

if (OB_SUCC(ret) && NULL == blkbuf_.blk_) {

if (OB_FAIL(alloc_block(blkbuf_, Block::min_buf_size(row)))) {

LOG_WARN("alloc block failed", K(ret));

}

}

if (OB_SUCC(ret)) {

const int64_t row_size = Block::row_store_size(row);

if (row_size > blkbuf_.buf_.remain()) {

if (OB_FAIL(switch_block(Block::min_buf_size(row)))) {

LOG_WARN("switch block failed", K(ret), K(row));

}

}

if (OB_SUCC(ret)) {

if (OB_FAIL(blkbuf_.blk_->add_row(blkbuf_.buf_, row, row_size))) {

LOG_WARN("add row to block failed", K(ret), K(row), K(row_size));

} else {

row_cnt_++;

}

}

}

// always keep row unchanged.

if (NULL != bak_projector_) {

r.projector_size_ = bak_projector_size;

r.projector_ = bak_projector_;

}

}

return ret;

}

int ObRARowStore::find_block_idx(Reader &reader, BlockIndex &bi, const int64_t row_id)

{

int ret = OB_SUCCESS;

if (!is_inited()) {

ret = OB_NOT_INIT;

LOG_WARN("not init", K(ret));

} else if (row_id < 0 || row_id >= save_row_cnt_) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("row should be saved", K(ret), K(row_id), K_(save_row_cnt));

} else {

bool found = false;

if (NULL != reader.idx_blk_) {

if (reader.ib_pos_ < 0 || reader.ib_pos_ >= reader.idx_blk_->cnt_) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("ib_pos out of range", K(ret), K(reader.ib_pos_), K(*reader.idx_blk_));

} else {

int64_t pos = reader.ib_pos_;

if (row_id > reader.idx_blk_->block_indexes_[pos].row_id_) {

pos += 1;

if (reader.idx_blk_->row_in_pos(row_id, pos)) {

found = true;

reader.ib_pos_ = pos;

}

} else {

pos -= 1;

if (reader.idx_blk_->row_in_pos(row_id, pos)) {

found = true;

reader.ib_pos_ = pos;

}

}

}

if (!found) {

reader.reset_cursor(file_size_);

} else {

bi = reader.idx_blk_->block_indexes_[reader.ib_pos_];

}

}

if (OB_FAIL(ret) || found) {

} else {

IndexBlock *ib = NULL;

if (NULL != idx_blk_ && idx_blk_->cnt_ > 0

&& row_id >= idx_blk_->block_indexes_[0].row_id_) {

ib = idx_blk_;

}

if (NULL == ib && blocks_.count() > 0) {

auto it = std::lower_bound(blocks_.begin(), blocks_.end(),

row_id, &BlockIndex::compare);

if (it == blocks_.end() || it->row_id_ != row_id) {

it--;

}

bi = *it;

if (!bi.is_idx_block_) {

found = true;

} else {

if (OB_FAIL(load_idx_block(reader, ib, bi))) {

LOG_WARN("load index block failed", K(ret), K(bi));

}

}

}

if (OB_FAIL(ret) || found) {

} else if (NULL == ib || ib->cnt_ <= 0) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("block index not found and index block is NULL or empty", K(ret));

} else {

auto it = std::lower_bound(&ib->block_indexes_[0], &ib->block_indexes_[ib->cnt_],

row_id, &BlockIndex::compare);

if (it == ib->block_indexes_ + ib->cnt_ || it->row_id_ != row_id) {

it--;

}

bi = *it;

reader.idx_blk_ = ib;

reader.ib_pos_ = it - ib->block_indexes_;

}

}

}

return ret;

}

int ObRARowStore::load_idx_block(Reader &reader, IndexBlock *&ib, const BlockIndex &bi)

{

int ret = OB_SUCCESS;

if (!is_inited()) {

ret = OB_NOT_INIT;

LOG_WARN("not init", K(ret));

} else if (!bi.is_idx_block_) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid block index", K(ret), K(bi));

} else {

if (!bi.on_disk_) {

ib = bi.idx_blk_;

} else {

if (bi.length_ > IndexBlock::INDEX_BLOCK_SIZE) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("invalid argument", K(ret), K(bi));

} else if (OB_FAIL(ensure_reader_buffer(

reader.idx_buf_, IndexBlock::INDEX_BLOCK_SIZE))) {

LOG_WARN("ensure reader buffer failed", K(ret));

} else if (OB_FAIL(read_file(

reader.idx_buf_.data(), bi.length_, bi.offset_))) {

LOG_WARN("read block index from file failed", K(ret), K(bi));

} else {

ib = reinterpret_cast<IndexBlock *>(reader.idx_buf_.data());

}

}

}

return ret;

}

int ObRARowStore::load_block(Reader &reader, const int64_t row_id)

{

int ret = OB_SUCCESS;

BlockIndex bi;

if (!is_inited()) {

ret = OB_NOT_INIT;

LOG_WARN("not init", K(ret));

} else if (row_id < 0 || row_id >= save_row_cnt_) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("row should be saved", K(ret), K(row_id), K_(save_row_cnt));

} else if (OB_FAIL(find_block_idx(reader, bi, row_id))) {

LOG_WARN("find block index failed", K(ret), K(row_id));

} else {

if (!bi.on_disk_) {

reader.blk_ = bi.blk_;

} else {

if (OB_FAIL(ensure_reader_buffer(reader.buf_, bi.length_))) {

LOG_WARN("ensure reader buffer failed", K(ret));

} else if (OB_FAIL(read_file(reader.buf_.data(), bi.length_, bi.offset_))) {

LOG_WARN("read block from file failed", K(ret), K(bi));

} else {

reader.blk_ = reinterpret_cast<Block *>(reader.buf_.data());

}

}

}

return ret;

}

int ObRARowStore::get_store_row(Reader &reader, const int64_t row_id, const StoreRow *&sr)

{

int ret = OB_SUCCESS;

if (!is_inited()) {

ret = OB_NOT_INIT;

LOG_WARN("not init", K(ret));

} else if (row_id < 0 || row_id >= row_cnt_) {

ret = OB_INDEX_OUT_OF_RANGE;

LOG_WARN("invalid of row_id", K(ret), K(row_id), K_(row_cnt));

} else {

if (reader.file_size_ != file_size_) {

reader.reset_cursor(file_size_);

}

if (NULL != reader.blk_ && reader.blk_->contain(row_id)) {

// found in previous visited block

} else if (row_id >= save_row_cnt_) {

// found in write block

reader.blk_ = blkbuf_.blk_;

} else {

if (NULL != reader.blk_) {

reader.blk_ = NULL;

}

if (OB_FAIL(load_block(reader, row_id))) {

LOG_WARN("load block failed", K(ret));

}

}

if (OB_SUCC(ret)) {

if (NULL == reader.blk_) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("null block", K(ret), K(row_id), K(*this));

} else if (OB_FAIL(reader.blk_->get_store_row(row_id, sr))) {

LOG_WARN("get row from block failed", K(ret), K(row_id), K(*reader.blk_));

}

}

}

return ret;

}

void ObRARowStore::Reader::reset()

{

const int64_t file_size = 0;

reset_cursor(file_size);

store_.free_blk_mem(buf_.data(), buf_.capacity());

buf_.reset();

store_.free_blk_mem(idx_buf_.data(), idx_buf_.capacity());

idx_buf_.reset();

}

void ObRARowStore::Reader::reuse()

{

reset_cursor(0);

buf_.reset();

idx_buf_.reset();

}

void ObRARowStore::Reader::reset_cursor(const int64_t file_size)

{

file_size_ = file_size;

idx_blk_ = NULL;

ib_pos_ = 0;

blk_ = NULL;

}

int ObRARowStore::Reader::get_row(const int64_t row_id, const ObNewRow *&row)

{

const StoreRow *sr = NULL;

int ret = OB_SUCCESS;

if (row_id < 0 || row_id >= get_row_cnt()) {

ret = OB_INDEX_OUT_OF_RANGE;

LOG_WARN("invalid row_id", K(ret), K(row_id), K(get_row_cnt()));

} else if (OB_FAIL(store_.get_store_row(*this, row_id, sr))) {

LOG_WARN("get store row failed", K(ret), K(row_id));

} else if (OB_ISNULL(sr)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("NULL store row returned", K(ret));

} else {

row_.count_ = sr->cnt_;

row_.cells_ = const_cast<ObObj *>(sr->cells());

if (OB_ISNULL(store_.projector_)) {

row_.projector_ = NULL;

row_.projector_size_ = 0;

} else {

row_.projector_ = store_.projector_;

row_.projector_size_ = store_.projector_size_;

}

row = &row_;

}

return ret;

}

int ObRARowStore::Reader::get_row(const int64_t row_id, const common::ObNewRow &row)

{

const StoreRow *sr = NULL;

int ret = OB_SUCCESS;

if (row_id < 0 || row_id >= get_row_cnt() || OB_ISNULL(row.cells_)) {

ret = OB_INDEX_OUT_OF_RANGE;

LOG_WARN("invalid row_id", K(ret), K(row_id), K(get_row_cnt()));

} else if (OB_FAIL(store_.get_store_row(*this, row_id, sr))) {

LOG_WARN("get store row failed", K(ret), K(row_id));

} else if (OB_ISNULL(sr)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("NULL store row returned", K(ret));

} else if (row.count_ < sr->cnt_) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("row cells count less than stored", K(ret), K(row.count_), K(sr->cnt_));

} else {

MEMCPY(row.cells_, sr->cells(), sr->cnt_ * sizeof(ObObj));

}

return ret;

}

int ObRARowStore::get_timeout(int64_t &timeout_ms)

{

int ret = OB_SUCCESS;

const int64_t timeout_us = THIS_WORKER.get_timeout_remain();

if (timeout_us / 1000 <= 0) {

ret = OB_TIMEOUT;

LOG_WARN("query is timeout", K(ret), K(timeout_us));

} else {

timeout_ms = timeout_us / 1000;

}

return ret;

}

int ObRARowStore::write_file(BlockIndex &bi, void *buf, int64_t size)

{

int ret = OB_SUCCESS;

int64_t timeout_ms = 0;

if (!is_inited()) {

ret = OB_NOT_INIT;

LOG_WARN("not init", K(ret));

} else if (size < 0 || (size > 0 && NULL == buf)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(size), KP(buf));

} else if (OB_FAIL(get_timeout(timeout_ms))) {

LOG_WARN("get timeout failed", K(ret));

} else {

if (!is_file_open()) {

if (OB_FAIL(FILE_MANAGER_INSTANCE_V2.alloc_dir(dir_id_))) {

LOG_WARN("alloc file directory failed", K(ret));

} else if (OB_FAIL(FILE_MANAGER_INSTANCE_V2.open(fd_, dir_id_))) {

LOG_WARN("open file failed", K(ret));

} else {

file_size_ = 0;

LOG_INFO("open file success", K_(fd), K_(dir_id));

}

}

ret = OB_E(EventTable::EN_8) ret;

}

if (OB_SUCC(ret) && size > 0) {

blocksstable::ObTmpFileIOInfo io;

io.fd_ = fd_;

io.buf_ = static_cast<char *>(buf);

io.size_ = size;

io.tenant_id_ = tenant_id_;

io.io_desc_.set_wait_event(ObWaitEventIds::ROW_STORE_DISK_WRITE);

io.io_timeout_ms_ = timeout_ms;

if (OB_FAIL(FILE_MANAGER_INSTANCE_V2.write(io))) {

LOG_WARN("write to file failed", K(ret), K(io), K(timeout_ms));

}

}

if (OB_SUCC(ret)) {

bi.on_disk_ = true;

bi.offset_ = file_size_;

file_size_ += size;

}

return ret;

}

int ObRARowStore::read_file(void *buf, const int64_t size, const int64_t offset)

{

int ret = OB_SUCCESS;

int64_t timeout_ms = 0;

if (!is_inited()) {

ret = OB_NOT_INIT;

LOG_WARN("not init", K(ret));

} else if (offset < 0 || size < 0 || (size > 0 && NULL == buf)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(size), K(offset), KP(buf));

} else if (OB_FAIL(get_timeout(timeout_ms))) {

LOG_WARN("get timeout failed", K(ret));

}

if (OB_SUCC(ret) && size > 0) {

blocksstable::ObTmpFileIOInfo io;

io.fd_ = fd_;

io.dir_id_ = dir_id_;

io.buf_ = static_cast<char *>(buf);

io.size_ = size;

io.tenant_id_ = tenant_id_;

io.io_desc_.set_wait_event(ObWaitEventIds::ROW_STORE_DISK_READ);

io.io_timeout_ms_ = timeout_ms;

blocksstable::ObTmpFileIOHandle handle;

if (OB_FAIL(FILE_MANAGER_INSTANCE_V2.pread(io, offset, handle))) {

LOG_WARN("read form file failed", K(ret), K(io), K(offset), K(timeout_ms));