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

add ddl kv multi row iterator

Browse Source 
Co-authored-by: suz-yang <suz.yang@foxmail.com>
This commit is contained in:
coolfishchen 2024-04-27 17:42:48 +00:00 committed by ob-robot
parent 7fa2e8dc39
commit 08350c392f
12 changed files with 164 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/storage/access/ob_block_sample_iterator.cpp
View File

@ -406,7 +406,7 @@ int ObBlockSampleRangeIterator::init_and_push_endkey_iterator(ObGetTableParam &g
} else {
STORAGE_LOG(WARN, "Fail to get next table iter", K(ret), K(get_table_param.tablet_iter_.table_iter()));
}
} else if (!table->is_sstable() || table->is_ddl_kv_sstable()) {
} else if (!table->is_sstable() || table->is_ddl_mem_sstable()) {
} else if (OB_ISNULL(table) || OB_ISNULL(sample_range_) || OB_ISNULL(allocator_)) {
ret = OB_ERR_UNEXPECTED;
STORAGE_LOG(WARN, "Unexpected null sstable", K(ret), KP(table), KP(sample_range_), KP(allocator_));

27
src/storage/access/ob_multiple_merge.cpp
View File

@ -31,7 +31,6 @@
#include "storage/column_store/ob_column_oriented_sstable.h"
#include "storage/tablet/ob_tablet.h"
#include "storage/tx/ob_trans_part_ctx.h"
#include "storage/ddl/ob_tablet_ddl_kv.h"
namespace oceanbase
{
@ -1316,34 +1315,16 @@ int ObMultipleMerge::prepare_tables_from_iterator(ObTableStoreIterator &table_it
}
}
if (OB_SUCC(ret) && need_table) {
ObITable *target_table_ptr = table_ptr;
if (table_ptr->no_data_to_read()) {
LOG_DEBUG("cur table is empty", K(ret), KPC(table_ptr));
continue;
} else if (table_ptr->is_memtable()) {
++memtable_cnt;
if (table_ptr->is_data_memtable()) {
read_released_memtable = read_released_memtable ||
TabletMemtableFreezeState::RELEASED == (static_cast<memtable::ObMemtable*>(table_ptr))->get_freeze_state();
} else if (table_ptr->is_direct_load_memtable()) {
ObDDLMemtable *ddl_memtable = nullptr;
if (OB_FAIL((static_cast<ObDDLKV*>(table_ptr)->get_first_ddl_memtable(ddl_memtable)))) {
if (ret == OB_ENTRY_NOT_EXIST) {
// memtable is null, ignore the ddl_kv
ret = OB_SUCCESS;
continue;
} else {
LOG_WARN("fail to get ddl memtable", K(ret));
}
} else {
target_table_ptr = ddl_memtable;
}
}
read_released_memtable = read_released_memtable ||
TabletMemtableFreezeState::RELEASED == (static_cast<ObITabletMemtable *>(table_ptr))->get_freeze_state();
}
if (OB_SUCC(ret)) {
if (OB_FAIL(tables_.push_back(target_table_ptr))) {
LOG_WARN("add table fail", K(ret), K(*table_ptr));
}
if (OB_FAIL(tables_.push_back(table_ptr))) {
LOG_WARN("add table fail", K(ret), K(*table_ptr));
}
}
} // end while

2
src/storage/access/ob_multiple_merge.h
View File

@ -35,8 +35,6 @@ namespace oceanbase
namespace storage
{
class ObBlockRowStore;
class ObDDLKV;
class ObDDLMemtable;
class ObMultipleMerge : public ObQueryRowIterator
{
public:

2
src/storage/blocksstable/index_block/ob_sstable_sec_meta_iterator.cpp
View File

@ -76,7 +76,7 @@ int ObSSTableSecMetaIterator::open(
{
int ret = OB_SUCCESS;
bool is_meta_root = false;
const bool is_ddl_mem_sstable = sstable.is_ddl_kv_sstable();
const bool is_ddl_mem_sstable = sstable.is_ddl_mem_sstable();
if (IS_INIT) {
ret = OB_INIT_TWICE;
LOG_WARN("Fail to open sstable secondary meta iterator", K(ret));

4
src/storage/ddl/ob_ddl_merge_task.cpp
View File

@ -1034,7 +1034,7 @@ int ObDDLMacroBlockIterator::open(ObSSTable *sstable, const ObDatumRange &query_
} else if (OB_UNLIKELY(nullptr == sstable || !query_range.is_valid() || !read_info.is_valid())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), KP(sstable), K(query_range), K(read_info));
} else if (sstable->is_ddl_kv_sstable()) { // ddl mem, scan keybtree
} else if (sstable->is_ddl_mem_sstable()) { // ddl mem, scan keybtree
ObDDLMemtable *ddl_memtable = static_cast<ObDDLMemtable *>(sstable);
if (OB_ISNULL(ddl_memtable)) {
ret = OB_ERR_UNEXPECTED;
@ -1085,7 +1085,7 @@ int ObDDLMacroBlockIterator::get_next(ObDataMacroBlockMeta &data_macro_meta, int
if (OB_UNLIKELY(!is_inited_)) {
ret = OB_NOT_INIT;
LOG_WARN("not init", K(ret), K(is_inited_));
} else if (sstable_->is_ddl_kv_sstable()) {
} else if (sstable_->is_ddl_mem_sstable()) {
ObDatumRowkeyWrapper tree_key;
ObBlockMetaTreeValue *tree_value = nullptr;
if (OB_FAIL(ddl_iter_.get_next(tree_key, tree_value))) {

68
src/storage/ddl/ob_tablet_ddl_kv.cpp
View File

@ -30,6 +30,8 @@
#include "storage/tablet/ob_tablet_create_delete_helper.h"
#include "storage/ddl/ob_direct_insert_sstable_ctx_new.h"
#include "storage/column_store/ob_column_oriented_sstable.h"
#include "storage/ddl/ob_tablet_ddl_kv_multi_version_row_iterator.h"
#include "storage/access/ob_sstable_multi_version_row_iterator.h"
using namespace oceanbase::storage;
using namespace oceanbase::blocksstable;
@ -165,7 +167,7 @@ int ObDDLMemtable::init_sstable_param(
}
} else {
if (table_key.table_type_ == ObITable::TableType::MINI_SSTABLE) {
sstable_param.table_key_.table_type_ = ObITable::TableType::DIRECT_LOAD_MEMTABLE;
sstable_param.table_key_.table_type_ = ObITable::TableType::DDL_MEM_MINI_SSTABLE;
} else {
sstable_param.table_key_.table_type_ = ObITable::TableType::DDL_MEM_SSTABLE;
}
@ -1595,6 +1597,30 @@ int ObDDLKV::get_frozen_schema_version(int64_t &schema_version) const
return ret;
}
#define ALLOCATE_DDL_KV_MULTI_VERSION_ROW_IETRATOR(class, query_range, ptr) \
if (OB_SUCC(ret)) { \
ObStoreRowIterator *row_scanner = nullptr; \
ALLOCATE_TABLE_STORE_ROW_IETRATOR(context, ObDDLKVMultiVersionRowIterator<class>, \
row_scanner); \
if (OB_SUCC(ret)) { \
if (OB_ISNULL(row_scanner)) { \
ret = OB_ERR_UNEXPECTED; \
STORAGE_LOG(WARN, "unexpected error, row_scanner is nullptr", K(ret), KP(row_scanner)); \
} else if (OB_FAIL(row_scanner->init(param, context, this, query_range))) { \
LOG_WARN("Fail to open row scanner", K(ret), K(param), K(context), KP(query_range)); \
} \
} \
if (OB_FAIL(ret)) { \
if (nullptr != row_scanner) { \
row_scanner->~ObStoreRowIterator(); \
FREE_TABLE_STORE_ROW_IETRATOR(context, row_scanner); \
row_scanner = nullptr; \
} \
} else { \
ptr = row_scanner; \
} \
}
int ObDDLKV::exist(const ObTableIterParam &param, ObTableAccessContext &context,
const blocksstable::ObDatumRowkey &rowkey, bool &is_exist, bool &has_found)
{
@ -1645,6 +1671,12 @@ int ObDDLKV::scan(const ObTableIterParam &param, ObTableAccessContext &context,
TCRLockGuard guard(lock_);
if (OB_UNLIKELY(!is_inited_)) {
ret = OB_NOT_INIT;
} else if (OB_UNLIKELY(
!param.is_valid()
|| !context.is_valid()
|| !key_range.is_valid())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("Invalid argument", K(ret), K(param), K(context), K(key_range));
} else if (OB_UNLIKELY(!is_inc_ddl_kv())) {
ret = OB_NOT_SUPPORTED;
LOG_WARN("not support get for full direct load", K(ret));
@ -1655,8 +1687,8 @@ int ObDDLKV::scan(const ObTableIterParam &param, ObTableAccessContext &context,
} else if (ddl_memtables_.count() != 1) {
ret = OB_NOT_SUPPORTED;
LOG_WARN("inc direct load do not support column store yet", K(ret));
} else if (OB_FAIL(ddl_memtables_.at(0)->scan(param, context, key_range, row_iter))) {
LOG_WARN("fail to get row", K(ret));
} else {
ALLOCATE_DDL_KV_MULTI_VERSION_ROW_IETRATOR(ObSSTableMultiVersionRowScanner, &key_range, row_iter);
}
return ret;
}
@ -1668,6 +1700,12 @@ int ObDDLKV::get(const storage::ObTableIterParam &param, storage::ObTableAccessC
TCRLockGuard guard(lock_);
if (OB_UNLIKELY(!is_inited_)) {
ret = OB_NOT_INIT;
} else if (OB_UNLIKELY(
!param.is_valid()
|| !context.is_valid()
|| !rowkey.is_valid())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("Invalid argument", K(ret), K(param), K(context), K(rowkey));
} else if (OB_UNLIKELY(!is_inc_ddl_kv())) {
ret = OB_NOT_SUPPORTED;
LOG_WARN("not support get for full direct load", K(ret));
@ -1678,8 +1716,8 @@ int ObDDLKV::get(const storage::ObTableIterParam &param, storage::ObTableAccessC
} else if (ddl_memtables_.count() != 1) {
ret = OB_NOT_SUPPORTED;
LOG_WARN("inc direct load do not support column store yet", K(ret));
} else if (OB_FAIL(ddl_memtables_.at(0)->get(param, context, rowkey, row_iter))) {
LOG_WARN("fail to get row", K(ret));
} else {
ALLOCATE_DDL_KV_MULTI_VERSION_ROW_IETRATOR(ObSSTableMultiVersionRowGetter, &rowkey, row_iter);
}
return ret;
}
@ -1691,6 +1729,12 @@ int ObDDLKV::multi_get(const ObTableIterParam &param, ObTableAccessContext &cont
TCRLockGuard guard(lock_);
if (OB_UNLIKELY(!is_inited_)) {
ret = OB_NOT_INIT;
} else if (OB_UNLIKELY(
!param.is_valid()
|| !context.is_valid()
|| 0 >= rowkeys.count())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("Invalid argument", K(ret), K(param), K(context), K(rowkeys));
} else if (OB_UNLIKELY(!is_inc_ddl_kv())) {
ret = OB_NOT_SUPPORTED;
LOG_WARN("not support get for full direct load", K(ret));
@ -1701,8 +1745,8 @@ int ObDDLKV::multi_get(const ObTableIterParam &param, ObTableAccessContext &cont
} else if (ddl_memtables_.count() != 1) {
ret = OB_NOT_SUPPORTED;
LOG_WARN("inc direct load do not support column store yet", K(ret));
} else if (OB_FAIL(ddl_memtables_.at(0)->multi_get(param, context, rowkeys, row_iter))) {
LOG_WARN("fail to get row", K(ret));
} else {
ALLOCATE_DDL_KV_MULTI_VERSION_ROW_IETRATOR(ObSSTableMultiVersionRowMultiGetter, &rowkeys, row_iter);
}
return ret;
}
@ -1714,6 +1758,12 @@ int ObDDLKV::multi_scan(const ObTableIterParam &param, ObTableAccessContext &con
TCRLockGuard guard(lock_);
if (OB_UNLIKELY(!is_inited_)) {
ret = OB_NOT_INIT;
} else if (OB_UNLIKELY(
!param.is_valid()
|| !context.is_valid()
|| 0 >= ranges.count())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("Invalid argument", K(ret), K(param), K(context), K(ranges));
} else if (OB_UNLIKELY(!is_inc_ddl_kv())) {
ret = OB_NOT_SUPPORTED;
LOG_WARN("not support get for full direct load", K(ret));
@ -1724,8 +1774,8 @@ int ObDDLKV::multi_scan(const ObTableIterParam &param, ObTableAccessContext &con
} else if (ddl_memtables_.count() != 1) {
ret = OB_NOT_SUPPORTED;
LOG_WARN("inc direct load do not support column store yet", K(ret));
} else if (OB_FAIL(ddl_memtables_.at(0)->multi_scan(param, context, ranges, row_iter))) {
LOG_WARN("fail to get row", K(ret));
} else {
ALLOCATE_DDL_KV_MULTI_VERSION_ROW_IETRATOR(ObSSTableMultiVersionRowMultiScanner, &ranges, row_iter);
}
return ret;
}

88
src/storage/ddl/ob_tablet_ddl_kv_multi_version_row_iterator.h Normal file
View File

@ -0,0 +1,88 @@
/**
* Copyright (c) 2024 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.
*/
#ifndef OB_STORAGE_OB_TABLET_DDL_KV_MULTI_VERSION_ROW_ITERATOR_H_
#define OB_STORAGE_OB_TABLET_DDL_KV_MULTI_VERSION_ROW_ITERATOR_H_
#include "storage/access/ob_store_row_iterator.h"
#include "storage/ddl/ob_tablet_ddl_kv.h"
namespace oceanbase
{
namespace storage
{
template <class T>
class ObDDLKVMultiVersionRowIterator : public ObStoreRowIterator
{
public:
ObDDLKVMultiVersionRowIterator() {}
virtual ~ObDDLKVMultiVersionRowIterator() {}
virtual void reuse()
{
iterator_.reuse();
ObStoreRowIterator::reuse();
}
virtual void reset()
{
iterator_.reset();
ObStoreRowIterator::reset();
}
// used for global query iterator pool, prepare for returning to pool
virtual void reclaim()
{
iterator_.reclaim();
ObStoreRowIterator::reclaim();
}
virtual int init(
const ObTableIterParam &param,
ObTableAccessContext &context,
ObITable *table,
const void *query_range)
{
int ret = common::OB_SUCCESS;
if (OB_ISNULL(query_range) || OB_ISNULL(table)) {
ret = OB_INVALID_ARGUMENT;
STORAGE_LOG(WARN, "invalid argument", K(ret), KP(query_range), KP(table));
} else if (!table->is_direct_load_memtable()) {
ret = OB_INVALID_ARGUMENT;
STORAGE_LOG(WARN, "invalid argument", K(ret), KPC(table));
} else {
ObDDLMemtable *ddl_memtable = nullptr;
if (OB_FAIL((static_cast<ObDDLKV *>(table)->get_first_ddl_memtable(ddl_memtable)))) {
STORAGE_LOG(WARN, "fail to get ddl memtable", K(ret));
} else if (OB_FAIL(iterator_.init(param, context, ddl_memtable, query_range))) {
STORAGE_LOG(WARN, "fail to init", K(ret));
} else {
type_ = iterator_.get_iter_type();
is_sstable_iter_ = false;
is_reclaimed_ = false;
}
}
return ret;
}
virtual int set_ignore_shadow_row() { return iterator_.set_ignore_shadow_row(); }
virtual bool can_blockscan() const { return iterator_.can_blockscan(); }
virtual bool can_batch_scan() const { return iterator_.can_batch_scan(); }
virtual int get_next_row(const blocksstable::ObDatumRow *&row)
{
return iterator_.get_next_row(row);
}
private:
T iterator_;
};
} // namespace storage
} // namespace oceanbase
#endif // OB_STORAGE_OB_TABLET_DDL_KV_MULTI_VERSION_ROW_ITERATOR_H_

3
src/storage/ob_i_table.cpp
View File

@ -75,7 +75,8 @@ const char* ObITable::table_type_name_[] =
"DDL_MERGE_CO",
"DDL_MERGE_CG",
"DDL_MEM_CO",
"DDL_MEM_CG"
"DDL_MEM_CG",
"DDL_MEM_MINI_SSTABLE"
};
uint64_t ObITable::TableKey::hash() const

7
src/storage/ob_i_table.h
View File

@ -105,6 +105,7 @@ public:
DDL_MERGE_CG_SSTABLE = 22, // used for column store ddl, for normal cg sstable, rowkey cg not supported
DDL_MEM_CO_SSTABLE = 23,
DDL_MEM_CG_SSTABLE = 24,
DDL_MEM_MINI_SSTABLE = 25,
// < add new sstable before here, See is_sstable()
MAX_TABLE_TYPE
@ -284,7 +285,6 @@ public:
virtual bool is_empty() const = 0;
virtual bool no_data_to_read() const { return is_empty(); }
virtual bool is_ddl_merge_empty_sstable() const { return is_empty() && is_ddl_merge_sstable(); }
virtual bool is_ddl_kv_sstable() const { return is_ddl_mem_sstable() || is_direct_load_memtable(); }
DECLARE_VIRTUAL_TO_STRING;
static bool is_sstable(const TableType table_type)
@ -309,7 +309,7 @@ public:
return ObITable::TableType::MINOR_SSTABLE == table_type
|| ObITable::TableType::MINI_SSTABLE == table_type
|| ObITable::TableType::REMOTE_LOGICAL_MINOR_SSTABLE == table_type
|| ObITable::TableType::DIRECT_LOAD_MEMTABLE == table_type;
|| ObITable::TableType::DDL_MEM_MINI_SSTABLE == table_type;
}
static bool is_multi_version_table(const TableType table_type)
@ -426,7 +426,8 @@ public:
{
return ObITable::TableType::DDL_MEM_SSTABLE == table_type
|| ObITable::TableType::DDL_MEM_CO_SSTABLE == table_type
|| ObITable::TableType::DDL_MEM_CG_SSTABLE == table_type;
|| ObITable::TableType::DDL_MEM_CG_SSTABLE == table_type
|| ObITable::TableType::DDL_MEM_MINI_SSTABLE == table_type;
}
static bool is_ddl_merge_sstable(const TableType table_type)
{

2
src/storage/ob_partition_range_spliter.cpp
View File

@ -400,7 +400,7 @@ int ObPartitionParallelRanger::init_macro_iters(ObRangeSplitInfo &range_info)
} else if (OB_UNLIKELY(!table->is_sstable())) {
ret = OB_ERR_UNEXPECTED;
STORAGE_LOG(WARN, "Unexpected table type", K(ret), KPC(table));
} else if (is_micro_level_ && !table->is_ddl_kv_sstable()) { // ddl kv not support endkey iterator of micro block
} else if (is_micro_level_ && !table->is_ddl_mem_sstable()) { // ddl kv not support endkey iterator of micro block
if (OB_ISNULL(buf = allocator_.alloc(sizeof(ObMicroEndkeyIterator)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
STORAGE_LOG(WARN, "Failed to alloc memory for endkey iter", K(ret));

2
src/storage/tablet/ob_table_store_util.cpp
View File

@ -230,7 +230,7 @@ int ObSSTableArray::inner_init(
if (OB_ISNULL(table)) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("unexpected null table ptr", K(ret));
} else if (OB_UNLIKELY(!table->is_sstable() && !table->is_ddl_kv_sstable())) {
} else if (OB_UNLIKELY(!table->is_sstable() && !table->is_ddl_mem_sstable())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("unexpected table type", K(ret), KPC(table));
} else if (OB_FAIL(static_cast<ObSSTable *>(table)->deep_copy(allocator, sstable_array_[i - start_pos]))) {

2
src/storage/tablet/ob_tablet_table_store_iterator.cpp
View File

@ -135,7 +135,7 @@ int ObTableStoreIterator::get_next(ObTableHandleV2 &table_handle)
} else {
if (OB_FAIL(get_ith_table(pos_, table))) {
LOG_WARN("fail to get ith table", K(ret), K(pos_));
} else if (table->is_memtable() || table->is_ddl_kv_sstable()) {
} else if (table->is_memtable() || table->is_ddl_mem_sstable()) {
ObTenantMetaMemMgr *t3m = MTL(ObTenantMetaMemMgr*);
if (OB_FAIL(table_handle.set_table(table, t3m, table->get_key().table_type_))) {
LOG_WARN("failed to set memtable to table handle", K(ret), KPC(table));