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

[CP] disable progressive merge for normal cg major

Browse Source
This commit is contained in:
yangqise7en 2024-09-26 11:46:38 +00:00 committed by ob-robot
parent 0dc70f6600
commit 1df5c3224b
9 changed files with 44 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
src/rootserver/ob_ddl_service.cpp
View File

@ -13558,38 +13558,6 @@ int ObDDLService::update_tables_attribute(ObIArray<ObTableSchema*> &new_table_sc
return ret;
}
int ObDDLService::check_need_add_progressive_round(
const uint64_t tenant_data_version,
const ObTableSchema &table_schema,
const AlterTableSchema &alter_table_schema,
bool &need_add_progressive_round)
{
int ret = OB_SUCCESS;
need_add_progressive_round = true;
bool is_column_store_schema = false;
if (tenant_data_version < DATA_VERSION_4_3_3_0) {
// do nothing
} else if (OB_FAIL(table_schema.get_is_column_store(is_column_store_schema))) {
LOG_WARN("failed to get is column store", KR(ret));
} else if (is_column_store_schema) {
AlterColumnSchema *alter_column_schema = nullptr;
ObTableSchema::const_column_iterator it_begin = alter_table_schema.column_begin();
ObTableSchema::const_column_iterator it_end = alter_table_schema.column_end();
need_add_progressive_round = false;
for (;OB_SUCC(ret) && it_begin != it_end; it_begin++) {
if (OB_ISNULL(alter_column_schema = static_cast<AlterColumnSchema *>(*it_begin))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("*it_begin is NULL", K(ret));
} else if (OB_DDL_ADD_COLUMN != alter_column_schema->alter_type_) {
need_add_progressive_round = true;
break;
}
}
}
return ret;
}
//fix me :Check whether the newly added index column covers the partition column --by rongxuan.lc
// It can be repaired after the featrue that add index in alter_table statement
int ObDDLService::alter_table_in_trans(obrpc::ObAlterTableArg &alter_table_arg,
@ -13703,31 +13671,20 @@ int ObDDLService::alter_table_in_trans(obrpc::ObAlterTableArg &alter_table_arg,
bool need_update_index_table = false;
AlterLocalityOp alter_locality_op = ALTER_LOCALITY_OP_INVALID;
bool need_progressive_for_alter_column = false;
if (alter_table_arg.is_alter_columns_) {
if (OB_FAIL(check_need_add_progressive_round(
tenant_data_version,
*orig_table_schema,
alter_table_arg.alter_table_schema_,
need_progressive_for_alter_column))) {
LOG_WARN("failed to check need progressive round", KR(ret));
}
} else if (alter_table_arg.is_alter_options_
&& alter_table_arg.need_progressive_merge()) {
if (alter_table_arg.is_alter_columns_
|| (alter_table_arg.is_alter_options_
&& alter_table_arg.need_progressive_merge())) {
if (alter_table_arg.alter_table_schema_.alter_option_bitset_.
has_member(ObAlterTableArg::ENCRYPTION) &&
alter_table_arg.alter_table_schema_.is_equal_encryption(*orig_table_schema)) {
// If the values before and after changing the encryption algorithm in the table are the same,
// the merge is not marked
} else {
need_progressive_for_alter_column = true;
}
}
if (OB_SUCC(ret) && need_progressive_for_alter_column) {
alter_table_arg.is_alter_options_ = true;
alter_table_arg.alter_table_schema_.set_progressive_merge_round(orig_table_schema->get_progressive_merge_round() + 1);
if (OB_FAIL(alter_table_arg.alter_table_schema_.alter_option_bitset_.add_member(ObAlterTableArg::PROGRESSIVE_MERGE_ROUND))) {
LOG_WARN("fail to add member progressive merge round", K(ret));
alter_table_arg.is_alter_options_ = true;
alter_table_arg.alter_table_schema_.set_progressive_merge_round(orig_table_schema->get_progressive_merge_round() + 1);
if (OB_FAIL(alter_table_arg.alter_table_schema_.alter_option_bitset_.add_member(ObAlterTableArg::PROGRESSIVE_MERGE_ROUND))) {
LOG_WARN("fail to add member progressive merge round", K(ret));
}
}
}
if (OB_SUCC(ret)) {

5
src/rootserver/ob_ddl_service.h
View File

@ -1576,11 +1576,6 @@ private:
obrpc::ObAlterTableRes &res,
const uint64_t tenant_data_version);
int check_need_add_progressive_round(
const uint64_t tenant_data_version,
const ObTableSchema &table_schema,
const AlterTableSchema &alter_table_schema,
bool &need_add_progressive_round);
int need_modify_not_null_constraint_validate(const obrpc::ObAlterTableArg &alter_table_arg,
bool &is_add_not_null_col,
bool &need_modify) const;

7
src/storage/blocksstable/index_block/ob_index_block_builder.cpp
View File

@ -34,6 +34,7 @@
namespace oceanbase
{
ERRSIM_POINT_DEF(EN_COMPACTION_DISABLE_SHARED_MACRO);
using namespace common;
using namespace storage;
using namespace compaction;
@ -1606,6 +1607,12 @@ int ObSSTableIndexBuilder::close_with_macro_seq(
if (index_store_desc_.get_desc().is_cg() && res.row_count_ > 50000) {
tmp_mode = DISABLE;
}
#ifdef ERRSIM
if (EN_COMPACTION_DISABLE_SHARED_MACRO) {
tmp_mode = DISABLE;
FLOG_INFO("ERRSIM EN_COMPACTION_DISABLE_SHARED_MACRO", KR(ret));
}
#endif
switch (tmp_mode) {
case ENABLE:
if (OB_FAIL(check_and_rewrite_sstable(res))) {

1
src/storage/column_store/ob_co_merge_ctx.h
View File

@ -151,7 +151,6 @@ struct ObCOTabletMergeCtx : public ObBasicTabletMergeCtx
OB_INLINE bool is_build_row_store() const { return static_param_.is_build_row_store(); }
int get_cg_schema_for_merge(const int64_t idx, const ObStorageColumnGroupSchema *&cg_schema_ptr);
const ObSSTableMergeHistory &get_merge_history() { return dag_net_merge_history_; }
INHERIT_TO_STRING_KV("ObCOTabletMergeCtx", ObBasicTabletMergeCtx,
K_(array_count), K_(exe_stat));
virtual int mark_cg_finish(const int64_t start_cg_idx, const int64_t end_cg_idx) { return OB_SUCCESS; }

2
src/storage/column_store/ob_co_merge_writer.cpp
View File

@ -236,7 +236,7 @@ int ObCOMergeWriter::basic_init(const blocksstable::ObDatumRow &default_row,
iter_ = OB_NEWx(ObDefaultRowIter, (&allocator_), default_row_);
} else if (merge_param.is_full_merge() || sstable->is_small_sstable() || only_use_row_table) {
iter_ = OB_NEWx(ObPartitionRowMergeIter, (&allocator_), allocator_, iter_co_build_row_store_);
} else if (MICRO_BLOCK_MERGE_LEVEL == merge_param.static_param_.merge_level_) {
} else if (MICRO_BLOCK_MERGE_LEVEL == merge_param.static_param_.get_merge_level_for_sstable(*sstable)) {
iter_ = OB_NEWx(ObPartitionMicroMergeIter, (&allocator_), allocator_);
} else {
iter_ = OB_NEWx(ObPartitionMacroMergeIter, (&allocator_), allocator_);

18
src/storage/compaction/ob_basic_tablet_merge_ctx.cpp
View File

@ -26,6 +26,7 @@
#include "share/schema/ob_tenant_schema_service.h"
#include "storage/tablet/ob_mds_schema_helper.h"
#include "storage/tablet/ob_mds_scan_param_helper.h"
#include "observer/ob_server_event_history_table_operator.h"
namespace oceanbase
{
@ -296,6 +297,23 @@ bool ObStaticMergeParam::is_build_row_store() const
return ObCOMajorMergePolicy::is_build_row_store_merge(co_major_merge_type_);
}
ObMergeLevel ObStaticMergeParam::get_merge_level_for_sstable(
const ObSSTable &sstable) const
{
ObMergeLevel ret_merge_level = merge_level_;
if (!is_full_merge_ && data_version_ >= DATA_VERSION_4_3_3_0) { // expect full merge
if (MACRO_BLOCK_MERGE_LEVEL == ret_merge_level && sstable.is_cg_sstable()) {
ret_merge_level = MICRO_BLOCK_MERGE_LEVEL;
LOG_INFO("for cg sstable, ignore macro merge level when progressive", K(sstable), K(ret_merge_level));
#ifdef ERRSIM
SERVER_EVENT_SYNC_ADD("merge_errsim", "cg_disable_progressive", "tablet_id", get_tablet_id(),
"sstable", sstable.get_key());
#endif
}
}
return ret_merge_level;
}
/*
* ObCtxMergeInfoCollector
*/

1
src/storage/compaction/ob_basic_tablet_merge_ctx.h
View File

@ -53,6 +53,7 @@ struct ObStaticMergeParam final
is_full_merge_ = is_full_merge;
merge_level_ = MACRO_BLOCK_MERGE_LEVEL;
}
ObMergeLevel get_merge_level_for_sstable(const ObSSTable &sstable) const;
private:
int init_multi_version_column_descs();

3
src/storage/compaction/ob_partition_merge_iter.cpp
View File

@ -927,7 +927,8 @@ bool ObPartitionMicroMergeIter::inner_check(const ObMergeParameter &merge_param)
if (OB_UNLIKELY(!is_major_or_meta_merge_type(static_param.get_merge_type()))) {
bret = false;
LOG_WARN_RET(OB_ERR_UNEXPECTED, "Unexpected merge type for major micro merge iter", K(bret), K(merge_param));
} else if (OB_UNLIKELY(static_param.merge_level_ != MICRO_BLOCK_MERGE_LEVEL)) {
} else if (OB_UNLIKELY(NULL == table_ || !table_->is_sstable()
|| merge_param.static_param_.get_merge_level_for_sstable(*static_cast<ObSSTable *>(table_)) != MICRO_BLOCK_MERGE_LEVEL)) {
bret = false;
LOG_WARN_RET(OB_ERR_UNEXPECTED, "Unexpected merge level for major micro merge iter", K(bret), K(merge_param));
} else if (OB_UNLIKELY(static_param.is_full_merge_)) {

7
src/storage/compaction/ob_progressive_merge_helper.cpp
View File

@ -110,6 +110,10 @@ int64_t ObProgressiveMergeMgr::get_result_progressive_merge_step(
#endif
}
result_step = progressive_merge_num_;
#ifdef ERRSIM
SERVER_EVENT_SYNC_ADD("merge_errsim", "end_progressive", K(tablet_id),
K(result_step), K_(progressive_merge_num));
#endif
}
return result_step;
}
@ -155,6 +159,9 @@ int ObProgressiveMergeHelper::init(
if (OB_FAIL(collect_macro_info(sstable, merge_param, rewrite_macro_cnt, reduce_macro_cnt, rewrite_block_cnt_for_progressive))) {
LOG_WARN("Fail to scan secondary meta", K(ret), K(merge_param));
} else if (static_param.data_version_ >= DATA_VERSION_4_3_3_0 && sstable.is_cg_sstable()) {
need_rewrite_block_cnt_ = 0;
LOG_INFO("skip rewrite macro for progressive in cg sstable", "sstable", sstable.get_key(), K(rewrite_block_cnt_for_progressive));
} else if (need_calc_progressive_merge()) {
if (rewrite_block_cnt_for_progressive > 0) {
need_rewrite_block_cnt_ = MAX(rewrite_block_cnt_for_progressive /