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

[FEAT MERGE]: use parameter _ob_ddl_temp_file_compress_func to control compression in ddl.

Browse Source
This commit is contained in:
Monk-Liu
2024-03-07 04:45:49 +00:00
committed by ob-robot
parent 5279be2c40
commit 92fca7def1
32 changed files with 491 additions and 780 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1222,7 +1222,8 @@ int ObTabletDirectLoadMgr::open_sstable_slice(
} else if (OB_UNLIKELY(!is_valid())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("unexpected err", K(ret), KPC(this));
} else if (OB_FAIL(prepare_schema_item_on_demand(sqc_build_ctx_.build_param_.runtime_only_param_.table_id_))) {
} else if (OB_FAIL(prepare_schema_item_on_demand(sqc_build_ctx_.build_param_.runtime_only_param_.table_id_,
sqc_build_ctx_.build_param_.runtime_only_param_.parallel_))) {
LOG_WARN("prepare table schema item on demand", K(ret), K(sqc_build_ctx_.build_param_));
} else {
ObDirectLoadSliceWriter *slice_writer = nullptr;
@ -1247,7 +1248,8 @@ int ObTabletDirectLoadMgr::open_sstable_slice(
return ret;
}
int ObTabletDirectLoadMgr::prepare_schema_item_on_demand(const uint64_t table_id)
int ObTabletDirectLoadMgr::prepare_schema_item_on_demand(const uint64_t table_id,
const int64_t parallel)
{
int ret = OB_SUCCESS;
bool is_schema_item_ready = ATOMIC_LOAD(&is_schema_item_ready_);
@ -1314,19 +1316,6 @@ int ObTabletDirectLoadMgr::prepare_schema_item_on_demand(const uint64_t table_id
}
}
}
if (OB_SUCC(ret)) {
// get compress type
uint64_t tenant_id = table_schema->get_tenant_id();
omt::ObTenantConfigGuard tenant_config(TENANT_CONF(tenant_id));
if (OB_UNLIKELY(!tenant_config.is_valid())) {
//tenant config获取失败时，租户不存在;返回默认值
ret = OB_ERR_UNEXPECTED;
LOG_WARN("fail get tenant_config", K(ret), K(tenant_id));
} else if (tenant_config->enable_store_compression) {
schema_item_.compress_type_ = table_schema->get_compressor_type();
}
LOG_INFO("load compress type is:", K(schema_item_.compress_type_), K(tenant_config->enable_store_compression));
}
}
if (OB_SUCC(ret)) {
is_schema_item_ready_ = true;
@ -1378,7 +1367,8 @@ 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_,
tablet_handle_, iter, schema_item_, direct_load_type_, column_items_, dir_id_, affected_rows, insert_monitor))) {
tablet_handle_, iter, schema_item_, direct_load_type_, column_items_, dir_id_,
sqc_build_ctx_.build_param_.runtime_only_param_.parallel_, affected_rows, insert_monitor))) {
LOG_WARN("fill sstable slice failed", K(ret), KPC(this));
}
}

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

@ -29,6 +29,7 @@
#include "storage/ddl/ob_direct_load_struct.h"
#include "storage/meta_mem/ob_tablet_pointer.h"
#include "storage/tablet/ob_tablet_meta.h"
#include "src/share/ob_ddl_common.h"
namespace oceanbase
{
@ -374,7 +375,8 @@ public:
K_(direct_load_type), K_(sqc_build_ctx), KPC(lob_mgr_handle_.get_obj()), K_(schema_item), K_(column_items), K_(lob_column_idxs));
private:
int prepare_schema_item_on_demand(const uint64_t table_id);
int prepare_schema_item_on_demand(const uint64_t table_id,
const int64_t parallel);
void calc_cg_idx(const int64_t thread_cnt, const int64_t thread_id, int64_t &strat_idx, int64_t &end_idx);
int fill_aggregated_column_group(
const int64_t start_idx,

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

@ -357,7 +357,8 @@ ObTabletDDLParam::~ObTabletDDLParam()
}
int ObChunkSliceStore::init(const int64_t rowkey_column_count, ObTabletHandle &tablet_handle,
ObArenaAllocator &allocator, const ObIArray<ObColumnSchemaItem> &col_array, const int64_t dir_id)
ObArenaAllocator &allocator, const ObIArray<ObColumnSchemaItem> &col_array, const int64_t dir_id,
const int64_t parallelism)
{
int ret = OB_SUCCESS;
if (OB_UNLIKELY(is_inited_)) {
@ -366,7 +367,7 @@ int ObChunkSliceStore::init(const int64_t rowkey_column_count, ObTabletHandle &t
} 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(prepare_datum_stores(MTL_ID(), tablet_handle, allocator, col_array, dir_id))) {
} else if (OB_FAIL(prepare_datum_stores(MTL_ID(), tablet_handle, allocator, col_array, dir_id, parallelism))) {
LOG_WARN("fail to prepare datum stores");
} else {
arena_allocator_ = &allocator;
@ -404,7 +405,8 @@ int64_t ObChunkSliceStore::calc_chunk_limit(const ObStorageColumnGroupSchema &cg
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 ObChunkSliceStore::prepare_datum_stores(const uint64_t tenant_id, ObTabletHandle &tablet_handle, ObIAllocator &allocator,
const ObIArray<ObColumnSchemaItem> &col_array, const int64_t dir_id, const int64_t parallelism)
{
int ret = OB_SUCCESS;
const int64_t chunk_mem_limit = 64 * 1024L; // 64K
@ -427,6 +429,11 @@ int ObChunkSliceStore::prepare_datum_stores(const uint64_t tenant_id, ObTabletHa
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 (OB_FAIL(ObDDLUtil::get_temp_store_compress_type(compressor_type,
parallelism,
compressor_type))) {
LOG_WARN("fail to get temp store compress type", K(ret));
}
if (cur_cg_schema.is_rowkey_column_group() || cur_cg_schema.is_all_column_group()) {
target_store_idx_ = i;
}
@ -450,7 +457,6 @@ int ObChunkSliceStore::prepare_datum_stores(const uint64_t tenant_id, ObTabletHa
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 {
@ -677,6 +683,7 @@ int ObDirectLoadSliceWriter::prepare_slice_store_if_need(
const int64_t schema_rowkey_column_num,
const bool is_column_store,
const int64_t dir_id,
const int64_t parallelism,
ObTabletHandle &tablet_handle,
const SCN &start_scn)
{
@ -696,7 +703,8 @@ int ObDirectLoadSliceWriter::prepare_slice_store_if_need(
} 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(), tablet_handle, allocator_, tablet_direct_load_mgr_->get_column_info(), dir_id))) {
} 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, parallelism))) {
LOG_WARN("init chunk slice store failed", K(ret));
} else {
slice_store_ = chunk_slice_store;
@ -911,7 +919,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*/, 1L/*unsued*/, unused_tablet_handle, start_scn))) {
false/*is_column_store*/, 1L/*unsued*/, 1L/*unused*/, 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));
@ -946,6 +954,7 @@ int ObDirectLoadSliceWriter::fill_sstable_slice(
const ObDirectLoadType &direct_load_type,
const ObArray<ObColumnSchemaItem> &column_items,
const int64_t dir_id,
const int64_t parallelism,
int64_t &affected_rows,
ObInsertMonitor *insert_monitor)
{
@ -996,7 +1005,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_, dir_id, tablet_handle, start_scn))) {
} else if (OB_FAIL(prepare_slice_store_if_need(schema_item.rowkey_column_num_, schema_item.is_column_store_, dir_id, parallelism, 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_) {

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

@ -217,11 +217,11 @@ struct ObDirectInsertRuntimeOnlyParam final
{
public:
ObDirectInsertRuntimeOnlyParam()
: exec_ctx_(nullptr), task_id_(0), table_id_(OB_INVALID_ID), schema_version_(0), task_cnt_(0), need_online_opt_stat_gather_(false), trans_id_(), seq_no_(0)
: exec_ctx_(nullptr), task_id_(0), table_id_(OB_INVALID_ID), schema_version_(0), task_cnt_(0), need_online_opt_stat_gather_(false), trans_id_(), seq_no_(0), parallel_(1)
{}
~ObDirectInsertRuntimeOnlyParam() = default;
bool is_valid() const { return OB_INVALID_ID != task_id_ && OB_INVALID_ID != table_id_ && schema_version_ > 0 && task_cnt_ >= 0; }
TO_STRING_KV(KP_(exec_ctx), K_(task_id), K_(table_id), K_(schema_version), K_(task_cnt), K_(need_online_opt_stat_gather), K_(trans_id), K_(seq_no));
TO_STRING_KV(KP_(exec_ctx), K_(task_id), K_(table_id), K_(schema_version), K_(task_cnt), K_(need_online_opt_stat_gather), K_(trans_id), K_(seq_no), K_(parallel));
public:
sql::ObExecContext *exec_ctx_;
int64_t task_id_;
@ -237,6 +237,7 @@ public:
// sequence number for the incremental direct load,
// fixed 0 for the full direct load.
int64_t seq_no_; //
int64_t parallel_; // used to decide wehter need to use compress temp data in rescan task.
};
// full parameters used by runtime execution
@ -405,14 +406,15 @@ public:
}
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);
const ObIArray<ObColumnSchemaItem> &col_schema, const int64_t dir_id, const int64_t parallelism);
virtual int append_row(const blocksstable::ObDatumRow &datum_row) override;
virtual int close() override;
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);
int prepare_datum_stores(const uint64_t tenant_id, ObTabletHandle &tablet_handle, ObIAllocator &allocator,
const ObIArray<ObColumnSchemaItem> &col_array, const int64_t dir_id, const int64_t parallelism);
int64_t calc_chunk_limit(const ObStorageColumnGroupSchema &cg_schema);
public:
bool is_inited_;
@ -477,6 +479,7 @@ public:
const ObDirectLoadType &direct_load_type,
const ObArray<ObColumnSchemaItem> &column_items,
const int64_t dir_id,
const int64_t parallelism,
int64_t &affected_rows,
ObInsertMonitor *insert_monitor = NULL);
int fill_lob_sstable_slice(
@ -535,6 +538,7 @@ private:
const int64_t schema_rowkey_column_num,
const bool is_slice_store,
const int64_t dir_id,
const int64_t parallelism,
ObTabletHandle &tablet_handle,
const share::SCN &start_scn);
int report_unique_key_dumplicated(

1
src/storage/direct_load/ob_direct_load_insert_table_ctx.cpp
View File

@ -157,6 +157,7 @@ int ObDirectLoadInsertTabletContext::open()
direct_load_param.runtime_only_param_.table_id_ = param_.table_id_;
direct_load_param.runtime_only_param_.schema_version_ = param_.schema_version_;
direct_load_param.runtime_only_param_.task_cnt_ = 1; // default value.
direct_load_param.runtime_only_param_.parallel_ = param_.reserved_parallel_;
if (OB_FAIL(sstable_insert_mgr->create_tablet_direct_load(
param_.context_id_, param_.execution_id_, direct_load_param))) {
LOG_WARN("create tablet manager failed", K(ret));