hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Use the corresponding column fill cg

Browse Source 
Co-authored-by: Charles0429 <xiezhenjiang@gmail.com>
Co-authored-by: Monk-Liu <1152761042@qq.com>
This commit is contained in:
renju96
2024-02-09 22:21:21 +00:00
committed by ob-robot
parent d969ad5db0
commit 19d5e20cf1
15 changed files with 355 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/sql/engine/basic/chunk_store/ob_block_iwriter.h
View File

@ -47,7 +47,7 @@ public:
virtual int add_row(const common::ObIArray<ObExpr*> &exprs, ObEvalCtx &ctx, ObChunkDatumStore::StoredRow **stored_row = nullptr) = 0;
virtual int add_row(const ObChunkDatumStore::StoredRow &src_sr, ObChunkDatumStore::StoredRow **dst_sr = nullptr) = 0;
virtual int add_row(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
virtual int add_row(const blocksstable::ObStorageDatum *storage_datums, const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size, ObChunkDatumStore::StoredRow **stored_row) = 0;
virtual int add_batch(const common::ObDatum **datums, const common::ObIArray<ObExpr *> &exprs,

2
src/sql/engine/basic/chunk_store/ob_compact_block_reader.cpp
View File

@ -62,7 +62,7 @@ int ObCompactBlockReader::get_row(const ObChunkDatumStore::StoredRow *&sr)
ret = OB_ITER_END;
} else if (cur_pos_in_blk_ > cur_blk_->raw_size_ - sizeof(ObTempBlockStore::Block)) {
ret = OB_INDEX_OUT_OF_RANGE;
LOG_WARN("invalid index", K(ret), K(cur_pos_in_blk_), KP(cur_blk_));
LOG_WARN("invalid index", K(ret), K(cur_pos_in_blk_), KP(cur_blk_), K(cur_row_in_blk_), K(cur_blk_->cnt_), K(row_meta_->column_offset_));
} else {
int64_t size = 0;
sr = nullptr;

2
src/sql/engine/basic/chunk_store/ob_compact_block_reader.h
View File

@ -112,6 +112,7 @@ public:
cur_row_in_blk_ = 0;
cur_row_offset_width_ = 0;
row_meta_ = nullptr;
cur_blk_ = nullptr;
}
void reuse()
@ -123,6 +124,7 @@ public:
cur_row_in_blk_ = 0;
cur_row_offset_width_ = 0;
row_info_.reset();
cur_blk_ = nullptr;
}
int get_row(const ObChunkDatumStore::StoredRow *&sr) override;

40
src/sql/engine/basic/chunk_store/ob_compact_block_writer.cpp
View File

@ -101,28 +101,30 @@ int ObCompactBlockWriter::add_row(const ObChunkDatumStore::StoredRow &src_sr, Ob
}
int ObCompactBlockWriter::add_row(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
int ObCompactBlockWriter::add_row(const blocksstable::ObStorageDatum *storage_datums,
const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size, ObChunkDatumStore::StoredRow **stored_row)
{
int ret = OB_SUCCESS;
if (OB_FAIL(ensure_init())) {
LOG_WARN("fail to ensure init", K(ret));
} else if (OB_FAIL(ensure_write(storage_datums, cnt, extra_size))) {
} else if (OB_FAIL(ensure_write(storage_datums, cg_schema, extra_size))) {
LOG_WARN("fail to ensure write", K(ret));
} else {
if ((cur_row_offset_width_ == BASE_OFFSET_SIZE) &&
OB_FAIL(inner_add_row<uint16_t>(storage_datums, cnt, extra_size, stored_row))) {
LOG_WARN("add row to block failed", K(ret), K(storage_datums), K(cnt), K(extra_size));
OB_FAIL(inner_add_row<uint16_t>(storage_datums, cg_schema, extra_size, stored_row))) {
LOG_WARN("add row to block failed", K(ret), K(storage_datums), K(cg_schema), K(extra_size));
} else if (cur_row_offset_width_ == EXTENDED_OFFSET_SIZE &&
OB_FAIL(inner_add_row<uint32_t>(storage_datums, cnt, extra_size, stored_row))) {
LOG_WARN("add row to block failed", K(ret), K(storage_datums), K(cnt), K(extra_size));
OB_FAIL(inner_add_row<uint32_t>(storage_datums, cg_schema, extra_size, stored_row))) {
LOG_WARN("add row to block failed", K(ret), K(storage_datums), K(cg_schema), K(extra_size));
}
}
return ret;
}
template <typename T>
int ObCompactBlockWriter::inner_add_row(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
int ObCompactBlockWriter::inner_add_row(const blocksstable::ObStorageDatum *storage_datums,
const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size, ObChunkDatumStore::StoredRow **stored_row)
{
int ret = OB_SUCCESS;
@ -136,8 +138,9 @@ int ObCompactBlockWriter::inner_add_row(const blocksstable::ObStorageDatum *stor
LOG_WARN("buf is null", K(ret));
} else {
T *var_offset_array = reinterpret_cast<T*>(row_info_.buf_ + HEAD_SIZE + row_info_.bitmap_size_);
for (int64_t i = 0; OB_SUCC(ret) && i < cnt; i++) {
if (OB_FAIL(inner_process_datum<T>(storage_datums[i], i, *row_meta_, row_info_))) {
for (int64_t i = 0; OB_SUCC(ret) && i < cg_schema.column_cnt_; i++) {
int64_t column_idx = cg_schema.column_idxs_ ? cg_schema.column_idxs_[i] : i;
if (OB_FAIL(inner_process_datum<T>(storage_datums[column_idx], i, *row_meta_, row_info_))) {
LOG_WARN("fail to process datum", K(ret));
}
}
@ -368,22 +371,24 @@ int ObCompactBlockWriter::get_row_stored_size(const ObChunkDatumStore::StoredRow
}
int ObCompactBlockWriter::get_row_stored_size(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
int ObCompactBlockWriter::get_row_stored_size(const blocksstable::ObStorageDatum *storage_datums,
const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size, uint64_t &size)
{
int ret = OB_SUCCESS;
int64_t head_size = sizeof(ObChunkDatumStore::StoredRow);
const ChunkRowMeta &row_meta = *get_meta();
int64_t bit_map_size = sql::ObBitVector::memory_size(row_meta.col_cnt_);
int64_t datum_size = sizeof(ObDatum) * cnt;
int64_t datum_size = sizeof(ObDatum) * cg_schema.column_cnt_;
int64_t data_size = 0;
int64_t offset_size = 0;
for (int64_t i = 0; i < cnt; ++i) {
for (int64_t i = 0; i < cg_schema.column_cnt_; ++i) {
if (row_meta.column_length_[i] != 0) {
data_size += row_meta.column_length_[i];
} else {
if (!storage_datums[i].is_null()) {
data_size += storage_datums[i].len_;
int64_t column_idx = cg_schema.column_idxs_ ? cg_schema.column_idxs_[i] : i;
if (!storage_datums[column_idx].is_null()) {
data_size += storage_datums[column_idx].len_;
}
}
}
@ -413,13 +418,14 @@ int ObCompactBlockWriter::ensure_write(const common::ObIArray<ObExpr*> &exprs, O
return ret;
}
int ObCompactBlockWriter::ensure_write(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
int ObCompactBlockWriter::ensure_write(const blocksstable::ObStorageDatum *storage_datums,
const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size)
{
int ret = OB_SUCCESS;
uint64_t row_size = 0;
if (OB_FAIL(get_row_stored_size(storage_datums, cnt, extra_size, row_size))) {
LOG_WARN("fail to get row_size", K(cnt), K(extra_size), K(row_size), K(ret));
if (OB_FAIL(get_row_stored_size(storage_datums, cg_schema, extra_size, row_size))) {
LOG_WARN("fail to get row_size", K(cg_schema), K(extra_size), K(row_size), K(ret));
} else if (OB_FAIL(ensure_write(row_size))) {
LOG_WARN("fail to call inner ensure write", K(ret));
}

8
src/sql/engine/basic/chunk_store/ob_compact_block_writer.h
View File

@ -104,7 +104,7 @@ public:
// if full, construct the block and use the block's block_mgr. return block
virtual int add_row(const ObChunkDatumStore::StoredRow &src_sr, ObChunkDatumStore::StoredRow **dst_sr = nullptr) override;
//virtual int try_add_row(const common::ObIArray<ObExpr*> &exprs, ObEvalCtx &ctx);
virtual int add_row(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
virtual int add_row(const blocksstable::ObStorageDatum *storage_datums, const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size, ObChunkDatumStore::StoredRow **stored_row) override;
virtual int add_batch(const common::ObDatum **datums, const common::ObIArray<ObExpr *> &exprs,
@ -127,14 +127,14 @@ protected:
*/
int ensure_write(const common::ObIArray<ObExpr*> &exprs, ObEvalCtx &ctx);
int ensure_write(const ObChunkDatumStore::StoredRow &stored_row);
int ensure_write(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
int ensure_write(const blocksstable::ObStorageDatum *storage_datums, const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size);
int ensure_write(const int64_t size);
// get the stored size in writer buffer for a row.
int get_row_stored_size(const common::ObIArray<ObExpr*> &exprs, ObEvalCtx &ctx, uint64_t &size);
int get_row_stored_size(const ObChunkDatumStore::StoredRow &sr, uint64_t &size);
int get_row_stored_size(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
int get_row_stored_size(const blocksstable::ObStorageDatum *storage_datums, const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size, uint64_t &size);
private:
@ -149,7 +149,7 @@ private:
template <typename T>
int inner_add_row(const common::ObIArray<ObExpr*> &exprs, ObEvalCtx &ctx);
template <typename T>
int inner_add_row(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
int inner_add_row(const blocksstable::ObStorageDatum *storage_datums, const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size, ObChunkDatumStore::StoredRow **stored_row);
inline int ensure_init()

10
src/sql/engine/basic/chunk_store/ob_compact_store.cpp
View File

@ -72,7 +72,7 @@ int ObCompactStore::inner_get_next_row(const ObChunkDatumStore::StoredRow *&sr)
ret = OB_ITER_END;
} else if (OB_FAIL(block_reader_.get_block(cur_blk_id_, tmp_blk))) {
if (ret != OB_ITER_END) {
LOG_WARN("fail to get block", K(ret));
LOG_WARN("fail to get block", K(ret), K(cur_blk_id_));
}
} else {
start_iter_ = true;
@ -83,12 +83,12 @@ int ObCompactStore::inner_get_next_row(const ObChunkDatumStore::StoredRow *&sr)
if (OB_FAIL(ret)) {
} else if (OB_FAIL(reader_->get_row(sr))) {
if (ret != OB_ITER_END) {
LOG_WARN("fail to get row", K(ret));
LOG_WARN("fail to get row", K(ret), K(cur_blk_id_));
} else if (cur_blk_id_ >= get_block_id_cnt()) {
ret = OB_ITER_END;
} else if (OB_FAIL(block_reader_.get_block(cur_blk_id_, tmp_blk))) {
if (ret != OB_ITER_END) {
LOG_WARN("fail to get block", K(ret));
LOG_WARN("fail to get block", K(ret), K(cur_blk_id_));
}
} else {
reader_->reuse();
@ -258,12 +258,12 @@ int ObCompactStore::add_row(const common::ObIArray<ObExpr *> &exprs, ObEvalCtx &
return ret;
}
int ObCompactStore::add_row(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
int ObCompactStore::add_row(const blocksstable::ObDatumRow &datum_row, const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size, ObChunkDatumStore::StoredRow **stored_row)
{
int ret = OB_SUCCESS;
if (inited_) {
if (OB_FAIL(writer_->add_row(storage_datums, cnt, extra_size, stored_row))) {
if (OB_FAIL(writer_->add_row(datum_row.storage_datums_, cg_schema, extra_size, stored_row))) {
LOG_WARN("fail to add row", K(ret));
} else {
row_cnt_++;

2
src/sql/engine/basic/chunk_store/ob_compact_store.h
View File

@ -76,7 +76,7 @@ public:
int add_row(const common::ObIArray<ObExpr *> &exprs, ObEvalCtx &ctx, ObChunkDatumStore::StoredRow **stored_row = nullptr);
int add_row(const ObChunkDatumStore::StoredRow &src_sr, ObChunkDatumStore::StoredRow **dst_sr = nullptr);
// for chunkslicestore.
int add_row(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
int add_row(const blocksstable::ObDatumRow &datum_row, const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size, ObChunkDatumStore::StoredRow **stored_row = nullptr);
int get_next_row(const ObChunkDatumStore::StoredRow *&sr);

45
src/sql/engine/basic/chunk_store/ob_default_block_writer.cpp
View File

@ -65,16 +65,16 @@ int ObDefaultBlockWriter::add_row(const ObChunkDatumStore::StoredRow &src_sr, Ob
return ret;
}
int ObDefaultBlockWriter::add_row(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
int ObDefaultBlockWriter::add_row(const blocksstable::ObStorageDatum *storage_datums, const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size, ObChunkDatumStore::StoredRow **stored_row)
{
int ret = OB_SUCCESS;
if (OB_FAIL(ensure_init())) {
LOG_WARN("fail to ensure init", K(ret));
} else if (OB_FAIL(ensure_write(storage_datums, cnt, extra_size))) {
} else if (OB_FAIL(ensure_write(storage_datums, cg_schema, extra_size))) {
LOG_WARN("fail to ensure write", K(ret));
} else if (OB_FAIL(inner_add_row(storage_datums, cnt, extra_size, stored_row))) {
LOG_WARN("add row to block failed", K(ret), K(storage_datums), K(cnt), K(extra_size));
} else if (OB_FAIL(inner_add_row(storage_datums, cg_schema, extra_size, stored_row))) {
LOG_WARN("add row to block failed", K(ret), K(storage_datums), K(cg_schema), K(extra_size));
}
return ret;
}
@ -210,30 +210,32 @@ int ObDefaultBlockWriter::add_batch(const common::ObDatum **datums, const common
return ret;
}
int ObDefaultBlockWriter::inner_add_row(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
int ObDefaultBlockWriter::inner_add_row(const blocksstable::ObStorageDatum *storage_datums, const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size, ObChunkDatumStore::StoredRow **dst_sr)
{
int ret = OB_SUCCESS;
int64_t head_size = sizeof(ObChunkDatumStore::StoredRow);
int64_t datum_size = sizeof(ObDatum) * cnt;
int64_t datum_size = sizeof(ObDatum) * cg_schema.column_cnt_;
int64_t row_size = head_size + datum_size + extra_size;
ObChunkDatumStore::StoredRow *sr = static_cast<ObChunkDatumStore::StoredRow *>((void*)get_cur_buf());
if (OB_ISNULL(sr)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("fail to get buffer", K(ret));
} else {
sr->cnt_ = cnt;
for (int64_t i = 0; i < cnt; ++i) {
const ObDatum *tmp_datum = static_cast<const ObDatum *>(&storage_datums[i]);
sr->cnt_ = cg_schema.column_cnt_;
for (int64_t i = 0; i < cg_schema.column_cnt_; ++i) {
int64_t column_idx = cg_schema.column_idxs_ ? cg_schema.column_idxs_[i] : i;
const ObDatum *tmp_datum = static_cast<const ObDatum *>(&storage_datums[column_idx]);
MEMCPY(sr->payload_ + i * sizeof(ObDatum), tmp_datum, sizeof(ObDatum));
}
char* data_start = sr->payload_ + datum_size + extra_size;
int64_t pos = 0;
for (int64_t i = 0; i < cnt; ++i) {
MEMCPY(data_start + pos, storage_datums[i].ptr_, storage_datums[i].len_);
for (int64_t i = 0; i < cg_schema.column_cnt_; ++i) {
int64_t column_idx = cg_schema.column_idxs_ ? cg_schema.column_idxs_[i] : i;
MEMCPY(data_start + pos, storage_datums[column_idx].ptr_, storage_datums[column_idx].len_);
sr->cells()[i].ptr_ = data_start + pos;
pos += storage_datums[i].len_;
row_size += storage_datums[i].len_;
pos += storage_datums[column_idx].len_;
row_size += storage_datums[column_idx].len_;
}
sr->row_size_ = row_size;
if (OB_FAIL(advance(row_size))) {
@ -364,13 +366,14 @@ int ObDefaultBlockWriter::prepare_blk_for_write(ObTempBlockStore::Block *blk)
return ret;
}
int ObDefaultBlockWriter::ensure_write(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
int ObDefaultBlockWriter::ensure_write(const blocksstable::ObStorageDatum *storage_datums,
const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size)
{
int ret = OB_SUCCESS;
uint64_t row_size;
if (OB_FAIL(get_row_stored_size(storage_datums, cnt, extra_size, row_size))) {
LOG_WARN("fail to get row_size", K(cnt), K(extra_size), K(row_size), K(ret));
if (OB_FAIL(get_row_stored_size(storage_datums, cg_schema, extra_size, row_size))) {
LOG_WARN("fail to get row_size", K(cg_schema), K(extra_size), K(row_size), K(ret));
} else if (OB_FAIL(ensure_write(row_size))) {
LOG_WARN("fail to call inner ensure write", K(ret));
}
@ -378,15 +381,17 @@ int ObDefaultBlockWriter::ensure_write(const blocksstable::ObStorageDatum *stora
return ret;
}
int ObDefaultBlockWriter::get_row_stored_size(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
int ObDefaultBlockWriter::get_row_stored_size(const blocksstable::ObStorageDatum *storage_datums,
const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size, uint64_t &size)
{
int ret = OB_SUCCESS;
int64_t head_size = sizeof(ObChunkDatumStore::StoredRow);
int64_t datum_size = sizeof(ObDatum) * cnt;
int64_t datum_size = sizeof(ObDatum) * cg_schema.column_cnt_;
int64_t data_size = 0;
for (int64_t i = 0; i < cnt; ++i) {
data_size += storage_datums[i].len_;
for (int64_t i = 0; i < cg_schema.column_cnt_; ++i) {
int64_t column_idx = cg_schema.column_idxs_ ? cg_schema.column_idxs_[i] : i;
data_size += storage_datums[column_idx].len_;
}
size = head_size + datum_size + extra_size + data_size;
return ret;

10
src/sql/engine/basic/chunk_store/ob_default_block_writer.h
View File

@ -39,7 +39,7 @@ public:
void reset() {}
int add_row(const common::ObIArray<ObExpr*> &exprs, ObEvalCtx &ctx, ObChunkDatumStore::StoredRow **stored_row = nullptr) override;
int add_row(const ObChunkDatumStore::StoredRow &src_sr, ObChunkDatumStore::StoredRow **dst_sr = nullptr) override;
int add_row(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
int add_row(const blocksstable::ObStorageDatum *storage_datums, const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size, ObChunkDatumStore::StoredRow **stored_row) override;
int close() override;
int add_batch(const common::ObDatum **datums, const common::ObIArray<ObExpr *> &exprs,
@ -64,7 +64,8 @@ private:
*/
int ensure_write(const common::ObIArray<ObExpr*> &exprs, ObEvalCtx &ctx);
int ensure_write(const ObChunkDatumStore::StoredRow &stored_row);
int ensure_write(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
int ensure_write(const blocksstable::ObStorageDatum *storage_datums,
const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size);
// before dump the block we need to unswizzling each row;
@ -73,7 +74,8 @@ private:
int ensure_write(const int64_t size);
// get the stored size in writer buffer for a row.
int get_row_stored_size(const common::ObIArray<ObExpr*> &exprs, ObEvalCtx &ctx, uint64_t &size);
int get_row_stored_size(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
int get_row_stored_size(const blocksstable::ObStorageDatum *storage_datums,
const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size, uint64_t &size);
inline int ensure_init()
{
@ -84,7 +86,7 @@ private:
return ret;
}
int inner_add_row(const common::ObIArray<ObExpr*> &exprs, ObEvalCtx &ctx, ObChunkDatumStore::StoredRow **stored_row = nullptr);
int inner_add_row(const blocksstable::ObStorageDatum *storage_datums, const int64_t cnt,
int inner_add_row(const blocksstable::ObStorageDatum *storage_datums, const ObStorageColumnGroupSchema &cg_schema,
const int64_t extra_size, ObChunkDatumStore::StoredRow **dst_sr);
};

3
src/sql/engine/basic/ob_temp_block_store.cpp
View File

@ -1023,6 +1023,9 @@ int ObTempBlockStore::ensure_reader_buffer(BlockReader &reader, ShrinkBuffer &bu
free_blk_mem(try_reuse_blk);
try_reuse_blk = NULL;
}
} else if (try_reuse_blk != NULL) {
free_blk_mem(try_reuse_blk);
try_reuse_blk = NULL;
}
}
return ret;

6
src/storage/ddl/ob_direct_insert_sstable_ctx_new.cpp
View File

@ -1102,7 +1102,7 @@ void ObTabletDirectLoadBuildCtx::reset_slice_ctx_on_demand()
ObTabletDirectLoadMgr::ObTabletDirectLoadMgr()
: is_inited_(false), is_schema_item_ready_(false), ls_id_(), tablet_id_(), table_key_(), data_format_version_(0),
lock_(), ref_cnt_(0), direct_load_type_(ObDirectLoadType::DIRECT_LOAD_INVALID), sqc_build_ctx_(),
column_items_(), lob_column_idxs_(), lob_col_types_(), tablet_handle_(), schema_item_()
column_items_(), lob_column_idxs_(), lob_col_types_(), tablet_handle_(), schema_item_(), dir_id_(0)
{
column_items_.set_attr(ObMemAttr(MTL_ID(), "DL_schema"));
lob_column_idxs_.set_attr(ObMemAttr(MTL_ID(), "DL_schema"));
@ -1172,6 +1172,8 @@ int ObTabletDirectLoadMgr::update(
if (OB_SUCC(ret)) {
if (OB_FAIL(sqc_build_ctx_.build_param_.assign(build_param))) {
LOG_WARN("assign build param failed", K(ret));
} else if (OB_FAIL(FILE_MANAGER_INSTANCE_V2.alloc_dir(dir_id_))) {
LOG_WARN("alloc dir id failed", K(ret));
} else {
ls_id_ = build_param.common_param_.ls_id_;
tablet_id_ = build_param.common_param_.tablet_id_;
@ -1361,7 +1363,7 @@ int ObTabletDirectLoadMgr::fill_sstable_slice(
ret = OB_ERR_UNEXPECTED;
LOG_WARN("unexpected err", K(ret), K(slice_info), K(is_schema_item_ready_));
} else if (OB_FAIL(slice_writer->fill_sstable_slice(start_scn, sqc_build_ctx_.build_param_.runtime_only_param_.table_id_, tablet_id_,
iter, schema_item_, direct_load_type_, column_items_, affected_rows, insert_monitor))) {
tablet_handle_, iter, schema_item_, direct_load_type_, column_items_, dir_id_, affected_rows, insert_monitor))) {
LOG_WARN("fill sstable slice failed", K(ret), KPC(this));
}
}

1
src/storage/ddl/ob_direct_insert_sstable_ctx_new.h
View File

@ -390,6 +390,7 @@ protected:
ObArray<common::ObObjMeta> lob_col_types_;
ObTabletHandle tablet_handle_;
ObTableSchemaItem schema_item_;
int64_t dir_id_;
};
class ObTabletFullDirectLoadMgr final : public ObTabletDirectLoadMgr

252
src/storage/ddl/ob_direct_load_struct.cpp
View File

@ -23,6 +23,7 @@
#include "sql/engine/pdml/static/ob_px_sstable_insert_op.h"
#include "storage/ddl/ob_direct_insert_sstable_ctx_new.h"
#include "storage/lob/ob_lob_util.h"
#include "storage/tablet/ob_tablet.h"
#include "sql/engine/expr/ob_expr_lob_utils.h"
#include "sql/das/ob_das_utils.h"
#include "sql/engine/basic/chunk_store/ob_compact_store.h"
@ -355,25 +356,18 @@ ObTabletDDLParam::~ObTabletDDLParam()
}
int ObChunkSliceStore::init(const int64_t rowkey_column_count, ObArenaAllocator &allocator,
const ObIArray<ObColumnSchemaItem> &col_array,
common::ObCompressorType compress_type)
int ObChunkSliceStore::init(const int64_t rowkey_column_count, ObTabletHandle &tablet_handle,
ObArenaAllocator &allocator, const ObIArray<ObColumnSchemaItem> &col_array, const int64_t dir_id)
{
int ret = OB_SUCCESS;
const int64_t chunk_mem_limit = 2 * 1024L * 1024L; // 2M
if (OB_UNLIKELY(is_inited_)) {
ret = OB_INIT_TWICE;
LOG_WARN("init twice", K(ret));
} else if (OB_UNLIKELY(rowkey_column_count <= 0)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalida argument", K(ret), K(rowkey_column_count));
} else if (OB_FAIL(datum_store_.init(chunk_mem_limit, col_array, MTL_ID(), ObCtxIds::DEFAULT_CTX_ID,
"DL_SLICE_STORE", true/*enable_dump*/, 0, false/*disable truncate*/,
compress_type == NONE_COMPRESSOR ? SORT_COMPACT_LEVEL : SORT_COMPRESSION_COMPACT_LEVEL,
compress_type))) {
LOG_WARN("failed to init chunk datum store", K(ret));
} else if (OB_FAIL(datum_store_.alloc_dir_id())) {
LOG_WARN("failed to alloc dir id", K(ret));
} else if (OB_FAIL(prepare_datum_stores(MTL_ID(), tablet_handle, allocator, col_array, dir_id))) {
LOG_WARN("fail to prepare datum stores");
} else {
arena_allocator_ = &allocator;
rowkey_column_count_ = rowkey_column_count;
@ -383,6 +377,112 @@ int ObChunkSliceStore::init(const int64_t rowkey_column_count, ObArenaAllocator
return ret;
}
void ObChunkSliceStore::reset()
{
int ret = OB_SUCCESS;
if (OB_NOT_NULL(arena_allocator_)) {
for (int64_t i = 0; OB_SUCC(ret) && i < datum_stores_.count(); ++i) {
sql::ObCompactStore *cur_store = datum_stores_.at(i);
cur_store->~ObCompactStore();
arena_allocator_->free(cur_store);
cur_store = nullptr;
}
}
datum_stores_.reset();
cg_schemas_.reset();
endkey_.reset();
target_store_idx_ = -1;
row_cnt_ = 0;
arena_allocator_ = nullptr;
is_inited_ = false;
}
int64_t ObChunkSliceStore::calc_chunk_limit(const ObStorageColumnGroupSchema &cg_schema)
{
const int64_t basic_column_cnt = 10;
const int64_t basic_chunk_memory_limit = 512L * 1024L; // 512KB
return ((cg_schema.column_cnt_ / basic_column_cnt) + 1) * basic_chunk_memory_limit;
}
int ObChunkSliceStore::prepare_datum_stores(const uint64_t tenant_id, ObTabletHandle &tablet_handle, ObIAllocator &allocator, const ObIArray<ObColumnSchemaItem> &col_array, const int64_t dir_id)
{
int ret = OB_SUCCESS;
const int64_t chunk_mem_limit = 64 * 1024L; // 64K
ObCompactStore *datum_store = nullptr;
void *buf = nullptr;
if (OB_UNLIKELY(tenant_id <= 0)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(tenant_id));
} else {
ObStorageSchema *storage_schema = nullptr;
if (OB_UNLIKELY(!tablet_handle.is_valid())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid tablet handle", K(ret), K(tablet_handle));
} else if (OB_FAIL(tablet_handle.get_obj()->load_storage_schema(allocator, storage_schema))) {
LOG_WARN("load storage schema failed", K(ret), K(tablet_handle));
} else {
const ObIArray<ObStorageColumnGroupSchema> &cg_schemas = storage_schema->get_column_groups();
for (int64_t i = 0; OB_SUCC(ret) && i < cg_schemas.count(); ++i) {
const ObStorageColumnGroupSchema &cur_cg_schema = cg_schemas.at(i);
ObCompressorType compressor_type = cur_cg_schema.compressor_type_;
compressor_type = NONE_COMPRESSOR == compressor_type ? (CS_ENCODING_ROW_STORE == cur_cg_schema.row_store_type_ ? ZSTD_1_3_8_COMPRESSOR : NONE_COMPRESSOR) : compressor_type;
if (cur_cg_schema.is_rowkey_column_group() || cur_cg_schema.is_all_column_group()) {
target_store_idx_ = i;
}
if (OB_ISNULL(buf = allocator.alloc(sizeof(ObCompactStore)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("allocate memory failed", K(ret));
} else {
datum_store = new (buf) ObCompactStore();
ObArray<ObColumnSchemaItem> cur_column_items;
cur_column_items.set_attr(ObMemAttr(tenant_id, "tmp_cg_item"));
for (int64_t j = 0; OB_SUCC(ret) && j < cur_cg_schema.column_cnt_; ++j) {
int64_t column_idx = cur_cg_schema.column_idxs_ ? cur_cg_schema.column_idxs_[j] : j; // all_cg column_idxs_ = null
if (column_idx >= col_array.count()) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("invalid column idex", K(ret), K(column_idx), K(col_array.count()), K(i), K(cur_cg_schema));
} else if (OB_FAIL(cur_column_items.push_back(col_array.at(column_idx)))) {
LOG_WARN("fail to push_back col_item", K(ret));
}
}
if (OB_FAIL(ret)) {
} else if (OB_FAIL(datum_store->init(chunk_mem_limit, cur_column_items, tenant_id, ObCtxIds::DEFAULT_CTX_ID,
"DL_SLICE_STORE", true/*enable_dump*/, 0, false/*disable truncate*/,
compressor_type == NONE_COMPRESSOR ? SORT_COMPACT_LEVEL : SORT_COMPRESSION_COMPACT_LEVEL,
compressor_type))) {
LOG_WARN("failed to init chunk datum store", K(ret));
} else {
datum_store->set_dir_id(dir_id);
LOG_INFO("set dir id", K(dir_id));
}
if (OB_SUCC(ret)) {
if (OB_FAIL(datum_stores_.push_back(datum_store))) {
LOG_WARN("fail to push back datum_store", K(ret));
}
}
if (OB_FAIL(ret)) {
if (OB_NOT_NULL(datum_store)) {
datum_store->~ObCompactStore();
allocator.free(datum_store);
datum_store = nullptr;
}
}
}
}
if (OB_SUCC(ret)) {
if (OB_FAIL(cg_schemas_.assign(cg_schemas))) {
LOG_WARN("fail to copy cg schemas", K(ret));
}
}
}
ObTabletObjLoadHelper::free(allocator, storage_schema);
}
LOG_INFO("init ObChunkSliceStore", K(*this));
return ret;
}
int ObChunkSliceStore::append_row(const blocksstable::ObDatumRow &datum_row)
{
int ret = OB_SUCCESS;
@ -392,8 +492,17 @@ int ObChunkSliceStore::append_row(const blocksstable::ObDatumRow &datum_row)
} else if (OB_UNLIKELY(!datum_row.is_valid() || datum_row.get_column_count() < rowkey_column_count_)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(datum_row), K(rowkey_column_count_));
} else if (OB_FAIL(datum_store_.add_row(datum_row.storage_datums_, datum_row.get_column_count(), 0/*extra_size*/))) {
LOG_WARN("chunk datum store add row failed", K(ret));
} else {
for (int64_t i = 0; OB_SUCC(ret) && i < cg_schemas_.count(); ++i) {
ObStorageColumnGroupSchema &cur_cg_schema = cg_schemas_.at(i);
sql::ObCompactStore *cur_store = datum_stores_.at(i);
if (OB_FAIL(cur_store->add_row(datum_row, cur_cg_schema, 0/*extra_size*/))) {
LOG_WARN("chunk datum store add row failed", K(ret), K(i), K(datum_row.get_column_count()), K(cur_cg_schema), K(cg_schemas_));
}
}
if (OB_SUCC(ret)) {
++row_cnt_;
}
}
return ret;
}
@ -404,9 +513,10 @@ int ObChunkSliceStore::close()
if (OB_UNLIKELY(!is_inited_)) {
ret = OB_NOT_INIT;
LOG_WARN("not init", K(ret));
} else if (datum_store_.get_row_cnt() > 0) { // save endkey
} else if (datum_stores_.count() > 0 && OB_NOT_NULL(datum_stores_.at(target_store_idx_)) && datum_stores_.at(target_store_idx_)->get_row_cnt() > 0) { // save endkey
const ObChunkDatumStore::StoredRow *stored_row = nullptr;
if (OB_FAIL(datum_store_.get_last_stored_row(stored_row))) {
ObCompactStore *target_store = datum_stores_.at(target_store_idx_);
if (OB_FAIL(target_store->get_last_stored_row(stored_row))) {
LOG_WARN("fail to get last stored row", K(ret));
} else if (OB_UNLIKELY(nullptr == stored_row || stored_row->cnt_ < rowkey_column_count_)) {
ret = OB_ERR_UNEXPECTED;
@ -430,10 +540,12 @@ int ObChunkSliceStore::close()
}
}
if (OB_SUCC(ret)) {
if (OB_FAIL(datum_store_.finish_add_row(true/*need_dump*/))) {
for (int64_t i = 0; OB_SUCC(ret) && i < datum_stores_.count(); ++i) {
if (OB_FAIL(datum_stores_.at(i)->finish_add_row(true/*need_dump*/))) {
LOG_WARN("finish add row failed", K(ret));
}
}
}
LOG_DEBUG("chunk slice store closed", K(ret), K(endkey_));
return ret;
}
@ -539,7 +651,8 @@ ObDirectLoadSliceWriter::~ObDirectLoadSliceWriter()
int ObDirectLoadSliceWriter::prepare_slice_store_if_need(
const int64_t schema_rowkey_column_num,
const bool is_column_store,
const ObCompressorType compress_type,
const int64_t dir_id,
ObTabletHandle &tablet_handle,
const SCN &start_scn)
{
int ret = OB_SUCCESS;
@ -552,15 +665,13 @@ int ObDirectLoadSliceWriter::prepare_slice_store_if_need(
if (is_column_store) {
need_column_store_ = true;
ObChunkSliceStore *chunk_slice_store = nullptr;
if (OB_FAIL(ret)) {
if (OB_UNLIKELY(!tablet_handle.is_valid())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid tablet handle", K(ret), K(tablet_handle));
} else if (OB_ISNULL(chunk_slice_store = OB_NEWx(ObChunkSliceStore, &allocator_))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("allocate memory for chunk slice store failed", K(ret));
} else if (OB_FAIL(chunk_slice_store->init(schema_rowkey_column_num +
ObMultiVersionRowkeyHelpper::get_extra_rowkey_col_cnt(), allocator_,
tablet_direct_load_mgr_->get_column_info(),
compress_type))) {
} else if (OB_FAIL(chunk_slice_store->init(schema_rowkey_column_num + ObMultiVersionRowkeyHelpper::get_extra_rowkey_col_cnt(), tablet_handle, allocator_, tablet_direct_load_mgr_->get_column_info(), dir_id))) {
LOG_WARN("init chunk slice store failed", K(ret));
} else {
slice_store_ = chunk_slice_store;
@ -754,6 +865,7 @@ int ObDirectLoadSliceWriter::fill_lob_into_macro_block(
info.trans_id_, info.seq_no_, timeout_ts, lob_inrow_threshold, info.src_tenant_id_, row_iter))) {
LOG_WARN("fail to prepare iters", K(ret), KP(row_iter), K(datum));
} else {
ObTabletHandle unused_tablet_handle; //lob no need to get storageschema with handle
while (OB_SUCC(ret)) {
const blocksstable::ObDatumRow *cur_row = nullptr;
if (OB_FAIL(THIS_WORKER.check_status())) {
@ -774,7 +886,7 @@ int ObDirectLoadSliceWriter::fill_lob_into_macro_block(
} else if (OB_FAIL(check_null(false/*is_index_table*/, ObLobMetaUtil::LOB_META_SCHEMA_ROWKEY_COL_CNT, *cur_row))) {
LOG_WARN("fail to check null value in row", KR(ret), KPC(cur_row));
} else if (OB_FAIL(prepare_slice_store_if_need(ObLobMetaUtil::LOB_META_SCHEMA_ROWKEY_COL_CNT,
false/*is_column_store*/, NONE_COMPRESSOR/*do not use compressort*/, start_scn))) {
false/*is_column_store*/, 1L/*unsued*/, unused_tablet_handle, start_scn))) {
LOG_WARN("prepare macro block writer failed", K(ret));
} else if (OB_FAIL(slice_store_->append_row(*cur_row))) {
LOG_WARN("macro block writer append row failed", K(ret), KPC(cur_row));
@ -803,10 +915,12 @@ int ObDirectLoadSliceWriter::fill_sstable_slice(
const SCN &start_scn,
const uint64_t table_id,
const ObTabletID &tablet_id,
ObTabletHandle &tablet_handle,
ObIStoreRowIterator *row_iter,
const ObTableSchemaItem &schema_item,
const ObDirectLoadType &direct_load_type,
const ObArray<ObColumnSchemaItem> &column_items,
const int64_t dir_id,
int64_t &affected_rows,
ObInsertMonitor *insert_monitor)
{
@ -857,8 +971,7 @@ int ObDirectLoadSliceWriter::fill_sstable_slice(
if (OB_FAIL(ret)) {
} else if (OB_FAIL(check_null(schema_item.is_index_table_, schema_item.rowkey_column_num_, *cur_row))) {
LOG_WARN("fail to check null value in row", KR(ret), KPC(cur_row));
} else if (OB_FAIL(prepare_slice_store_if_need(schema_item.rowkey_column_num_, schema_item.is_column_store_,
schema_item.compress_type_, start_scn))) {
} else if (OB_FAIL(prepare_slice_store_if_need(schema_item.rowkey_column_num_, schema_item.is_column_store_, dir_id, tablet_handle, start_scn))) {
LOG_WARN("prepare macro block writer failed", K(ret));
} else if (OB_FAIL(slice_store_->append_row(*cur_row))) {
if (is_full_direct_load_task && OB_ERR_PRIMARY_KEY_DUPLICATE == ret && schema_item.is_unique_index_) {
@ -978,65 +1091,31 @@ int ObDirectLoadSliceWriter::fill_column_group(const ObStorageSchema *storage_sc
LOG_WARN("fil cg task canceled", K(ret), K(is_canceled_));
} else {
const ObIArray<ObStorageColumnGroupSchema> &cg_schemas = storage_schema->get_column_groups();
const int64_t MAX_CO_BATCH_SIZE = 10; // todo @qilu: add opt hint for batch_cnt
ObArray<ObCOSliceWriter *> co_ddl_writers;
co_ddl_writers.set_attr(ObMemAttr(MTL_ID(), "DL_co_writers"));
ObTimeGuard tg("fill_column_group", 1000L * 1000L * 600L); // 10 mins
FLOG_INFO("[DDL_FILL_CG] fill column group start",
"tablet_id", tablet_direct_load_mgr_->get_tablet_id(),
"row_count", chunk_slice_store->get_row_count(),
"column_group_count", cg_schemas.count());
// 1. reserve writers
const int64_t batch_count = MIN(MAX_CO_BATCH_SIZE, cg_schemas.count());
for (int64_t i = 0; OB_SUCC(ret) && i < batch_count; ++i) {
ObCOSliceWriter *tmp_writer = nullptr;
if (OB_ISNULL(tmp_writer = OB_NEWx(ObCOSliceWriter, &allocator_))) {
ObCOSliceWriter *cur_writer = nullptr;
if (OB_ISNULL(cur_writer = OB_NEWx(ObCOSliceWriter, &allocator_))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("allocate memory for co writer failed", K(ret));
} else if (OB_FAIL(co_ddl_writers.reserve(batch_count))) {
LOG_WARN("fail to reserve writers array", K(ret), K(batch_count));
} else if (OB_FAIL(co_ddl_writers.push_back(tmp_writer))) {
LOG_WARN("push back co writer failed", K(ret));
tmp_writer->~ObCOSliceWriter();
allocator_.free(tmp_writer);
}
}
int64_t cg_idx = 0;
while (OB_SUCC(ret) && cg_idx < cg_schemas.count()) {
tg.click("batch_fill");
int64_t current_batch_count = batch_count;
for (int64_t i = 0; OB_SUCC(ret) && i < batch_count; ++i) {
if (cg_idx >= cg_schemas.count()) {
current_batch_count = i;
break;
} else {
// 2. rescan and write
for (int64_t cg_idx = 0; OB_SUCC(ret) && cg_idx < cg_schemas.count(); ++cg_idx) {
const ObStorageColumnGroupSchema &cg_schema = cg_schemas.at(cg_idx);
ObCOSliceWriter *cur_writer = co_ddl_writers.at(i);
cur_writer->reset();
if (OB_FAIL(cur_writer->init(storage_schema, cg_idx, tablet_direct_load_mgr_, start_seq_, row_offset_, start_scn))) {
LOG_WARN("init co ddl writer failed", K(ret), K(i), K(cg_idx), KPC(this));
LOG_WARN("init co ddl writer failed", K(ret), KPC(cur_writer), K(cg_idx), KPC(this));
} else {
++cg_idx;
}
}
}
if (OB_SUCC(ret)) {
// 2. rescan and write
sql::ObCompactStore * cur_datum_store = chunk_slice_store->datum_stores_.at(cg_idx);
const ObChunkDatumStore::StoredRow *stored_row = nullptr;
bool has_next = false;
chunk_slice_store->datum_store_.rescan();
int64_t begin_ts = ObTimeUtility::fast_current_time();
while (OB_SUCC(ret) && OB_SUCC(chunk_slice_store->datum_store_.has_next(has_next)) && has_next) {
int64_t row_count = 0;
if (row_count > 0 && row_count % (10L * 10000L) == 0) { // print log per 10w records
int64_t curr_ts = ObTimeUtility::fast_current_time();
FLOG_INFO("[DDL_FILL_CG] rescan and fill", "tablet_id", tablet_direct_load_mgr_->get_tablet_id(),
"start_cg_idx", cg_idx - current_batch_count,
K(current_batch_count), K(row_count), "cost_time_us", curr_ts - begin_ts);
begin_ts = curr_ts;
}
if (OB_FAIL(chunk_slice_store->datum_store_.get_next_row(stored_row))) {
while (OB_SUCC(ret) && OB_SUCC(cur_datum_store->has_next(has_next)) && has_next) {
if (OB_FAIL(cur_datum_store->get_next_row(stored_row))) {
if (OB_ITER_END == ret) {
ret = OB_SUCCESS;
break;
@ -1044,48 +1123,35 @@ int ObDirectLoadSliceWriter::fill_column_group(const ObStorageSchema *storage_sc
LOG_WARN("get next row failed", K(ret));
}
} else {
++row_count;
if (OB_NOT_NULL(insert_monitor)) {
insert_monitor->inserted_cg_row_cnt_ = insert_monitor->inserted_cg_row_cnt_ + current_batch_count;
insert_monitor->inserted_cg_row_cnt_ += 1;
}
for (int64_t i = 0; OB_SUCC(ret) && i < current_batch_count; ++i) {
ObCOSliceWriter *cur_writer = co_ddl_writers.at(i);
if (OB_FAIL(cur_writer->append_row(stored_row))) {
LOG_WARN("append row failed", K(ret), KPC(stored_row), K(row_count));
LOG_WARN("append row failed", K(ret), KPC(stored_row));
} else {
if (OB_NOT_NULL(insert_monitor)) {
insert_monitor->inserted_cg_row_cnt_ += 1;
}
}
}
}
}
if (OB_SUCC(ret)) {
// 3. close writers
for (int64_t i = 0; OB_SUCC(ret) && i < current_batch_count; ++i) {
ObCOSliceWriter *cur_writer = co_ddl_writers.at(i);
if (OB_FAIL(cur_writer->close())) {
LOG_WARN("close co ddl writer failed", K(ret));
}
}
}
FLOG_INFO("[DDL_FILL_CG] finish cg batch", "tablet_id", tablet_direct_load_mgr_->get_tablet_id(),
"next_cg_idx", cg_idx, "total_cg_count", cg_schemas.count(), K(current_batch_count));
}
tg.click("fill_end");
// 4. free writers, ignore ret
for (int64_t i = 0; i < co_ddl_writers.count(); ++i) {
ObCOSliceWriter *cur_writer = co_ddl_writers.at(i);
}
if (OB_NOT_NULL(cur_writer)) {
cur_writer->~ObCOSliceWriter();
allocator_.free(cur_writer);
}
}
co_ddl_writers.reset();
FLOG_INFO("[DDL_FILL_CG] fill column group finished",
"tablet_id", tablet_direct_load_mgr_->get_tablet_id(),
"row_count", chunk_slice_store->get_row_count(),
"column_group_count", cg_schemas.count(),
"time_cost_us", tg.get_diff());
"column_group_count", cg_schemas.count());
}
return ret;
}
@ -1182,21 +1248,15 @@ int ObCOSliceWriter::project_cg_row(const ObStorageColumnGroupSchema &cg_schema,
if (OB_UNLIKELY(!cg_schema.is_valid() || nullptr == stored_row)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(cg_schema), KP(stored_row));
} else if (cg_schema.column_cnt_ > stored_row->cnt_ || cg_row.get_column_count() != cg_schema.column_cnt_) {
} else if (cg_schema.column_cnt_ != stored_row->cnt_ || cg_row.get_column_count() != cg_schema.column_cnt_) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("column count not match", K(ret), K(stored_row->cnt_), K(cg_row), K(cg_schema));
} else {
for (int64_t i = 0; OB_SUCC(ret) && i < cg_schema.column_cnt_; ++i) {
int64_t column_idx = cg_schema.column_idxs_ ? cg_schema.column_idxs_[i] : i;
if (column_idx >= stored_row->cnt_) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("invalid column idex", K(ret));
} else {
const ObDatum &cur_datum = stored_row->cells()[column_idx];
const ObDatum &cur_datum = stored_row->cells()[i];
cg_row.storage_datums_[i].set_datum(cur_datum);
}
}
}
return ret;
}

33
src/storage/ddl/ob_direct_load_struct.h
View File

@ -287,7 +287,7 @@ public:
inline int64_t get_lob_slice_id() { return lob_slice_id_; }
inline share::ObTabletCacheInterval &get_lob_id_cache() { return lob_id_cache_; }
private:
static const int64_t AUTO_INC_CACHE_SIZE = 100000; // 10w.
static const int64_t AUTO_INC_CACHE_SIZE = 5000000; // 500w.
ObArenaAllocator lob_allocator_;
sql::ObPxMultiPartSSTableInsertOp *op_;
share::ObLSID ls_id_;
@ -398,19 +398,29 @@ public:
class ObChunkSliceStore : public ObTabletSliceStore
{
public:
ObChunkSliceStore() : is_inited_(false), arena_allocator_(nullptr), rowkey_column_count_(0) {}
virtual ~ObChunkSliceStore() {}
int init(const int64_t rowkey_column_count, ObArenaAllocator &allocator,
const ObIArray<ObColumnSchemaItem> &col_schema,
common::ObCompressorType compress_type = NONE_COMPRESSOR);
ObChunkSliceStore() : is_inited_(false), target_store_idx_(-1), row_cnt_(0), arena_allocator_(nullptr), cg_schemas_(), datum_stores_(), rowkey_column_count_(0)
{
cg_schemas_.set_attr(ObMemAttr(MTL_ID(), "ChunkSlicStoreC"));
datum_stores_.set_attr(ObMemAttr(MTL_ID(), "ChunkSlicStoreD"));
}
virtual ~ObChunkSliceStore() { reset(); }
int init(const int64_t rowkey_column_count, ObTabletHandle &tablet_handle, ObArenaAllocator &allocator,
const ObIArray<ObColumnSchemaItem> &col_schema, const int64_t dir_id);
virtual int append_row(const blocksstable::ObDatumRow &datum_row) override;
virtual int close() override;
virtual int64_t get_row_count() const { return datum_store_.get_row_cnt(); }
TO_STRING_KV(K(is_inited_), KP(arena_allocator_), K(datum_store_), K(endkey_), K(rowkey_column_count_));
void reset();
virtual int64_t get_row_count() const { return row_cnt_; }
TO_STRING_KV(K(is_inited_), K(target_store_idx_), K(row_cnt_), KP(arena_allocator_), K(datum_stores_), K(endkey_), K(rowkey_column_count_), K(cg_schemas_));
private:
int prepare_datum_stores(const uint64_t tenant_id, ObTabletHandle &tablet_handle, ObIAllocator &allocator, const ObIArray<ObColumnSchemaItem> &col_array, const int64_t dir_id);
int64_t calc_chunk_limit(const ObStorageColumnGroupSchema &cg_schema);
public:
bool is_inited_;
int64_t target_store_idx_;
int64_t row_cnt_;
ObArenaAllocator *arena_allocator_;
sql::ObCompactStore datum_store_;
ObArray<ObStorageColumnGroupSchema> cg_schemas_;
ObArray<sql::ObCompactStore *> datum_stores_;
blocksstable::ObDatumRowkey endkey_;
int64_t rowkey_column_count_;
};
@ -460,10 +470,12 @@ public:
const share::SCN &start_scn,
const uint64_t table_id,
const ObTabletID &curr_tablet_id,
ObTabletHandle &tablet_handle,
ObIStoreRowIterator *row_iter,
const ObTableSchemaItem &schema_item,
const ObDirectLoadType &direct_load_type,
const ObArray<ObColumnSchemaItem> &column_items,
const int64_t dir_id,
int64_t &affected_rows,
ObInsertMonitor *insert_monitor = NULL);
int fill_lob_sstable_slice(
@ -516,7 +528,8 @@ private:
int prepare_slice_store_if_need(
const int64_t schema_rowkey_column_num,
const bool is_slice_store,
const ObCompressorType compress_type,
const int64_t dir_id,
ObTabletHandle &tablet_handle,
const share::SCN &start_scn);
int report_unique_key_dumplicated(
const int ret_code,

140
unittest/storage/ddl/test_chunk_compact_store.cpp
View File

@ -720,78 +720,78 @@ TEST_F(TestCompactChunk, test_rescan_get_last_row_compact)
}
}
TEST_F(TestCompactChunk, test_rescan_add_storagedatum)
{
int ret = OB_SUCCESS;
ObCompactStore cs_chunk;
cs_chunk.init(1, 1,
ObCtxIds::DEFAULT_CTX_ID, "SORT_CACHE_CTX", true, 0, false/*disable trunc*/, share::SORT_COMPACT_LEVEL);
ChunkRowMeta row_meta(allocator_);
row_meta.col_cnt_ = COLUMN_CNT;
row_meta.fixed_cnt_ = 0;
row_meta.var_data_off_ = 0;
row_meta.column_length_.prepare_allocate(COLUMN_CNT);
row_meta.column_offset_.prepare_allocate(COLUMN_CNT);
for (int64_t i = 0; i < COLUMN_CNT; i++) {
if (i != COLUMN_CNT) {
row_meta.column_length_[i] = 0;
row_meta.column_offset_[i] = 0;
} else {
row_meta.column_length_[i] = 0;
row_meta.column_offset_[i] = 0;
}
}
cs_chunk.set_meta(&row_meta);
StoredRow **sr;
ret = row_generate_.get_stored_row_irregular(sr);
ASSERT_EQ(ret, OB_SUCCESS);
// TEST_F(TestCompactChunk, test_rescan_add_storagedatum)
// {
// int ret = OB_SUCCESS;
// ObCompactStore cs_chunk;
// cs_chunk.init(1, 1,
// ObCtxIds::DEFAULT_CTX_ID, "SORT_CACHE_CTX", true, 0, false/*disable trunc*/, share::SORT_COMPACT_LEVEL);
// ChunkRowMeta row_meta(allocator_);
// row_meta.col_cnt_ = COLUMN_CNT;
// row_meta.fixed_cnt_ = 0;
// row_meta.var_data_off_ = 0;
// row_meta.column_length_.prepare_allocate(COLUMN_CNT);
// row_meta.column_offset_.prepare_allocate(COLUMN_CNT);
// for (int64_t i = 0; i < COLUMN_CNT; i++) {
// if (i != COLUMN_CNT) {
// row_meta.column_length_[i] = 0;
// row_meta.column_offset_[i] = 0;
// } else {
// row_meta.column_length_[i] = 0;
// row_meta.column_offset_[i] = 0;
// }
// }
// cs_chunk.set_meta(&row_meta);
// StoredRow **sr;
// ret = row_generate_.get_stored_row_irregular(sr);
// ASSERT_EQ(ret, OB_SUCCESS);
char *buf = reinterpret_cast<char*>(sr);
int64_t pos = 0;
for (int64_t i = 0; OB_SUCC(ret) && i < BATCH_SIZE; i++) {
StoredRow *tmp_sr = (StoredRow *)(buf + pos);
ObStorageDatum ssr[COLUMN_CNT];
for (int64_t k = 0; OB_SUCC(ret) && k < COLUMN_CNT; k++) {
ssr[k].shallow_copy_from_datum(tmp_sr->cells()[k]);
}
ret = cs_chunk.add_row(ssr, COLUMN_CNT, 0);
ASSERT_EQ(ret, OB_SUCCESS);
pos += tmp_sr->row_size_;
// get last row
const StoredRow *cur_sr = nullptr;
ret = cs_chunk.get_last_stored_row(cur_sr);
ASSERT_EQ(ret, OB_SUCCESS);
int64_t res = 0;
for (int64_t k = 0; k < cur_sr->cnt_; k++) {
ObDatum cur_cell = cur_sr->cells()[k];
res += *(int64_t *)(cur_cell.ptr_);
}
OB_ASSERT(res == ((1024 * i * COLUMN_CNT) + ((COLUMN_CNT - 1) * COLUMN_CNT / 2)));
}
// char *buf = reinterpret_cast<char*>(sr);
// int64_t pos = 0;
// for (int64_t i = 0; OB_SUCC(ret) && i < BATCH_SIZE; i++) {
// StoredRow *tmp_sr = (StoredRow *)(buf + pos);
// ObStorageDatum ssr[COLUMN_CNT];
// for (int64_t k = 0; OB_SUCC(ret) && k < COLUMN_CNT; k++) {
// ssr[k].shallow_copy_from_datum(tmp_sr->cells()[k]);
// }
// ret = cs_chunk.add_row(ssr, COLUMN_CNT, 0);
// ASSERT_EQ(ret, OB_SUCCESS);
// pos += tmp_sr->row_size_;
// // get last row
// const StoredRow *cur_sr = nullptr;
// ret = cs_chunk.get_last_stored_row(cur_sr);
// ASSERT_EQ(ret, OB_SUCCESS);
// int64_t res = 0;
// for (int64_t k = 0; k < cur_sr->cnt_; k++) {
// ObDatum cur_cell = cur_sr->cells()[k];
// res += *(int64_t *)(cur_cell.ptr_);
// }
// OB_ASSERT(res == ((1024 * i * COLUMN_CNT) + ((COLUMN_CNT - 1) * COLUMN_CNT / 2)));
// }
ret = cs_chunk.finish_add_row();
ASSERT_EQ(ret, OB_SUCCESS);
for (int j = 0; OB_SUCC(ret) && j < 2; j++ ) {
int64_t total_res = 0;
cs_chunk.rescan();
for (int64_t i = 0; OB_SUCC(ret) && i < BATCH_SIZE; i++) {
int64_t result = 0;
const StoredRow *cur_sr = nullptr;
ret = cs_chunk.get_next_row(cur_sr);
if (ret == OB_ITER_END) {
ret = OB_SUCCESS;
}
ASSERT_EQ(ret, OB_SUCCESS);
for (int64_t k = 0; k < cur_sr->cnt_; k++) {
ObDatum cur_cell = cur_sr->cells()[k];
result += *(int64_t *)(cur_cell.ptr_);
total_res += *(int64_t *)(cur_cell.ptr_);
}
OB_ASSERT(result == ((1024 * i * COLUMN_CNT) + ((COLUMN_CNT - 1) * COLUMN_CNT / 2)));
}
OB_ASSERT(total_res == ((1024 * (BATCH_SIZE-1) * BATCH_SIZE * COLUMN_CNT / 2) + BATCH_SIZE * ((COLUMN_CNT - 1) * COLUMN_CNT / 2)));
}
}
// ret = cs_chunk.finish_add_row();
// ASSERT_EQ(ret, OB_SUCCESS);
// for (int j = 0; OB_SUCC(ret) && j < 2; j++ ) {
// int64_t total_res = 0;
// cs_chunk.rescan();
// for (int64_t i = 0; OB_SUCC(ret) && i < BATCH_SIZE; i++) {
// int64_t result = 0;
// const StoredRow *cur_sr = nullptr;
// ret = cs_chunk.get_next_row(cur_sr);
// if (ret == OB_ITER_END) {
// ret = OB_SUCCESS;
// }
// ASSERT_EQ(ret, OB_SUCCESS);
// for (int64_t k = 0; k < cur_sr->cnt_; k++) {
// ObDatum cur_cell = cur_sr->cells()[k];
// result += *(int64_t *)(cur_cell.ptr_);
// total_res += *(int64_t *)(cur_cell.ptr_);
// }
// OB_ASSERT(result == ((1024 * i * COLUMN_CNT) + ((COLUMN_CNT - 1) * COLUMN_CNT / 2)));
// }
// OB_ASSERT(total_res == ((1024 * (BATCH_SIZE-1) * BATCH_SIZE * COLUMN_CNT / 2) + BATCH_SIZE * ((COLUMN_CNT - 1) * COLUMN_CNT / 2)));
// }
// }
}