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

[FEAT MERGE] column store ddl

Browse Source 
Co-authored-by: simonjoylet <simonjoylet@gmail.com>
Co-authored-by: Monk-Liu <1152761042@qq.com>
Co-authored-by: renju96 <fcbrenju@163.com>
This commit is contained in:
Charles0429
2024-02-07 23:09:01 +00:00
committed by ob-robot
parent fd359dc4a4
commit 1c20161f72
354 changed files with 29599 additions and 7741 deletions
Download Patch File Download Diff File Expand all files Collapse all files

168
src/share/ob_ddl_common.cpp
View File

@ -30,6 +30,7 @@
#include "storage/tx_storage/ob_ls_map.h"
#include "rootserver/ob_root_service.h"
#include "rootserver/ddl_task/ob_ddl_task.h"
#include "storage/column_store/ob_column_oriented_sstable.h"
using namespace oceanbase::share;
using namespace oceanbase::common;
@ -711,7 +712,8 @@ int ObDDLUtil::generate_build_replica_sql(
const bool use_heap_table_ddl_plan,
const bool use_schema_version_hint_for_src_table,
const ObColumnNameMap *col_name_map,
ObSqlString &sql_string)
ObSqlString &sql_string,
const SortCompactLevel compact_level)
{
int ret = OB_SUCCESS;
ObSchemaGetterGuard schema_guard;
@ -918,8 +920,6 @@ int ObDDLUtil::generate_build_replica_sql(
}
}
if (OB_SUCC(ret)) {
ObArenaAllocator allocator("ObDDLTmp");
ObString new_dest_database_name;
@ -962,8 +962,8 @@ int ObDDLUtil::generate_build_replica_sql(
}
if (OB_FAIL(ret)) {
} else if (oracle_mode) {
if (OB_FAIL(sql_string.assign_fmt("INSERT /*+ monitor enable_parallel_dml parallel(%ld) opt_param('ddl_execution_id', %ld) opt_param('ddl_task_id', %ld) opt_param('enable_newsort', 'false') use_px */INTO \"%.*s\".\"%.*s\"(%.*s) SELECT /*+ index(\"%.*s\" primary) %.*s */ %.*s from \"%.*s\".\"%.*s\" as of scn %ld %.*s",
real_parallelism, execution_id, task_id,
if (OB_FAIL(sql_string.assign_fmt("INSERT /*+ monitor enable_parallel_dml parallel(%ld) opt_param('ddl_execution_id', %ld) opt_param('ddl_task_id', %ld) opt_param('compact_sort_level', %ld) opt_param('enable_newsort', 'false') use_px */INTO \"%.*s\".\"%.*s\"(%.*s) SELECT /*+ index(\"%.*s\" primary) %.*s */ %.*s from \"%.*s\".\"%.*s\" as of scn %ld %.*s",
real_parallelism, execution_id, task_id, static_cast<int64_t>(compact_level),
static_cast<int>(new_dest_database_name.length()), new_dest_database_name.ptr(), static_cast<int>(new_dest_table_name.length()), new_dest_table_name.ptr(),
static_cast<int>(insert_column_sql_string.length()), insert_column_sql_string.ptr(),
static_cast<int>(new_source_table_name.length()), new_source_table_name.ptr(),
@ -974,8 +974,8 @@ int ObDDLUtil::generate_build_replica_sql(
LOG_WARN("fail to assign sql string", K(ret));
}
} else {
if (OB_FAIL(sql_string.assign_fmt("INSERT /*+ monitor enable_parallel_dml parallel(%ld) opt_param('ddl_execution_id', %ld) opt_param('ddl_task_id', %ld) opt_param('enable_newsort', 'false') use_px */INTO `%.*s`.`%.*s`(%.*s) SELECT /*+ index(`%.*s` primary) %.*s */ %.*s from `%.*s`.`%.*s` as of snapshot %ld %.*s",
real_parallelism, execution_id, task_id,
if (OB_FAIL(sql_string.assign_fmt("INSERT /*+ monitor enable_parallel_dml parallel(%ld) opt_param('ddl_execution_id', %ld) opt_param('ddl_task_id', %ld) opt_param('compact_sort_level', %ld), opt_param('enable_newsort', 'false') use_px */INTO `%.*s`.`%.*s`(%.*s) SELECT /*+ index(`%.*s` primary) %.*s */ %.*s from `%.*s`.`%.*s` as of snapshot %ld %.*s",
real_parallelism, execution_id, task_id, static_cast<int64_t>(compact_level),
static_cast<int>(new_dest_database_name.length()), new_dest_database_name.ptr(), static_cast<int>(new_dest_table_name.length()), new_dest_table_name.ptr(),
static_cast<int>(insert_column_sql_string.length()), insert_column_sql_string.ptr(),
static_cast<int>(new_source_table_name.length()), new_source_table_name.ptr(),
@ -1424,12 +1424,17 @@ int64_t ObDDLUtil::get_default_ddl_tx_timeout()
}
int ObDDLUtil::get_data_format_version(
int ObDDLUtil::get_data_information(
const uint64_t tenant_id,
const uint64_t task_id,
int64_t &data_format_version)
uint64_t &data_format_version,
int64_t &snapshot_version,
share::ObDDLTaskStatus &task_status)
{
int ret = OB_SUCCESS;
data_format_version = 0;
snapshot_version = 0;
task_status = share::ObDDLTaskStatus::PREPARE;
if (OB_UNLIKELY(OB_INVALID_ID == tenant_id || task_id <= 0
|| nullptr == GCTX.sql_proxy_)) {
ret = OB_INVALID_ARGUMENT;
@ -1440,7 +1445,7 @@ int ObDDLUtil::get_data_format_version(
SMART_VAR(ObMySQLProxy::MySQLResult, res) {
ObSqlString query_string;
sqlclient::ObMySQLResult *result = NULL;
if (OB_FAIL(query_string.assign_fmt(" SELECT ddl_type, UNHEX(message) as message_unhex FROM %s WHERE task_id = %lu",
if (OB_FAIL(query_string.assign_fmt(" SELECT snapshot_version, ddl_type, UNHEX(message) as message_unhex, status FROM %s WHERE task_id = %lu",
OB_ALL_DDL_TASK_STATUS_TNAME, task_id))) {
LOG_WARN("assign sql string failed", K(ret));
} else if (OB_FAIL(GCTX.sql_proxy_->read(res, tenant_id, query_string.ptr()))) {
@ -1452,10 +1457,14 @@ int ObDDLUtil::get_data_format_version(
LOG_WARN("get next row failed", K(ret));
} else {
int64_t pos = 0;
int cur_task_status = 0;
ObDDLType ddl_type = ObDDLType::DDL_INVALID;
ObString task_message;
EXTRACT_UINT_FIELD_MYSQL(*result, "snapshot_version", snapshot_version, uint64_t);
EXTRACT_INT_FIELD_MYSQL(*result, "ddl_type", ddl_type, ObDDLType);
EXTRACT_VARCHAR_FIELD_MYSQL(*result, "message_unhex", task_message);
EXTRACT_INT_FIELD_MYSQL(*result, "status", cur_task_status, int);
task_status = static_cast<share::ObDDLTaskStatus>(cur_task_status);
if (ObDDLType::DDL_CREATE_INDEX == ddl_type) {
SMART_VAR(rootserver::ObIndexBuildTask, task) {
if (OB_FAIL(task.deserlize_params_from_message(tenant_id, task_message.ptr(), task_message.length(), pos))) {
@ -2343,6 +2352,145 @@ int ObCheckTabletDataComplementOp::check_and_wait_old_complement_task(
return ret;
}
int ObCODDLUtil::get_base_cg_idx(const storage::ObStorageSchema *storage_schema, int64_t &base_cg_idx)
{
int ret = OB_SUCCESS;
base_cg_idx = -1;
if (OB_UNLIKELY(nullptr == storage_schema)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), KP(storage_schema));
} else {
bool found_base_cg_idx = false;
const ObIArray<ObStorageColumnGroupSchema> &cg_schemas = storage_schema->get_column_groups();
for (int64_t i = 0; OB_SUCC(ret) && !found_base_cg_idx && i < cg_schemas.count(); ++i) {
const ObStorageColumnGroupSchema &cur_cg_schmea = cg_schemas.at(i);
if (cur_cg_schmea.is_all_column_group() || cur_cg_schmea.is_rowkey_column_group()) {
base_cg_idx = i;
found_base_cg_idx = true;
}
}
if (OB_SUCC(ret) && !found_base_cg_idx) {
ret = OB_ENTRY_NOT_EXIST;
LOG_WARN("base columng group schema not found", K(ret));
}
}
LOG_DEBUG("get base cg idx", K(ret), K(base_cg_idx));
return ret;
}
int ObCODDLUtil::get_column_checksums(
const storage::ObCOSSTableV2 *co_sstable,
const storage::ObStorageSchema *storage_schema,
ObIArray<int64_t> &column_checksums)
{
int ret = OB_SUCCESS;
column_checksums.reset();
int64_t column_count = 0;
if (OB_UNLIKELY(nullptr == co_sstable || nullptr == storage_schema)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), KP(co_sstable), KP(storage_schema));
} else if (OB_FAIL(storage_schema->get_stored_column_count_in_sstable(column_count))) {
LOG_WARN("fail to get_stored_column_count_in_sstable", K(ret), KPC(storage_schema));
} else {
const common::ObIArray<ObStorageColumnGroupSchema> &column_groups = storage_schema->get_column_groups();
ObArray<bool/*checksum_ready*/> checksum_ready_array;
if (OB_FAIL(checksum_ready_array.reserve(column_count))) {
LOG_WARN("reserve checksum ready array failed", K(ret), K(column_count));
} else if (OB_FAIL(column_checksums.reserve(column_count))) {
LOG_WARN("reserve checksum array failed", K(ret), K(column_count));
}
for (int64_t i = 0; i < column_count && OB_SUCC(ret); i ++) {
if (OB_FAIL(checksum_ready_array.push_back(false))) {
LOG_WARN("push back ready flag failed", K(ret), K(i));
} else if (OB_FAIL(column_checksums.push_back(0))) {
LOG_WARN("fail to push back column checksum", K(ret), K(i));
}
}
ObSSTableWrapper cg_sstable_wrapper;
ObSSTable *cg_sstable = nullptr;
for (int64_t i = 0; !co_sstable->is_empty_co_table() && i < column_groups.count() && OB_SUCC(ret); i++) {
const ObStorageColumnGroupSchema &column_group = column_groups.at(i);
ObSSTableMetaHandle cg_table_meta_hdl;
if (column_group.is_all_column_group()) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("unexpected column_group", K(ret), K(i));
} else if (OB_FAIL(co_sstable->fetch_cg_sstable(i, cg_sstable_wrapper))) {
LOG_WARN("fail to get cg sstable", K(ret), K(i));
} else if (OB_FAIL(cg_sstable_wrapper.get_sstable(cg_sstable))) {
LOG_WARN("get sstable failed", K(ret));
} else if (OB_UNLIKELY(cg_sstable == nullptr || !cg_sstable->is_valid())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("unexpec cg sstable", K(ret), KPC(cg_sstable));
} else if (OB_FAIL(cg_sstable->get_meta(cg_table_meta_hdl))) {
LOG_WARN("fail to get meta", K(ret), KPC(cg_sstable));
} else if (OB_UNLIKELY(cg_table_meta_hdl.get_sstable_meta().get_col_checksum_cnt() != column_group.get_column_count())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("unexpected col_checksum_cnt", K(ret),
K(cg_table_meta_hdl.get_sstable_meta().get_col_checksum_cnt()), K(column_group.get_column_count()));
} else {
for (int64_t j = 0; j < column_group.get_column_count() && OB_SUCC(ret); j++) {
const uint16_t column_idx = column_group.column_idxs_[j];
if (column_idx < 0 || column_idx >= column_checksums.count()) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("invalid column index", K(ret), K(i), K(j), K(column_idx), K(column_checksums.count()));
} else {
int64_t &column_checksum = column_checksums.at(column_idx);
bool &is_checksum_ready = checksum_ready_array.at(column_idx);
if (!is_checksum_ready) {
column_checksum = cg_table_meta_hdl.get_sstable_meta().get_col_checksum()[j];
is_checksum_ready = true;
} else if (OB_UNLIKELY(column_checksum != cg_table_meta_hdl.get_sstable_meta().get_col_checksum()[j])) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("unexpected col_checksum_cnt", K(ret), K(column_checksum), K(cg_table_meta_hdl.get_sstable_meta().get_col_checksum()[j]));
}
}
}
}
}
}
return ret;
}
int ObCODDLUtil::is_rowkey_based_co_sstable(
const storage::ObCOSSTableV2 *co_sstable,
const storage::ObStorageSchema *storage_schema,
bool &is_rowkey_based)
{
int ret = OB_SUCCESS;
is_rowkey_based = false;
if (OB_UNLIKELY(nullptr == co_sstable || nullptr == storage_schema)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), KP(co_sstable), KP(storage_schema));
} else {
const int64_t base_cg_idx = co_sstable->get_key().get_column_group_id();
if (base_cg_idx < 0 || base_cg_idx >= storage_schema->get_column_groups().count()) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("invalid base column group index", K(ret), K(base_cg_idx));
} else {
is_rowkey_based = storage_schema->get_column_groups().at(base_cg_idx).is_rowkey_column_group();
}
}
return ret;
}
int ObCODDLUtil::need_column_group_store(const storage::ObStorageSchema &table_schema, bool &need_column_group)
{
int ret = OB_SUCCESS;
need_column_group = table_schema.get_column_group_count() > 1;
return ret;
}
int ObCODDLUtil::need_column_group_store(const schema::ObTableSchema &table_schema, bool &need_column_group)
{
int ret = OB_SUCCESS;
if (OB_FAIL(table_schema.get_is_column_store(need_column_group))) {
SHARE_LOG(WARN, "fail to check whether table is column store", K(ret));
}
return ret;
}
//record trace_id
ObDDLEventInfo::ObDDLEventInfo()
: addr_(GCTX.self_addr()),